test_event_pusher_upload.py 1.4 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152
  1. import sys
  2. import os
  3. sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
  4. import numpy as np
  5. import pytest
  6. from event_pusher import EventPusher
  7. def test_upload_numpy_image(monkeypatch):
  8. config = {"device_id": "test-device", "base_url": "http://localhost"}
  9. pusher = EventPusher(config)
  10. called = {"path": None, "existed": False}
  11. def fake_upload(path):
  12. called["path"] = path
  13. called["existed"] = os.path.exists(path)
  14. return "http://example.com/image.jpg"
  15. monkeypatch.setattr(pusher, "_upload_image", fake_upload)
  16. img = np.zeros((100, 100, 3), dtype=np.uint8)
  17. url = pusher.upload_numpy_image(img)
  18. assert url == "http://example.com/image.jpg"
  19. assert called["path"] is not None
  20. assert called["existed"]
  21. def test_push_tracking_capture(monkeypatch):
  22. config = {"device_id": "test-device", "base_url": "http://localhost"}
  23. pusher = EventPusher(config)
  24. captured = {}
  25. def fake_post(url, json):
  26. captured["url"] = url
  27. captured["json"] = json
  28. class Resp:
  29. status_code = 200
  30. return Resp()
  31. monkeypatch.setattr(pusher, "_post", fake_post)
  32. pusher.push_tracking_capture(
  33. batch_time=1234567890.0,
  34. captures=[{"track_id": 1, "ptz_image_url": "url1"}]
  35. )
  36. assert captured["json"]["eventType"] == "TRACKING_CAPTURE"
  37. assert captured["json"]["data"]["captureCount"] == 1