yovision/tests/test_sense_capacity_contract.py

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("$requiredSamples = [int][Math]::Ceiling($targetSeconds / $sampleSeconds)", self.script)
        self.assertIn("$nextSampleAt = ($samples + 1) * $sampleSeconds", 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)
        self.assertIn("go1\\.26\\.5", self.script)
        self.assertIn("& go '-C' $senseDirectory 'version'", 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)
        self.assertIn("$item.Process.Dispose()", self.script)
        self.assertIn("$deleteWatch.Elapsed.TotalSeconds -ge 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()