165 lines
5.5 KiB
Python
165 lines
5.5 KiB
Python
from __future__ import annotations
|
|
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from cold_display_guard.config import load_settings, save_config_document
|
|
|
|
|
|
class ConfigTests(unittest.TestCase):
|
|
def test_loads_settings_from_toml(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
path = Path(tmpdir) / "config.toml"
|
|
path.write_text(
|
|
"""
|
|
camera_id = "cam_a"
|
|
|
|
[thresholds]
|
|
max_dwell_seconds = 30
|
|
trash_confirmation_seconds = 4
|
|
|
|
[layout]
|
|
rows = 1
|
|
cols = 2
|
|
""".strip(),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
settings = load_settings(path)
|
|
|
|
self.assertEqual(settings.camera_id, "cam_a")
|
|
self.assertEqual(settings.max_dwell_seconds, 30)
|
|
self.assertEqual(settings.trash_confirmation_seconds, 4)
|
|
self.assertEqual(settings.zone_ids, ("r1c1", "r1c2"))
|
|
|
|
def test_loads_numeric_zone_ids_for_custom_zone_count(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
path = Path(tmpdir) / "config.toml"
|
|
path.write_text(
|
|
"""
|
|
camera_id = "cam_numeric"
|
|
|
|
[thresholds]
|
|
max_dwell_seconds = 1200
|
|
trash_confirmation_seconds = 120
|
|
|
|
[layout]
|
|
zone_count = 3
|
|
zone_ids = ["1", "2", "3"]
|
|
""".strip(),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
settings = load_settings(path)
|
|
|
|
self.assertEqual(settings.camera_id, "cam_numeric")
|
|
self.assertEqual(settings.max_dwell_seconds, 1200)
|
|
self.assertEqual(settings.zone_ids, ("1", "2", "3"))
|
|
|
|
def test_rejects_more_than_ten_numeric_food_zones(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
path = Path(tmpdir) / "config.toml"
|
|
path.write_text(
|
|
"""
|
|
[layout]
|
|
zone_ids = ["1", "2", "3", "4", "5", "6", "7", "8", "9", "10", "11"]
|
|
""".strip(),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
with self.assertRaisesRegex(ValueError, "1 to 10"):
|
|
load_settings(path)
|
|
|
|
def test_loads_numeric_zone_ids_from_zone_count_without_explicit_ids(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
path = Path(tmpdir) / "config.toml"
|
|
path.write_text(
|
|
"""
|
|
[layout]
|
|
zone_count = 4
|
|
""".strip(),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
settings = load_settings(path)
|
|
|
|
self.assertEqual(settings.zone_ids, ("1", "2", "3", "4"))
|
|
|
|
def test_rejects_numeric_zone_count_that_conflicts_with_zone_ids(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
path = Path(tmpdir) / "config.toml"
|
|
path.write_text(
|
|
"""
|
|
[layout]
|
|
zone_count = 5
|
|
zone_ids = ["1", "2", "3"]
|
|
""".strip(),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
with self.assertRaisesRegex(ValueError, "zone_count"):
|
|
load_settings(path)
|
|
|
|
def test_save_config_document_round_trips_zone_count_and_numeric_labels(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
path = Path(tmpdir) / "config.toml"
|
|
save_config_document(
|
|
path,
|
|
{
|
|
"layout": {"zone_count": 2, "zone_ids": ["1", "2"]},
|
|
"zones": [
|
|
{"id": "1", "label": "区域 1", "polygon": [[0, 0], [1, 0], [1, 1]]},
|
|
{"id": "2", "label": "区域 2", "polygon": [[0, 0], [0.5, 0], [0.5, 1]]},
|
|
],
|
|
"trash": {"roi": [[0, 0], [1, 0], [1, 1]]},
|
|
},
|
|
)
|
|
text = path.read_text(encoding="utf-8")
|
|
|
|
self.assertIn("zone_count = 2", text)
|
|
self.assertIn('label = "区域 1"', text)
|
|
self.assertIn("[trash]", text)
|
|
self.assertNotIn('"trash"', text.split("[layout]", maxsplit=1)[1].split("[[zones]]", maxsplit=1)[0])
|
|
|
|
def test_save_config_document_writes_webhooks_and_case_sink(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
path = Path(tmpdir) / "config.toml"
|
|
save_config_document(
|
|
path,
|
|
{
|
|
"webhooks": {
|
|
"enabled": True,
|
|
"event_url": "https://example.com/events",
|
|
"case_url": "https://example.com/cases",
|
|
"callback_token": "secret",
|
|
"connect_timeout_seconds": 3,
|
|
"read_timeout_seconds": 5,
|
|
"retry_max_attempts": 4,
|
|
"retry_backoff_seconds": 30,
|
|
"retry_max_backoff_seconds": 300,
|
|
"retry_batch_limit": 12,
|
|
},
|
|
"case_sink": {"path": "logs/cases.jsonl"},
|
|
"webhook_retry_sink": {"path": "logs/webhook_retry.jsonl"},
|
|
},
|
|
)
|
|
text = path.read_text(encoding="utf-8")
|
|
|
|
self.assertIn("[webhooks]", text)
|
|
self.assertIn('event_url = "https://example.com/events"', text)
|
|
self.assertIn('case_url = "https://example.com/cases"', text)
|
|
self.assertIn('callback_token = "secret"', text)
|
|
self.assertIn("retry_max_attempts = 4", text)
|
|
self.assertIn("retry_backoff_seconds = 30", text)
|
|
self.assertIn("retry_max_backoff_seconds = 300", text)
|
|
self.assertIn("retry_batch_limit = 12", text)
|
|
self.assertIn("[case_sink]", text)
|
|
self.assertIn('path = "logs/cases.jsonl"', text)
|
|
self.assertIn("[webhook_retry_sink]", text)
|
|
self.assertIn('path = "logs/webhook_retry.jsonl"', text)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|