5cb530e153
재부팅 뒤 워치독이 5분마다 콘솔 창에 실패만 뿌리던 문제의 원인 세 가지를 고친다. - 워치독이 DB 다운을 감지하고 직접 복구한다. postgres(vignette-dev-db) 기동은 boot 담당이라 start-public-runtime.ps1 재호출로는 절대 복구되지 않았고, 그 결과 워치독은 고칠 수 없는 대상에 start를 무한 재시도하며 실패만 기록했다. db를 health check 항목에 넣고, 재시작 전에 컨테이너를 되살리며, 복구 실패 시에는 runtime 재시작을 시도하지 않고 종료한다. - 작업 액션을 wscript 런처(watch-public-runtime-task.vbs) 경유로 등록한다. powershell.exe를 직접 등록하면 -WindowStyle Hidden이어도 conhost 창이 매 실행 번쩍이고, 5분 주기에서는 그것이 곧 화면을 가리는 창이 된다. 런처는 pin 인자를 해석하지 않고 전달만 하며 provenance 검증은 기존대로 watchdog이 수행한다. - boot이 web preview 상태를 보고 -SkipWebRestart를 조건부로 붙인다. 무조건 스킵하면 재부팅 직후처럼 vite가 죽은 상태에서 boot 경로로는 web이 영영 복구되지 않았다. hidden trigger는 액션이 wscript 런처를 거치는지 함께 검증하도록 맞췄다.
642 lines
28 KiB
Python
642 lines
28 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("-File $startScript")
|
|
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_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()
|