github-action/test_sync.py at main · datalumo/github-action · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
import unittest
from pathlib import Path
from tempfile import TemporaryDirectory
from unittest.mock import patch

import sync


class CollectTests(unittest.TestCase):
    def test_external_id_drops_the_extension(self):
        self.assertEqual(sync.external_id_for("introduction.md"), "introduction")
        self.assertEqual(sync.external_id_for("guides/install.mdx"), "guides/install")

    def test_collect_uses_the_first_heading_and_slug(self):
        with TemporaryDirectory() as raw:
            root = Path(raw)
            (root / "introduction.md").write_text("# Welcome to Datalumo\n\nBody.\n")
            (root / "nested").mkdir()
            (root / "nested" / "install.md").write_text("No heading here.\n")

            pages = sync.collect(root)

            self.assertEqual(
                [page["external_id"] for page in pages],
                ["introduction", "nested/install"],
            )
            self.assertEqual(pages[0]["name"], "Welcome to Datalumo")
            self.assertEqual(pages[1]["name"], "Install")
            self.assertEqual(pages[0]["content_mime"], "text/markdown")

    def test_collect_rejects_two_files_with_the_same_slug(self):
        with TemporaryDirectory() as raw:
            root = Path(raw)
            (root / "intro.md").write_text("# Intro\n")
            (root / "intro.html").write_text("<h1>Intro</h1>")

            with self.assertRaises(SystemExit) as raised:
                sync.collect(root)

            self.assertIn("intro", str(raised.exception))

    def test_citation_urls_do_not_include_the_file_extension(self):
        pages = [{"external_id": "introduction"}]
        sync.add_urls(pages, "https://datalumo.app/docs/")
        self.assertEqual(pages[0]["source_url"], "https://datalumo.app/docs/introduction")


class SyncTests(unittest.TestCase):
    def test_sync_pushes_prunes_and_indexes(self):
        pages = [{"external_id": "introduction", "name": "Introduction", "content": "# Introduction"}]
        calls: list[tuple[str, str]] = []

        def fake_request(api_url, org, source, token, method, path, body=None):
            calls.append((method, path))
            if method == "GET" and path.startswith("/pages?"):
                return 200, {
                    "data": [
                        {"external_id": "introduction"},
                        {"external_id": "removed-page"},
                    ],
                    "meta": {"next_cursor": None},
                }
            return 202, {}

        with patch.object(sync, "api_request", side_effect=fake_request):
            result = sync.sync("https://datalumo.app", "org", "docs", "token", pages, log=lambda *_: None)

        self.assertEqual(result, {"pushed": 1, "deleted": 1})
        self.assertIn(("POST", "/pages/batch"), calls)
        self.assertTrue(any(method == "DELETE" and path.endswith("/removed-page") for method, path in calls))
        self.assertNotIn(("DELETE", "/pages/introduction"), calls)
        self.assertIn(("POST", "/index"), calls)

    def test_sync_follows_the_list_cursor(self):
        pages = [{"external_id": "keep"}]
        calls: list[str] = []

        def fake_request(api_url, org, source, token, method, path, body=None):
            calls.append(path)
            if method == "GET" and "cursor=" not in path:
                return 200, {
                    "data": [{"external_id": "keep"}],
                    "meta": {"next_cursor": "next"},
                }
            if method == "GET":
                return 200, {
                    "data": [{"external_id": "gone"}],
                    "meta": {"next_cursor": None},
                }
            return 202, {}

        with patch.object(sync, "api_request", side_effect=fake_request):
            result = sync.sync("https://datalumo.app", "org", "docs", "token", pages, log=lambda *_: None)

        self.assertEqual(result["deleted"], 1)
        self.assertTrue(any("cursor=next" in path for path in calls))

    def test_sync_refuses_an_empty_set(self):
        with self.assertRaises(SystemExit) as raised:
            sync.sync("https://datalumo.app", "org", "docs", "token", [], log=lambda *_: None)

        self.assertIn("empty", str(raised.exception))


if __name__ == "__main__":
    unittest.main()