static-site-deploy/tests/test_static_site.py

import importlib.util
from io import BytesIO
import json
from pathlib import Path
import sys
import tempfile
import threading
import unittest
from unittest import mock

MODULE_PATH = Path(__file__).resolve().parents[1] / "static_site.py"
SPEC = importlib.util.spec_from_file_location("static_site_deploy", MODULE_PATH)
assert SPEC and SPEC.loader
static_site = importlib.util.module_from_spec(SPEC)
sys.modules[SPEC.name] = static_site
SPEC.loader.exec_module(static_site)

DeployError = static_site.DeployError
apply_bunny_plan = static_site.apply_bunny_plan
bunny_plan = static_site.bunny_plan
collect_files = static_site.collect_files
manifest_for = static_site.manifest_for
parse_manifest = static_site.parse_manifest
purge_bunny_gateway = static_site.purge_bunny_gateway
run_concurrently = static_site.run_concurrently


def write_site(root: Path) -> None:
    (root / "assets").mkdir()
    (root / "index.html").write_text("<h1>Hello</h1>")
    (root / "assets" / "app.js").write_text("console.log('hello')")


class StaticSiteTests(unittest.TestCase):
    def test_collect_files_requires_index(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            with self.assertRaisesRegex(DeployError, "index.html"):
                collect_files(Path(directory))

    def test_collect_files_rejects_symlinks(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            write_site(root)
            (root / "linked").symlink_to(root / "index.html")
            with self.assertRaisesRegex(DeployError, "symbolic links"):
                collect_files(root)

    def test_manifest_round_trip(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            write_site(root)
            files = collect_files(root)
            manifest = manifest_for(files)
            parsed = parse_manifest(json.dumps(manifest).encode())
            self.assertEqual(
                parsed["index.html"]["sha256"], files["index.html"].sha256
            )

    def test_manifest_rejects_unsafe_remote_path(self) -> None:
        raw = b'{"version": 1, "files": {"../other-site/index.html": {}}}'
        with self.assertRaisesRegex(DeployError, "unsafe path"):
            parse_manifest(raw)

    def test_bunny_plan_uploads_assets_before_html_and_removes_stale(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            write_site(root)
            files = collect_files(root)
            remote = {
                "index.html": {"sha256": "old", "size": 1},
                "assets/app.js": {"sha256": "old", "size": 1},
                "old.css": {"sha256": "old", "size": 1},
            }
            upload, delete = bunny_plan(files, remote)
            self.assertEqual(upload, ["assets/app.js", "index.html"])
            self.assertEqual(delete, ["old.css"])

    def test_bunny_plan_skips_unchanged_files(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            write_site(root)
            files = collect_files(root)
            remote = manifest_for(files)["files"]
            self.assertEqual(bunny_plan(files, remote), ([], []))

    def test_bunny_transfers_use_multiple_workers(self) -> None:
        barrier = threading.Barrier(4)
        completed: list[int] = []

        def transfer(item: int) -> None:
            barrier.wait(timeout=2)
            completed.append(item)

        run_concurrently([1, 2, 3, 4], transfer, workers=4)

        self.assertCountEqual(completed, [1, 2, 3, 4])

    def test_bunny_transfer_order_keeps_documents_and_manifest_safe(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            write_site(root)
            (root / "about.html").write_text("<h1>About</h1>")
            files = collect_files(root)
            events: list[tuple[str, str]] = []
            storage = mock.Mock()
            storage.upload.side_effect = lambda item: events.append(
                ("upload", item.path)
            )
            storage.delete.side_effect = lambda path: events.append(("delete", path))
            storage.upload_manifest.side_effect = lambda _manifest: events.append(
                ("manifest", static_site.MANIFEST_NAME)
            )

            apply_bunny_plan(
                storage,
                files,
                ["assets/app.js", "about.html", "index.html"],
                ["stale.css"],
                workers=2,
            )

        asset_position = events.index(("upload", "assets/app.js"))
        document_positions = {
            events.index(("upload", "about.html")),
            events.index(("upload", "index.html")),
        }
        delete_position = events.index(("delete", "stale.css"))
        self.assertTrue(
            all(asset_position < position for position in document_positions)
        )
        self.assertTrue(
            all(position < delete_position for position in document_positions)
        )
        self.assertEqual(events[-1], ("manifest", static_site.MANIFEST_NAME))

    def test_gateway_requires_a_plain_https_origin(self) -> None:
        invalid = (
            "http://purge.example",
            "https://purge.example/path",
            "https://user@purge.example",
            "https://purge.example?token=secret",
        )
        for endpoint in invalid:
            with self.subTest(endpoint=endpoint):
                with self.assertRaisesRegex(DeployError, "HTTPS origin"):
                    purge_bunny_gateway(endpoint, "example-org", "password")

    @mock.patch.object(static_site.time, "sleep")
    @mock.patch.object(static_site.urllib.request, "urlopen")
    def test_storage_retries_transient_new_zone_authentication(
        self, urlopen: mock.Mock, sleep: mock.Mock
    ) -> None:
        unauthorized = static_site.urllib.error.HTTPError(
            "https://storage.example/example/.static-site-manifest.json",
            401,
            "Unauthorized",
            {},
            BytesIO(b'{"HttpCode":401,"Message":"Unauthorized"}'),
        )
        response = mock.MagicMock()
        response.__enter__.return_value.read.return_value = b"manifest"
        urlopen.side_effect = [unauthorized, response]
        storage = static_site.BunnyStorage(
            "example", "new-password", "storage.example"
        )

        self.assertEqual(storage.get_manifest(), b"manifest")
        sleep.assert_called_once_with(1)
        self.assertEqual(urlopen.call_count, 2)

    @mock.patch.object(static_site.time, "sleep")
    @mock.patch.object(static_site.urllib.request, "urlopen")
    def test_storage_retries_transient_connection_errors(
        self, urlopen: mock.Mock, sleep: mock.Mock
    ) -> None:
        response = mock.MagicMock()
        response.__enter__.return_value.read.return_value = b"manifest"
        urlopen.side_effect = [
            static_site.urllib.error.URLError("connection reset"),
            response,
        ]
        storage = static_site.BunnyStorage(
            "example", "storage-password", "storage.example"
        )

        self.assertEqual(storage.get_manifest(), b"manifest")
        sleep.assert_called_once_with(1)
        self.assertEqual(urlopen.call_count, 2)

    @mock.patch.object(static_site.urllib.request, "urlopen")
    def test_gateway_reuses_storage_password(self, urlopen: mock.Mock) -> None:
        urlopen.return_value.__enter__.return_value = object()
        purge_bunny_gateway(
            "https://purge.example", "example-org", "storage-password"
        )
        request = urlopen.call_args.args[0]
        self.assertEqual(
            request.full_url,
            "https://purge.example/v1/static-sites/example-org/purge",
        )
        self.assertEqual(request.method, "POST")
        self.assertEqual(
            request.get_header("Authorization"), "Bearer storage-password"
        )

    @mock.patch.object(static_site, "purge_bunny_gateway")
    @mock.patch.object(static_site, "BunnyStorage")
    def test_unchanged_deployment_retries_gateway_purge(
        self, storage_class: mock.Mock, purge_gateway: mock.Mock
    ) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            write_site(root)
            files = collect_files(root)
            storage_class.return_value.get_manifest.return_value = json.dumps(
                manifest_for(files)
            ).encode()
            args = static_site.argparse.Namespace(
                directory=root,
                zone="example-org",
                password="storage-password",
                endpoint="storage.bunnycdn.com",
                purge_endpoint="https://purge.example",
                pull_zone_id="",
                api_key="",
                dry_run=False,
            )
            result = static_site.deploy_bunny(args)

        self.assertFalse(result["changed"])
        self.assertTrue(result["purged"])
        purge_gateway.assert_called_once_with(
            "https://purge.example", "example-org", "storage-password"
        )


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