truf-server/tests/container_import_stop_e2e.py

"""Real PostgreSQL lifecycle regression on a fresh, disposable, offline volume.

Only truf-import-stop-test-<32 hex digits> volumes are accepted. Provision the
volume with container_runtime.py first. No scanner, provider, or user data is
opened. --expect-schema-refusal records the pre-fix behavior and withdraws the
synthetic fault before cleanup; the default requires identity-safe shutdown.
"""

import argparse
import json
import os
from pathlib import Path
import re
import runpy
import socket
import subprocess
import sys
import time


def require(value, check):
    if not value:
        raise AssertionError(check)


def emit(**values):
    print(json.dumps(values, sort_keys=True), flush=True)


def main():
    parser = argparse.ArgumentParser(allow_abbrev=False)
    parser.add_argument('--expect-schema-refusal', action='store_true')
    args = parser.parse_args()
    require(sys.platform == 'linux' and os.geteuid() == 10001, 'test_identity')
    require(Path(__file__) == Path('/opt/truf/tests/container_import_stop_e2e.py'), 'test_image')
    require({name for _, name in socket.if_nameindex()} == {'lo'}, 'offline_test')
    mounts = [line.split() for line in Path('/proc/self/mountinfo').read_text().splitlines()]
    data = [row for row in mounts if row[4] == '/data' or row[4].startswith('/data/')]
    require(len(data) == 1 and data[0][4] == '/data'
            and re.fullmatch(r'/var/lib/docker/volumes/truf-import-stop-test-[a-f0-9]{32}/_data', data[0][3])
            and data[0][data[0].index('-') + 1] == 'ext4', 'disposable_test_volume')
    os.umask(0o077)
    runtime = runpy.run_path('/opt/truf/app/container_runtime.py')
    runtime['require_container']()
    require(not any(Path('/data/postgres-linux').iterdir()), 'fresh_cluster')
    require(not Path('/data/runtime-linux/postgres/cluster_identity.json').exists(), 'fresh_identity')
    config_path = runtime['DEFAULT_CONFIG']
    config = runtime['prepare_environment'](config_path)
    import postgres_runtime as pg
    import psycopg
    from runtime_security import ClusterAuthorityLock

    def cli(action):
        child = subprocess.Popen(runtime['_bootstrap_command'](
            'postgres-runtime', action, '--config', str(config_path)),
            stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        try:
            code = child.wait(timeout=60)
        except subprocess.TimeoutExpired:
            emit(stage=action, status='FAILED_HOLD', reason='lifecycle_child_timeout')
            # The child, not a timeout, owns compensation of an uncertain start.
            code = child.wait()
        require(code == 0, 'lifecycle_cli_' + action)

    def connect():
        return psycopg.connect(os.environ['SCANNER_DB_URL'], autocommit=True,
                              connect_timeout=3, options='-c statement_timeout=5000')

    def stop(backend, stage):
        started = time.monotonic()
        result = backend.stop()
        emit(stage=stage, completed=result.completed, stopped=result.stopped,
             elapsed_ms=round((time.monotonic() - started) * 1000),
             recovering_refusal='online identity is unavailable' in result.detail)
        return result.completed is True and result.stopped is True

    def cleanup(backend):
        while not stop(backend, 'cleanup'):
            emit(status='FAILED_HOLD', reason='stop_unconfirmed')
            time.sleep(5)
        require(backend.probe().kind == pg.ProbeKind.STOPPED, 'offline_proof')
        backend.close()

    cli('initialize-empty')
    cli('maintenance-start')
    with ClusterAuthorityLock(config, endpoint_dsn=os.environ['SCANNER_DB_URL']):
        backend = pg.PostgresBackend(config)
        try:
            require(backend.probe().kind == pg.ProbeKind.READY, 'handoff_ready')
            require(stop(backend, 'healthy_handoff'), 'healthy_handoff_stop')
        finally:
            cleanup(backend)

    cli('maintenance-start')
    with ClusterAuthorityLock(config, endpoint_dsn=os.environ['SCANNER_DB_URL']):
        backend = pg.PostgresBackend(config)
        fault_installed = False
        stopped = False
        try:
            require(backend.probe().kind == pg.ProbeKind.READY, 'second_handoff_ready')
            with connect() as connection:
                connection.execute('GRANT CREATE ON SCHEMA public TO PUBLIC')
            fault_installed = True
            require(backend.probe().kind != pg.ProbeKind.READY, 'unsafe_schema_not_ready')
            stopped = stop(backend, 'schema_failure_handoff')
            require(stopped is not args.expect_schema_refusal, 'schema_failure_stop_expectation')
        finally:
            # Withdraw only this synthetic fault, never weaken a production gate.
            if fault_installed and not stopped:
                with connect() as connection:
                    connection.execute('REVOKE CREATE ON SCHEMA public FROM PUBLIC')
            cleanup(backend)
    require(not Path('/data/initialized.json').exists(), 'no_application_initialization')
    emit(status='passed', expected_schema_refusal=args.expect_schema_refusal)
    return 0


if __name__ == '__main__':
    try:
        result = main()
    except BaseException as error:
        emit(status='failed', error_type=type(error).__name__)
        result = 1
    raise SystemExit(result)