469 lines
20 KiB
Python
469 lines
20 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
|
|
SCRIPT_PATH = Path(__file__).with_name("run-g7-external-proof-window.py")
|
|
SPEC = importlib.util.spec_from_file_location("run_g7_external_proof_window", SCRIPT_PATH)
|
|
assert SPEC is not None and SPEC.loader is not None
|
|
MODULE = importlib.util.module_from_spec(SPEC)
|
|
sys.modules[SPEC.name] = MODULE
|
|
SPEC.loader.exec_module(MODULE)
|
|
|
|
|
|
def args(**overrides):
|
|
base = {
|
|
"wss_url": "wss://api.example.test/voice/ws",
|
|
"origin": "https://web.example.test",
|
|
"allowed_browser_origins": ["https://web.example.test"],
|
|
"admin_runtime_url": "https://api.example.test/admin/voice-runtime",
|
|
"topology_mode": "linux-compose",
|
|
"compose_project": "vignette",
|
|
"api_container": "vignette-api-1",
|
|
"api_image_digest": "sha256:" + "a" * 64,
|
|
"caddy_container": "vignette-proxy-1",
|
|
"caddy_image_digest": "sha256:" + "b" * 64,
|
|
"repo_root": None,
|
|
"git_sha": None,
|
|
"git_tree_sha": None,
|
|
"runner_script_sha256": None,
|
|
"collector_script_sha256": None,
|
|
"checker_script_sha256": None,
|
|
"psutil_version": None,
|
|
"api_pid": None,
|
|
"api_executable_name": None,
|
|
"api_executable_sha256": None,
|
|
"api_cwd": None,
|
|
"api_listen_port": None,
|
|
"cloudflared_pid": None,
|
|
"cloudflared_executable_name": None,
|
|
"cloudflared_executable_sha256": None,
|
|
"cloudflared_cwd": None,
|
|
"expected_stt_provider": "local_whisper",
|
|
"expected_stt_model": "small",
|
|
"expected_tts_provider": "melotts",
|
|
"expected_tts_model": "melotts-korean",
|
|
"microphone_device": "",
|
|
"confirm_physical_capture": False,
|
|
"duration_seconds": 3_120.0,
|
|
"runtime_interval_seconds": 100.0,
|
|
"topology_interval_seconds": 100.0,
|
|
"human_voice_gain": Path("pack.json"),
|
|
"out_dir": Path("out"),
|
|
"rehearse": False,
|
|
}
|
|
base.update(overrides)
|
|
return SimpleNamespace(**base)
|
|
|
|
|
|
class FakeCompleted:
|
|
def __init__(self, returncode: int) -> None:
|
|
self.returncode = returncode
|
|
self.stdout = ""
|
|
self.stderr = ""
|
|
|
|
|
|
class HostAlignmentTest(unittest.TestCase):
|
|
def test_matching_transport_hosts_are_accepted(self) -> None:
|
|
host = MODULE.assert_single_host(
|
|
"wss://api.example.test/voice/ws",
|
|
"https://api.example.test/admin/voice-runtime",
|
|
)
|
|
self.assertEqual(host, "api.example.test")
|
|
|
|
def test_mismatched_transport_hosts_fail_closed_without_echoing_hosts(self) -> None:
|
|
with self.assertRaises(MODULE.WindowError) as caught:
|
|
MODULE.assert_single_host(
|
|
"wss://api.example.test/voice/ws",
|
|
"https://private-tenant.example.test/admin/voice-runtime",
|
|
)
|
|
self.assertEqual("transport_hosts_must_match", str(caught.exception))
|
|
self.assertNotIn("private-tenant", str(caught.exception))
|
|
|
|
def test_browser_origin_is_independent_from_transport_host(self) -> None:
|
|
legs = MODULE.plan_legs(
|
|
args(microphone_device="mic", confirm_physical_capture=True),
|
|
Path("out"),
|
|
admin_probe=lambda origin: ["/admin/voice-runtime"],
|
|
)
|
|
self.assertEqual(legs[0].argv[legs[0].argv.index("--origin") + 1],
|
|
"https://web.example.test")
|
|
|
|
def test_plan_rejects_a_split_transport_window(self) -> None:
|
|
with self.assertRaises(MODULE.WindowError) as caught:
|
|
MODULE.plan_legs(
|
|
args(admin_runtime_url="https://elsewhere.test/admin/voice-runtime",
|
|
microphone_device="mic", confirm_physical_capture=True),
|
|
Path("out"),
|
|
admin_probe=lambda origin: ["/admin/voice-runtime"],
|
|
)
|
|
self.assertEqual("transport_hosts_must_match", str(caught.exception))
|
|
|
|
def test_browser_origin_requires_an_explicit_allowlist_match(self) -> None:
|
|
with self.assertRaises(MODULE.WindowError) as caught:
|
|
MODULE.plan_legs(
|
|
args(origin="https://unlisted.example.test"),
|
|
Path("out"),
|
|
admin_probe=lambda origin: ["/admin/voice-runtime"],
|
|
)
|
|
self.assertEqual("browser_origin_not_allowed", str(caught.exception))
|
|
|
|
def test_added_origin_does_not_remove_the_public_default(self) -> None:
|
|
normalized = MODULE.assert_browser_origin_allowed(
|
|
MODULE.DEFAULT_BROWSER_ORIGIN, ["https://preview.example.test"]
|
|
)
|
|
self.assertEqual(MODULE.DEFAULT_BROWSER_ORIGIN, normalized)
|
|
|
|
def test_transport_schemes_fail_closed_without_reflecting_urls(self) -> None:
|
|
cases = (
|
|
(
|
|
"ws://private-wss.example.test/voice/ws?token=secret",
|
|
"https://api.example.test/admin/voice-runtime",
|
|
"wss_url_scheme_invalid",
|
|
),
|
|
(
|
|
"wss://api.example.test/voice/ws",
|
|
"http://private-admin.example.test/admin/voice-runtime?email=secret",
|
|
"admin_runtime_url_scheme_invalid",
|
|
),
|
|
)
|
|
for wss_url, admin_url, expected in cases:
|
|
with self.subTest(expected=expected):
|
|
with self.assertRaises(MODULE.WindowError) as caught:
|
|
MODULE.assert_transport_schemes(wss_url, admin_url)
|
|
self.assertEqual(expected, str(caught.exception))
|
|
self.assertNotIn("secret", str(caught.exception))
|
|
|
|
def test_browser_origin_query_is_rejected_without_echoing_it(self) -> None:
|
|
secret_query = "https://web.example.test?email=person@example.test"
|
|
with self.assertRaises(MODULE.WindowError) as caught:
|
|
MODULE.assert_browser_origin_allowed(
|
|
secret_query, ["https://web.example.test"]
|
|
)
|
|
self.assertEqual("browser_origin_invalid", str(caught.exception))
|
|
self.assertNotIn("person@example.test", str(caught.exception))
|
|
|
|
def test_public_parser_defaults_match_browser_and_transport_contract(self) -> None:
|
|
parser = MODULE.build_parser()
|
|
parsed = parser.parse_args(["--out-dir", "out"])
|
|
self.assertEqual(MODULE.DEFAULT_WSS_URL, parsed.wss_url)
|
|
self.assertEqual(MODULE.DEFAULT_BROWSER_ORIGIN, parsed.origin)
|
|
self.assertEqual(MODULE.DEFAULT_ADMIN_RUNTIME_URL, parsed.admin_runtime_url)
|
|
self.assertEqual(3_120.0, parsed.duration_seconds)
|
|
self.assertIsNone(parsed.allowed_browser_origins)
|
|
help_text = parser.format_help()
|
|
for expected in (
|
|
"https://vignette.chanpaca.net",
|
|
"wss://api-vignette.chanpaca.net/voice/ws",
|
|
"stt_provider=local_whisper stt_model=small",
|
|
"tts_provider=melotts tts_model=melotts-korean",
|
|
"3120초",
|
|
):
|
|
self.assertIn(expected, help_text)
|
|
|
|
|
|
class SamplePlanTest(unittest.TestCase):
|
|
def test_samples_cover_the_whole_window(self) -> None:
|
|
self.assertEqual(MODULE.sample_plan(3_120.0, 100.0), 33)
|
|
|
|
def test_fractional_interval_always_gets_a_terminal_sample(self) -> None:
|
|
self.assertEqual(MODULE.sample_plan(3_121.0, 100.0), 33)
|
|
|
|
def test_invalid_plan_fails_closed(self) -> None:
|
|
for duration, interval in ((0, 100.0), (3_000.0, 0)):
|
|
with self.subTest(duration=duration, interval=interval):
|
|
with self.assertRaises(MODULE.WindowError):
|
|
MODULE.sample_plan(duration, interval)
|
|
|
|
|
|
class ConsentGateTest(unittest.TestCase):
|
|
def test_production_soak_requires_device_and_confirmation(self) -> None:
|
|
for device, confirm in (("", True), ("mic", False), ("", False)):
|
|
with self.subTest(device=device, confirm=confirm):
|
|
with self.assertRaises(MODULE.WindowError) as ctx:
|
|
MODULE.build_soak_leg(
|
|
args(microphone_device=device, confirm_physical_capture=confirm),
|
|
Path("soak.json"),
|
|
)
|
|
self.assertEqual(
|
|
str(ctx.exception), "physical_microphone_consent_required"
|
|
)
|
|
|
|
def test_confirmed_production_soak_passes_the_flags_through(self) -> None:
|
|
leg = MODULE.build_soak_leg(
|
|
args(microphone_device="mic-1", confirm_physical_capture=True),
|
|
Path("soak.json"),
|
|
)
|
|
self.assertEqual(leg.name, "voice_soak")
|
|
self.assertIn("--confirm-physical-capture", leg.argv)
|
|
self.assertEqual(leg.argv[leg.argv.index("--microphone-device") + 1], "mic-1")
|
|
self.assertNotIn("--preflight-only", leg.argv)
|
|
|
|
def test_rehearse_never_opens_a_microphone(self) -> None:
|
|
leg = MODULE.build_soak_leg(args(rehearse=True), Path("soak.json"))
|
|
self.assertEqual(leg.name, "voice_soak_preflight")
|
|
self.assertIn("--preflight-only", leg.argv)
|
|
self.assertNotIn("--confirm-physical-capture", leg.argv)
|
|
self.assertNotIn("--microphone-device", leg.argv)
|
|
|
|
def test_three_thousand_second_capture_is_rejected_without_skew_margin(self) -> None:
|
|
with self.assertRaises(MODULE.WindowError) as ctx:
|
|
MODULE.validate(args(duration_seconds=3_000.0))
|
|
self.assertEqual(str(ctx.exception), "production_window_too_short")
|
|
|
|
def test_fifty_two_minute_capture_is_accepted(self) -> None:
|
|
MODULE.validate(args(duration_seconds=3_120.0))
|
|
|
|
def test_production_run_requires_the_human_pack(self) -> None:
|
|
with self.assertRaises(MODULE.WindowError) as ctx:
|
|
MODULE.validate(args(human_voice_gain=None))
|
|
self.assertEqual(str(ctx.exception), "human_voice_gain_pack_required")
|
|
|
|
def test_rehearse_relaxes_only_the_two_human_inputs(self) -> None:
|
|
MODULE.validate(args(rehearse=True, duration_seconds=30.0, human_voice_gain=None))
|
|
|
|
|
|
class LegCompositionTest(unittest.TestCase):
|
|
def test_expected_providers_default_to_the_decided_local_stack(self) -> None:
|
|
leg = MODULE.build_soak_leg(args(rehearse=True), Path("soak.json"))
|
|
self.assertEqual(
|
|
leg.argv[leg.argv.index("--expected-stt-provider") + 1], "local_whisper"
|
|
)
|
|
self.assertEqual(
|
|
leg.argv[leg.argv.index("--expected-tts-provider") + 1], "melotts"
|
|
)
|
|
|
|
def test_topology_leg_pins_the_public_host_from_the_wss_url(self) -> None:
|
|
leg = MODULE.build_topology_leg(args(), Path("topology.json"))
|
|
self.assertEqual(
|
|
leg.argv[leg.argv.index("--public-host") + 1], "api.example.test"
|
|
)
|
|
|
|
def test_windows_topology_leg_passes_exact_process_pins(self) -> None:
|
|
windows_args = args(
|
|
topology_mode="windows-host",
|
|
compose_project=None,
|
|
api_container=None,
|
|
api_image_digest=None,
|
|
caddy_container=None,
|
|
caddy_image_digest=None,
|
|
repo_root=r"D:\workspace\vignette",
|
|
git_sha="5" * 40,
|
|
git_tree_sha="6" * 40,
|
|
runner_script_sha256="7" * 64,
|
|
collector_script_sha256="8" * 64,
|
|
checker_script_sha256="9" * 64,
|
|
psutil_version="7.0.0",
|
|
api_pid=301,
|
|
api_executable_name="python.exe",
|
|
api_executable_sha256="3" * 64,
|
|
api_cwd=r"D:\workspace\vignette\apps\api",
|
|
api_listen_port=8001,
|
|
cloudflared_pid=302,
|
|
cloudflared_executable_name="cloudflared.exe",
|
|
cloudflared_executable_sha256="4" * 64,
|
|
cloudflared_cwd=r"D:\workspace\vignette",
|
|
)
|
|
MODULE.validate(windows_args)
|
|
leg = MODULE.build_topology_leg(windows_args, Path("topology.json"))
|
|
|
|
self.assertEqual("windows-host", leg.argv[leg.argv.index("--topology-mode") + 1])
|
|
self.assertEqual("301", leg.argv[leg.argv.index("--api-pid") + 1])
|
|
self.assertEqual(
|
|
"6" * 40, leg.argv[leg.argv.index("--git-tree-sha") + 1]
|
|
)
|
|
self.assertEqual(
|
|
"7" * 64, leg.argv[leg.argv.index("--runner-script-sha256") + 1]
|
|
)
|
|
self.assertEqual(
|
|
"8" * 64,
|
|
leg.argv[leg.argv.index("--collector-script-sha256") + 1],
|
|
)
|
|
self.assertEqual(
|
|
"9" * 64, leg.argv[leg.argv.index("--checker-script-sha256") + 1]
|
|
)
|
|
self.assertEqual("7.0.0", leg.argv[leg.argv.index("--psutil-version") + 1])
|
|
self.assertEqual(
|
|
"4" * 64,
|
|
leg.argv[leg.argv.index("--cloudflared-executable-sha256") + 1],
|
|
)
|
|
self.assertNotIn("--compose-project", leg.argv)
|
|
|
|
def test_each_topology_mode_requires_its_own_pins(self) -> None:
|
|
cases = (
|
|
(args(topology_mode="unknown"), "topology_mode_invalid"),
|
|
(args(compose_project=None), "topology_argument_required:compose-project"),
|
|
(
|
|
args(
|
|
topology_mode="windows-host",
|
|
compose_project=None,
|
|
api_container=None,
|
|
api_image_digest=None,
|
|
caddy_container=None,
|
|
caddy_image_digest=None,
|
|
),
|
|
"topology_argument_required:repo-root",
|
|
),
|
|
)
|
|
for invalid, failure in cases:
|
|
with self.subTest(failure=failure):
|
|
with self.assertRaises(MODULE.WindowError) as caught:
|
|
MODULE.validate(invalid)
|
|
self.assertEqual(failure, str(caught.exception))
|
|
|
|
def test_all_three_legs_share_the_margin_extended_window_length(self) -> None:
|
|
legs = MODULE.plan_legs(
|
|
args(rehearse=True, duration_seconds=3_120.0), Path("out"),
|
|
admin_probe=lambda origin: ["/admin/voice-runtime"],
|
|
)
|
|
self.assertEqual([leg.name for leg in legs][1:], ["runtime", "topology"])
|
|
for leg in legs[1:]:
|
|
samples = int(leg.argv[leg.argv.index("--samples") + 1])
|
|
interval = float(leg.argv[leg.argv.index("--interval-seconds") + 1])
|
|
self.assertGreaterEqual((samples - 1) * interval, 3_120.0)
|
|
|
|
|
|
class WindowExecutionTest(unittest.TestCase):
|
|
def test_all_three_legs_start_before_any_finishes(self) -> None:
|
|
started: list[str] = []
|
|
|
|
def runner(argv, **kwargs):
|
|
started.append(argv[3])
|
|
return FakeCompleted(0)
|
|
|
|
legs = MODULE.plan_legs(args(rehearse=True), Path("out"), admin_probe=lambda origin: ["/admin/voice-runtime"])
|
|
results = MODULE.run_window(legs, timeout=60, runner=runner)
|
|
self.assertEqual(len(started), 3)
|
|
self.assertTrue(all(result.ok for result in results))
|
|
|
|
def test_one_failing_leg_does_not_cancel_the_others(self) -> None:
|
|
def runner(argv, **kwargs):
|
|
return FakeCompleted(1 if "capture-g7-runtime-evidence.py" in argv[3] else 0)
|
|
|
|
legs = MODULE.plan_legs(args(rehearse=True), Path("out"), admin_probe=lambda origin: ["/admin/voice-runtime"])
|
|
results = MODULE.run_window(legs, timeout=60, runner=runner)
|
|
self.assertEqual(len(results), 3)
|
|
failed = [r.name for r in results if not r.ok]
|
|
self.assertEqual(failed, ["runtime"])
|
|
|
|
|
|
class AdminEndpointPreflightTest(unittest.TestCase):
|
|
"""공개 런타임이 구버전이라 endpoint 가 없으면 50분을 버리기 전에 막는다."""
|
|
|
|
ADMIN_URL = "https://api.example.test/admin/voice-runtime"
|
|
|
|
def test_missing_endpoint_is_rejected_before_the_window_starts(self) -> None:
|
|
with self.assertRaises(MODULE.WindowError) as ctx:
|
|
MODULE.assert_admin_endpoint_deployed(
|
|
self.ADMIN_URL, paths_probe=lambda origin: ["/health", "/voice/speech"]
|
|
)
|
|
self.assertEqual(str(ctx.exception), "admin_voice_runtime_not_deployed")
|
|
|
|
def test_deployed_endpoint_is_accepted(self) -> None:
|
|
origin = MODULE.assert_admin_endpoint_deployed(
|
|
self.ADMIN_URL,
|
|
paths_probe=lambda origin: ["/health", "/admin/voice-runtime"],
|
|
)
|
|
self.assertEqual(origin, "https://api.example.test")
|
|
|
|
def test_unauthenticated_status_is_never_used_as_the_signal(self) -> None:
|
|
"""Cloudflare 는 자동화 클라이언트에 모든 경로를 403 으로 준다.
|
|
|
|
상태 코드로 판정하면 배포 누락을 통과시킨다. spec 만 본다.
|
|
"""
|
|
|
|
probed: list[str] = []
|
|
|
|
def probe(origin: str):
|
|
probed.append(origin)
|
|
return ["/admin/voice-runtime"]
|
|
|
|
MODULE.assert_admin_endpoint_deployed(self.ADMIN_URL, paths_probe=probe)
|
|
self.assertEqual(probed, ["https://api.example.test"])
|
|
|
|
def test_wrong_admin_url_path_fails_closed(self) -> None:
|
|
with self.assertRaises(MODULE.WindowError) as ctx:
|
|
MODULE.assert_admin_endpoint_deployed(
|
|
"https://api.example.test/admin/something-else",
|
|
paths_probe=lambda origin: ["/admin/voice-runtime"],
|
|
)
|
|
self.assertEqual(str(ctx.exception), "admin_runtime_url_unexpected_path")
|
|
|
|
def test_plan_refuses_a_deployment_without_the_endpoint(self) -> None:
|
|
with self.assertRaises(MODULE.WindowError) as ctx:
|
|
MODULE.plan_legs(
|
|
args(rehearse=True), Path("out"), admin_probe=lambda origin: ["/health"]
|
|
)
|
|
self.assertEqual(str(ctx.exception), "admin_voice_runtime_not_deployed")
|
|
|
|
|
|
class ReportTest(unittest.TestCase):
|
|
def _results(self, codes=(0, 0, 0)):
|
|
names = ("voice_soak", "runtime", "topology")
|
|
return [
|
|
MODULE.LegResult(name, code, Path(f"{name}.json"))
|
|
for name, code in zip(names, codes)
|
|
]
|
|
|
|
def test_rehearse_never_reports_the_gate_as_closed(self) -> None:
|
|
report = MODULE.summarize(
|
|
self._results(), rehearse=True, checker_returncode=0
|
|
)
|
|
self.assertFalse(report["gate_closed"])
|
|
self.assertEqual(report["mode"], "rehearse")
|
|
|
|
def test_production_gate_closes_only_on_checker_exit_zero(self) -> None:
|
|
closed = MODULE.summarize(
|
|
self._results(), rehearse=False, checker_returncode=0
|
|
)
|
|
open_gate = MODULE.summarize(
|
|
self._results(), rehearse=False, checker_returncode=1
|
|
)
|
|
self.assertTrue(closed["gate_closed"])
|
|
self.assertFalse(open_gate["gate_closed"])
|
|
|
|
def test_production_process_exit_is_bound_to_checker_and_closed_gate(self) -> None:
|
|
closed = MODULE.summarize(
|
|
self._results(), rehearse=False, checker_returncode=0
|
|
)
|
|
checker_failed = MODULE.summarize(
|
|
self._results(), rehearse=False, checker_returncode=1
|
|
)
|
|
leg_failed = MODULE.summarize(
|
|
self._results((0, 1, 0)), rehearse=False, checker_returncode=0
|
|
)
|
|
self.assertEqual(0, MODULE.process_exit_code(closed))
|
|
self.assertEqual(1, MODULE.process_exit_code(checker_failed))
|
|
self.assertEqual(1, MODULE.process_exit_code(leg_failed))
|
|
|
|
def test_rehearse_exit_reports_leg_health_but_never_closes_gate(self) -> None:
|
|
passed = MODULE.summarize(
|
|
self._results(), rehearse=True, checker_returncode=None
|
|
)
|
|
failed = MODULE.summarize(
|
|
self._results((0, 1, 0)), rehearse=True, checker_returncode=None
|
|
)
|
|
self.assertEqual(0, MODULE.process_exit_code(passed))
|
|
self.assertEqual(1, MODULE.process_exit_code(failed))
|
|
self.assertFalse(passed["gate_closed"])
|
|
|
|
def test_failed_leg_is_reported(self) -> None:
|
|
report = MODULE.summarize(
|
|
self._results((0, 1, 0)), rehearse=False, checker_returncode=None
|
|
)
|
|
self.assertFalse(report["all_legs_passed"])
|
|
self.assertFalse(report["gate_closed"])
|
|
|
|
def test_checker_argv_passes_all_four_artifacts(self) -> None:
|
|
argv = MODULE.build_checker_argv(self._results(), Path("pack.json"))
|
|
for flag in ("--voice-soak", "--runtime", "--topology", "--human-voice-gain"):
|
|
self.assertIn(flag, argv)
|
|
self.assertEqual(argv[argv.index("--human-voice-gain") + 1], "pack.json")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|