cesnimda
218 lines
12 KiB
Python
218 lines
12 KiB
Python
"""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']:
|
|
if '{{json .NetworkSettings.Ports}}' in args:
|
|
return json.dumps({'8080/tcp': [] if failure == 'port' else [{'HostIp': '127.0.0.1', 'HostPort': '5302'}]})
|
|
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):
|
|
calls = self.exercise()
|
|
network = next(args for args in calls if args[:3] == ['docker', 'network', 'create'])
|
|
self.assertNotIn('--internal', network)
|
|
|
|
def test_api_startup_failure_still_cleans_every_owned_resource(self):
|
|
self.exercise('startup')
|
|
|
|
def test_unpublished_api_port_fails_and_cleans_every_owned_resource(self):
|
|
self.exercise('port')
|
|
|
|
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()
|