cesnimda
60 lines
3.1 KiB
Python
60 lines
3.1 KiB
Python
"""Offline failure-path checks; these do not qualify a database or renderer."""
|
|
import contextlib
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
spec = importlib.util.spec_from_file_location('linux_run', Path(__file__).with_name('linux-run.py'))
|
|
runner = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(runner)
|
|
|
|
class CleanupTests(unittest.TestCase):
|
|
def exercise(self, *, sql_failure=False, log_failure=False):
|
|
calls=[]
|
|
migration_reads=0
|
|
def command(args, **kwargs):
|
|
calls.append(args)
|
|
if args==['docker','inspect',runner.DB]:
|
|
return json.dumps([{'Config':{'Image':runner.PIN,'Labels':{'com.docker.compose.project':'jobtracker-it-p9-requal','com.docker.compose.service':'mariadb-test'}},
|
|
'State':{'Running':True,'Health':{'Status':'healthy'}},
|
|
'HostConfig':{'Tmpfs':{'/var/lib/mysql':''},'PortBindings':{'3306/tcp':[{'HostIp':'127.0.0.1','HostPort':'33308'}]}}}])
|
|
if args[:2]==['docker','logs'] and log_failure: raise OSError('Synthetic log capture failure')
|
|
return 'synthetic identity'
|
|
def sql(query):
|
|
nonlocal migration_reads
|
|
if query=='SELECT VERSION()': return '12.2.2-MariaDB-test'
|
|
if '__EFMigrationsHistory' in query:
|
|
migration_reads+=1
|
|
if migration_reads==2 and sql_failure: raise OSError('Synthetic post-run SQL failure')
|
|
return '1'
|
|
return '0'
|
|
def urlopen(url, **kwargs):
|
|
return io.BytesIO(json.dumps({'version':runner.VERSION} if url.endswith('/health') else {'status':'ready'}).encode())
|
|
with tempfile.TemporaryDirectory(prefix='p9-linux-guard-') as directory:
|
|
out=Path(directory)/'linux'
|
|
with patch.object(runner,'OUT',out), patch.object(runner,'command',command), patch.object(runner,'sql',sql), \
|
|
patch.object(runner.urllib.request,'urlopen',urlopen), patch.object(runner.subprocess,'run',return_value=SimpleNamespace(returncode=0)), \
|
|
patch.object(runner.sys,'argv',['linux-run.py']), contextlib.redirect_stdout(io.StringIO()):
|
|
with self.assertRaises(SystemExit) as failure: runner.main()
|
|
self.assertEqual(failure.exception.code,1)
|
|
record=json.loads((out/'process.json').read_text(encoding='utf-8'))
|
|
self.assertEqual(record['browserExitCode'],0)
|
|
self.assertEqual(record['exitCode'],1)
|
|
self.assertTrue(record['cleanup']['apiRemoved'])
|
|
self.assertTrue(record['cleanup']['schemaRemoved'])
|
|
self.assertTrue(any(args[:3]==['docker','rm','--force'] for args in calls))
|
|
self.assertEqual(json.loads((out.parent/'linux-renderer-exit.json').read_text())['exitCode'],1)
|
|
|
|
def test_post_run_sql_failure_cannot_pass(self):
|
|
self.exercise(sql_failure=True)
|
|
|
|
def test_log_failure_does_not_suppress_removal(self):
|
|
self.exercise(log_failure=True)
|
|
|
|
if __name__=='__main__': unittest.main()
|