"""部署切换事务：构建失败不动旧服务，schema 不兼容在停服前拒绝，中断后对账回滚。"""
import importlib.util
import json
import os
import unittest
from pathlib import Path

SOURCE = Path(__file__).resolve().parents[1]
_spec = importlib.util.spec_from_file_location(
    "tsi", SOURCE / "tests" / "test_svc_identity.py")
tsi = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(tsi)


def caps(target, floor):
    return (
        '{"protocol":1,"product":"agent-system","version":"stub",'
        '"identity":{"ops":["ensure","verify"],"open":"existing-no-migrate","dsn":"env"},'
        '"cli":{"unknownSubcommand":"reject"},'
        '"schema":{"target":%d,"compatFloor":%d}}\n' % (target, floor)
    )


class DeployTxnTests(unittest.TestCase):
    def setUp(self):
        self.fx = tsi.IdentityTests("test_T1_bare_restart_stable_identity")
        self.fx.setUp()

    def tearDown(self):
        self.fx.tearDown()
        self.fx.doCleanups()

    def test_schema_incompatible_refuses_before_stop(self):
        fx = self.fx
        old = fx.root / "stub-old"
        new = fx.root / "stub-new"
        text = (SOURCE / "tests" / "test_svc_recover.py").read_text()
        # 复用桩的 identity/serve，只替换 version --json。
        old.write_text(text[text.find("STUB = r'''") and 0:])
        # 直接基于已写入的 recover 桩文件内容不够稳。用 fx.stub 再替换 JSON。
        base = fx.stub.read_text()
        marker = '\'{"protocol":1,"product":"agent-system","version":"stub","identity":{"ops":["ensure","verify"],"open":"existing-no-migrate","dsn":"env"},"cli":{"unknownSubcommand":"reject"}}\''
        old.write_text(base.replace(marker, "'" + caps(1, 0).strip() + "'"))
        new.write_text(base.replace(marker, "'" + caps(5, 5).strip() + "'"))
        old.chmod(0o755)
        new.chmod(0o755)
        fx.stub.write_text(old.read_text())
        fx.stub.chmod(0o755)
        fx.seed_agent("txn", 9471)
        fx.start_ok("txn")
        pid = fx.pid_of("txn")
        script = fx.root / "scripts" / "svc.sh"
        probe = fx.svc(
            "txn", "status", check=False)
        self.assertEqual(probe.returncode, 0, probe.stdout)
        out = __import__("subprocess").run(
            ["bash", "-c",
             'source "%s" >/dev/null 2>&1; ENV_NAME=txn; ENV_DIR="%s"; ENV_BIN="%s"; '
             'if svc_schema_compatible "%s" "%s"; then echo OK; else echo NO; fi'
             % (script, fx.runtime("txn"), fx.runtime("txn") / "bin" / "agent-system",
                fx.runtime("txn") / "bin" / "agent-system", new)],
            cwd=fx.root, env=fx.env, text=True,
            stdout=__import__("subprocess").PIPE,
            stderr=__import__("subprocess").STDOUT, timeout=30)
        self.assertIn("NO", out.stdout, out.stdout)
        self.assertEqual(fx.pid_of("txn"), pid)
        fx.svc("txn", "stop")

    def test_forward_only_requires_confirm_and_backup(self):
        fx = self.fx
        fx.seed_agent("fwd", 9472)
        fx.start_ok("fwd")
        iid = json.loads((fx.runtime("fwd") / "manifest.json").read_text())["instance_id"]
        base = fx.stub.read_text()
        marker = '\'{"protocol":1,"product":"agent-system","version":"stub","identity":{"ops":["ensure","verify"],"open":"existing-no-migrate","dsn":"env"},"cli":{"unknownSubcommand":"reject"}}\''
        installed = fx.root / "stub-installed"
        candidate = fx.root / "stub-forward"
        installed.write_text(base.replace(marker, "'" + caps(1, 0).strip() + "'"))
        candidate.write_text(base.replace(marker, "'" + caps(5, 5).strip() + "'"))
        installed.chmod(0o755)
        candidate.chmod(0o755)
        script = fx.root / "scripts" / "svc.sh"
        envdir = fx.runtime("fwd")

        def run(extra):
            return __import__("subprocess").run(
                ["bash", "-c",
                 'source "%s" >/dev/null 2>&1; ENV_NAME=fwd; ENV_DIR="%s"; '
                 'ROOT="%s"; %s '
                 'if svc_schema_compatible "%s" "%s"; then echo OK; else echo NO; fi'
                 % (script, envdir, fx.root, extra, installed, candidate)],
                cwd=fx.root, env=fx.env, text=True,
                stdout=__import__("subprocess").PIPE,
                stderr=__import__("subprocess").STDOUT, timeout=30).stdout

        self.assertIn("NO", run(""))
        self.assertIn("NO", run('CLI_MIGRATE_SCHEMA=1; CLI_CONFIRM="%s";' % iid))
        backup = fx.root / "runtime-backups" / "fwd" / "20261012T000000Z"
        backup.mkdir(parents=True)
        (backup / "ok").write_text("snapshot\n")
        self.assertIn("OK", run('CLI_MIGRATE_SCHEMA=1; CLI_CONFIRM="%s";' % iid))
        fx.svc("fwd", "stop")


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