"""Offline lifecycle failure checks. These tests never call Docker, .NET or a browser.""" import contextlib import copy import importlib.util import io import json from pathlib import Path from types import SimpleNamespace import tempfile import unittest from unittest.mock import patch spec = importlib.util.spec_from_file_location('cv_ux_linux', Path(__file__).with_name('linux-run.py')) runner = importlib.util.module_from_spec(spec) spec.loader.exec_module(runner) IMAGE = 'sha256:' + 'b' * 64 SOURCE = {'JobTrackerApi/Program.cs': 'a' * 64} COMMIT = 'c' * 40 PROJECT = 'jobtracker-it-' + 'a' * 32 DB = PROJECT + '-mariadb-test-1' def database(port=33309): return {'Name': '/' + DB, 'Config': {'Image': runner.PIN, 'Labels': { 'com.docker.compose.project': PROJECT, 'com.docker.compose.service': 'mariadb-test'}}, 'State': {'Running': True, 'Health': {'Status': 'healthy'}}, 'HostConfig': {'Tmpfs': {'/var/lib/mysql': ''}, 'NetworkMode': 'bridge', 'PortBindings': {'3306/tcp': [{'HostIp': '127.0.0.1', 'HostPort': str(port)}]}}, 'Mounts': [{'Destination': '/var/lib/mysql', 'Type': 'tmpfs'}]} def build_record(): return {'exitCode': 0, 'baseImageId': runner.BASE, 'imageId': IMAGE, 'applicationPath': runner.APP, 'sources': SOURCE, 'sourceSha256': runner.identity(SOURCE), 'commit': COMMIT, 'publish': {'JobTrackerApi.dll': 'f' * 64}, 'publishSha256': 'e' * 64} class GuardTests(unittest.TestCase): def test_mutable_or_wrong_image_and_source_are_rejected(self): for image in ('jobtracker:latest', 'sha256:abc', 'sha256:' + 'd' * 64): with self.assertRaises(ValueError): runner.verify_build(build_record(), image, SOURCE) with self.assertRaises(ValueError): runner.verify_build(build_record(), IMAGE, {'different-source': '0' * 64}) def test_only_generated_schema_can_be_removed(self): self.assertEqual(runner.schema_name('jt_it_' + 'a' * 32), 'jt_it_' + 'a' * 32) for name in ('jobtracker', 'jt_it_abc', 'jt_it_' + 'a' * 32 + ';DROP DATABASE x'): with self.assertRaises(ValueError): runner.schema_name(name) def test_wrong_database_identity_or_persistent_storage_is_rejected(self): runner.guard_db(database(), DB, PROJECT, 33309) mutations = [('image', lambda d: d['Config'].update(Image='mariadb:latest')), ('project', lambda d: d['Config']['Labels'].update({'com.docker.compose.project': 'production'})), ('port', lambda d: d['HostConfig']['PortBindings'].update({'3306/tcp': [{'HostIp': '0.0.0.0', 'HostPort': '33309'}]})), ('persistent', lambda d: d['Mounts'][0].update(Type='volume')), ('privileged', lambda d: d['HostConfig'].update(Privileged=True)), ('host', lambda d: d['HostConfig'].update(NetworkMode='host'))] for label, mutate in mutations: with self.subTest(label=label): value = copy.deepcopy(database()) mutate(value) with self.assertRaises(ValueError): runner.guard_db(value, DB, PROJECT, 33309) def test_existing_or_outside_evidence_root_is_rejected(self): with tempfile.TemporaryDirectory() as directory, patch.object(runner, 'ROOT', Path(directory)): root = Path(directory) / '.artifacts/cv-ux-01/attempt' runner.fresh(root) with self.assertRaises(ValueError): runner.fresh(root) with self.assertRaises(ValueError): runner.fresh(Path(directory) / 'outside') class LifecycleTests(unittest.TestCase): def test_image_preparation_uses_current_portable_publish_and_offline_new_app_directory(self): calls = [] with tempfile.TemporaryDirectory() as directory: root = Path(directory) out = root / '.artifacts/cv-ux-01/image' class Commands: def __init__(self, output): pass def run(self, args, **kwargs): calls.append(args) if 'publish' in args: publish = Path(args[args.index('--output') + 1]) publish.mkdir() (publish / 'JobTrackerApi.dll').write_bytes(b'current compiled API') (publish / 'JobTrackerApi.deps.json').write_text('{}') if args[:3] == ['docker', 'image', 'inspect']: if '{{json .Config.Labels}}' in args: return '{}' return runner.BASE if args[-1] == runner.BASE or args[-1].startswith('jobtracker-cv-ux-native:') else IMAGE return COMMIT if args == ['git', 'rev-parse', 'HEAD'] else '' with patch.object(runner, 'ROOT', root), patch.object(runner, 'Commands', Commands), \ patch.object(runner, 'sources', return_value=SOURCE), contextlib.redirect_stdout(io.StringIO()): code = runner.build(SimpleNamespace(output=str(out), dotnet='local-dotnet')) self.assertEqual(code, 0) record = json.loads((out / 'build.json').read_text()) self.assertEqual(record['imageId'], IMAGE) self.assertEqual(record['sources'], SOURCE) self.assertIn('WORKDIR ' + runner.APP, (out / 'Dockerfile').read_text()) self.assertNotEqual(runner.APP, '/app') publish = next(args for args in calls if 'publish' in args) self.assertIn('--no-restore', publish) self.assertIn('/p:UseAppHost=false', publish) docker = next(args for args in calls if args[:2] == ['docker', 'build']) self.assertEqual(docker[docker.index('--network') + 1], 'none') self.assertIn('--pull=false', docker) def exercise(self, failure=None): calls = [] with tempfile.TemporaryDirectory() as directory: root = Path(directory) out = root / '.artifacts/cv-ux-01/renderer' saved = root / 'build.json' saved.write_text(json.dumps(build_record()), encoding='utf-8') class Commands: def __init__(self, output): self.out = output def run(self, args, **kwargs): calls.append(args) if args[:3] == ['docker', 'image', 'inspect'] and '--format' not in args: return json.dumps([{'Id': IMAGE, 'Config': {'WorkingDir': runner.APP, 'Labels': { 'io.jobjakt.cv-ux.source-sha256': runner.identity(SOURCE), 'io.jobjakt.cv-ux.publish-sha256': 'e' * 64}}}]) if args[:2] == ['docker', 'inspect'] and args[-1] == DB: return json.dumps([database()]) if args[:2] == ['docker', 'inspect']: return IMAGE if '--entrypoint' in args and args[args.index('--entrypoint') + 1] == 'sha256sum': return 'f' * 64 + ' ' + runner.APP + '/JobTrackerApi.dll' if args[:2] == ['docker', 'exec']: query = args[-1] if query == 'SELECT VERSION()': return '12.2.2-MariaDB-offline-test' if '__EFMigrationsHistory' in query: return '20261025090000_AddCoverLetterDocumentDesign' return '0' if args[:2] == ['docker', 'logs'] and failure == 'log': raise OSError('Synthetic log failure') if args[:3] == ['docker', 'rm', '--force'] and failure == 'cleanup': raise OSError('Synthetic removal failure') if args[0] == 'node': if failure == 'browser': raise OSError('Synthetic browser failure') runner.write(out / 'linux-renderer.json', {'stats': {'expected': 6, 'unexpected': 0, 'flaky': 0, 'skipped': 0}, 'errors': []}) for name, fixtures in [('fidelity', [{}] * 16), ('letter-fidelity', [{'fixture': name} for name in ['light-en-a4', 'dark-nb-a4', 'light-nb-letter', 'dark-en-letter']])]: (out / name).mkdir() runner.write(out / name / 'measurements.json', {'fixtures': fixtures}) (out / 'settings-fidelity').mkdir() for index in range(4): runner.write(out / 'settings-fidelity' / f'{index}-measurements.json', {'fixtures': [{}, {}]}) return 'synthetic-runtime-identity' def urlopen(url, **kwargs): if failure == 'startup': raise OSError('Synthetic startup failure') return io.BytesIO(json.dumps({'version': 'cv-ux-linux-' + runner.identity(SOURCE)[:12]} if url.endswith('/health') else {'status': 'ready'}).encode()) options = SimpleNamespace(output=str(out), image=IMAGE, build_record=str(saved)) with patch.object(runner, 'ROOT', root), patch.object(runner, 'Commands', Commands), \ patch.object(runner, 'sources', return_value=SOURCE), \ patch.object(runner, 'tree', side_effect=lambda p: build_record()['publish'] if p.name == 'publish' else {'index.html': '1' * 64}), \ patch.object(runner, 'free_port', side_effect=lambda port=0: port or 33309), \ patch.object(runner.uuid, 'uuid4', return_value=SimpleNamespace(hex='a' * 32)), \ patch.object(runner.urllib.request, 'urlopen', urlopen), \ patch.object(runner, 'auth_identity', return_value={'appVersion': 'synthetic', 'appCommitSha': COMMIT}), \ patch.object(runner.time, 'monotonic', side_effect=[0, 121] if failure == 'startup' else lambda: 0), \ contextlib.redirect_stdout(io.StringIO()): code = runner.run(options) record = json.loads((out / 'process.json').read_text(encoding='utf-8')) self.assertEqual(code, 0 if failure is None else 1) self.assertEqual(record['exitCode'], code) if failure == 'cleanup': self.assertIn('apiRemovedError', record['cleanup']) self.assertNotIn('schemaRemoved', record['cleanup']) else: self.assertTrue(record['cleanup']['apiRemoved']) self.assertTrue(record['cleanup']['schemaRemoved']) self.assertTrue(record['cleanup']['databaseRemoved']) self.assertTrue(any(args[:3] == ['docker', 'rm', '--force'] for args in calls)) return calls def test_success_records_positive_cleanup(self): self.exercise() def test_api_startup_failure_still_cleans_every_owned_resource(self): self.exercise('startup') def test_log_failure_cannot_suppress_removal_or_pass(self): self.exercise('log') def test_api_removal_failure_prevents_schema_drop_and_pass(self): self.exercise('cleanup') def test_browser_failure_cleans_resources_and_stays_failed(self): self.exercise('browser') if __name__ == '__main__': unittest.main()