cesnimda
91 lines
4.5 KiB
Python
91 lines
4.5 KiB
Python
"""Run the explicit MariaDB persistence lane, never a SQLite fallback."""
|
|
import argparse
|
|
import datetime
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import shutil
|
|
import socket
|
|
import subprocess
|
|
import uuid
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
|
|
def compose_command(project):
|
|
if not re.fullmatch(r'jobtracker-it-[a-z0-9]+', project):
|
|
raise ValueError('Only generated integration projects may be managed.')
|
|
return ['docker', 'compose', '--project-name', project, '--project-directory', str(ROOT),
|
|
'-f', str(ROOT / 'docker-compose.database.yml'), '--profile', 'integration']
|
|
|
|
def test_connection(port):
|
|
if not 0 < port < 65536: raise ValueError('Invalid integration port.')
|
|
return f'Server=127.0.0.1;Port={port};User ID=root;Password=jt-integration-only;SslMode=None'
|
|
|
|
def existing_connection(environment):
|
|
value = environment.get('JOBTRACKER_MARIADB_TEST_ADMIN', '')
|
|
if not value.strip(): raise ValueError('--existing-server requires explicit JOBTRACKER_MARIADB_TEST_ADMIN.')
|
|
# The fixture independently enforces loopback TCP, exact server version and generated schemas.
|
|
return value
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--existing-server', action='store_true', help='Use an explicitly configured isolated loopback server; fixture guards still apply.')
|
|
parser.add_argument('--output', default=str(ROOT / '.artifacts/mariadb/integration'))
|
|
args = parser.parse_args()
|
|
environment = os.environ.copy()
|
|
output = Path(args.output).resolve()
|
|
output.mkdir(parents=True, exist_ok=True)
|
|
project = 'jobtracker-it-' + uuid.uuid4().hex
|
|
compose = compose_command(project)
|
|
managed = not args.existing_server
|
|
started = datetime.datetime.now(datetime.timezone.utc).isoformat()
|
|
commands = []
|
|
code = 1
|
|
def run(command, name):
|
|
commands.append({'command': command, 'log': str(output / name)})
|
|
with (output / name).open('w', encoding='utf-8') as log:
|
|
result = subprocess.run(command, cwd=ROOT, env=environment, stdout=log, stderr=subprocess.STDOUT)
|
|
commands[-1]['exitCode'] = result.returncode
|
|
return result.returncode
|
|
try:
|
|
if managed:
|
|
with socket.socket() as probe:
|
|
probe.bind(('127.0.0.1', 0))
|
|
port = probe.getsockname()[1]
|
|
environment['MARIADB_TEST_PORT'] = str(port)
|
|
environment['JOBTRACKER_MARIADB_TEST_ADMIN'] = test_connection(port)
|
|
code = run(compose + ['up', '-d', '--wait', '--wait-timeout', '120', 'mariadb-test'], 'compose-start.log')
|
|
else:
|
|
existing_connection(environment)
|
|
code = 0
|
|
if code == 0:
|
|
dotnet = environment.get('DOTNET_HOST_PATH') or shutil.which('dotnet')
|
|
if not dotnet: raise ValueError('Pinned .NET SDK is required; set DOTNET_HOST_PATH.')
|
|
project_file = 'JobTrackerApi.MariaDb.Tests/JobTrackerApi.MariaDb.Tests.csproj'
|
|
code = run([dotnet, 'restore', project_file, '--locked-mode'], 'restore.log')
|
|
if code == 0:
|
|
code = run([dotnet, 'test', project_file, '--configuration', 'Release', '--no-restore',
|
|
'--logger', 'trx;LogFileName=mariadb.trx', '--results-directory', str(output)], 'tests.log')
|
|
except Exception:
|
|
code = 1
|
|
raise
|
|
finally:
|
|
if managed:
|
|
# Never use down/-v: this generated project only owns its disposable test service.
|
|
run(compose + ['logs', '--no-color', 'mariadb-test'], 'server.log')
|
|
cleanup = run(compose + ['rm', '--stop', '--force', 'mariadb-test'], 'compose-cleanup.log')
|
|
if cleanup:
|
|
print('Test container cleanup failed; inspect compose-cleanup.log.')
|
|
if code == 0: code = cleanup
|
|
record = {'startedUtc':started, 'finishedUtc':datetime.datetime.now(datetime.timezone.utc).isoformat(),
|
|
'mode':'compose' if managed else 'explicit-existing-loopback', 'expectedVersion':environment.get('JOBTRACKER_MARIADB_TEST_VERSION','12.2.2'),
|
|
'project':project if managed else None, 'exitCode':code, 'commands':commands}
|
|
(output / 'run.json').write_text(json.dumps(record, indent=2)+'\n', encoding='utf-8')
|
|
print(json.dumps({'exitCode':code, 'evidence':str(output), 'mode':record['mode']}))
|
|
return code
|
|
|
|
if __name__ == '__main__':
|
|
try: raise SystemExit(main())
|
|
except ValueError as error: raise SystemExit(str(error))
|