751 lines
33 KiB
Python
751 lines
33 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
|
|
SCRIPTS = Path(__file__).resolve().parent
|
|
WATCHDOG = SCRIPTS / "watch-public-runtime.ps1"
|
|
INSTALLER = SCRIPTS / "install-public-runtime-task.ps1"
|
|
BOOT = SCRIPTS / "boot-public-runtime.ps1"
|
|
BOOT_REGISTER = SCRIPTS / "register-boot-task.ps1"
|
|
HIDDEN_TRIGGER = SCRIPTS / "watch-public-runtime-hidden.vbs"
|
|
TASK_LAUNCHER = SCRIPTS / "watch-public-runtime-task.vbs"
|
|
START = SCRIPTS / "start-public-runtime.ps1"
|
|
VOICE_PROBE = SCRIPTS / "probe-public-voice-sidecars.py"
|
|
REPO_ROOT = SCRIPTS.parent
|
|
RUNBOOK = REPO_ROOT / "docs" / "ops" / "public-runtime-watchdog.md"
|
|
LOCAL_DEVELOPMENT = REPO_ROOT / "docs" / "guides" / "local-development.md"
|
|
WATCHDOG_SOURCE = WATCHDOG.read_text(encoding="utf-8")
|
|
INSTALLER_SOURCE = INSTALLER.read_text(encoding="utf-8")
|
|
BOOT_SOURCE = BOOT.read_text(encoding="utf-8")
|
|
BOOT_REGISTER_SOURCE = BOOT_REGISTER.read_text(encoding="utf-8")
|
|
HIDDEN_TRIGGER_SOURCE = HIDDEN_TRIGGER.read_text(encoding="utf-8")
|
|
START_SOURCE = START.read_text(encoding="utf-8")
|
|
RUNBOOK_SOURCE = RUNBOOK.read_text(encoding="utf-8")
|
|
LOCAL_DEVELOPMENT_SOURCE = LOCAL_DEVELOPMENT.read_text(encoding="utf-8")
|
|
|
|
|
|
def sha256(path: Path) -> str:
|
|
return hashlib.sha256(path.read_bytes()).hexdigest()
|
|
|
|
|
|
class PublicRuntimeWatchdogProvenanceTest(unittest.TestCase):
|
|
def test_watchdog_requires_exact_source_and_script_pins(self) -> None:
|
|
for expected in (
|
|
"[string]$StableSourceRoot",
|
|
"[string]$ExpectedSourceCommit",
|
|
"[string]$ExpectedSourceTree",
|
|
"[string]$ExpectedWatchdogSha256",
|
|
"[string]$ExpectedStartScriptSha256",
|
|
'symbolic-ref --quiet HEAD',
|
|
'"status", "--porcelain=v1", "--untracked-files=normal"',
|
|
'"ls-files", "--error-unmatch"',
|
|
):
|
|
with self.subTest(expected=expected):
|
|
self.assertIn(expected, WATCHDOG_SOURCE)
|
|
self.assertGreaterEqual(WATCHDOG_SOURCE.count("[Parameter(Mandatory = $true)]"), 5)
|
|
|
|
def test_provenance_gate_precedes_health_and_runtime_recovery(self) -> None:
|
|
gate_comment = WATCHDOG_SOURCE.index(
|
|
"# health probe, failcount 기록, 프로세스 재기동보다 먼저"
|
|
)
|
|
gate = WATCHDOG_SOURCE.index("Assert-StableSourceProvenance", gate_comment)
|
|
health = WATCHDOG_SOURCE.index("$checks = @(")
|
|
restart = WATCHDOG_SOURCE.index("& $startScript @startArgs")
|
|
failcount_write = WATCHDOG_SOURCE.index("Set-FailCount 0", health)
|
|
self.assertLess(gate, health)
|
|
self.assertLess(gate, restart)
|
|
self.assertLess(gate, failcount_write)
|
|
|
|
def test_installer_pins_a_detached_clean_release_in_task_action(self) -> None:
|
|
for expected in (
|
|
"[Parameter(Mandatory = $true)]",
|
|
"[string]$StableSourceRoot",
|
|
"symbolic-ref --quiet HEAD",
|
|
'"status", "--porcelain=v1", "--untracked-files=normal"',
|
|
'"-ExpectedSourceCommit $sourceCommit"',
|
|
'"-ExpectedSourceTree $sourceTree"',
|
|
'"-ExpectedWatchdogSha256 $watchdogSha256"',
|
|
'"-ExpectedStartScriptSha256 $startScriptSha256"',
|
|
"-WorkingDirectory $resolvedSourceRoot",
|
|
"Watchdog installer is not executing from the pinned stable source root",
|
|
):
|
|
with self.subTest(expected=expected):
|
|
self.assertIn(expected, INSTALLER_SOURCE)
|
|
dirty_gate = INSTALLER_SOURCE.index("Stable source is not clean")
|
|
task_registration = INSTALLER_SOURCE.index("Register-ScheduledTask")
|
|
self.assertLess(dirty_gate, task_registration)
|
|
|
|
def test_boot_recovery_uses_the_same_gate_before_docker_or_process_mutation(self) -> None:
|
|
for expected in (
|
|
"[string]$StableSourceRoot",
|
|
"[string]$ExpectedSourceCommit",
|
|
"[string]$ExpectedSourceTree",
|
|
"[string]$ExpectedBootScriptSha256",
|
|
"[string]$ExpectedStartScriptSha256",
|
|
"symbolic-ref --quiet HEAD",
|
|
'"status", "--porcelain=v1", "--untracked-files=normal"',
|
|
):
|
|
with self.subTest(expected=expected):
|
|
self.assertIn(expected, BOOT_SOURCE)
|
|
gate_comment = BOOT_SOURCE.index(
|
|
"# Docker/DB/process mutation보다 먼저 stable source를 매 실행 재검증한다."
|
|
)
|
|
gate = BOOT_SOURCE.index("Assert-StableSourceProvenance", gate_comment)
|
|
docker_mutation = BOOT_SOURCE.index("docker.exe update")
|
|
runtime_recovery = BOOT_SOURCE.index("& $startScript @startArgs |")
|
|
self.assertLess(gate, docker_mutation)
|
|
self.assertLess(gate, runtime_recovery)
|
|
|
|
def test_boot_task_registration_pins_the_same_detached_release(self) -> None:
|
|
for expected in (
|
|
"[string]$StableSourceRoot",
|
|
"symbolic-ref --quiet HEAD",
|
|
'"status", "--porcelain=v1", "--untracked-files=normal"',
|
|
'"-ExpectedSourceCommit $sourceCommit"',
|
|
'"-ExpectedSourceTree $sourceTree"',
|
|
'"-ExpectedBootScriptSha256 $bootScriptSha256"',
|
|
'"-ExpectedStartScriptSha256 $startScriptSha256"',
|
|
"-WorkingDirectory $resolvedSourceRoot",
|
|
"Boot task registrar is not executing from the pinned stable source root",
|
|
):
|
|
with self.subTest(expected=expected):
|
|
self.assertIn(expected, BOOT_REGISTER_SOURCE)
|
|
dirty_gate = BOOT_REGISTER_SOURCE.index("Stable source is not clean")
|
|
task_registration = BOOT_REGISTER_SOURCE.index("Register-ScheduledTask")
|
|
self.assertLess(dirty_gate, task_registration)
|
|
|
|
def test_task_limits_cover_a_full_cold_start_without_overlap(self) -> None:
|
|
self.assertIn("-ExecutionTimeLimit (New-TimeSpan -Minutes 60)", BOOT_REGISTER_SOURCE)
|
|
self.assertIn("-MultipleInstances IgnoreNew", BOOT_REGISTER_SOURCE)
|
|
self.assertIn("-ExecutionTimeLimit (New-TimeSpan -Minutes 60)", INSTALLER_SOURCE)
|
|
self.assertIn("-MultipleInstances IgnoreNew", INSTALLER_SOURCE)
|
|
self.assertNotIn("-RestartCount", INSTALLER_SOURCE)
|
|
self.assertNotIn("-RestartInterval", INSTALLER_SOURCE)
|
|
|
|
# 최초 health 진단 + DB 복구 + engine/voice/API/web의 각 bounded wait와
|
|
# process-stop 여유를 모두 합쳐도 작업 스케줄러 60분보다 5분 이상 짧다.
|
|
initial_health_seconds = 210
|
|
recovery_seconds = 60 + 90 + (300 * 2) + 180 + 90 + 600 + 30 + (15 * 4)
|
|
safety_margin_seconds = 300
|
|
task_limit_seconds = 60 * 60
|
|
self.assertLess(
|
|
initial_health_seconds + recovery_seconds + safety_margin_seconds,
|
|
task_limit_seconds,
|
|
)
|
|
boot_prerequisite_seconds = 360 + 90
|
|
self.assertLess(
|
|
boot_prerequisite_seconds + recovery_seconds + safety_margin_seconds,
|
|
task_limit_seconds,
|
|
)
|
|
|
|
def test_hard_down_bypasses_transient_failure_debounce(self) -> None:
|
|
hard_down = WATCHDOG_SOURCE.index("$hardDown =")
|
|
debounce = WATCHDOG_SOURCE.index(
|
|
"if ($failCount -lt $FailuresBeforeRestart -and -not $hardDown)"
|
|
)
|
|
restart = WATCHDOG_SOURCE.index("& $startScript @startArgs")
|
|
self.assertLess(hard_down, debounce)
|
|
self.assertLess(debounce, restart)
|
|
self.assertIn('($FailedNames -contains "cloudflared")', WATCHDOG_SOURCE)
|
|
self.assertIn('($FailedNames -contains "api")', WATCHDOG_SOURCE)
|
|
self.assertIn('($FailedNames -contains "web-preview")', WATCHDOG_SOURCE)
|
|
self.assertIn('($FailedNames -contains "voice-sidecars")', WATCHDOG_SOURCE)
|
|
self.assertIn(
|
|
"immediate restart: public runtime hard-down detected",
|
|
WATCHDOG_SOURCE,
|
|
)
|
|
|
|
def test_hard_down_classifier_distinguishes_total_and_transient_failures(self) -> None:
|
|
powershell = shutil.which("powershell.exe")
|
|
if powershell is None:
|
|
self.skipTest("Windows PowerShell 5.1 is not available")
|
|
|
|
classifier = WATCHDOG_SOURCE[
|
|
WATCHDOG_SOURCE.index("function Test-PublicRuntimeHardDown") :
|
|
WATCHDOG_SOURCE.index("# engine 판정은", WATCHDOG_SOURCE.index("function Test-PublicRuntimeHardDown"))
|
|
].strip()
|
|
with tempfile.TemporaryDirectory() as temporary_directory:
|
|
harness = Path(temporary_directory) / "hard-down.ps1"
|
|
harness.write_text(
|
|
classifier
|
|
+ r'''
|
|
if (-not (Test-PublicRuntimeHardDown -FailedNames @('cloudflared'))) { exit 2 }
|
|
if (-not (Test-PublicRuntimeHardDown -FailedNames @('api','web-preview','voice-sidecars'))) { exit 3 }
|
|
if (Test-PublicRuntimeHardDown -FailedNames @('engine')) { exit 4 }
|
|
exit 0
|
|
''',
|
|
encoding="utf-8-sig",
|
|
)
|
|
completed = subprocess.run(
|
|
[
|
|
powershell,
|
|
"-NoLogo",
|
|
"-NoProfile",
|
|
"-NonInteractive",
|
|
"-ExecutionPolicy",
|
|
"Bypass",
|
|
"-File",
|
|
str(harness),
|
|
],
|
|
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_check_only_exits_before_failcount_log_or_recovery_mutation(self) -> None:
|
|
failed = WATCHDOG_SOURCE.index("$failed = @(")
|
|
check_only = WATCHDOG_SOURCE.index("if ($CheckOnly)", failed)
|
|
inner_health = WATCHDOG_SOURCE.index("if ($failed.Count -eq 0)", check_only)
|
|
normal_health = WATCHDOG_SOURCE.index(
|
|
"if ($failed.Count -eq 0)", inner_health + 1
|
|
)
|
|
fail_count = WATCHDOG_SOURCE.index("$failCount =", check_only)
|
|
restart = WATCHDOG_SOURCE.index("& $startScript @startArgs", fail_count)
|
|
self.assertLess(check_only, fail_count)
|
|
self.assertLess(check_only, restart)
|
|
check_only_contract = WATCHDOG_SOURCE[check_only:normal_health]
|
|
self.assertIn("Write-Output", check_only_contract)
|
|
self.assertNotIn("Set-FailCount", check_only_contract)
|
|
self.assertNotIn("Write-WatchdogLog", check_only_contract)
|
|
self.assertNotIn("Restore-DatabaseContainer", check_only_contract)
|
|
|
|
def test_boot_streams_start_script_progress_without_nested_powershell_buffering(self) -> None:
|
|
self.assertIn("& $startScript @startArgs |", BOOT_SOURCE)
|
|
self.assertIn('Write-BootLog (" pub> " + $_)', BOOT_SOURCE)
|
|
self.assertIn("start-public-runtime.ps1 failed", BOOT_SOURCE)
|
|
self.assertNotIn("$out = & powershell.exe", BOOT_SOURCE)
|
|
|
|
def test_watchdog_and_boot_probe_exact_local_voice_stack(self) -> None:
|
|
for source in (WATCHDOG_SOURCE, BOOT_SOURCE):
|
|
for expected in (
|
|
'"--component", "all"',
|
|
'"--stt-provider", "local_whisper"',
|
|
'"--stt-model", "small"',
|
|
'"--stt-device", "cpu"',
|
|
'"--tts-provider", "melotts"',
|
|
'"--tts-model", "melotts-korean"',
|
|
'"scripts\\probe-public-voice-sidecars.py"',
|
|
"Test-VoiceSidecarStack",
|
|
):
|
|
with self.subTest(source=source[:32], expected=expected):
|
|
self.assertIn(expected, source)
|
|
self.assertIn("(Test-VoiceSidecarStack)", WATCHDOG_SOURCE)
|
|
self.assertIn("$voiceSidecarsAfter = Test-VoiceSidecarStack", WATCHDOG_SOURCE)
|
|
self.assertIn('-Name "voice-api"', WATCHDOG_SOURCE)
|
|
self.assertIn("$voiceApiAfter = Test-JsonHealth", WATCHDOG_SOURCE)
|
|
self.assertIn(
|
|
"(Test-ApiControlPlaneHealthy) -and (Test-EngineHealthy) -and (Test-VoiceApiHealthy) -and (Test-VoiceSidecarStack)",
|
|
BOOT_SOURCE,
|
|
)
|
|
self.assertGreaterEqual(BOOT_SOURCE.count("Test-VoiceApiHealthy"), 3)
|
|
self.assertIn('"-WhisperPort", $WhisperPort', BOOT_SOURCE)
|
|
self.assertIn('"-MeloTtsPort", $MeloTtsPort', BOOT_SOURCE)
|
|
# web preview는 살아 있을 때만 스킵한다. 무조건 -SkipWebRestart면 재부팅 후
|
|
# boot 경로로 web이 복구되지 않는다.
|
|
self.assertIn("$webHealthy = Test-WebPreviewHealthy", BOOT_SOURCE)
|
|
self.assertIn('if ($webHealthy) {\n $startArgs += "-SkipWebRestart"', BOOT_SOURCE)
|
|
self.assertIn("scripts/probe-public-voice-sidecars.py", INSTALLER_SOURCE)
|
|
self.assertIn("scripts/probe-public-voice-sidecars.py", BOOT_REGISTER_SOURCE)
|
|
|
|
def test_old_openai_api_is_not_healthy_when_exact_sidecars_are_ready(self) -> None:
|
|
powershell = shutil.which("powershell.exe")
|
|
if powershell is None:
|
|
self.skipTest("Windows PowerShell 5.1 is not available")
|
|
|
|
watchdog_contract = WATCHDOG_SOURCE[
|
|
WATCHDOG_SOURCE.index("function Test-VoiceApiReady") :
|
|
WATCHDOG_SOURCE.index("function Test-VoiceSidecarStack")
|
|
].strip()
|
|
boot_contract = BOOT_SOURCE[
|
|
BOOT_SOURCE.index("function Test-VoiceApiReady") :
|
|
BOOT_SOURCE.index("function Test-VoiceApiHealthy")
|
|
].strip()
|
|
self.assertEqual(watchdog_contract, boot_contract)
|
|
|
|
with tempfile.TemporaryDirectory() as temporary_directory:
|
|
harness = Path(temporary_directory) / "voice-api-contract.ps1"
|
|
harness.write_text(
|
|
watchdog_contract
|
|
+ r'''
|
|
$openAi = [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}
|
|
}
|
|
if (Test-VoiceApiReady -Health $openAi) { throw 'old OpenAI API was accepted' }
|
|
$local = [pscustomobject]@{
|
|
status='ok';available=$true;stt_available=$true;tts_available=$true
|
|
stt_provider='local_whisper';stt_model='small'
|
|
tts_provider='melotts';tts_model='melotts-korean'
|
|
limits=[pscustomobject]@{uvicorn_ws_max_queue=4}
|
|
}
|
|
if (-not (Test-VoiceApiReady -Health $local)) { throw 'exact local API was rejected' }
|
|
$local.limits.uvicorn_ws_max_queue = 5
|
|
if (Test-VoiceApiReady -Health $local) { throw 'wrong websocket queue was accepted' }
|
|
''',
|
|
encoding="utf-8-sig",
|
|
)
|
|
completed = subprocess.run(
|
|
[
|
|
powershell,
|
|
"-NoLogo",
|
|
"-NoProfile",
|
|
"-NonInteractive",
|
|
"-ExecutionPolicy",
|
|
"Bypass",
|
|
"-File",
|
|
str(harness),
|
|
],
|
|
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_failed_native_sidecar_probe_returns_unhealthy_under_powershell_51(self) -> None:
|
|
powershell = shutil.which("powershell.exe")
|
|
python = shutil.which("python.exe")
|
|
if powershell is None or python is None:
|
|
self.skipTest("Windows PowerShell 5.1 and Python are required")
|
|
|
|
watchdog_contract = WATCHDOG_SOURCE[
|
|
WATCHDOG_SOURCE.index("function Test-VoiceSidecarStack") :
|
|
WATCHDOG_SOURCE.index("function Test-CloudflaredProcess")
|
|
].strip()
|
|
|
|
with tempfile.TemporaryDirectory() as temporary_directory:
|
|
harness = Path(temporary_directory) / "voice-sidecar-failure-contract.ps1"
|
|
python_escaped = python.replace("'", "''")
|
|
probe_escaped = str(VOICE_PROBE).replace("'", "''")
|
|
harness.write_text(
|
|
"$ErrorActionPreference = 'Stop'\n"
|
|
+ f"$Python = '{python_escaped}'\n"
|
|
+ f"$voiceSidecarProbe = '{probe_escaped}'\n"
|
|
+ "$WhisperPort = 1\n"
|
|
+ "$MeloTtsPort = 2\n"
|
|
+ watchdog_contract
|
|
+ r'''
|
|
$result = Test-VoiceSidecarStack
|
|
if ($result.Ok) { throw 'unreachable sidecars were accepted' }
|
|
if ($result.Detail -ne 'exact readiness probe failed') {
|
|
throw "unexpected detail: $($result.Detail)"
|
|
}
|
|
''',
|
|
encoding="utf-8-sig",
|
|
)
|
|
completed = subprocess.run(
|
|
[
|
|
powershell,
|
|
"-NoLogo",
|
|
"-NoProfile",
|
|
"-NonInteractive",
|
|
"-ExecutionPolicy",
|
|
"Bypass",
|
|
"-File",
|
|
str(harness),
|
|
],
|
|
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_boolean_sidecar_probes_survive_native_stderr_under_powershell_51(self) -> None:
|
|
powershell = shutil.which("powershell.exe")
|
|
python = shutil.which("python.exe")
|
|
if powershell is None or python is None:
|
|
self.skipTest("Windows PowerShell 5.1 and Python are required")
|
|
|
|
contracts = (
|
|
(
|
|
START_SOURCE[
|
|
START_SOURCE.index("function Test-VoiceSidecarReady") :
|
|
START_SOURCE.index("function Wait-VoiceSidecarReady")
|
|
].strip(),
|
|
"$result = Test-VoiceSidecarReady -Component 'stt'",
|
|
"start",
|
|
),
|
|
(
|
|
BOOT_SOURCE[
|
|
BOOT_SOURCE.index("function Test-VoiceSidecarStack") :
|
|
BOOT_SOURCE.index('Write-BootLog "================ boot start')
|
|
].strip(),
|
|
"$result = Test-VoiceSidecarStack",
|
|
"boot",
|
|
),
|
|
)
|
|
python_escaped = python.replace("'", "''")
|
|
probe_escaped = str(VOICE_PROBE).replace("'", "''")
|
|
|
|
with tempfile.TemporaryDirectory() as temporary_directory:
|
|
for contract, invocation, role in contracts:
|
|
harness = Path(temporary_directory) / f"{role}-probe-contract.ps1"
|
|
harness.write_text(
|
|
"$ErrorActionPreference = 'Stop'\n"
|
|
+ f"$Python = '{python_escaped}'\n"
|
|
+ f"$VoiceSidecarProbe = '{probe_escaped}'\n"
|
|
+ f"$voiceSidecarProbe = '{probe_escaped}'\n"
|
|
+ "$WhisperPort = 1\n"
|
|
+ "$MeloTtsPort = 2\n"
|
|
+ "$WhisperModel = 'small'\n"
|
|
+ "$WhisperLanguage = 'ko'\n"
|
|
+ "$WhisperDevice = 'cpu'\n"
|
|
+ "$MeloTtsModel = 'melotts-korean'\n"
|
|
+ "$MeloTtsLanguage = 'KR'\n"
|
|
+ contract
|
|
+ "\n"
|
|
+ invocation
|
|
+ "\nif ($result) { throw 'unreachable sidecars were accepted' }\n",
|
|
encoding="utf-8-sig",
|
|
)
|
|
completed = subprocess.run(
|
|
[
|
|
powershell,
|
|
"-NoLogo",
|
|
"-NoProfile",
|
|
"-NonInteractive",
|
|
"-ExecutionPolicy",
|
|
"Bypass",
|
|
"-File",
|
|
str(harness),
|
|
],
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
errors="replace",
|
|
timeout=30,
|
|
check=False,
|
|
)
|
|
with self.subTest(role=role):
|
|
self.assertEqual(
|
|
completed.returncode,
|
|
0,
|
|
msg=f"stdout={completed.stdout}\nstderr={completed.stderr}",
|
|
)
|
|
|
|
def test_hidden_trigger_never_executes_a_workspace_script_directly(self) -> None:
|
|
for expected in (
|
|
"VignettePublicRuntimeWatchdog",
|
|
"-StableSourceRoot",
|
|
"-ExpectedSourceCommit",
|
|
"-ExpectedSourceTree",
|
|
"-ExpectedWatchdogSha256",
|
|
"-ExpectedStartScriptSha256",
|
|
"WorkingDirectory",
|
|
"Pinned watchdog script path does not match its working directory",
|
|
"Pinned watchdog source root does not match its working directory",
|
|
"Start-ScheduledTask",
|
|
"Legacy shared-worktree watchdog action is forbidden",
|
|
):
|
|
with self.subTest(expected=expected):
|
|
self.assertIn(expected, HIDDEN_TRIGGER_SOURCE)
|
|
self.assertIn(
|
|
"Join-Path $root 'scripts\\watch-public-runtime.ps1'",
|
|
HIDDEN_TRIGGER_SOURCE,
|
|
)
|
|
self.assertNotIn('objShell.Run "powershell.exe', HIDDEN_TRIGGER_SOURCE)
|
|
self.assertNotIn("D:\\workspace\\vignette", HIDDEN_TRIGGER_SOURCE)
|
|
marker_gate = HIDDEN_TRIGGER_SOURCE.index("foreach($marker in $required)")
|
|
task_trigger = HIDDEN_TRIGGER_SOURCE.rindex("Start-ScheduledTask")
|
|
self.assertLess(marker_gate, task_trigger)
|
|
|
|
def test_active_docs_only_show_the_pinned_release_workflow(self) -> None:
|
|
stale_commands = (
|
|
"powershell -NoProfile -ExecutionPolicy Bypass -File scripts\\watch-public-runtime.ps1 -CheckOnly",
|
|
"powershell -NoProfile -ExecutionPolicy Bypass -File scripts\\install-public-runtime-task.ps1 -RunNow",
|
|
)
|
|
for source in (RUNBOOK_SOURCE, LOCAL_DEVELOPMENT_SOURCE):
|
|
for stale in stale_commands:
|
|
with self.subTest(document=source[:32], stale=stale):
|
|
self.assertNotIn(stale, source)
|
|
for expected in (
|
|
"detached HEAD",
|
|
"-StableSourceRoot",
|
|
"-ExpectedSourceCommit",
|
|
"-ExpectedSourceTree",
|
|
"-ExpectedWatchdogSha256",
|
|
"-ExpectedStartScriptSha256",
|
|
"--porcelain=v1 --untracked-files=normal",
|
|
):
|
|
with self.subTest(expected=expected):
|
|
self.assertIn(expected, RUNBOOK_SOURCE)
|
|
self.assertIn("-StableSourceRoot", LOCAL_DEVELOPMENT_SOURCE)
|
|
self.assertIn("watch-public-runtime-hidden.vbs", LOCAL_DEVELOPMENT_SOURCE)
|
|
|
|
@unittest.skipUnless(shutil.which("git.exe"), "git.exe is unavailable")
|
|
@unittest.skipUnless(shutil.which("powershell.exe"), "Windows PowerShell 5.1 is unavailable")
|
|
def test_dirty_detached_release_fails_before_health_or_runtime_mutation(self) -> None:
|
|
with tempfile.TemporaryDirectory(prefix="vignette-watchdog-") as temp:
|
|
root = Path(temp)
|
|
scripts = root / "scripts"
|
|
scripts.mkdir()
|
|
copied_watchdog = scripts / WATCHDOG.name
|
|
copied_installer = scripts / INSTALLER.name
|
|
copied_boot = scripts / BOOT.name
|
|
copied_boot_register = scripts / BOOT_REGISTER.name
|
|
copied_start = scripts / START.name
|
|
copied_voice_probe = scripts / VOICE_PROBE.name
|
|
copied_task_launcher = scripts / TASK_LAUNCHER.name
|
|
shutil.copy2(WATCHDOG, copied_watchdog)
|
|
shutil.copy2(INSTALLER, copied_installer)
|
|
shutil.copy2(BOOT, copied_boot)
|
|
shutil.copy2(BOOT_REGISTER, copied_boot_register)
|
|
shutil.copy2(START, copied_start)
|
|
shutil.copy2(VOICE_PROBE, copied_voice_probe)
|
|
shutil.copy2(TASK_LAUNCHER, copied_task_launcher)
|
|
|
|
self._git(root, "init")
|
|
self._git(root, "config", "user.name", "Watchdog Contract Test")
|
|
self._git(root, "config", "user.email", "watchdog-test@example.invalid")
|
|
self._git(
|
|
root,
|
|
"add",
|
|
"scripts/watch-public-runtime.ps1",
|
|
"scripts/install-public-runtime-task.ps1",
|
|
"scripts/boot-public-runtime.ps1",
|
|
"scripts/register-boot-task.ps1",
|
|
"scripts/start-public-runtime.ps1",
|
|
"scripts/probe-public-voice-sidecars.py",
|
|
"scripts/watch-public-runtime-task.vbs",
|
|
)
|
|
self._git(root, "commit", "-m", "watchdog fixture")
|
|
self._git(root, "checkout", "--detach")
|
|
commit = self._git(root, "rev-parse", "HEAD").stdout.strip()
|
|
tree = self._git(root, "rev-parse", "HEAD^{tree}").stdout.strip()
|
|
watchdog_hash = sha256(copied_watchdog)
|
|
boot_hash = sha256(copied_boot)
|
|
start_hash = sha256(copied_start)
|
|
|
|
with copied_start.open("ab") as stream:
|
|
stream.write(b"\n# dirty fixture\n")
|
|
|
|
completed = subprocess.run(
|
|
[
|
|
shutil.which("powershell.exe") or "powershell.exe",
|
|
"-NoProfile",
|
|
"-ExecutionPolicy",
|
|
"Bypass",
|
|
"-File",
|
|
str(copied_watchdog),
|
|
"-StableSourceRoot",
|
|
str(root),
|
|
"-ExpectedSourceCommit",
|
|
commit,
|
|
"-ExpectedSourceTree",
|
|
tree,
|
|
"-ExpectedWatchdogSha256",
|
|
watchdog_hash,
|
|
"-ExpectedStartScriptSha256",
|
|
start_hash,
|
|
"-CheckOnly",
|
|
"-SkipPublicHealth",
|
|
"-SkipCloudflaredRestart",
|
|
],
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
errors="replace",
|
|
timeout=30,
|
|
)
|
|
detail = f"{completed.stdout}\n{completed.stderr}"
|
|
self.assertNotEqual(completed.returncode, 0, detail)
|
|
self.assertIn("Stable source is not clean", detail)
|
|
self.assertFalse((root / "public-runtime-watchdog.failcount").exists())
|
|
self.assertFalse((root / "public-runtime-watchdog.log").exists())
|
|
|
|
boot_completed = subprocess.run(
|
|
[
|
|
shutil.which("powershell.exe") or "powershell.exe",
|
|
"-NoProfile",
|
|
"-ExecutionPolicy",
|
|
"Bypass",
|
|
"-File",
|
|
str(copied_boot),
|
|
"-StableSourceRoot",
|
|
str(root),
|
|
"-ExpectedSourceCommit",
|
|
commit,
|
|
"-ExpectedSourceTree",
|
|
tree,
|
|
"-ExpectedBootScriptSha256",
|
|
boot_hash,
|
|
"-ExpectedStartScriptSha256",
|
|
start_hash,
|
|
],
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
errors="replace",
|
|
timeout=30,
|
|
)
|
|
boot_detail = f"{boot_completed.stdout}\n{boot_completed.stderr}"
|
|
self.assertNotEqual(boot_completed.returncode, 0, boot_detail)
|
|
self.assertIn("Stable source is not clean", boot_detail)
|
|
self.assertFalse((root / "boot-public-runtime.log").exists())
|
|
|
|
for registrar in (copied_installer, copied_boot_register):
|
|
register_completed = subprocess.run(
|
|
[
|
|
shutil.which("powershell.exe") or "powershell.exe",
|
|
"-NoProfile",
|
|
"-ExecutionPolicy",
|
|
"Bypass",
|
|
"-File",
|
|
str(registrar),
|
|
"-StableSourceRoot",
|
|
str(root),
|
|
],
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
errors="replace",
|
|
timeout=30,
|
|
)
|
|
register_detail = (
|
|
f"{register_completed.stdout}\n{register_completed.stderr}"
|
|
)
|
|
with self.subTest(registrar=registrar.name):
|
|
self.assertNotEqual(register_completed.returncode, 0, register_detail)
|
|
self.assertIn("Stable source is not clean", register_detail)
|
|
|
|
def test_windows_powershell_51_parser_accepts_scripts(self) -> None:
|
|
powershell = shutil.which("powershell.exe")
|
|
if powershell is None:
|
|
self.skipTest("Windows PowerShell 5.1 is unavailable")
|
|
for script in (WATCHDOG, INSTALLER, BOOT, BOOT_REGISTER):
|
|
escaped = str(script.resolve()).replace("'", "''")
|
|
command = (
|
|
"$tokens=$null; $errors=$null; "
|
|
"[System.Management.Automation.Language.Parser]::ParseFile("
|
|
f"'{escaped}', [ref]$tokens, [ref]$errors) | Out-Null; "
|
|
"if ($errors.Count -gt 0) { $errors | ForEach-Object { Write-Error $_ }; exit 1 }; "
|
|
"exit 0"
|
|
)
|
|
completed = subprocess.run(
|
|
[powershell, "-NoProfile", "-Command", command],
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
errors="replace",
|
|
)
|
|
with self.subTest(script=script.name):
|
|
self.assertEqual(completed.returncode, 0, completed.stderr)
|
|
|
|
def test_windows_script_host_accepts_hidden_trigger_syntax(self) -> None:
|
|
cscript = shutil.which("cscript.exe")
|
|
if cscript is None:
|
|
self.skipTest("Windows Script Host is unavailable")
|
|
completed = subprocess.run(
|
|
[cscript, "//nologo", str(HIDDEN_TRIGGER), "syntax-only"],
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
errors="replace",
|
|
)
|
|
self.assertEqual(completed.returncode, 0, completed.stderr)
|
|
|
|
def test_hidden_trigger_command_accepts_only_a_structurally_pinned_task(self) -> None:
|
|
cscript = shutil.which("cscript.exe")
|
|
powershell = shutil.which("powershell.exe")
|
|
if cscript is None or powershell is None:
|
|
self.skipTest("Windows Script Host or PowerShell 5.1 is unavailable")
|
|
rendered = subprocess.run(
|
|
[cscript, "//nologo", str(HIDDEN_TRIGGER), "print-command"],
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
errors="replace",
|
|
)
|
|
self.assertEqual(rendered.returncode, 0, rendered.stderr)
|
|
prefix = (
|
|
"powershell.exe -NoProfile -ExecutionPolicy Bypass "
|
|
'-WindowStyle Hidden -Command "'
|
|
)
|
|
wrapped = rendered.stdout.strip()
|
|
self.assertTrue(wrapped.startswith(prefix), wrapped)
|
|
self.assertTrue(wrapped.endswith('"'), wrapped)
|
|
command = wrapped[len(prefix) : -1].replace('""', '"')
|
|
|
|
root = r"C:\Pinned Vignette Release"
|
|
# 액션은 wscript 런처를 거친다 — powershell.exe 직접 실행은 conhost 창이 번쩍인다.
|
|
arguments = (
|
|
f'"{root}\\scripts\\watch-public-runtime-task.vbs" '
|
|
f'-File "{root}\\scripts\\watch-public-runtime.ps1" '
|
|
f'-StableSourceRoot "{root}" '
|
|
f"-ExpectedSourceCommit {'a' * 40} "
|
|
f"-ExpectedSourceTree {'b' * 40} "
|
|
f"-ExpectedWatchdogSha256 {'c' * 64} "
|
|
f"-ExpectedStartScriptSha256 {'d' * 64}"
|
|
)
|
|
fixture = (
|
|
"$script:watchdogTriggered=$false;"
|
|
"function Get-ScheduledTask { param($TaskName,$ErrorAction) "
|
|
"$action=[pscustomobject]@{"
|
|
"Execute='C:\\Windows\\System32\\wscript.exe';"
|
|
f"WorkingDirectory='{root}';Arguments='{arguments}'"
|
|
"};[pscustomobject]@{Actions=@($action)}};"
|
|
"function Start-ScheduledTask { param($TaskName) "
|
|
"$script:watchdogTriggered=$true };"
|
|
f"{command};"
|
|
"if(-not $script:watchdogTriggered){exit 9};exit 0"
|
|
)
|
|
checked = subprocess.run(
|
|
[powershell, "-NoProfile", "-Command", fixture],
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
errors="replace",
|
|
)
|
|
self.assertEqual(checked.returncode, 0, checked.stderr)
|
|
|
|
def _git(self, root: Path, *args: str) -> subprocess.CompletedProcess[str]:
|
|
completed = subprocess.run(
|
|
[shutil.which("git.exe") or "git.exe", "-C", str(root), *args],
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
errors="replace",
|
|
)
|
|
self.assertEqual(completed.returncode, 0, completed.stderr)
|
|
return completed
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|