cesnimda
45 lines
3.0 KiB
Python
45 lines
3.0 KiB
Python
"""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()
|