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 typing import Any
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, "finalize_bunny_deployment")
    @mock.patch.object(static_site, "start_bunny_deployment")
    @mock.patch.object(static_site, "BunnyStorage")
    def test_unchanged_deployment_is_finalized_without_forcing_a_purge(
        self,
        storage_class: mock.Mock,
        start_deployment: mock.Mock,
        finalize_deployment: 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()
            start_deployment.return_value = {"deployment_id": "deployment-id"}
            finalize_deployment.return_value = {
                "deployment_id": "deployment-id",
                "status": "unchanged",
                "purged": False,
            }
            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,
                workers=4,
                revision="abc1234",
                idempotency_key="test-run-123",
            )
            result = static_site.deploy_bunny(args)

        self.assertFalse(result["changed"])
        self.assertFalse(result["purged"])
        start_deployment.assert_called_once()
        finalize_deployment.assert_called_once()
        storage_class.return_value.upload_manifest.assert_not_called()

    @mock.patch.object(static_site, "abort_bunny_deployment")
    @mock.patch.object(static_site, "finalize_bunny_deployment")
    @mock.patch.object(static_site, "start_bunny_deployment")
    @mock.patch.object(static_site, "apply_bunny_plan")
    @mock.patch.object(static_site, "BunnyStorage")
    def test_failed_upload_releases_the_deployment_lease(
        self,
        storage_class: mock.Mock,
        apply_plan: mock.Mock,
        start_deployment: mock.Mock,
        finalize_deployment: mock.Mock,
        abort_deployment: mock.Mock,
    ) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            write_site(root)
            storage_class.return_value.get_manifest.return_value = None
            start_deployment.return_value = {"deployment_id": "deployment-id"}
            apply_plan.side_effect = static_site.DeployError("upload failed")
            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,
                workers=4,
                revision="abc1234",
                idempotency_key="test-run-123",
            )

            with self.assertRaisesRegex(static_site.DeployError, "upload failed"):
                static_site.deploy_bunny(args)

        finalize_deployment.assert_not_called()
        abort_deployment.assert_called_once_with(
            "https://purge.example",
            "example-org",
            "storage-password",
            "deployment-id",
            "test-run-123",
        )


MARKDOWN_RULES = '''
[[edge_rules]]
name = "markdown"
url = "*.md"
response_headers = { "Content-Type" = "text/markdown; charset=utf-8" }

[[edge_rules]]
name = "old-blog"
url = ["*/blog/*"]
redirect = { to = "https://example.org{{path}}", status = 301 }
'''


def gateway_args(root: Path, edge_rules: Path | None) -> Any:
    return 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,
        workers=4,
        revision="abc1234",
        idempotency_key="",
        edge_rules=edge_rules,
    )


class EdgeRuleTests(unittest.TestCase):
    def write_rules(self, directory: str, text: str) -> Path:
        path = Path(directory) / "bunny.toml"
        path.write_text(text)
        return path

    def test_loads_declared_rules(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            rules = static_site.load_edge_rules(
                self.write_rules(directory, MARKDOWN_RULES)
            )
        self.assertEqual(
            rules,
            [
                {
                    "name": "markdown",
                    "url": "*.md",
                    "response_headers": {
                        "Content-Type": "text/markdown; charset=utf-8"
                    },
                },
                {
                    "name": "old-blog",
                    "url": ["*/blog/*"],
                    "redirect": {"to": "https://example.org{{path}}", "status": 301},
                },
            ],
        )

    def test_rejects_malformed_files(self) -> None:
        # Rule contents are validated by the gateway; see handler_test.ts.
        cases = {
            "unknown keys": 'headers = "x"\n' + MARKDOWN_RULES,
            "must be \\[\\[edge_rules\\]\\] tables": 'edge_rules = ["x"]\n',
            "unsupported value": MARKDOWN_RULES + "since = 2026-10-02\n",
            "not valid TOML": "[[edge_rules]\n",
        }
        for message, text in cases.items():
            with self.subTest(message), tempfile.TemporaryDirectory() as directory:
                with self.assertRaisesRegex(DeployError, message):
                    static_site.load_edge_rules(self.write_rules(directory, text))

    def test_edge_rules_need_the_gateway(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            write_site(root)
            args = gateway_args(root, self.write_rules(directory, MARKDOWN_RULES))
            args.purge_endpoint = ""
            with self.assertRaisesRegex(DeployError, "BUNNY_PURGE_ENDPOINT"):
                static_site.deploy_bunny(args)

    @mock.patch.object(static_site, "finalize_bunny_deployment")
    @mock.patch.object(static_site, "start_bunny_deployment")
    @mock.patch.object(static_site, "BunnyStorage")
    def test_finalization_sends_declared_rules_or_none(
        self,
        storage_class: mock.Mock,
        start_deployment: mock.Mock,
        finalize_deployment: mock.Mock,
    ) -> None:
        storage_class.return_value.get_manifest.return_value = None
        start_deployment.return_value = {"deployment_id": "deployment-id"}
        finalize_deployment.return_value = {"purged": True}
        keys = []
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory) / "dist"
            root.mkdir()
            write_site(root)
            rules = self.write_rules(directory, MARKDOWN_RULES)
            for path, expected in ((rules, ["markdown", "old-blog"]), (None, [])):
                result = static_site.deploy_bunny(gateway_args(root, path))
                for sent in (
                    start_deployment.call_args.args[5],
                    finalize_deployment.call_args.args[7],
                ):
                    self.assertEqual([rule["name"] for rule in sent], expected)
                self.assertEqual(result["edge_rules"], expected)
                keys.append(start_deployment.call_args.args[3])
        # Changing only the rules must not replay the previous deployment.
        self.assertNotEqual(keys[0], keys[1])


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