fix(cv): preserve rewrite language and qualify responsive document controls
This commit is contained in:
cesnimda
@@ -0,0 +1,210 @@
|
||||
"""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()
|
||||
Reference in New Issue
Block a user