Eric Wendland
ffb46edc65
All checks were successful
Test / test (push) Successful in 6s
How a site answers missing paths now belongs to its repository like its edge rules: bunny.toml declares [site] not_found as "default", "spa" or a page from the build such as "/404.html". The action reads and sends the table on start and finalize; the gateway validates it before the upload and applies it to the Storage Zone at finalization, recording and purging only when it changed. A deployment without [site] resets the default; older action versions leave the setting alone. The console's Settings tab only shows the applied value, and the admin API no longer sets it. The maintainer guide and llms.txt describe the setting under Configuration.
517 lines
20 KiB
Python
517 lines
20 KiB
Python
from contextlib import redirect_stdout
|
|
import importlib.util
|
|
from io import BytesIO, StringIO
|
|
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 }
|
|
'''
|
|
|
|
|
|
SITE_SECTION = '''
|
|
[site]
|
|
not_found = "/404.html"
|
|
'''
|
|
|
|
|
|
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, site = static_site.load_site_config(
|
|
self.write_rules(directory, MARKDOWN_RULES + SITE_SECTION)
|
|
)
|
|
self.assertEqual(site, {"not_found": "/404.html"})
|
|
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 a \\[site\\] table": 'site = "spa"\n',
|
|
"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_site_config(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 + SITE_SECTION)
|
|
cases = (
|
|
(rules, ["markdown", "old-blog"], {"not_found": "/404.html"}),
|
|
(None, [], {}),
|
|
)
|
|
for path, expected, site in cases:
|
|
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)
|
|
# No [site] means the defaults, which reset earlier settings.
|
|
self.assertEqual(start_deployment.call_args.args[6], site)
|
|
self.assertEqual(finalize_deployment.call_args.args[8], site)
|
|
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()
|
|
|
|
|
|
def identity_args(root: Path) -> Any:
|
|
args = gateway_args(root, None)
|
|
args.password = ""
|
|
args.idempotency_key = "test-run-identity"
|
|
return args
|
|
|
|
|
|
class IdentityTests(unittest.TestCase):
|
|
@mock.patch.object(static_site, "finalize_bunny_deployment")
|
|
@mock.patch.object(static_site, "start_bunny_deployment")
|
|
@mock.patch.object(static_site, "BunnyStorage")
|
|
@mock.patch.object(static_site, "request_identity_token")
|
|
def test_identity_deploys_with_the_password_the_gateway_hands_out(
|
|
self,
|
|
request_token: mock.Mock,
|
|
storage_class: mock.Mock,
|
|
start_deployment: mock.Mock,
|
|
finalize_deployment: mock.Mock,
|
|
) -> None:
|
|
request_token.side_effect = ["jwt-start", "jwt-finalize"]
|
|
start_deployment.return_value = {
|
|
"deployment_id": "deployment-id",
|
|
"storage_password": "zone-password",
|
|
"storage_endpoint": "la.storage.bunnycdn.com",
|
|
}
|
|
storage_class.return_value.get_manifest.return_value = None
|
|
finalize_deployment.return_value = {"purged": True}
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
write_site(root)
|
|
output = StringIO()
|
|
with redirect_stdout(output), mock.patch.object(static_site, "apply_bunny_plan"):
|
|
result = static_site.deploy_bunny(identity_args(root))
|
|
|
|
request_token.assert_called_with("https://purge.example")
|
|
self.assertEqual(start_deployment.call_args.args[2], "jwt-start")
|
|
self.assertEqual(finalize_deployment.call_args.args[2], "jwt-finalize")
|
|
storage_class.assert_called_once_with(
|
|
"example-org", "zone-password", "la.storage.bunnycdn.com"
|
|
)
|
|
self.assertIn("::add-mask::zone-password", output.getvalue())
|
|
self.assertNotIn("zone-password", json.dumps(result))
|
|
self.assertTrue(result["purged"])
|
|
|
|
def test_identity_needs_the_gateway(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
write_site(root)
|
|
args = identity_args(root)
|
|
args.purge_endpoint = ""
|
|
with self.assertRaisesRegex(DeployError, "required without the Gandalf gateway"):
|
|
static_site.deploy_bunny(args)
|
|
|
|
@mock.patch.object(static_site, "start_bunny_deployment")
|
|
def test_identity_dry_run_stays_local(self, start_deployment: mock.Mock) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
write_site(root)
|
|
args = identity_args(root)
|
|
args.dry_run = True
|
|
result = static_site.deploy_bunny(args)
|
|
start_deployment.assert_not_called()
|
|
self.assertEqual(result["files"], 2)
|
|
self.assertIsNone(result["changed"])
|
|
|
|
def test_identity_token_request_explains_missing_permissions(self) -> None:
|
|
with mock.patch.dict(static_site.os.environ, {}, clear=True):
|
|
with self.assertRaisesRegex(DeployError, "id-token: write"):
|
|
static_site.request_identity_token("https://gandalf.example")
|
|
|
|
@mock.patch.object(static_site.urllib.request, "urlopen")
|
|
def test_identity_token_request_asks_for_the_gateway_audience(
|
|
self, urlopen: mock.Mock
|
|
) -> None:
|
|
urlopen.return_value.__enter__.return_value.read.return_value = (
|
|
b'{"count": 1, "value": "jwt"}'
|
|
)
|
|
environment = {
|
|
"ACTIONS_ID_TOKEN_REQUEST_URL": "https://ci.example/token?api-version=2",
|
|
"ACTIONS_ID_TOKEN_REQUEST_TOKEN": "request-token",
|
|
}
|
|
with mock.patch.dict(static_site.os.environ, environment, clear=True):
|
|
token = static_site.request_identity_token("https://gandalf.example")
|
|
self.assertEqual(token, "jwt")
|
|
request = urlopen.call_args.args[0]
|
|
self.assertEqual(
|
|
request.full_url,
|
|
"https://ci.example/token?api-version=2&audience=https%3A%2F%2Fgandalf.example",
|
|
)
|
|
self.assertEqual(request.get_header("Authorization"), "bearer request-token")
|