74 lines
3.9 KiB
Python
74 lines
3.9 KiB
Python
|
import hashlib
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
import sqlite3
|
||
|
|
import subprocess
|
||
|
|
import sys
|
||
|
|
import threading
|
||
|
|
import unittest
|
||
|
|
from http.client import HTTPConnection
|
||
|
|
from pathlib import Path
|
||
|
|
from tempfile import TemporaryDirectory
|
||
|
|
|
||
|
|
from app.main import create_server
|
||
|
|
from app.config import ConfigError, load_config
|
||
|
|
|
||
|
|
ROOT = Path(__file__).resolve().parents[3]
|
||
|
|
SCRIPTS = ROOT / "scripts"
|
||
|
|
|
||
|
|
|
||
|
|
class Phase15OpsTests(unittest.TestCase):
|
||
|
|
def test_config_defaults_are_safe_and_production_requires_explicit_secret(self):
|
||
|
|
with TemporaryDirectory() as tmp:
|
||
|
|
cfg = load_config({"DATA_DIR": tmp})
|
||
|
|
self.assertFalse(cfg.outreach_enabled)
|
||
|
|
self.assertEqual(cfg.app_env, "development")
|
||
|
|
with self.assertRaises(ConfigError):
|
||
|
|
load_config({"APP_ENV": "production", "DATA_DIR": tmp})
|
||
|
|
cfg = load_config({"APP_ENV": "production", "DATA_DIR": tmp, "SESSION_SECRET": "a" * 32})
|
||
|
|
self.assertEqual(cfg.session_secret, "a" * 32)
|
||
|
|
|
||
|
|
def test_health_and_readiness_report_database_state(self):
|
||
|
|
with TemporaryDirectory() as tmp:
|
||
|
|
db_path = str(Path(tmp) / "state.db")
|
||
|
|
server = create_server("127.0.0.1", 0, db_path)
|
||
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
||
|
|
thread.start()
|
||
|
|
try:
|
||
|
|
conn = HTTPConnection("127.0.0.1", server.server_port, timeout=3)
|
||
|
|
for path in ("/api/v1/health/live", "/api/v1/health/ready"):
|
||
|
|
conn.request("GET", path)
|
||
|
|
response = conn.getresponse()
|
||
|
|
payload = json.loads(response.read())
|
||
|
|
self.assertEqual(response.status, 200)
|
||
|
|
self.assertEqual(payload["status"], "ok")
|
||
|
|
self.assertFalse(payload["outreach_enabled"])
|
||
|
|
finally:
|
||
|
|
server.shutdown(); server.server_close(); thread.join(timeout=2)
|
||
|
|
|
||
|
|
def test_backup_restore_integrity_and_checksum(self):
|
||
|
|
with TemporaryDirectory() as tmp:
|
||
|
|
root = Path(tmp); db = root / "prospects.db"; backups = root / "backups"
|
||
|
|
conn = sqlite3.connect(db); conn.execute("CREATE TABLE facts (value TEXT)"); conn.execute("INSERT INTO facts VALUES ('before')"); conn.commit(); conn.close()
|
||
|
|
result = subprocess.run([str(SCRIPTS / "backup_sqlite.sh"), str(db), str(backups), "2"], text=True, capture_output=True, check=True)
|
||
|
|
backup = Path(result.stdout.strip().splitlines()[-1])
|
||
|
|
self.assertTrue(backup.exists()); self.assertTrue(Path(str(backup) + ".sha256").exists())
|
||
|
|
self.assertEqual(hashlib.sha256(backup.read_bytes()).hexdigest(), Path(str(backup) + ".sha256").read_text().split()[0])
|
||
|
|
conn = sqlite3.connect(db); conn.execute("UPDATE facts SET value='after'"); conn.commit(); conn.close()
|
||
|
|
subprocess.run([str(SCRIPTS / "restore_sqlite.sh"), str(backup), str(db), "--confirm-restore"], text=True, capture_output=True, check=True)
|
||
|
|
conn = sqlite3.connect(db); self.assertEqual(conn.execute("SELECT value FROM facts").fetchone()[0], "before"); conn.close()
|
||
|
|
self.assertTrue(any(p.name.startswith("pre-restore-") for p in backups.glob("*.db")))
|
||
|
|
|
||
|
|
def test_backup_output_does_not_contain_secret_values(self):
|
||
|
|
with TemporaryDirectory() as tmp:
|
||
|
|
root = Path(tmp); db = root / "prospects.db"; backups = root / "backups"
|
||
|
|
conn = sqlite3.connect(db); conn.execute("CREATE TABLE logs (message TEXT)"); conn.execute("INSERT INTO logs VALUES ('password=should-not-be-logged')"); conn.commit(); conn.close()
|
||
|
|
result = subprocess.run([str(SCRIPTS / "backup_sqlite.sh"), str(db), str(backups), "1"], text=True, capture_output=True, check=True)
|
||
|
|
combined = result.stdout + result.stderr
|
||
|
|
self.assertNotIn("should-not-be-logged", combined)
|
||
|
|
self.assertNotIn("password=", combined)
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
unittest.main()
|