Dave Kowalski
099a48f2b5
Test / test (pull_request) Failing after 5s
Authored-By: OpenAI (GPT-5) <noreply@openai.com>
193 lines
8.7 KiB
Python
193 lines
8.7 KiB
Python
import contextlib
|
|
import io
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
import yaml
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT / "scripts"))
|
|
|
|
import deploy
|
|
import utils
|
|
|
|
|
|
def write_config(directory, value):
|
|
(Path(directory) / "site.yaml").write_text(yaml.safe_dump(value), encoding="utf-8")
|
|
|
|
|
|
def v2_config():
|
|
return yaml.safe_load((ROOT / "tests" / "fixtures" / "v2-site.yaml").read_text())
|
|
|
|
|
|
class ConfigTests(unittest.TestCase):
|
|
def parse(self, config):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
write_config(directory, config)
|
|
return utils.parse_site_yaml(directory, "baseline.fritzlab.net")
|
|
|
|
def reject(self, config, message):
|
|
with self.assertRaisesRegex(SystemExit, "1"):
|
|
with contextlib.redirect_stderr(io.StringIO()) as stderr:
|
|
self.parse(config)
|
|
self.assertIn(message, stderr.getvalue())
|
|
|
|
def test_legacy_config_uses_explicit_compatibility_translation(self):
|
|
config = self.parse({"domain": "example.fritzlab.net", "type": "static",
|
|
"middlewares": ["authentik-forwardauth"]})
|
|
self.assertEqual("single-surface-v1", config["schema"])
|
|
self.assertTrue(config["compatibility"])
|
|
self.assertEqual("baseline.fritzlab.net", config["artifacts"]["site"]["bucket"])
|
|
self.assertEqual("/", config["routes"][0]["path"])
|
|
|
|
def test_v2_sorts_longest_prefix_and_assigns_cache_contracts(self):
|
|
config = self.parse(v2_config())
|
|
self.assertEqual(["/dist/channels", "/dist/releases", "/"],
|
|
[route["path"] for route in config["routes"]])
|
|
self.assertEqual("public, max-age=31536000, immutable",
|
|
config["artifacts"]["releases"]["cache_control"])
|
|
self.assertEqual("public, max-age=0, must-revalidate",
|
|
config["artifacts"]["channels"]["cache_control"])
|
|
|
|
def test_rejects_partial_v2(self):
|
|
self.reject({"domain": "x.example", "schema": "v2", "routes": []},
|
|
"declared together")
|
|
|
|
def test_rejects_public_catch_all(self):
|
|
config = v2_config()
|
|
config["routes"][0] = {"path": "/", "artifact": "portal", "access": "public"}
|
|
config["artifacts"]["portal"]["cache"] = "revalidated-channel"
|
|
self.reject(config, "public catch-all")
|
|
|
|
def test_rejects_duplicate_paths(self):
|
|
config = v2_config()
|
|
config["routes"][1]["path"] = "/dist/releases"
|
|
self.reject(config, "duplicate route path")
|
|
|
|
def test_rejects_bucket_reuse_across_access(self):
|
|
config = v2_config()
|
|
config["artifacts"]["portal"]["bucket"] = "baseline-releases"
|
|
self.reject(config, "bucket is reused across access classes")
|
|
|
|
def test_rejects_credential_reuse_across_artifacts(self):
|
|
config = v2_config()
|
|
config["artifacts"]["portal"]["credentials"] = {
|
|
"access_key_env": "RELEASES_ACCESS_KEY", "secret_key_env": "RELEASES_SECRET_KEY"}
|
|
self.reject(config, "publication credentials are reused")
|
|
|
|
def test_requires_explicit_root_policy(self):
|
|
config = v2_config()
|
|
config["routes"] = [route for route in config["routes"] if route["path"] != "/"]
|
|
del config["artifacts"]["portal"]
|
|
self.reject(config, "explicit / access policy")
|
|
|
|
def test_rejects_overlapping_sources_across_access(self):
|
|
config = v2_config()
|
|
config["artifacts"]["portal"]["source"] = "."
|
|
self.reject(config, "sources overlap across access classes")
|
|
|
|
def test_rejects_unknown_top_level_key(self):
|
|
config = v2_config()
|
|
config["surprise"] = True
|
|
self.reject(config, "unknown site.yaml keys")
|
|
|
|
|
|
class ManifestTests(unittest.TestCase):
|
|
def test_split_routes_render_one_certificate_and_distinct_backends(self):
|
|
with tempfile.TemporaryDirectory() as config_dir, tempfile.TemporaryDirectory() as out:
|
|
write_config(config_dir, v2_config())
|
|
config = utils.parse_site_yaml(config_dir, "baseline.fritzlab.net")
|
|
app_dir = Path(out) / "app"
|
|
deploy.render_site_manifests("baseline.fritzlab.net", ROOT, app_dir,
|
|
app_dir / "manifests", config)
|
|
manifests = app_dir / "manifests"
|
|
self.assertEqual(1, len(list(manifests.glob("certificate*.yaml"))))
|
|
self.assertEqual(3, len(list(manifests.glob("service-*.yaml"))))
|
|
self.assertEqual(3, len(list(manifests.glob("ingress-*.yaml"))))
|
|
for path in manifests.glob("*.yaml"):
|
|
list(yaml.safe_load_all(path.read_text(encoding="utf-8")))
|
|
services = "".join(path.read_text() for path in manifests.glob("service-*.yaml"))
|
|
self.assertIn("baseline-releases.web.sjc001.fritzlab.net", services)
|
|
self.assertIn('service.passhostheader: "false"', services)
|
|
portal_ingress = next(manifests.glob("ingress-*-portal.yaml")).read_text()
|
|
self.assertIn("authentik-forwardauth@file", portal_ingress)
|
|
|
|
|
|
class PublicationTests(unittest.TestCase):
|
|
def artifact(self, immutable=False):
|
|
return {
|
|
"name": "release", "bucket": "release-bucket",
|
|
"credentials": {"access_key_env": "TEST_ACCESS",
|
|
"secret_key_env": "TEST_SECRET"},
|
|
"cache_control": "public, max-age=31536000, immutable" if immutable
|
|
else "public, max-age=0, must-revalidate",
|
|
"immutable": immutable,
|
|
}
|
|
|
|
@mock.patch.dict(os.environ, {"TEST_ACCESS": "access", "TEST_SECRET": "secret"})
|
|
@mock.patch("deploy.urlopen")
|
|
def test_immutable_put_is_conditional_and_signed_in_process(self, urlopen_mock):
|
|
response = urlopen_mock.return_value.__enter__.return_value
|
|
response.read.return_value = b""
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory) / "bundle.js"
|
|
path.write_bytes(b"release")
|
|
deploy._conditional_put("http://garage", self.artifact(True),
|
|
"dist/releases/bundle.js", path,
|
|
"application/javascript", "digest")
|
|
request = urlopen_mock.call_args.args[0]
|
|
self.assertEqual("*", request.headers["If-none-match"])
|
|
self.assertNotIn("secret", request.full_url)
|
|
|
|
@mock.patch.dict(os.environ, {"TEST_ACCESS": "access", "TEST_SECRET": "secret"})
|
|
@mock.patch("deploy.run")
|
|
def test_revalidated_publication_sets_cache_on_sync_and_restamp(self, run_mock):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
deploy._replaceable_sync(Path(directory), self.artifact(), "http://garage", [])
|
|
commands = [call.args[0] for call in run_mock.call_args_list]
|
|
self.assertEqual(["sync", "cp"], [command[4] for command in commands])
|
|
self.assertTrue(all("public, max-age=0, must-revalidate" in command
|
|
for command in commands))
|
|
|
|
@mock.patch.dict(os.environ, {"TEST_ACCESS": "access", "TEST_SECRET": "secret"})
|
|
@mock.patch("deploy._conditional_put")
|
|
@mock.patch("deploy.subprocess.run")
|
|
def test_immutable_identical_retry_skips_and_changed_key_fails(self, subprocess_mock,
|
|
put_mock):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
source = Path(directory)
|
|
content = b"release"
|
|
(source / "bundle.js").write_bytes(content)
|
|
digest = __import__("hashlib").sha256(content).hexdigest()
|
|
subprocess_mock.return_value = subprocess.CompletedProcess(
|
|
[], 0, stdout=json.dumps({
|
|
"Metadata": {"sha256": digest},
|
|
"CacheControl": "public, max-age=31536000, immutable"}), stderr="")
|
|
deploy._immutable_sync(source, self.artifact(True), "http://garage", [])
|
|
put_mock.assert_not_called()
|
|
subprocess_mock.return_value = subprocess.CompletedProcess(
|
|
[], 0, stdout='{"Metadata":{"sha256":"different"}}', stderr="")
|
|
with self.assertRaises(SystemExit):
|
|
deploy._immutable_sync(source, self.artifact(True), "http://garage", [])
|
|
|
|
|
|
class CredentialSafetyTests(unittest.TestCase):
|
|
def test_git_askpass_reads_token_only_from_inherited_fd(self):
|
|
token = "credential-sentinel"
|
|
with utils.git_auth(token, "ci-bot") as auth:
|
|
self.assertNotIn(token, str(auth["env"]))
|
|
result = subprocess.run(
|
|
[auth["env"]["GIT_ASKPASS"], "Password"],
|
|
capture_output=True, text=True, check=True, **auth)
|
|
self.assertEqual(token, result.stdout)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|