#!/usr/bin/env python3 """Control-plane tests only: fake engine, never upstream or backend acceptance.""" import json import os from pathlib import Path import subprocess import tempfile import unittest ROOT = Path(__file__).resolve().parents[1] FAKE = r'''#!/usr/bin/env python3 import json, os, pathlib, sys root = pathlib.Path(os.environ['FAKE_ROOT']) a = sys.argv[1:] with (root/'calls').open('a') as out: out.write(json.dumps(a)+'\n') if a[:2] == ['image','exists']: sys.exit(1 if os.environ.get('MISSING_IMAGE') else 0) elif a[0] == 'run': name = a[a.index('--name')+1] (root/name).write_text(json.dumps(a)) print(name) elif a[:2] == ['container','exists']: sys.exit(0 if (root/a[2]).exists() else 1) elif a[0] == 'logs': print('Immich Microservices is running') elif a[0] == 'exec': if a[-1] == '/contract.mjs': sys.exit(int(os.environ.get('CONTRACT_EXIT','0'))) elif a[0] == 'inspect': if '--format' in a: print('{}'); sys.exit(0) result=[] def size(s): return int(s[:-1])*{'m':1024**2,'g':1024**3}[s[-1]] for name in a[1:]: args=json.loads((root/name).read_text()) def option(key): return next(x.split('=',1)[1] for x in reversed(args) if x.startswith(key+'=')) mounts=[]; scratch={} for i,value in enumerate(args[:-1]): if value=='--volume': source,target,mode=args[i+1].rsplit(':',2) mounts.append(dict(Type='bind',Source=source,Destination=target,RW=mode!='ro')) if value=='--tmpfs': target,options=args[i+1].split(':',1); scratch[target]=options caps=next((x.split('=',1)[1].upper().split(',') for x in args if x.startswith('--cap-add=')),[]) network=option('--network'); network='none' if network=='none' else 'container:'+'d'*64 image=next(x for x in args if len(x)==64 and set(x)<=set('abc')) result.append(dict(Name=name,Id='d'*64,Image=image, Config={'Labels':{'org.archipelago.cloud-fixture':'immich274-v1'}}, HostConfig=dict(NetworkMode=network,Privileged=False,ReadonlyRootfs=True,PidMode='private',IpcMode='private', Memory=size(option('--memory')),MemorySwap=size(option('--memory-swap')),PidsLimit=int(option('--pids-limit')), NanoCpus=int(option('--cpus'))*10**9,SecurityOpt=['no-new-privileges'],Tmpfs=scratch), EffectiveCaps=caps or None,BoundingCaps=caps or None,Mounts=mounts,State={'Running':True,'Pid':123})) tamper=os.environ.get('TAMPER') if tamper=='network': result[0]['HostConfig']['NetworkMode']='host' if tamper=='memory': result[2]['HostConfig']['Memory']=8*1024**3 if tamper=='capability': result[2]['EffectiveCaps']=['SYS_ADMIN'] if tamper=='missing-caps': del result[2]['EffectiveCaps'] if tamper=='mount': result[2]['Mounts'].append(dict(Type='bind',Source='/synthetic',Destination='/root',RW=True)) if tamper=='image': result[2]['Image']='e'*64 print(json.dumps(result)) elif a[0] == 'rm': if not os.environ.get('FAIL_CLEANUP'): (root/a[-1]).unlink(missing_ok=True) else: sys.exit(99) ''' class ImmichRunner(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.path = Path(self.temp.name) tool = self.path / 'podman' tool.write_text(FAKE) tool.chmod(0o755) self.env = dict(os.environ, PATH=f'{self.path}:' + os.environ['PATH'], FAKE_ROOT=str(self.path), ARCHY_CLOUD_IMMICH_IMAGE='a'*64, ARCHY_CLOUD_IMMICH_DB_IMAGE='b'*64, ARCHY_CLOUD_IMMICH_REDIS_IMAGE='c'*64) def run_fixture(self): return subprocess.run(['bash', str(ROOT/'tests/cloud/run-immich-disposable.sh')], env=self.env, capture_output=True, text=True, timeout=15) def calls(self): path = self.path/'calls' return [json.loads(line) for line in path.read_text().splitlines()] if path.exists() else [] def test_mutable_or_short_images_refused_before_start(self): for value in ('immich:release', 'abc123', 'a'*63, 'A'*64): self.env['ARCHY_CLOUD_IMMICH_IMAGE'] = value self.assertEqual(self.run_fixture().returncode, 2) self.assertFalse(self.calls()) def test_missing_local_image_never_pulls_or_starts(self): self.env['MISSING_IMAGE'] = '1' self.assertNotEqual(self.run_fixture().returncode, 0) self.assertTrue(all(call[:2] == ['image', 'exists'] for call in self.calls())) def test_three_service_isolation_and_namespace_cleanup_order(self): result = self.run_fixture() self.assertEqual(result.returncode, 0, result.stderr) calls = self.calls() runs = [call for call in calls if call[0] == 'run'] self.assertEqual(len(runs), 3) db, redis, server = runs owner = db[db.index('--name')+1] self.assertIn('--network=none', db) for call in (redis, server): self.assertIn('--network=container:'+owner, call) for call in runs: for flag in ('--pull=never', '--image-volume=ignore', '--read-only', '--pid=private', '--ipc=private', '--http-proxy=false', '--no-hosts', '--cap-drop=all', '--security-opt=no-new-privileges', '--timeout=420'): self.assertIn(flag, call) self.assertFalse(any(arg in ('-p', '--publish', '--privileged') or 'network=host' in arg or 'podman.sock' in arg or 'docker.sock' in arg for arg in call)) mounts = [call[i+1] for i, value in enumerate(call[:-1]) if value == '--volume'] self.assertTrue(all(value.endswith(':ro') for value in mounts)) memory = next(value.split('=',1)[1] for value in call if value.startswith('--memory=')) self.assertIn('--memory-swap='+memory, call) self.assertTrue(any(value.endswith(':/etc/hosts:ro') for value in mounts)) self.assertTrue(any(value.endswith(':/etc/resolv.conf:ro') for value in mounts)) self.assertIn('--memory=1g', db) self.assertIn('--memory=128m', redis) self.assertIn('--memory=2g', server) names = [call[call.index('--name')+1] for call in runs] removed = [call[-1] for call in calls if call[0] == 'rm'] self.assertEqual(removed, names[::-1]) contract = next(i for i, call in enumerate(calls) if call[0] == 'exec' and call[-1] == '/contract.mjs') self.assertTrue(any(call[0] == 'logs' for call in calls[:contract])) for call in runs: hosts = next(value.split(':/etc/hosts:')[0] for value in call if value.endswith(':/etc/hosts:ro')) self.assertFalse(Path(hosts).exists(), 'Private hosts scratch must be removed') def test_inspect_drift_refused_before_protocol_execution(self): for tamper in ('network', 'memory', 'capability', 'missing-caps', 'mount', 'image'): with self.subTest(tamper=tamper): self.env['TAMPER'] = tamper calls = self.path / 'calls' calls.unlink(missing_ok=True) result = self.run_fixture() self.assertEqual(result.returncode, 2, result.stderr) self.assertFalse(any(call[0] == 'exec' and call[-1] == '/contract.mjs' for call in self.calls())) self.assertEqual(len([call for call in self.calls() if call[0] == 'rm']), 3) def test_browser_mode_requires_precompiled_artifact_before_engine_calls(self): self.env['ARCHY_CLOUD_BROWSER_QUALIFY'] = '1' self.env.pop('ARCHY_TEST_REUSE_ARTIFACT', None) self.assertNotEqual(self.run_fixture().returncode, 0) self.assertFalse(self.calls()) def test_contract_failure_propagates_and_still_cleans_all_services(self): self.env['CONTRACT_EXIT'] = '7' result = self.run_fixture() self.assertEqual(result.returncode, 7, result.stderr) self.assertEqual(len([call for call in self.calls() if call[0] == 'rm']), 3) self.assertNotIn('PASS portable', result.stdout) def test_cleanup_failure_cannot_report_successful_exit(self): self.env['FAIL_CLEANUP'] = '1' result = self.run_fixture() self.assertEqual(result.returncode, 1, result.stderr) self.assertIn('Cleanup failed:', result.stderr) if __name__ == '__main__': unittest.main()