Files
infra-docker-configs/servers/tools-pve/asr-pipeline/tests/test_portability.py
T
Codex 3c70861463
Build image / build (push) Canceled after 0s
Build ASR Pipeline / image (push) Canceled after 0s
feat(asr-pipeline): merge PR #1 webauth/CSRF/locking hardening and web_password fallback
2026-11-02 06:37:33 -08:00

48 lines
2.0 KiB
Python

from __future__ import annotations
import os
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
class PortabilityTests(unittest.TestCase):
def probe(self, code, overrides):
env = {**os.environ, **overrides, "PYTHONPATH": str(ROOT),
"PYTHONDONTWRITEBYTECODE": "1"}
return subprocess.check_output([sys.executable, "-c", code],
env=env, text=True).strip()
def test_workspace_and_notifier_are_explicit(self):
with tempfile.TemporaryDirectory() as tmp:
result = self.probe(
"from asrpipe.config import CFG; print(CFG.obsidian); print(CFG.notifier)",
{"ASR_WORKSPACE": tmp, "ASR_NOTIFIER": tmp + "/notify.py"})
self.assertEqual(result.splitlines(), [tmp, tmp + "/notify.py"])
def test_watcher_does_not_assume_claude_workspace(self):
with tempfile.TemporaryDirectory() as tmp:
result = self.probe(
"import runpy; from pathlib import Path; "
"d=runpy.run_path('recordings/bin/meeting-watch.py',run_name='probe'); "
"print(d['CONTROL_DB']); print(d['ASR_INBOX'])",
{"ASR_STATE_DIR": tmp + "/state", "ASR_VAULT": tmp + "/vault"})
self.assertEqual(result.splitlines(),
[tmp + "/state/intake.db", tmp + "/vault/.asr-inbox"])
def test_notifier_accepts_external_webhook_without_sending(self):
result = self.probe(
"import runpy; d=runpy.run_path('scripts/meeting-status-notify.py',run_name='probe'); "
"print(d['URL']); print(d['secret']() == 'synthetic-only')",
{"MEETING_STATUS_WEBHOOK_URL": "http://example.invalid/hook",
"MEETING_STATUS_WEBHOOK_SECRET": "synthetic-only"})
self.assertEqual(result.splitlines(), ["http://example.invalid/hook", "True"])
if __name__ == "__main__":
unittest.main()