70 lines
3.3 KiB
Python
70 lines
3.3 KiB
Python
import re
|
||
import unittest
|
||
from pathlib import Path
|
||
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
SCRIPT = ROOT / "Sense" / "scripts" / "t014-capacity.ps1"
|
||
TASK = ROOT / "docs" / "tasks" / "T-014.md"
|
||
|
||
|
||
class SenseCapacityContractTests(unittest.TestCase):
|
||
@classmethod
|
||
def setUpClass(cls):
|
||
cls.script = SCRIPT.read_text(encoding="utf-8")
|
||
cls.task = TASK.read_text(encoding="utf-8")
|
||
|
||
def test_uses_sixteen_independent_copy_publishers(self):
|
||
self.assertIn("$sourceCount = 16", self.script)
|
||
self.assertIn("foreach ($index in 1..$sourceCount)", self.script)
|
||
self.assertIn("'-c', 'copy'", self.script)
|
||
self.assertIn("$faultIndexes = @(5, 6, 7, 8)", self.script)
|
||
self.assertNotIn("ip_camera.env", self.script)
|
||
|
||
def test_exercises_real_control_api_and_quota_boundary(self):
|
||
self.assertIn("/api/v1/sites/site-t014/devices:batchDesiredState", self.script)
|
||
self.assertIn("desired_state = 'disabled'", self.script)
|
||
self.assertIn("t014-overflow-17", self.script)
|
||
self.assertIn("quota_exceeded", self.script)
|
||
self.assertIn("'enabled'", self.script)
|
||
self.assertIn("'disabled'", self.script)
|
||
|
||
def test_formal_window_and_functional_gates_are_fixed(self):
|
||
self.assertRegex(self.script, r"\[int\]\$ObservationMinutes = 30")
|
||
self.assertIn("$sampleSeconds = 10", self.script)
|
||
self.assertIn("formal_eligible = $ObservationMinutes -ge 30", self.script)
|
||
self.assertIn("Wait-Convergence 'enabled'", self.script)
|
||
self.assertIn("Wait-Convergence 'disabled'", self.script)
|
||
self.assertIn("maximum_unconverged", self.script)
|
||
|
||
def test_isolated_postgres_and_cleanup_are_guarded(self):
|
||
for migration in range(2, 12):
|
||
self.assertRegex(self.script, rf"'{migration:03d}_[^']+\.sql'")
|
||
self.assertIn("PostgreSQL\\) 17\\.10$", self.script)
|
||
self.assertIn("yovision-t014", self.script)
|
||
self.assertIn("[IO.Path]::GetFileName($resolvedSession) -notlike 'session-*'", self.script)
|
||
self.assertIn("the existing PostgreSQL listener on port 5432 changed", self.script)
|
||
self.assertNotRegex(self.script, r"Invoke-Checked \$script:pgCtl .* 'start'")
|
||
self.assertIn("$cleanupWatch.Elapsed.TotalSeconds -lt 10", self.script)
|
||
|
||
def test_runtime_secrets_and_sensitive_inventory_are_not_reported(self):
|
||
self.assertIn("New-SecretToken", self.script)
|
||
self.assertIn("token_sha256", self.script)
|
||
self.assertIn("capacity result output must stay outside the repository", self.script)
|
||
result_block = self.script.split("$resultJSON = [ordered]@{", 1)[1]
|
||
for forbidden in ("controlToken", "senseDSN", "adminDatabaseDSN", "sourceRTSPPort", "endpoint_ref", "credential_ref"):
|
||
self.assertNotIn(forbidden, result_block)
|
||
self.assertNotRegex(self.script, r"(?i)rtsp://[^\s\"']+:[^\s\"']+@")
|
||
|
||
def test_task_preserves_product_boundaries(self):
|
||
self.assertIn("deps: [T-012]", self.task)
|
||
self.assertIn("默认配额 16、允许范围 1～128", self.task)
|
||
self.assertIn("129 仍非法", self.task)
|
||
self.assertIn("不代表 16 台真实摄像头", self.task)
|
||
self.assertIn("不解除 T-007", self.task)
|
||
self.assertIn("WireGuard T-013 暂缓", self.task)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|