from __future__ import annotations import unittest import shutil import subprocess import sys import tempfile from pathlib import Path SCRIPTS = Path(__file__).resolve().parent PUBLIC_RUNTIME = (SCRIPTS / "start-public-runtime.ps1").read_text(encoding="utf-8") WHISPER_START = (SCRIPTS / "start-local-whisper-stt.ps1").read_text( encoding="utf-8" ) class PublicRuntimeVoiceContractTest(unittest.TestCase): def test_public_provider_model_and_loopback_environment_are_explicit(self) -> None: for expected in ( '$WhisperModel = "small"', '$WhisperDevice = "cpu"', '$MeloTtsModel = "melotts-korean"', '$env:VIGNETTE_VOICE_STT_PROVIDER = "local_whisper"', '$env:VIGNETTE_LOCAL_WHISPER_STT_MODEL = $WhisperModel', '$env:VIGNETTE_VOICE_TTS_PROVIDER = "melotts"', '"ws://127.0.0.1:$WhisperPort/v1/listen"', '"http://127.0.0.1:$MeloTtsPort"', ): with self.subTest(expected=expected): self.assertIn(expected, PUBLIC_RUNTIME) def test_existing_listeners_require_exact_protocol_metadata(self) -> None: self.assertIn('Test-VoiceSidecarReady -Component "stt"', PUBLIC_RUNTIME) self.assertIn('Test-VoiceSidecarReady -Component "tts"', PUBLIC_RUNTIME) self.assertIn("if (Test-PortListener -Port $WhisperPort)", PUBLIC_RUNTIME) self.assertIn("if (Test-PortListener -Port $MeloTtsPort)", PUBLIC_RUNTIME) self.assertIn("does not expose the exact local_whisper", PUBLIC_RUNTIME) self.assertIn("does not expose the exact melotts", PUBLIC_RUNTIME) def test_sidecar_fail_closed_gate_precedes_api_mutation(self) -> None: sidecar_gate = PUBLIC_RUNTIME.index( '# 두 sidecar를 한 번 더 함께 검사해 개별 probe 사이의 TOCTOU를 닫는다.' ) api_stop = PUBLIC_RUNTIME.index("$apiStoppedProcessIds = @(") self.assertLess(sidecar_gate, api_stop) self.assertLess( PUBLIC_RUNTIME.index("Voice sidecar readiness changed"), api_stop, ) def test_api_skip_and_post_start_checks_require_exact_voice_health(self) -> None: for expected in ( '$Health.stt_provider -eq "local_whisper"', "$Health.stt_model -eq $WhisperModel", '$Health.tts_provider -eq "melotts"', "$Health.tts_model -eq $MeloTtsModel", "$Health.limits.uvicorn_ws_max_queue -eq 4", "$health.engine -eq $true", ): with self.subTest(expected=expected): self.assertIn(expected, PUBLIC_RUNTIME) self.assertGreaterEqual( PUBLIC_RUNTIME.count("Test-VoiceApiReady -Health"), 3, ) def test_api_uvicorn_websocket_queue_is_pinned(self) -> None: start = PUBLIC_RUNTIME.index("$apiStoppedProcessIds = @(") end = PUBLIC_RUNTIME.index("Start-Sleep -Seconds 3", start) api_section = PUBLIC_RUNTIME[start:end] self.assertIn('"app.main:app"', api_section) self.assertIn('"--ws", "websockets"', api_section) self.assertIn('"--ws-max-queue", "4"', api_section) def test_standalone_whisper_launcher_matches_public_cpu_default(self) -> None: self.assertIn("[string]$Model = 'small'", WHISPER_START) self.assertIn("[string]$Device = 'cpu'", WHISPER_START) class FreshPublicProvenanceContractTest(unittest.TestCase): def test_fresh_mode_is_explicit_and_fail_closed(self) -> None: for expected in ( "[switch]$RequireFreshPublicProvenance", 'throw "-RequireFreshPublicProvenance requires -ForceApiRestart"', 'throw "-RequireFreshPublicProvenance forbids -SkipCloudflaredRestart"', 'throw "-RequireFreshPublicProvenance requires -SkipEngineRestart;', 'throw "-RequireFreshPublicProvenance requires -SkipWebRestart;', 'throw "-RequireFreshPublicProvenance forbids DNS route mutation"', "Fresh public promotion requires the canonical HTTPS public health URL", "ExpectedSourceCommit", "ExpectedSourceTree", "ExpectedPythonSha256", "ExpectedCloudflaredSha256", "ExpectedCloudflaredConfigSha256", ): with self.subTest(expected=expected): self.assertIn(expected, PUBLIC_RUNTIME) def test_fresh_gate_precedes_every_runtime_mutation(self) -> None: gate = PUBLIC_RUNTIME.index("Assert-FreshPublicProvenanceContract `") receipt_preflight = PUBLIC_RUNTIME.index( "$resolvedRuntimeProvenancePath = Initialize-RuntimeProvenanceOutput" ) exact_ingress = PUBLIC_RUNTIME.index("-RequireUnchanged", gate) engine_probe = PUBLIC_RUNTIME.index("$engineHealth = Get-JsonHealth") self.assertLess(gate, receipt_preflight) self.assertLess(receipt_preflight, exact_ingress) self.assertLess(gate, exact_ingress) self.assertLess(exact_ingress, engine_probe) self.assertIn("requires detached HEAD", PUBLIC_RUNTIME) self.assertIn("requires a clean stable source", PUBLIC_RUNTIME) def test_receipt_preflight_proves_sibling_atomic_replace_capability(self) -> None: preflight = PUBLIC_RUNTIME[ PUBLIC_RUNTIME.index("function Initialize-RuntimeProvenanceOutput") : PUBLIC_RUNTIME.index("function Write-Utf8TextAtomically") ] for expected in ( "output preflight failed before runtime mutation", "[System.IO.FileMode]::Open", "[System.IO.FileAccess]::ReadWrite", "[System.IO.File]::Replace($probeSource, $probeTarget, $probeBackup)", "[System.IO.File]::Delete($probeTarget)", ): with self.subTest(expected=expected): self.assertIn(expected, preflight) def test_receipt_commit_is_atomic_and_failure_is_explicit(self) -> None: writer = PUBLIC_RUNTIME[ PUBLIC_RUNTIME.index("function Write-Utf8TextAtomically") : PUBLIC_RUNTIME.index("function Assert-FreshPublicProvenanceContract") ] for expected in ( "[System.IO.FileMode]::CreateNew", "$stream.Flush($true)", "[System.IO.File]::Replace($temporaryPath, $OutputPath, $backupPath)", "[System.IO.File]::Move($temporaryPath, $OutputPath)", "[System.IO.File]::Delete($temporaryPath)", ): with self.subTest(expected=expected): self.assertIn(expected, writer) receipt = PUBLIC_RUNTIME[PUBLIC_RUNTIME.index("$provenance = [ordered]@{") :] self.assertIn("Write-Utf8TextAtomically `", receipt) self.assertIn("no atomic passed receipt was published", receipt) self.assertNotIn( "[System.IO.File]::WriteAllText(\n $resolvedRuntimeProvenancePath", receipt, ) def test_receipt_helpers_work_in_windows_powershell(self) -> None: powershell = shutil.which("powershell.exe") if powershell is None: self.skipTest("Windows PowerShell 5.1 is not available") function_source = PUBLIC_RUNTIME[ PUBLIC_RUNTIME.index("function Initialize-RuntimeProvenanceOutput") : PUBLIC_RUNTIME.index("function Assert-FreshPublicProvenanceContract") ] with tempfile.TemporaryDirectory() as temporary_directory: root = Path(temporary_directory) functions_path = root / "receipt-functions.ps1" harness_path = root / "receipt-harness.ps1" functions_path.write_text(function_source, encoding="utf-8-sig") quoted_functions = str(functions_path).replace("'", "''") quoted_root = str(root).replace("'", "''") harness_path.write_text( f"""$ErrorActionPreference = 'Stop' . '{quoted_functions}' $target = Join-Path '{quoted_root}' 'nested\\receipt.json' $resolved = Initialize-RuntimeProvenanceOutput -OutputPath $target Write-Utf8TextAtomically -OutputPath $resolved -Value '{{"status":"passed"}}' $receipt = Get-Content -LiteralPath $resolved -Raw -Encoding UTF8 | ConvertFrom-Json if ($receipt.status -ne 'passed') {{ throw 'atomic receipt content mismatch' }} Write-Utf8TextAtomically -OutputPath $resolved -Value '{{"status":"replaced"}}' $replacement = Get-Content -LiteralPath $resolved -Raw -Encoding UTF8 | ConvertFrom-Json if ($replacement.status -ne 'replaced') {{ throw 'atomic replacement mismatch' }} $failedPath = Write-FailedFreshPromotionEvidence ` -OutputPath $resolved ` -FailureStage 'receipt_publish' ` -RollbackSucceeded $true ` -RollbackResult @{{local_health=$true;public_health=$true}} ` -SourceCommit ('a' * 40) ` -SourceTree ('b' * 40) $failed = Get-Content -LiteralPath $failedPath -Raw -Encoding UTF8 | ConvertFrom-Json if ((Split-Path -Leaf $failedPath) -notlike '*.failed.log') {{ throw 'failure evidence suffix mismatch' }} if ($failed.status -ne 'failed_rolled_back') {{ throw 'failure evidence status mismatch' }} if ($failed.failure_stage -ne 'receipt_publish') {{ throw 'failure evidence stage mismatch' }} if (-not $failed.rollback.succeeded) {{ throw 'failure evidence rollback mismatch' }} if (@(Get-ChildItem -LiteralPath (Split-Path -Parent $resolved) -Filter '*.tmp').Count -ne 0) {{ throw 'temporary receipt files were not cleaned' }} [System.IO.File]::SetAttributes($resolved, [System.IO.FileAttributes]::ReadOnly) $preflightFailed = $false try {{ Initialize-RuntimeProvenanceOutput -OutputPath $resolved | Out-Null }} catch {{ $preflightFailed = $true }} finally {{ [System.IO.File]::SetAttributes($resolved, [System.IO.FileAttributes]::Normal) }} if (-not $preflightFailed) {{ throw 'read-only receipt preflight did not fail' }} $preserved = Get-Content -LiteralPath $resolved -Raw -Encoding UTF8 | ConvertFrom-Json if ($preserved.status -ne 'replaced') {{ throw 'failed preflight changed the prior receipt' }} """, encoding="utf-8-sig", ) completed = subprocess.run( [ powershell, "-NoLogo", "-NoProfile", "-NonInteractive", "-ExecutionPolicy", "Bypass", "-File", str(harness_path), ], capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=30, check=False, ) self.assertEqual( completed.returncode, 0, msg=f"stdout={completed.stdout}\nstderr={completed.stderr}", ) def test_force_api_restart_keeps_regular_healthy_semantics(self) -> None: self.assertIn( "if ($apiControlPlaneReady -and -not $ForceApiRestart)", PUBLIC_RUNTIME, ) fresh_guard = PUBLIC_RUNTIME.index( 'throw "-RequireFreshPublicProvenance requires -ForceApiRestart"' ) api_skip = PUBLIC_RUNTIME.index( "if ($apiControlPlaneReady -and -not $ForceApiRestart)" ) self.assertLess(fresh_guard, api_skip) def test_api_and_tunnel_are_bounded_replacements_with_exact_cwds(self) -> None: api = PUBLIC_RUNTIME.index("$apiStoppedProcessIds = @(") cloud = PUBLIC_RUNTIME.index("$cloudflaredStoppedProcessIds = @(") receipt = PUBLIC_RUNTIME.index("$provenance = [ordered]@{") api_section = PUBLIC_RUNTIME[api:cloud] cloud_section = PUBLIC_RUNTIME[cloud:receipt] self.assertIn("Stop-UvicornByPort `", api_section) self.assertIn("-TimeoutSec $ProcessStopTimeoutSeconds", api_section) self.assertIn("-WorkingDirectory $ApiDir", api_section) self.assertIn("did not receive a replacement PID", api_section) self.assertIn("Stop-ProcessesBounded `", cloud_section) self.assertIn("-TimeoutSec $ProcessStopTimeoutSeconds", cloud_section) self.assertIn("-WorkingDirectory $Workspace", cloud_section) self.assertIn("Cloudflared did not receive a replacement PID", cloud_section) self.assertNotIn("Cloudflared already running; skipping", PUBLIC_RUNTIME) def test_tunnel_command_and_config_are_pinned_before_receipt(self) -> None: for expected in ( '"tunnel", "--config", $resolvedCloudflaredConfig, "run"', "command line is not pinned to the expected config", "Pinned cloudflared config drifted before provenance receipt", "finalConfigSha256", ): with self.subTest(expected=expected): self.assertIn(expected, PUBLIC_RUNTIME) def test_fresh_tunnel_stop_matches_only_the_resolved_full_config_path(self) -> None: cloud = PUBLIC_RUNTIME.index("$resolvedCloudflaredConfig =") start = PUBLIC_RUNTIME.index( "$cloudflaredProcess = Start-Process", cloud ) cloud_section = PUBLIC_RUNTIME[cloud:start] self.assertIn("if ($RequireFreshPublicProvenance)", cloud_section) self.assertIn("-ConfigPath $resolvedCloudflaredConfig `", cloud_section) self.assertIn("-ExactPath", cloud_section) matcher = PUBLIC_RUNTIME[ PUBLIC_RUNTIME.index("function Get-CloudflaredProcessesForConfig") : PUBLIC_RUNTIME.index("function Wait-ProcessIdentity") ] self.assertIn("-not $ExactPath", matcher) self.assertIn("IndexOf($ConfigPath", matcher) def test_receipt_projects_no_raw_command_or_config_contents(self) -> None: projection = PUBLIC_RUNTIME[ PUBLIC_RUNTIME.index("function ConvertTo-SafeProcessIdentity") : PUBLIC_RUNTIME.index("function Save-ManagedEnvironment") ] self.assertIn("command_line_sha256 =", projection) self.assertNotIn("command_line =", projection) self.assertNotIn("executable_path =", projection) receipt = PUBLIC_RUNTIME.index("$provenance = [ordered]@{") receipt_section = PUBLIC_RUNTIME[receipt:] self.assertIn("api = $safeApiIdentity", receipt_section) self.assertIn("cloudflared = $safeCloudflaredIdentity", receipt_section) self.assertNotIn("api = $apiFinalIdentity", receipt_section) self.assertNotIn("cloudflared = $cloudflaredFinalIdentity", receipt_section) config_projection = receipt_section[ receipt_section.index("config = [ordered]@{") : receipt_section.index("replacement = [ordered]@{") ].lower() self.assertNotIn("token", config_projection) self.assertNotIn("credential", config_projection) self.assertNotIn("contents", config_projection) def test_process_start_and_command_hash_match_topology_psutil_algorithm(self) -> None: identity = PUBLIC_RUNTIME[ PUBLIC_RUNTIME.index("function Wait-ProcessIdentity") : PUBLIC_RUNTIME.index("function ConvertTo-SafeProcessIdentity") ] self.assertIn("datetime.fromtimestamp(p.create_time(), UTC)", identity) self.assertIn("argv=p.cmdline()", identity) self.assertIn("chr(0).join(argv)", identity) self.assertIn("argument_list = $argumentList", identity) self.assertIn("environment = $processEnvironment", identity) self.assertIn("command_line_sha256 = $commandLineSha256", identity) def test_prior_process_argv_and_environment_round_trip_in_powershell_51(self) -> None: powershell = shutil.which("powershell.exe") if powershell is None: self.skipTest("Windows PowerShell 5.1 is not available") function_source = PUBLIC_RUNTIME[ PUBLIC_RUNTIME.index("function Save-CompleteProcessEnvironment") : PUBLIC_RUNTIME.index("function Restore-PriorPublicRuntime") ] with tempfile.TemporaryDirectory() as temporary_directory: root = Path(temporary_directory) functions_path = root / "rollback-process-functions.ps1" harness_path = root / "rollback-process-harness.ps1" functions_path.write_text(function_source, encoding="utf-8-sig") quoted_functions = str(functions_path).replace("'", "''") quoted_root = str(root).replace("'", "''") quoted_python = sys.executable.replace("'", "''") harness_path.write_text( f"""$ErrorActionPreference = 'Stop' . '{quoted_functions}' $root = '{quoted_root}' $python = '{quoted_python}' if ($null -eq (Get-Command Get-FileHash -ErrorAction SilentlyContinue)) {{ function Get-FileHash {{ param([string]$LiteralPath, [string]$Algorithm) $stream = [IO.File]::OpenRead($LiteralPath) try {{ $hasher = [Security.Cryptography.SHA256]::Create() try {{ $hash = $hasher.ComputeHash($stream) }} finally {{ $hasher.Dispose() }} }} finally {{ $stream.Dispose() }} [pscustomobject]@{{Hash = ([BitConverter]::ToString($hash)).Replace('-', '')}} }} }} $pythonSha256 = (Get-FileHash -LiteralPath $python -Algorithm SHA256).Hash.ToLowerInvariant() $originalEnvironment = Save-CompleteProcessEnvironment try {{ [Environment]::SetEnvironmentVariable('VIGNETTE_ROLLBACK_ARGV_TEST', 'caller', 'Process') [Environment]::SetEnvironmentVariable('VIGNETTE_CALLER_ONLY', 'caller-only', 'Process') $callerEnvironment = Save-CompleteProcessEnvironment [Environment]::SetEnvironmentVariable('VIGNETTE_ROLLBACK_ARGV_TEST', 'prior', 'Process') [Environment]::SetEnvironmentVariable('VIGNETTE_CALLER_ONLY', $null, 'Process') $priorEnvironment = Save-CompleteProcessEnvironment Set-CompleteProcessEnvironment -Environment $callerEnvironment $output = Join-Path $root 'argv-result.json' $code = "import json,os,sys; open(sys.argv[1], 'w', encoding='utf-8').write(json.dumps({{'argv':sys.argv[2:],'marker':os.environ.get('VIGNETTE_ROLLBACK_ARGV_TEST'),'leak':os.environ.get('VIGNETTE_CALLER_ONLY')}}, ensure_ascii=True))" $expected = @('plain', 'space value', 'quote"value', 'trailing\', '', 'slashes\\before"quote') $identity = [ordered]@{{ executable_path = $python executable_sha256 = $pythonSha256 cwd = $root argument_list = @('-X', 'utf8', '-c', $code, $output) + $expected environment = $priorEnvironment }} $process = Start-PinnedPriorProcess ` -Identity $identity ` -Role 'argv-roundtrip' ` -StdoutLog (Join-Path $root 'child.out.log') ` -StderrLog (Join-Path $root 'child.err.log') $process.WaitForExit() if (-not (Test-Path -LiteralPath $output -PathType Leaf)) {{ throw 'child did not write argv result' }} $actual = Get-Content -LiteralPath $output -Raw -Encoding UTF8 | ConvertFrom-Json if (@($actual.argv).Count -ne $expected.Count) {{ throw 'argument count mismatch' }} for ($i = 0; $i -lt $expected.Count; $i++) {{ if ($actual.argv[$i] -cne $expected[$i]) {{ throw "argument mismatch at $i" }} }} if ($actual.marker -cne 'prior') {{ throw 'prior environment was not inherited' }} if ($null -ne $actual.leak) {{ throw 'caller-only environment leaked into prior process' }} if ($env:VIGNETTE_ROLLBACK_ARGV_TEST -cne 'caller') {{ throw 'caller environment marker was not restored' }} if ($env:VIGNETTE_CALLER_ONLY -cne 'caller-only') {{ throw 'caller-only environment was not restored' }} }} finally {{ Set-CompleteProcessEnvironment -Environment $originalEnvironment }} """, encoding="utf-8-sig", ) completed = subprocess.run( [ powershell, "-NoLogo", "-NoProfile", "-NonInteractive", "-ExecutionPolicy", "Bypass", "-File", str(harness_path), ], capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=30, check=False, ) self.assertEqual( completed.returncode, 0, msg=f"stdout={completed.stdout}\nstderr={completed.stderr}", ) def test_rollback_restores_prior_openai_voice_contract_in_powershell_51(self) -> None: powershell = shutil.which("powershell.exe") if powershell is None: self.skipTest("Windows PowerShell 5.1 is not available") voice_contract_source = PUBLIC_RUNTIME[ PUBLIC_RUNTIME.index("function Test-PriorVoiceApiReady") : PUBLIC_RUNTIME.index("function Test-RequiredOpenApiPaths") ] rollback_source = PUBLIC_RUNTIME[ PUBLIC_RUNTIME.index("function Restore-PriorPublicRuntime") : PUBLIC_RUNTIME.index("function Stop-NodeByPortHint") ] with tempfile.TemporaryDirectory() as temporary_directory: root = Path(temporary_directory) functions_path = root / "rollback-contract-functions.ps1" harness_path = root / "rollback-contract-harness.ps1" functions_path.write_text( voice_contract_source + rollback_source, encoding="utf-8-sig", ) quoted_functions = str(functions_path).replace("'", "''") quoted_root = str(root).replace("'", "''") harness_path.write_text( f"""$ErrorActionPreference = 'Stop' . '{quoted_functions}' $events = New-Object System.Collections.ArrayList $root = '{quoted_root}' function Stop-UvicornByPort {{ param($AppImport,$Port,$TimeoutSec); $null=$events.Add('stop-api'); 11 }} function Restore-ManagedEnvironment {{ param($Snapshot); $null=$events.Add('restore-env') }} function Start-PinnedPriorProcess {{ param($Identity,$Role,$StdoutLog,$StderrLog) $null=$events.Add("start-$Role") if ($Role -eq 'api') {{ [pscustomobject]@{{Id=101}} }} else {{ [pscustomobject]@{{Id=202}} }} }} function Wait-ProcessIdentity {{ param($ProcessId,$Role,$ExpectedCwd,$TimeoutSec) $null=$events.Add("identity-$Role") [ordered]@{{ pid=[int]$ProcessId; started_at_utc='2026-08-09T00:00:00Z'; executable_name='runtime.exe'; executable_sha256=('a' * 64); command_line_sha256=('b' * 64); cwd=$root }} }} function Wait-JsonHealth {{ param($Uri,$IsHealthy,$TimeoutSec) $null=$events.Add("health-$Uri") if ($Uri -like '*/voice/health') {{ $payload=[pscustomobject]@{{ status='ok';available=$true;stt_available=$true;tts_available=$true; stt_provider='openai';stt_model='gpt-4o-mini-transcribe'; tts_provider='openai';tts_model='gpt-4o-mini-tts'; limits=[pscustomobject]@{{uvicorn_ws_max_queue=4}} }} }} else {{ $payload=[pscustomobject]@{{environment='prod';db=$true;engine=$true}} }} if (-not (& $IsHealthy $payload)) {{ throw "health predicate failed: $Uri" }} $payload }} function Test-VoiceSidecarReady {{ param($Component); $true }} function Get-CloudflaredProcessesForConfig {{ param($ConfigPath,[switch]$ExactPath); @([pscustomobject]@{{ProcessId=303}}) }} function Stop-ProcessesBounded {{ param($Processes,$TimeoutSec,$Role); $null=$events.Add('stop-cloudflared'); 303 }} function ConvertTo-SafeProcessIdentity {{ param($Identity); $Identity }} $configPath = Join-Path $root 'cloudflared.yml' [IO.File]::WriteAllText($configPath, 'tunnel: test') $priorVoice = [ordered]@{{ status='ok';available=$true;stt_available=$true;tts_available=$true; stt_provider='openai';stt_model='gpt-4o-mini-transcribe'; tts_provider='openai';tts_model='gpt-4o-mini-tts';uvicorn_ws_max_queue=4 }} $priorApi=[ordered]@{{executable_sha256=('a' * 64);command_line_sha256=('b' * 64);cwd=$root}} $priorCloud=[ordered]@{{executable_sha256=('a' * 64);command_line_sha256=('b' * 64);cwd=$root}} $result = Restore-PriorPublicRuntime ` -PriorApi $priorApi ` -PriorCloudflared $priorCloud ` -PriorLocalVoiceContract $priorVoice ` -PriorPublicVoiceContract $priorVoice ` -EnvironmentSnapshot @{{}} ` -ConfigPath $configPath ` -ApiPortValue 8001 ` -HealthUrl 'https://api-vignette.chanpaca.net/health' ` -VoiceHealthUrl 'https://api-vignette.chanpaca.net/voice/health' ` -TimeoutSec 5 foreach ($field in @('local_health','local_voice_health','public_health','public_voice_health')) {{ if (-not $result[$field]) {{ throw "rollback result missing $field" }} }} if ($events.IndexOf('stop-api') -ge $events.IndexOf('start-api')) {{ throw 'API restore ordering mismatch' }} if ($events.IndexOf('stop-cloudflared') -ge $events.IndexOf('start-cloudflared')) {{ throw 'cloudflared restore ordering mismatch' }} if ($events.IndexOf('start-api') -ge $events.IndexOf('health-http://127.0.0.1:8001/health')) {{ throw 'local health ordering mismatch' }} if ($events.IndexOf('start-cloudflared') -ge $events.IndexOf('health-https://api-vignette.chanpaca.net/health')) {{ throw 'public health ordering mismatch' }} $changed=[pscustomobject]@{{ status='ok';available=$true;stt_available=$true;tts_available=$true; stt_provider='openai';stt_model='changed-model';tts_provider='openai';tts_model='gpt-4o-mini-tts'; limits=[pscustomobject]@{{uvicorn_ws_max_queue=4}} }} if (Test-VoiceHealthContract -Health $changed -Expected $priorVoice) {{ throw 'voice contract drift was accepted' }} """, encoding="utf-8-sig", ) completed = subprocess.run( [ powershell, "-NoLogo", "-NoProfile", "-NonInteractive", "-ExecutionPolicy", "Bypass", "-File", str(harness_path), ], capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=30, check=False, ) self.assertEqual( completed.returncode, 0, msg=f"stdout={completed.stdout}\nstderr={completed.stderr}", ) def test_engine_log_rotation_cannot_dirty_the_stable_release(self) -> None: self.assertIn('Destination "$logFile.$rotateStamp.bak.log"', PUBLIC_RUNTIME) self.assertNotIn('Destination "$logFile.$rotateStamp.bak"', PUBLIC_RUNTIME) def test_failure_evidence_keeps_a_clean_detached_git_source(self) -> None: self.assertIn('$failedPath = "$OutputPath.failed.log"', PUBLIC_RUNTIME) self.assertNotIn('$failedPath = "$OutputPath.failed.json"', PUBLIC_RUNTIME) with tempfile.TemporaryDirectory() as temporary_directory: root = Path(temporary_directory) (root / ".gitignore").write_text("*.log\n", encoding="utf-8") (root / "tracked.txt").write_text("release\n", encoding="utf-8") commands = ( ["git", "init", "--quiet"], ["git", "config", "user.name", "Yun Chan"], ["git", "config", "user.email", "yun.chan@example.invalid"], ["git", "add", ".gitignore", "tracked.txt"], ["git", "commit", "--quiet", "-m", "테스트 기준"], ["git", "checkout", "--quiet", "--detach", "HEAD"], ) for command in commands: subprocess.run( command, cwd=root, capture_output=True, check=True, ) failure_evidence = root / "public-runtime-launch-provenance.log.failed.log" failure_evidence.write_text('{"status":"failed_rolled_back"}\n', encoding="utf-8") ignored = subprocess.run( ["git", "check-ignore", "--quiet", failure_evidence.name], cwd=root, check=False, ) self.assertEqual(ignored.returncode, 0) status = subprocess.run( ["git", "status", "--porcelain", "--untracked-files=normal"], cwd=root, capture_output=True, check=True, ) self.assertEqual(status.stdout, b"") def test_fresh_cutover_requires_and_restores_a_pinned_prior_runtime(self) -> None: prior_capture = PUBLIC_RUNTIME.index("$priorApiProcesses = @(") mutation = PUBLIC_RUNTIME.index('$freshFailureStage = "api_cutover"') self.assertLess(prior_capture, mutation) for expected in ( "requires exactly one prior API process for transactional rollback", "requires exactly one prior cloudflared process for transactional rollback", "$freshPriorApiIdentity = Wait-ProcessIdentity", "$freshPriorCloudflaredIdentity = Wait-ProcessIdentity", "$freshEnvironmentSnapshot = Save-ManagedEnvironment", "Restore-PriorPublicRuntime `", "Start-PinnedPriorProcess `", "Restored prior API identity drift", "Restored prior cloudflared identity drift", "$freshPromotionCommitted = $true", ): with self.subTest(expected=expected): self.assertIn(expected, PUBLIC_RUNTIME) def test_fresh_transaction_has_no_engine_web_sidecar_or_dns_mutation(self) -> None: contract = PUBLIC_RUNTIME[ PUBLIC_RUNTIME.index("function Assert-FreshPublicProvenanceContract") : PUBLIC_RUNTIME.index("$freshMutationStarted = $false") ] for expected in ( "if (-not $SkipEngineRestart)", "if (-not $SkipWebRestart)", "if ($RouteCloudflareDns)", "$CanonicalPublicHealthUrl", ): with self.subTest(expected=expected): self.assertIn(expected, contract) api_mutation = PUBLIC_RUNTIME.index('$freshFailureStage = "api_cutover"') preflight = PUBLIC_RUNTIME[:api_mutation] self.assertIn("requires exact healthy voice sidecars as an unchanged precondition", preflight) self.assertIn("-Uri $CanonicalPublicHealthUrl", preflight) self.assertIn("-Uri $CanonicalPublicVoiceHealthUrl", preflight) self.assertIn("$health.engine -eq $true", preflight) stt_start = PUBLIC_RUNTIME.index("& $WhisperStartScript `") tts_start = PUBLIC_RUNTIME.index("& $MeloTtsStartScript `") self.assertLess( PUBLIC_RUNTIME.index("Fresh public promotion will not mutate local_whisper"), stt_start, ) self.assertLess( PUBLIC_RUNTIME.index("Fresh public promotion will not mutate MeloTTS"), tts_start, ) def test_success_and_rollback_require_canonical_public_engine_and_voice(self) -> None: rollback = PUBLIC_RUNTIME[ PUBLIC_RUNTIME.index("function Restore-PriorPublicRuntime") : PUBLIC_RUNTIME.index("function Stop-NodeByPortHint") ] self.assertGreaterEqual(rollback.count("$health.engine -eq $true"), 2) self.assertIn("Test-VoiceSidecarReady", rollback) self.assertIn("Test-VoiceHealthContract", rollback) success_gate = PUBLIC_RUNTIME.index('$freshFailureStage = "public_health_validation"') receipt = PUBLIC_RUNTIME.index("$provenance = [ordered]@{") committed = PUBLIC_RUNTIME.index("$freshPromotionCommitted = $true") self.assertLess(success_gate, receipt) self.assertLess(receipt, committed) success_section = PUBLIC_RUNTIME[success_gate:receipt] self.assertIn("-Uri $CanonicalPublicHealthUrl", success_section) self.assertIn("-Uri $CanonicalPublicVoiceHealthUrl", success_section) self.assertIn("-Uri $CanonicalPublicOpenApiUrl", success_section) self.assertIn("Test-RequiredOpenApiPaths", success_section) self.assertIn('"/admin/voice-runtime"', PUBLIC_RUNTIME) self.assertIn("$health.engine -eq $true", success_section) self.assertIn("Test-VoiceSidecarReady", success_section) def test_failed_cutover_emits_metadata_only_rollback_evidence(self) -> None: failure_writer = PUBLIC_RUNTIME[ PUBLIC_RUNTIME.index("function Write-FailedFreshPromotionEvidence") : PUBLIC_RUNTIME.index("function Assert-FreshPublicProvenanceContract") ] for expected in ( 'schema_version = "vignette.public-runtime-launch-failure.v1"', '"failed_rolled_back"', '"failed_rollback"', "failure_stage = $FailureStage", "attempted = $true", "succeeded = $RollbackSucceeded", "Write-Utf8TextAtomically", ): with self.subTest(expected=expected): self.assertIn(expected, failure_writer) self.assertNotIn("command_line", failure_writer) self.assertNotIn("executable_path", failure_writer) def test_receipt_is_direct_input_for_windows_topology_capture(self) -> None: receipt = PUBLIC_RUNTIME.index( 'schema_version = "vignette.public-runtime-launch-provenance.v1"' ) receipt_section = PUBLIC_RUNTIME[receipt:] for expected in ( "git_commit = $ExpectedSourceCommit.ToLowerInvariant()", "git_tree = $ExpectedSourceTree.ToLowerInvariant()", "api_stopped_pids", "cloudflared_stopped_pids", "api_new_pid", "cloudflared_new_pid", "started_at_utc", "executable_sha256", "command_line_sha256", "api_pid", "api_executable_name", "api_cwd", "cloudflared_pid", "cloudflared_executable_name", "cloudflared_cwd", "psutil_version", ): with self.subTest(expected=expected): self.assertIn(expected, receipt_section) if __name__ == "__main__": unittest.main()