姚章浩
111 lines
4.8 KiB
Python
111 lines
4.8 KiB
Python
"""Exercise the deployment shell with fake Docker; never touch a daemon."""
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
|
|
SCRIPT = Path(__file__).with_name("run-release-executor.sh")
|
|
|
|
|
|
class ReleaseExecutorTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.root = Path(self.tmp.name)
|
|
self.bin = self.root / "bin"
|
|
self.bin.mkdir()
|
|
self.calls = self.root / "calls.jsonl"
|
|
for name, text in {
|
|
"id": "#!/bin/sh\nprintf '0\\n'\n",
|
|
"sleep": "#!/bin/sh\nexit 0\n",
|
|
"docker": """#!/usr/bin/env python3
|
|
import json,os,sys
|
|
with open(os.environ['CALLS'],'a') as f: f.write(json.dumps(sys.argv[1:])+'\\n')
|
|
a=sys.argv[1:]
|
|
if a[0]=='inspect':
|
|
if '--format' in a: print(os.environ.get('CONTAINER_STATE','running 0'))
|
|
else: sys.exit(0)
|
|
""",
|
|
}.items():
|
|
p = self.bin / name
|
|
p.write_text(text)
|
|
p.chmod(0o755)
|
|
self.envfile = self.root / "executor.env"
|
|
self.token = self.root / "token"
|
|
pgpass = self.root / "pgpass"
|
|
pgpass.write_text("fixture-password-not-real")
|
|
pgpass.chmod(0o600)
|
|
self.env = dict(os.environ, PATH=str(self.bin) + os.pathsep + os.environ["PATH"],
|
|
CALLS=str(self.calls), RELEASE_EXECUTOR_IMAGE="test:fixture",
|
|
RELEASE_EXECUTOR_ENV=str(self.envfile), RELEASE_PGPASS_FILE=str(pgpass),
|
|
RELEASE_PG_SOCKET_DIR=str(self.root), RELEASE_GATE_STATE=str(self.root),
|
|
RELEASE_ARCHIVE_ROOT=str(self.root / "archive"),
|
|
RELEASE_JOURNAL_ROOT=str(self.root / "journal"),
|
|
RELEASE_BUNDLE_DIR=str(self.root / "bundles"),
|
|
RELEASE_AUTH_TOKEN_FILE=str(self.token))
|
|
|
|
def run_script(self, enabled=True, author=True, path=True, token=True, mode=0o600):
|
|
lines = ["LIBRARY_WORKER_ROLE=release-executor", "PGPASSFILE=/run/secrets/governance_pgpass",
|
|
"PRODUCTION_RELEASE_REQUESTS_ENABLED=" + str(enabled).lower()]
|
|
if author:
|
|
lines.append("PRODUCTION_RELEASE_AUTHORIZER_REF=fixture-owner")
|
|
if path:
|
|
lines.append("PRODUCTION_RELEASE_AUTH_TOKEN_FILE=/run/secrets/production_release_auth_token")
|
|
self.envfile.write_text("\n".join(lines) + "\n")
|
|
self.envfile.chmod(0o600)
|
|
if token:
|
|
self.token.write_text("fixture-only-not-real")
|
|
self.token.chmod(mode)
|
|
result = subprocess.run(["bash", str(SCRIPT)], env=self.env, text=True, capture_output=True)
|
|
calls = [json.loads(line) for line in self.calls.read_text().splitlines()] if self.calls.exists() else []
|
|
self.assertNotIn("fixture-only-not-real", result.stdout + result.stderr)
|
|
return result, calls
|
|
|
|
def test_enabled_missing_author_rejected_before_replacing(self):
|
|
r, calls = self.run_script(author=False)
|
|
self.assertNotEqual(r.returncode, 0)
|
|
self.assertFalse(any(c[0] in ("rm", "run") for c in calls))
|
|
|
|
def test_enabled_missing_token_path_rejected_before_replacing(self):
|
|
r, calls = self.run_script(path=False)
|
|
self.assertNotEqual(r.returncode, 0)
|
|
self.assertFalse(any(c[0] in ("rm", "run") for c in calls))
|
|
|
|
def test_enabled_missing_token_file_rejected_before_replacing(self):
|
|
r, calls = self.run_script(token=False)
|
|
self.assertNotEqual(r.returncode, 0)
|
|
self.assertFalse(any(c[0] in ("rm", "run") for c in calls))
|
|
|
|
def test_enabled_wrong_token_permissions_rejected_before_replacing(self):
|
|
r, calls = self.run_script(mode=0o644)
|
|
self.assertNotEqual(r.returncode, 0)
|
|
self.assertFalse(any(c[0] in ("rm", "run") for c in calls))
|
|
|
|
def test_enabled_mounts_private_token_readonly(self):
|
|
r, calls = self.run_script()
|
|
self.assertEqual(r.returncode, 0, r.stderr)
|
|
run = next(c for c in calls if c[0] == "run")
|
|
self.assertIn(str(self.token) + ":/run/secrets/production_release_auth_token:ro", run)
|
|
|
|
def test_disabled_needs_no_authorization_material(self):
|
|
r, calls = self.run_script(enabled=False, author=False, path=False, token=False)
|
|
self.assertEqual(r.returncode, 0, r.stderr)
|
|
self.assertTrue(any(c[0] == "run" for c in calls))
|
|
|
|
def test_crash_loop_is_not_success(self):
|
|
self.env["CONTAINER_STATE"] = "restarting 3"
|
|
r, _ = self.run_script()
|
|
self.assertNotEqual(r.returncode, 0)
|
|
self.assertNotIn("启动完成", r.stdout)
|
|
|
|
def test_already_restarted_is_not_success(self):
|
|
self.env["CONTAINER_STATE"] = "running 1"
|
|
r, _ = self.run_script()
|
|
self.assertNotEqual(r.returncode, 0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|