姚章浩
79 lines
3.8 KiB
Python
79 lines
3.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Execute deployment scripts with fake external CLIs; never touch Docker/SSH."""
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
|
|
HERE = Path(__file__).resolve().parent
|
|
|
|
|
|
class DeploymentTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.root = Path(self.temp.name)
|
|
self.trace = self.root / "trace.jsonl"
|
|
self.env = dict(os.environ, PATH=str(self.root) + ":" + os.environ["PATH"], TRACE=str(self.trace))
|
|
self.stub("docker", "if 'inspect' in sys.argv: print('network=172.21.0.3 ')")
|
|
self.stub("ssh", "")
|
|
|
|
def stub(self, name, body):
|
|
path = self.root / name
|
|
path.write_text("#!/usr/bin/env python3\nimport os,sys,json\nwith open(os.environ['TRACE'],'a') as f: f.write(json.dumps(sys.argv)+chr(10))\n" + body + "\n")
|
|
path.chmod(0o755)
|
|
|
|
def calls(self):
|
|
return [json.loads(line) for line in self.trace.read_text().splitlines()] if self.trace.exists() else []
|
|
|
|
def run_script(self, name):
|
|
return subprocess.run(["bash", str(HERE / name)], env=self.env, capture_output=True, text=True)
|
|
|
|
def test_invalid_socket_rejected_before_ssh(self):
|
|
self.env['REMOTE_SOCKET'] = '/run/kgtest-pg/pg.sock'
|
|
result = self.run_script('pg-forward.sh')
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertFalse(any(Path(c[0]).name == 'ssh' for c in self.calls()))
|
|
|
|
def test_executor_mounts_socket_and_secret(self):
|
|
self.stub('id', "print('0')")
|
|
self.stub('sleep', '')
|
|
self.stub('docker', "if sys.argv[1:2] == ['inspect']:\n if '--format' in sys.argv: print('running 0')\n else: sys.exit(1)")
|
|
secret = self.root / 'pgpass'
|
|
secret.write_text('localhost:5432:governance:tester:example-only\n')
|
|
secret.chmod(0o600)
|
|
envfile = self.root / 'executor.env'
|
|
envfile.write_text('LIBRARY_WORKER_ROLE=release-executor\nPGPASSFILE=/run/secrets/governance_pgpass\n')
|
|
envfile.chmod(0o600)
|
|
self.env.update(RELEASE_EXECUTOR_IMAGE='fixture:only', RELEASE_EXECUTOR_ENV=str(envfile),
|
|
RELEASE_ARCHIVE_ROOT=str(self.root / 'archive'), RELEASE_JOURNAL_ROOT=str(self.root / 'journal'),
|
|
RELEASE_GATE_STATE=str(self.root), RELEASE_PG_SOCKET_DIR=str(self.root),
|
|
RELEASE_PGPASS_FILE=str(secret), RELEASE_BUNDLE_DIR=str(self.root / 'bundles'))
|
|
result = self.run_script('run-release-executor.sh')
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
call = next(c for c in self.calls() if c[1:2] == ['run'])
|
|
self.assertEqual(call[-1], '/app/worker')
|
|
self.assertEqual(call[call.index('--user') + 1], '0:0')
|
|
self.assertEqual(call[call.index('--network') + 1], 'weknora-test-v071-network')
|
|
self.assertIn(str(self.root) + ':/run/kgtest-pg:ro', call)
|
|
self.assertIn(str(secret) + ':/run/secrets/governance_pgpass:ro', call)
|
|
bundles = self.root / 'bundles'
|
|
self.assertIn(str(bundles) + ':' + str(bundles), call)
|
|
self.assertEqual(bundles.stat().st_mode & 0o777, 0o700)
|
|
self.assertNotIn('example-only', result.stdout + result.stderr + json.dumps(self.calls()))
|
|
|
|
def test_libpq_socket_directory_contract(self):
|
|
result = self.run_script("pg-forward.sh")
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
ssh = [c for c in self.calls() if Path(c[0]).name == "ssh"]
|
|
forward = ssh[-1]
|
|
self.assertEqual(forward[forward.index("-R") + 1], "/run/kgtest-pg/.s.PGSQL.5432:172.21.0.3:5432")
|
|
self.assertIn("host=/run/kgtest-pg&", result.stdout)
|
|
self.assertNotIn("host=/run/kgtest-pg/.s.PGSQL", result.stdout)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|