Kunpeng
25fa0c5825
Backend:
- Persist alert events and robot delivery counters in SQLite (data/app.db,
YOLO_DB_PATH override); EventStore keeps its interface, DeliveryStatsStore
uses read/write-through UPSERT; in-memory fallback remains for standalone
AlertManager use
- Serve /api/detect from a sync endpoint and serialize YOLO inference with a
lock, so concurrent requests no longer block the event loop
- Unify single-image and session detection: requests without session_id
share the fixed single-image session and run full frame confirmation
- Store UTC Z-suffixed timestamps for sortable string comparison; close
executor and database on shutdown via FastAPI lifespan
Frontend:
- Split app.js into ES modules (main.js + modules/{dom,api,ui,charts,store}
+ views/{dashboard,inspection,events,robots,robotCards}), no build step
- Escape all server data interpolated into innerHTML; guard missing
event.classes; replace lazy element ID list with memoized qs()
- Remove hardcoded fake stats (trend badge, device donut segment)
Docs:
- Rewrite README: accurate weight policy (only trained best.pt committed,
no Git LFS), SQLite persistence, single-worker note, data asset inventory
- Align models/pretrained/README.md with actual files; add YOLO_DB_PATH to
.env.example; add persistence unit tests; ruff clean
81 lines
2.9 KiB
Python
81 lines
2.9 KiB
Python
from __future__ import annotations
|
|
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from backend.alerting import AlertManager
|
|
from backend.storage import Database, DeliveryStatsStore
|
|
|
|
|
|
class AlertManagerRobotTests(unittest.TestCase):
|
|
def test_channel_status_does_not_expose_webhook(self) -> None:
|
|
manager = AlertManager(
|
|
wechat_webhook_url="https://example.test/wechat-secret",
|
|
feishu_webhook_url="https://example.test/feishu-secret",
|
|
feishu_secret="signing-secret",
|
|
)
|
|
|
|
statuses = {item["id"]: item for item in manager.channel_status}
|
|
|
|
self.assertTrue(statuses["wechat"]["configured"])
|
|
self.assertTrue(statuses["feishu"]["configured"])
|
|
self.assertTrue(statuses["feishu"]["signed"])
|
|
self.assertNotIn("webhook", repr(statuses).lower())
|
|
self.assertNotIn("example.test", repr(statuses))
|
|
|
|
@patch("backend.alerting.post_json", return_value={})
|
|
def test_wechat_connection_test_updates_delivery_stats(
|
|
self,
|
|
mocked_post_json,
|
|
) -> None:
|
|
manager = AlertManager(
|
|
wechat_webhook_url="https://example.test/hook"
|
|
)
|
|
|
|
result = manager.test_channel("wechat")
|
|
|
|
self.assertEqual(result["last_status"], "success")
|
|
self.assertEqual(result["delivered"], 1)
|
|
self.assertEqual(result["tests"], 1)
|
|
self.assertIsNotNone(result["last_test_at"])
|
|
mocked_post_json.assert_called_once()
|
|
|
|
def test_unconfigured_channel_cannot_send_test(self) -> None:
|
|
manager = AlertManager()
|
|
|
|
with self.assertRaisesRegex(ValueError, "尚未配置"):
|
|
manager.test_channel("wechat")
|
|
|
|
def test_unknown_channel_is_rejected(self) -> None:
|
|
manager = AlertManager()
|
|
|
|
with self.assertRaisesRegex(ValueError, "Unsupported"):
|
|
manager.test_channel("unknown")
|
|
|
|
def test_delivery_stats_persist_across_store_instances(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp_dir:
|
|
db_path = Path(tmp_dir) / "app.db"
|
|
first_db = Database(db_path)
|
|
store = DeliveryStatsStore(first_db)
|
|
store.record("wechat", success=True)
|
|
store.record("wechat", success=False, error="boom")
|
|
store.record("feishu", success=True, test=True)
|
|
first_db.close()
|
|
|
|
second_db = Database(db_path)
|
|
reloaded = DeliveryStatsStore(second_db)
|
|
stats = reloaded.get_all()
|
|
second_db.close()
|
|
|
|
self.assertEqual(stats["wechat"]["delivered"], 1)
|
|
self.assertEqual(stats["wechat"]["failed"], 1)
|
|
self.assertEqual(stats["wechat"]["last_status"], "failed")
|
|
self.assertEqual(stats["wechat"]["last_error"], "boom")
|
|
self.assertEqual(stats["feishu"]["tests"], 1)
|
|
self.assertIsNotNone(stats["feishu"]["last_test_at"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main() |