#!/usr/bin/env python3
"""Run the production eligibility SQL against controlled PostgreSQL 18 fixtures.

Explicit opt-in integration test: creates/removes only its own disposable Podman
container. No host ports, volumes, cluster credentials or live databases used.
The postgres:18 image must already be present locally.
"""
from pathlib import Path
import os
import json
import shlex
import subprocess
import tempfile
import time
import uuid

ROOT = Path(__file__).resolve().parents[1]
with tempfile.TemporaryDirectory(prefix='mc-score-sql-') as directory:
    query_file = Path(directory) / 'query.sql'
    env = dict(os.environ, QUERY_FILE=str(query_file))
    extract = '. ' + shlex.quote(str(ROOT / 'refactored/managed-container.d/postgres.sh')) + '''
as_service_user() { printf '%s' "${@: -1}" > "$QUERY_FILE"; echo t; }
mc_pg_promotion_eligible
'''
    subprocess.run(['bash', '-c', extract], env=env, check=True)
    query = query_file.read_text()
    subprocess.run(['bash', '-c', extract.replace('mc_pg_promotion_eligible', 'mc_pg_standby_observation')], env=env, check=True)
    observation_query = query_file.read_text()
    name = 'mc-score-sql-' + uuid.uuid4().hex[:12]
    subprocess.run(['podman', 'run', '-d', '--pull=never', '--name', name,
                    '--network=none', '--tmpfs', '/var/lib/postgresql',
                    '-e', 'POSTGRES_HOST_AUTH_METHOD=trust',
                    'docker.io/library/postgres:18'], check=True, capture_output=True)
    try:
        for _ in range(40):
            ready = subprocess.run(['podman', 'exec', name, 'pg_isready', '-U', 'postgres'],
                                   capture_output=True)
            if ready.returncode == 0:
                break
            time.sleep(0.5)
        else:
            raise RuntimeError('Disposable PostgreSQL did not become ready')
        def sql(text):
            return subprocess.check_output(['podman', 'exec', '-i', name, 'psql',
                                            '-X', '-U', 'postgres', '-v', 'ON_ERROR_STOP=1',
                                            '-Atq'], input=text, text=True).strip()
        assert sql('SHOW server_version_num;').startswith('18')
        sql('''
CREATE SCHEMA score_test;
CREATE TABLE score_test.state (recovery bool, paused bool, replay pg_lsn);
INSERT INTO score_test.state VALUES (true,false,'0/20');
CREATE TABLE score_test.pg_stat_wal_receiver (status text, flushed_lsn pg_lsn);
CREATE FUNCTION score_test.pg_is_in_recovery() RETURNS bool LANGUAGE sql
AS 'SELECT recovery FROM score_test.state';
CREATE FUNCTION score_test.pg_is_wal_replay_paused() RETURNS bool LANGUAGE plpgsql AS $$
BEGIN
  IF NOT (SELECT recovery FROM score_test.state) THEN
    RAISE EXCEPTION 'replay pause is undefined on a primary';
  END IF;
  RETURN (SELECT paused FROM score_test.state);
END $$;
CREATE FUNCTION score_test.pg_last_wal_replay_lsn() RETURNS pg_lsn LANGUAGE sql
AS 'SELECT replay FROM score_test.state';
''')
        cases = [
            ('caught-up streaming standby', True, False, '0/20', 'streaming', '0/20', 't'),
            ('replay ahead of receiver snapshot', True, False, '0/30', 'streaming', '0/20', 't'),
            ('replay lag', True, False, '0/10', 'streaming', '0/20', 'f'),
            ('paused replay', True, True, '0/20', 'streaming', '0/20', 'f'),
            ('no replay position', True, False, None, 'streaming', '0/20', 'f'),
            ('unknown flushed position', True, False, '0/20', 'streaming', None, 'f'),
            ('receiver stopped', True, False, '0/20', 'stopping', '0/20', 'f'),
            ('no receiver after primary failure', True, False, '0/20', None, None, 'f'),
            ('primary skips recovery-only function', False, False, '0/20', None, None, 'f'),
        ]
        def literal(value):
            return 'NULL' if value is None else "'" + value + "'"
        for label, recovery, paused, replay, status, flushed, expected in cases:
            setup = f'''SET search_path=score_test,pg_catalog;
UPDATE score_test.state SET recovery={str(recovery).lower()},paused={str(paused).lower()},replay={literal(replay)};
TRUNCATE score_test.pg_stat_wal_receiver;
'''
            if status is not None:
                setup += f'INSERT INTO score_test.pg_stat_wal_receiver VALUES ({literal(status)},{literal(flushed)});\n'
            result = sql(setup + query)
            assert result == expected, (label, result, expected)
            observation = json.loads(sql(setup + observation_query))
            assert (observation['strict'] is True) == (expected == 't'), (label, observation)
            assert observation['identity'][0].isdigit() and observation['identity'][1].startswith('/')
            assert observation['recovery'] == recovery
            assert observation['receiver'] == status
            print('PASS strict and observation SQL:', label)
    finally:
        subprocess.run(['podman', 'rm', '-f', name], check=True, capture_output=True)