Eric Wendland
c242db40a1
All checks were successful
Test / test (push) Successful in 9s
Without BUNNY_STORAGE_PASSWORD, the action now requests the running workflow's OIDC identity token (Forgejo with enable-openid-connect, GitHub with id-token: write) for the gateway's audience and starts the deployment with it. It takes the Storage Zone password from the gateway's answer, masks it in the log, plans and uploads with it, and finalizes or aborts with a fresh token. A dry run in this mode stays local. Password deployments work as before. action.yml finds its script under FORGEJO_ACTION_PATH or GITHUB_ACTION_PATH, so the action also runs on GitHub Actions.
502 lines
20 KiB
Python
502 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 }
|
|
'''
|
|
|
|
|
|
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()
|
|
|
|
|
|
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")
|