kangshifu-board/test/test_reconcile_unified_monitoring.py

"""Reconciliation acceptance against a disposable loopback PostgreSQL instance."""
from contextlib import closing
import importlib.util
import json
import os
from pathlib import Path
import sqlite3
import subprocess
import sys
import tempfile
from types import SimpleNamespace
import unittest

ROOT = Path(__file__).resolve().parents[1]
SCRIPT = ROOT / 'scripts/reconcile-unified-monitoring.py'
RUNTIME = ROOT / 'docs/unified-database-migration-20261109/release-r3/package'
spec = importlib.util.spec_from_file_location('reconcile', SCRIPT)
repair = importlib.util.module_from_spec(spec)
spec.loader.exec_module(repair)
store = repair.load_store(RUNTIME)
CONFIG = os.environ.get('KSF_TEST_UNIFIED_CONFIG')
GENERATION = 'a' * 64
WORKER = 'b' * 32


@unittest.skipUnless(CONFIG, 'requires disposable local PostgreSQL')
class ReconciliationTests(unittest.TestCase):
    def setUp(self):
        config = store.configuration(CONFIG)
        self.assertEqual(config['connection'].get('host'), '127.0.0.1')
        self.assertEqual(int(config['connection'].get('port', 0)), 55439)
        self.assertEqual(os.environ.get('KSF_UNIFIED_TEST_DISPOSABLE'), '1')
        with store.connect_pg(CONFIG) as pg:
            for name, in pg.execute("SELECT nspname FROM pg_namespace WHERE nspname LIKE 'ksfv_%' OR nspname='ksf_voc_control'").fetchall():
                pg.execute('DROP SCHEMA ' + store.identifier(name) + ' CASCADE')
        self.temp = tempfile.TemporaryDirectory()
        self.addCleanup(self.temp.cleanup)
        self.root = Path(self.temp.name)
        (self.root / 'source').mkdir()
        self.source = self.root / 'source/monitoring.sqlite'
        research = self.root / 'research.sqlite'
        with closing(sqlite3.connect(research)) as c:
            c.executescript('CREATE TABLE meta(key TEXT PRIMARY KEY,value TEXT);'
                            'CREATE TABLE products(id TEXT PRIMARY KEY,title TEXT);')
            c.executemany('INSERT INTO meta VALUES(?,?)', [('generation', GENERATION),
                ('schema', 'ksf.voc-library.v2'), ('status', 'ready')])
            c.execute('INSERT INTO products VALUES(?,?)', ('p1', 'unchanged'))
            c.commit()
        with closing(sqlite3.connect(self.source)) as c:
            c.executescript('CREATE TABLE meta(key TEXT PRIMARY KEY,value TEXT);'
                'CREATE TABLE pipeline_workers(id TEXT PRIMARY KEY,generation TEXT NOT NULL,'
                'heartbeat TEXT NOT NULL,interval_seconds INTEGER NOT NULL,state TEXT NOT NULL);'
                'CREATE TABLE events(id INTEGER PRIMARY KEY,body TEXT NOT NULL);')
            c.execute('INSERT INTO events VALUES(1,?)', ('original-event',))
            c.execute('INSERT INTO pipeline_workers VALUES(?,?,?,?,?)',
                      (WORKER, GENERATION, '2026-11-01T13:24:55+00:00', 10, 'WATCHING'))
            c.commit()
        store.migrate(research, CONFIG, apply=True)
        store.migrate(self.source, CONFIG, 'monitoring', apply=True)
        self.modify('UPDATE pipeline_workers SET heartbeat=? WHERE id=?',
                    ('2026-11-01T13:46:57+00:00', WORKER))
        self.counter = 0

    def modify(self, sql, args=()):
        with closing(sqlite3.connect(self.source)) as c:
            c.execute(sql, args)
            c.commit()

    def run_step(self, command, plan=None):
        self.counter += 1
        return repair.run(SimpleNamespace(command=command, runtime_root=str(RUNTIME),
            source=str(self.source), config=CONFIG, expected_research_generation=GENERATION,
            out=str(self.root / ('result-' + str(self.counter))), plan=str(plan) if plan else None))

    def plan(self):
        value = self.run_step('plan')
        return self.root / ('result-' + str(self.counter)) / 'plan.json', value

    def descriptor(self):
        with store.connect_pg(CONFIG, readonly=True) as pg:
            return repair.brief(store.descriptor(pg, 'monitoring'))

    def test_original_r3_reapply_rejects_changed_source(self):
        with self.assertRaisesRegex(ValueError, 'EXISTING_DATASET_REQUIRES_EXPECTED_GENERATION'):
            store.migrate(self.source, CONFIG, 'monitoring', apply=True)

    def test_plan_apply_full_verify_retry_and_prior_snapshot_retained(self):
        before = self.descriptor()
        path, plan = self.plan()
        self.assertEqual(before, self.descriptor())
        self.assertEqual(len(plan['workerChanges']), 1)
        result = self.run_step('apply', path)
        self.assertEqual(result['operation'], 'new-sealed-monitoring-snapshot')
        self.assertTrue(result['allRowsVerified'])
        self.assertEqual(result['businessRowsChanged'], 0)
        after = self.descriptor()
        self.assertNotEqual(before['generation'], after['generation'])
        self.assertEqual(after['revision'], 1)
        self.assertEqual(self.run_step('apply', path)['operation'], 'already-converged')
        self.assertEqual(self.run_step('verify', path)['operation'], 'already-converged')
        with store.connect_pg(CONFIG, readonly=True) as pg:
            self.assertIsNotNone(pg.execute('SELECT 1 FROM pg_namespace WHERE nspname=%s',
                                           (before['storage_schema'],)).fetchone())
            old = pg.execute('SELECT count(*) FROM ksf_voc_control.snapshots WHERE dataset=\'monitoring\'').fetchone()[0]
            self.assertEqual(old, 2)

    def test_non_worker_change_rejected_without_target_writes(self):
        before = self.descriptor()
        self.modify('UPDATE events SET body=? WHERE id=1', ('must-not-import',))
        with self.assertRaisesRegex(ValueError, 'NON_WORKER_DATA_CHANGED'):
            self.plan()
        self.assertEqual(before, self.descriptor())

    def test_changed_source_after_plan_rejected(self):
        path, _ = self.plan()
        before = self.descriptor()
        self.modify('UPDATE pipeline_workers SET heartbeat=?', ('2026-11-01T14:00:00+00:00',))
        with self.assertRaisesRegex(ValueError, 'PLAN_OR_SOURCE_CHANGED'):
            self.run_step('apply', path)
        self.assertEqual(before, self.descriptor())

    def test_pg_revision_changed_after_plan_rejected(self):
        path, _ = self.plan()
        with closing(store.MonitoringConnection(CONFIG)) as c:
            c.execute('UPDATE events SET body=? WHERE id=1', ('new-pg-event',))
            c.commit()
        before = self.descriptor()
        with self.assertRaisesRegex(ValueError, 'NON_WORKER_DATA_CHANGED'):
            self.run_step('apply', path)
        self.assertEqual(before, self.descriptor())

    def test_new_worker_and_stop_status_preserve_all_other_tables(self):
        self.modify('INSERT INTO pipeline_workers VALUES(?,?,?,?,?)',
                    ('c' * 32, GENERATION, '2026-11-01T13:46:58+00:00', 10, 'STOPPED'))
        self.modify('UPDATE pipeline_workers SET state=? WHERE id=?', ('STOPPED', WORKER))
        path, plan = self.plan()
        self.assertEqual(len(plan['workerChanges']), 2)
        self.assertTrue(self.run_step('apply', path)['allRowsVerified'])

    def test_history_removal_and_backward_heartbeat_rejected(self):
        self.modify('UPDATE pipeline_workers SET heartbeat=?', ('2026-11-01T13:00:00+00:00',))
        with self.assertRaisesRegex(ValueError, 'WORKER_HEARTBEAT_MOVED_BACKWARDS'):
            self.plan()
        self.modify('DELETE FROM pipeline_workers')
        with self.assertRaisesRegex(ValueError, 'WORKER_HISTORY_REMOVAL_REJECTED'):
            self.plan()

    def test_schema_change_rejected(self):
        self.modify('ALTER TABLE events ADD COLUMN extra TEXT')
        with self.assertRaisesRegex(ValueError, 'SOURCE_SCHEMA_CHANGED'):
            self.plan()

    def test_verify_before_apply_rejected(self):
        path, _ = self.plan()
        with self.assertRaisesRegex(ValueError, 'TARGET_NOT_CONVERGED'):
            self.run_step('verify', path)

    def test_cli_invocation_uses_sealed_module(self):
        result = subprocess.run([sys.executable, '-B', str(SCRIPT), 'plan',
            '--runtime-root', str(RUNTIME), '--source', str(self.source), '--config', CONFIG,
            '--expected-research-generation', GENERATION, '--out', str(self.root / 'cli-plan')],
            capture_output=True, timeout=30)
        self.assertEqual(result.returncode, 0, result.stdout.decode())
        self.assertTrue(json.loads(result.stdout)['ok'])
        self.assertEqual(json.loads((self.root / 'cli-plan/plan.json').read_text('utf-8'))['storeSha256'], repair.R3_STORE_SHA)


if __name__ == '__main__':
    unittest.main()