290 lines
12 KiB
Python
290 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
|
|
SCRIPTS = Path(__file__).resolve().parent
|
|
HELPER = SCRIPTS / "public-runtime-upload-root.ps1"
|
|
PROBE = SCRIPTS / "probe-public-runtime-upload-root.py"
|
|
START = SCRIPTS / "start-public-runtime.ps1"
|
|
BOOT = SCRIPTS / "boot-public-runtime.ps1"
|
|
WATCH = SCRIPTS / "watch-public-runtime.ps1"
|
|
INSTALL = SCRIPTS / "install-public-runtime-task.ps1"
|
|
REGISTER = SCRIPTS / "register-boot-task.ps1"
|
|
HIDDEN = SCRIPTS / "watch-public-runtime-hidden.vbs"
|
|
|
|
|
|
def _read(path: Path) -> str:
|
|
return path.read_text(encoding="utf-8")
|
|
|
|
|
|
def _run_powershell(script: Path) -> subprocess.CompletedProcess[str]:
|
|
powershell = shutil.which("powershell.exe")
|
|
if powershell is None:
|
|
raise unittest.SkipTest("Windows PowerShell 5.1 is unavailable")
|
|
return subprocess.run(
|
|
[
|
|
powershell,
|
|
"-NoLogo",
|
|
"-NoProfile",
|
|
"-NonInteractive",
|
|
"-ExecutionPolicy",
|
|
"Bypass",
|
|
"-File",
|
|
str(script),
|
|
],
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
errors="replace",
|
|
timeout=30,
|
|
)
|
|
|
|
|
|
class PublicRuntimeUploadRootContractTest(unittest.TestCase):
|
|
def test_call_chain_owns_one_explicit_upload_root(self) -> None:
|
|
self.assertTrue(HELPER.is_file())
|
|
self.assertTrue(PROBE.is_file())
|
|
start = _read(START)
|
|
boot = _read(BOOT)
|
|
watch = _read(WATCH)
|
|
install = _read(INSTALL)
|
|
register = _read(REGISTER)
|
|
hidden = _read(HIDDEN)
|
|
|
|
for source in (start, boot, watch, install, register):
|
|
with self.subTest(source=source[:24]):
|
|
self.assertIn("[string]$UserUploadDir", source)
|
|
self.assertIn("Resolve-PublicRuntimeUploadRoot", source)
|
|
self.assertIn("scripts\\public-runtime-upload-root.ps1", source)
|
|
self.assertIn("scripts\\probe-public-runtime-upload-root.py", source)
|
|
|
|
self.assertIn("$env:USER_UPLOAD_DIR = $resolvedUserUploadDir", start)
|
|
self.assertIn('"USER_UPLOAD_DIR"', start)
|
|
self.assertIn("user_upload_root = $resolvedUserUploadDir", start)
|
|
self.assertIn('"-UserUploadDir", $resolvedUserUploadDir', boot)
|
|
self.assertIn("UserUploadDir = $resolvedUserUploadDir", watch)
|
|
self.assertIn('"-UserUploadDir `"$resolvedUserUploadDir`""', install)
|
|
self.assertIn('"-UserUploadDir `"$resolvedUserUploadDir`""', register)
|
|
self.assertIn("'-UserUploadDir'", hidden)
|
|
|
|
def test_watchdog_check_only_proves_running_api_storage_contract(self) -> None:
|
|
watch = _read(WATCH)
|
|
provenance_comment = watch.index(
|
|
"# health probe, failcount 기록, 프로세스 재기동보다 먼저"
|
|
)
|
|
source_gate = watch.index("Assert-StableSourceProvenance", provenance_comment)
|
|
helper_load = watch.index(". $uploadRootContract", source_gate)
|
|
upload_preflight = watch.index(
|
|
"$resolvedUserUploadDir = Resolve-PublicRuntimeUploadRoot", helper_load
|
|
)
|
|
health_checks = watch.index("$checks = @(")
|
|
runtime_probe = watch.index("Test-PublicRuntimeApiUploadRoot", health_checks)
|
|
check_only = watch.index("if ($CheckOnly)", health_checks)
|
|
restart = watch.index("& $startScript @startArgs")
|
|
self.assertLess(source_gate, helper_load)
|
|
self.assertLess(helper_load, upload_preflight)
|
|
self.assertLess(upload_preflight, health_checks)
|
|
self.assertLess(health_checks, runtime_probe)
|
|
self.assertLess(runtime_probe, check_only)
|
|
self.assertLess(check_only, restart)
|
|
self.assertIn('(Test-PublicRuntimeApiUploadRoot `', watch[health_checks:check_only])
|
|
self.assertIn('($FailedNames -contains "api-upload-root")', watch)
|
|
|
|
def test_boot_and_registrars_verify_source_before_loading_upload_helper(self) -> None:
|
|
start = _read(START)
|
|
fresh_gate = start.index("Assert-FreshPublicProvenanceContract `", start.index("if ($RequireFreshPublicProvenance)"))
|
|
start_helper = start.index(". $uploadRootContract", fresh_gate)
|
|
self.assertLess(fresh_gate, start_helper)
|
|
|
|
boot = _read(BOOT)
|
|
boot_comment = boot.index(
|
|
"# Docker/DB/process mutation보다 먼저 stable source를 매 실행 재검증한다."
|
|
)
|
|
boot_gate = boot.index("Assert-StableSourceProvenance", boot_comment)
|
|
self.assertLess(boot_gate, boot.index(". $uploadRootContract", boot_gate))
|
|
|
|
for path, helper_name in (
|
|
(INSTALL, "$uploadRootContract"),
|
|
(REGISTER, "$UploadRootContract"),
|
|
):
|
|
source = _read(path)
|
|
clean_gate = source.index('$dirty = Invoke-GitText -Arguments @("status"')
|
|
helper_load = source.index(f". {helper_name}")
|
|
task_mutation = source.index("Register-ScheduledTask")
|
|
with self.subTest(path=path.name):
|
|
self.assertLess(clean_gate, helper_load)
|
|
self.assertLess(helper_load, task_mutation)
|
|
|
|
@unittest.skipUnless(shutil.which("git.exe"), "git.exe is unavailable")
|
|
def test_path_preflight_rejects_relative_and_source_overlap_in_powershell_51(
|
|
self,
|
|
) -> None:
|
|
self.assertTrue(HELPER.is_file())
|
|
with tempfile.TemporaryDirectory(prefix="vignette-upload-contract-") as temp:
|
|
root = Path(temp)
|
|
source = root / "source"
|
|
source.mkdir()
|
|
subprocess.run(
|
|
[shutil.which("git.exe") or "git.exe", "-C", str(source), "init"],
|
|
check=True,
|
|
capture_output=True,
|
|
)
|
|
upload = root / "stable-uploads"
|
|
file_target = root / "not-a-directory"
|
|
file_target.write_text("x", encoding="utf-8")
|
|
harness = root / "upload-root-harness.ps1"
|
|
quoted_helper = str(HELPER).replace("'", "''")
|
|
quoted_root = str(root).replace("'", "''")
|
|
quoted_source = str(source).replace("'", "''")
|
|
quoted_upload = str(upload).replace("'", "''")
|
|
quoted_file = str(file_target).replace("'", "''")
|
|
harness.write_text(
|
|
f"""$ErrorActionPreference = 'Stop'
|
|
. '{quoted_helper}'
|
|
function Assert-Rejected([string]$Candidate) {{
|
|
$rejected = $false
|
|
try {{
|
|
Resolve-PublicRuntimeUploadRoot `
|
|
-SourceRoot '{quoted_source}' `
|
|
-UploadRoot $Candidate `
|
|
-CreateIfMissing `
|
|
-ProbeWritable | Out-Null
|
|
}} catch {{
|
|
$rejected = $true
|
|
}}
|
|
if (-not $rejected) {{ throw "unsafe upload root accepted: $Candidate" }}
|
|
}}
|
|
Assert-Rejected 'relative\\uploads'
|
|
Assert-Rejected (Join-Path '{quoted_source}' 'uploads')
|
|
Assert-Rejected '{quoted_source}'
|
|
Assert-Rejected '{quoted_file}'
|
|
$junctionTarget = Join-Path '{quoted_root}' 'junction-target'
|
|
$junction = Join-Path '{quoted_root}' 'upload-junction'
|
|
[System.IO.Directory]::CreateDirectory($junctionTarget) | Out-Null
|
|
try {{
|
|
New-Item -ItemType Junction -Path $junction -Target $junctionTarget | Out-Null
|
|
Assert-Rejected (Join-Path $junction 'uploads')
|
|
}} finally {{
|
|
if ([System.IO.Directory]::Exists($junction)) {{
|
|
[System.IO.Directory]::Delete($junction)
|
|
}}
|
|
}}
|
|
$resolved = Resolve-PublicRuntimeUploadRoot `
|
|
-SourceRoot '{quoted_source}' `
|
|
-UploadRoot '{quoted_upload}' `
|
|
-CreateIfMissing `
|
|
-ProbeWritable
|
|
$expected = (Resolve-Path -LiteralPath '{quoted_upload}').Path
|
|
if ($resolved -cne $expected) {{ throw "resolved root mismatch: $resolved" }}
|
|
if (@(Get-ChildItem -LiteralPath $resolved -Force -Filter '.vignette-write-probe-*.tmp').Count -ne 0) {{
|
|
throw 'writability probe was not cleaned'
|
|
}}
|
|
""",
|
|
encoding="utf-8-sig",
|
|
)
|
|
completed = _run_powershell(harness)
|
|
self.assertEqual(
|
|
completed.returncode,
|
|
0,
|
|
msg=f"stdout={completed.stdout}\nstderr={completed.stderr}",
|
|
)
|
|
|
|
def test_user_upload_environment_is_in_fresh_rollback_snapshot(self) -> None:
|
|
start = _read(START)
|
|
functions = start[
|
|
start.index("function Save-ManagedEnvironment") :
|
|
start.index("function Save-CompleteProcessEnvironment")
|
|
]
|
|
with tempfile.TemporaryDirectory(prefix="vignette-upload-rollback-") as temp:
|
|
root = Path(temp)
|
|
function_file = root / "environment-functions.ps1"
|
|
harness = root / "environment-rollback.ps1"
|
|
function_file.write_text(functions, encoding="utf-8-sig")
|
|
quoted_functions = str(function_file).replace("'", "''")
|
|
harness.write_text(
|
|
f"""$ErrorActionPreference = 'Stop'
|
|
. '{quoted_functions}'
|
|
$original = [Environment]::GetEnvironmentVariable('USER_UPLOAD_DIR', 'Process')
|
|
try {{
|
|
[Environment]::SetEnvironmentVariable('USER_UPLOAD_DIR', 'C:\\prior-uploads', 'Process')
|
|
$snapshot = Save-ManagedEnvironment -Names @('USER_UPLOAD_DIR')
|
|
[Environment]::SetEnvironmentVariable('USER_UPLOAD_DIR', 'C:\\fresh-uploads', 'Process')
|
|
Restore-ManagedEnvironment -Snapshot $snapshot
|
|
if ($env:USER_UPLOAD_DIR -cne 'C:\\prior-uploads') {{ throw 'prior upload root was not restored' }}
|
|
|
|
[Environment]::SetEnvironmentVariable('USER_UPLOAD_DIR', $null, 'Process')
|
|
$missing = Save-ManagedEnvironment -Names @('USER_UPLOAD_DIR')
|
|
[Environment]::SetEnvironmentVariable('USER_UPLOAD_DIR', 'C:\\fresh-uploads', 'Process')
|
|
Restore-ManagedEnvironment -Snapshot $missing
|
|
if ($null -ne [Environment]::GetEnvironmentVariable('USER_UPLOAD_DIR', 'Process')) {{
|
|
throw 'missing upload root was not restored as missing'
|
|
}}
|
|
}} finally {{
|
|
[Environment]::SetEnvironmentVariable('USER_UPLOAD_DIR', $original, 'Process')
|
|
}}
|
|
""",
|
|
encoding="utf-8-sig",
|
|
)
|
|
completed = _run_powershell(harness)
|
|
self.assertEqual(
|
|
completed.returncode,
|
|
0,
|
|
msg=f"stdout={completed.stdout}\nstderr={completed.stderr}",
|
|
)
|
|
|
|
def test_process_probe_reports_missing_and_mismatched_upload_roots(self) -> None:
|
|
self.assertTrue(PROBE.is_file())
|
|
spec = importlib.util.spec_from_file_location("upload_probe", PROBE)
|
|
self.assertIsNotNone(spec)
|
|
self.assertIsNotNone(spec.loader if spec else None)
|
|
module = importlib.util.module_from_spec(spec)
|
|
assert spec is not None and spec.loader is not None
|
|
spec.loader.exec_module(module)
|
|
|
|
expected = r"C:\Users\tester\AppData\Local\Vignette\public-runtime\uploads"
|
|
argv = [
|
|
"python.exe",
|
|
"-m",
|
|
"uvicorn",
|
|
"app.main:app",
|
|
"--host",
|
|
"127.0.0.1",
|
|
"--port",
|
|
"8001",
|
|
"--workers",
|
|
"1",
|
|
]
|
|
passed = module.evaluate_snapshots(
|
|
[{"pid": 10, "argv": argv, "user_upload_root": expected}],
|
|
expected_root=expected,
|
|
api_port=8001,
|
|
)
|
|
self.assertEqual(passed[0], 0)
|
|
self.assertEqual(passed[1]["status"], "passed")
|
|
|
|
missing = module.evaluate_snapshots(
|
|
[{"pid": 10, "argv": argv, "user_upload_root": None}],
|
|
expected_root=expected,
|
|
api_port=8001,
|
|
)
|
|
self.assertNotEqual(missing[0], 0)
|
|
self.assertEqual(missing[1]["reason"], "user_upload_dir_missing")
|
|
|
|
mismatched = module.evaluate_snapshots(
|
|
[{"pid": 10, "argv": argv, "user_upload_root": r"D:\\release\\uploads"}],
|
|
expected_root=expected,
|
|
api_port=8001,
|
|
)
|
|
self.assertNotEqual(mismatched[0], 0)
|
|
self.assertEqual(mismatched[1]["reason"], "user_upload_dir_drift")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|