test: harden database qualification startup and cleanup evidence
MariaDB persistence qualification / mariadb (push) Failing after 7s
MariaDB persistence qualification / mariadb (push) Failing after 7s
This commit is contained in:
@@ -0,0 +1,44 @@
|
||||
"""Offline ownership regression; no server or browser is started."""
|
||||
import contextlib, importlib.util, io, json, tempfile, unittest
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
spec=importlib.util.spec_from_file_location('lane',Path(__file__).with_name('mariadb-lane.py'))
|
||||
lane=importlib.util.module_from_spec(spec); spec.loader.exec_module(lane)
|
||||
|
||||
class SchemaOwnership(unittest.TestCase):
|
||||
def exercise(self,create_code):
|
||||
queries=[]
|
||||
fixture={'Name':'/'+lane.CONTAINER,'Id':'task-id','Image':'image-id',
|
||||
'Config':{'Image':lane.IMAGE,'Labels':{'com.docker.compose.project':'jobtracker-it-p9-requal','com.docker.compose.service':'mariadb-test'}},
|
||||
'State':{'Running':True},'NetworkSettings':{'Ports':{'3306/tcp':[{'HostIp':'127.0.0.1','HostPort':'33308'}]}},
|
||||
'HostConfig':{'Tmpfs':{'/var/lib/mysql':''}},'Mounts':[]}
|
||||
def run(command,**kwargs):
|
||||
if command[2:4]==['docker','inspect']: return SimpleNamespace(returncode=0,stdout=json.dumps([fixture]),stderr='')
|
||||
query=next(value[len('--execute='):] for value in command if value.startswith('--execute='))
|
||||
queries.append(query)
|
||||
code=create_code if query.startswith('CREATE DATABASE') else 0
|
||||
output='12.2.2-MariaDB-ubu2404' if 'VERSION()' in query else '0'
|
||||
return SimpleNamespace(returncode=code,stdout=output,stderr='')
|
||||
original=Path.write_text
|
||||
def write(path,*args,**kwargs):
|
||||
if create_code==0 and path.name.endswith('-create-schema.log'): raise OSError('Synthetic log-write failure')
|
||||
return original(path,*args,**kwargs)
|
||||
socket=MagicMock(); socket.__enter__.return_value.connect_ex.return_value=1
|
||||
with tempfile.TemporaryDirectory(prefix='p9-schema-owner-',dir=lane.ROOT/'.artifacts/p9-requal-01') as directory:
|
||||
out=Path(directory)
|
||||
self.assertTrue(out.resolve().is_relative_to(lane.ROOT/'.artifacts/p9-requal-01'))
|
||||
with patch.object(lane,'OUT',out), patch.object(lane.subprocess,'run',run), patch.object(lane.socket,'socket',return_value=socket), \
|
||||
patch.object(Path,'write_text',write), patch.object(lane.sys,'argv',['mariadb-lane.py','mariadb-observations']), \
|
||||
patch.object(lane.subprocess,'Popen',side_effect=AssertionError('No browser after setup failure')), contextlib.redirect_stdout(io.StringIO()):
|
||||
code=lane.main()
|
||||
self.assertEqual(code,1)
|
||||
record=json.loads((out/'mariadb-observations-exit.json').read_text(encoding='utf-8'))
|
||||
self.assertEqual(any(query.startswith('DROP DATABASE') for query in queries),create_code==0,record)
|
||||
self.assertEqual(record['cleanup'].get('schemaAbsent',False),create_code==0)
|
||||
|
||||
def test_created_schema_is_owned_before_log_write(self): self.exercise(0)
|
||||
def test_failed_create_never_authorizes_drop(self): self.exercise(1)
|
||||
|
||||
if __name__=='__main__': unittest.main()
|
||||
Reference in New Issue
Block a user