165 lines
8.3 KiB
Python
165 lines
8.3 KiB
Python
#!/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()
|