Behavision/tests/test_api_cameras.py

"""Camera CRUD over HTTP against a stub engine (no models, no camera).

Needs httpx for starlette's TestClient; skipped if absent so a bare checkout
still runs the suite.
"""
import threading

import pytest

pytest.importorskip("httpx")

from fastapi.testclient import TestClient  # noqa: E402

from behavision.api import create_app  # noqa: E402
from behavision.cameras import CameraStore  # noqa: E402
from behavision.config import Config  # noqa: E402
from behavision.engine import Engine  # noqa: E402

SECRET = "sup3r-s3cret-rtsp-pw"


class FakeWorker:
    def __init__(self, cam_cfg):
        self.cam_cfg = cam_cfg
        self.rcfg = Config().recognition.merged(cam_cfg.tuning)
        self.commission = None
    def start(self): pass
    def stop(self): pass
    def is_alive(self): return True
    def join(self, timeout=None): pass
    def stats(self): return {"camera_id": self.cam_cfg.id, "connected": True,
                             "url": self.cam_cfg.safe_url()}
    def latest_jpeg(self): return None
    def latest_jpeg_since(self, known_ts): return None, known_ts


@pytest.fixture
def client(tmp_path, monkeypatch):
    monkeypatch.setattr("behavision.engine.CameraWorker",
                        lambda cam, *a, **kw: FakeWorker(cam))
    monkeypatch.setattr("behavision.engine.FaceDetector", lambda *a, **kw: object())
    eng = object.__new__(Engine)
    eng.cfg = Config()
    eng.cfg.app.data_dir = tmp_path
    eng.cfg.api.username = eng.cfg.api.password = ""   # loopback: no auth
    eng.encoder = eng.gallery = eng.bus = eng.attributes = eng.store = None
    eng._lock = threading.RLock()
    eng.workers, eng.detectors = {}, {}
    eng.started_at, eng._running = 1.0, True
    eng.camera_store = CameraStore(tmp_path / "cameras.json")
    return TestClient(create_app(eng)), eng


def _add(c, cid="cam1", **kw):
    body = dict(id=cid, host="192.168.0.138", path="/ch0_0.264",
                username="admin", password=SECRET)
    body.update(kw)
    return c.post("/api/cameras", json=body)


def test_add_camera_starts_it_without_a_restart(client):
    c, eng = client
    r = _add(c)
    assert r.status_code == 201
    assert cid_in(c, "cam1")
    assert "cam1" in eng.workers          # running, not just stored


def cid_in(c, cid):
    return any(x["id"] == cid for x in c.get("/api/cameras").json())


def test_password_never_appears_in_any_response(client):
    """The assertion this whole phase exists to satisfy."""
    c, _ = client
    _add(c)
    for resp in (c.get("/api/cameras"),
                 c.patch("/api/cameras/cam1", json={"port": 8554}),
                 c.post("/api/cameras", json={"id": "cam2", "host": "10.0.0.5",
                                              "password": SECRET})):
        assert SECRET not in resp.text, resp.url


def test_listing_reports_password_presence_not_the_value(client):
    c, _ = client
    _add(c)
    entry = c.get("/api/cameras").json()[0]
    assert entry["has_password"] is True
    assert "password" not in entry
    assert "*****" in entry["url"]


def test_duplicate_id_conflicts(client):
    c, _ = client
    _add(c)
    assert _add(c).status_code == 409


def test_camera_with_no_source_is_rejected(client):
    c, eng = client
    assert c.post("/api/cameras", json={"id": "bad"}).status_code == 400
    assert eng.camera_store.list() == []   # nothing persisted


def test_id_is_required(client):
    c, _ = client
    assert c.post("/api/cameras", json={"host": "10.0.0.1"}).status_code == 400


def test_edit_restarts_the_worker(client):
    c, eng = client
    _add(c)
    first = eng.workers["cam1"]
    r = c.patch("/api/cameras/cam1", json={"host": "10.0.0.77"})
    assert r.status_code == 200
    assert eng.workers["cam1"] is not first
    assert eng.camera_store.get("cam1").host == "10.0.0.77"


def test_edit_keeps_the_password_when_not_resent(client):
    """The UI never receives the password, so it cannot echo it back on save."""
    c, eng = client
    _add(c)
    c.patch("/api/cameras/cam1", json={"port": 8554})
    assert eng.camera_store.get("cam1").password == SECRET


def test_delete_removes_from_store_and_engine(client):
    c, eng = client
    _add(c)
    assert c.delete("/api/cameras/cam1").status_code == 200
    assert eng.camera_store.get("cam1") is None
    assert "cam1" not in eng.workers
    assert c.delete("/api/cameras/cam1").status_code == 404


def test_edit_unknown_camera_404s(client):
    c, _ = client
    assert c.patch("/api/cameras/nope", json={"port": 1}).status_code == 404


def test_test_endpoint_does_not_save(client):
    c, eng = client
    r = c.post("/api/cameras/test", json={"id": "probe", "host": "127.0.0.1",
                                          "port": 1, "path": "/none"})
    assert r.status_code == 200
    assert r.json()["ok"] is False        # nothing listening on port 1
    assert eng.camera_store.list() == []  # and nothing was persisted


def test_test_endpoint_reports_bad_input_instead_of_raising(client):
    c, _ = client
    body = c.post("/api/cameras/test", json={"id": "probe"}).json()
    assert body["ok"] is False and "url, host or webcam" in body["error"]


# -- placement commissioning ------------------------------------------------

def test_commission_starts_and_reports_progress(client):
    c, eng = client
    _add(c)
    r = c.post("/api/cameras/cam1/commission", json={"seconds": 30})
    assert r.status_code == 200
    assert r.json()["verdict"] == "running"
    assert eng.workers["cam1"].commission is not None

    again = c.get("/api/cameras/cam1/commission")
    assert again.status_code == 200
    assert again.json()["camera_id"] == "cam1"


def test_commission_judges_against_the_cameras_own_gate(client):
    """A per-camera loosened gate must drive the verdict, or the wizard would
    grade an overhead camera against a threshold it never runs under."""
    c, eng = client
    _add(c, cid="overhead", tuning={"min_enroll_quality": 0.40})
    r = c.post("/api/cameras/overhead/commission", json={"seconds": 30})
    assert r.json()["gate"] == 0.40


def test_commission_on_unknown_camera_404s(client):
    c, _ = client
    assert c.post("/api/cameras/nope/commission",
                  json={"seconds": 5}).status_code == 404
    assert c.get("/api/cameras/nope/commission").status_code == 404


def test_reading_a_check_that_never_ran_404s(client):
    c, _ = client
    _add(c)
    assert c.get("/api/cameras/cam1/commission").status_code == 404


def test_commission_can_be_cancelled(client):
    c, eng = client
    _add(c)
    c.post("/api/cameras/cam1/commission", json={"seconds": 600})
    assert c.delete("/api/cameras/cam1/commission").status_code == 200
    assert eng.workers["cam1"].commission.report()["cancelled"] is True


def test_tuning_round_trips_through_add_and_edit(client):
    """model_copy(update=...) does not coerce, so a `tuning` dict arriving as
    JSON used to be stored raw and then fail the first time the camera asked
    it for thresholds. It must come back as a real CameraTuning."""
    from behavision.config import CameraTuning

    c, eng = client
    _add(c, cid="overhead", tuning={"min_enroll_quality": 0.40})
    assert c.get("/api/cameras").json()[0]["tuning"]["min_enroll_quality"] == 0.40

    r = c.patch("/api/cameras/overhead",
                json={"tuning": {"min_enroll_quality": 0.35}})
    assert r.status_code == 200
    stored = eng.camera_store.get("overhead")
    assert isinstance(stored.tuning, CameraTuning)
    assert stored.tuning.min_enroll_quality == 0.35


def test_an_inverted_per_camera_pair_is_rejected_not_stored(client):
    """enroll >= match would make one camera contradict every other one, and
    they all write into the same gallery."""
    c, _ = client
    r = _add(c, cid="bad",
             tuning={"enroll_threshold": 0.8, "match_threshold": 0.5})
    assert r.status_code == 400
    assert c.get("/api/cameras").json() == []


# "This computer's own camera" — the demo case, and a capability the engine
# has always had with nothing able to reach it.
#
# It matters most for showing the product to somebody. A laptop's own camera
# gives real recognition, of real faces, in the room, depending on no network
# at all — where pointing a demo machine at a camera in another building
# depends on two internet connections and a tunnel staying up while you talk.
def test_this_computers_own_camera_can_be_added(client):
    c, eng = client
    r = c.post("/api/cameras", json={"id": "laptop", "webcam": 0})
    assert r.status_code == 201, r.text
    assert "laptop" in eng.workers

    cam = next(x for x in c.get("/api/cameras").json() if x["id"] == "laptop")
    # Returned, or the form cannot tell a webcam camera from a half-filled
    # RTSP one when somebody opens it to edit.
    assert cam["webcam"] == 0
    assert eng.camera_store.get("laptop").source() == 0


def test_a_second_camera_index_is_kept(client):
    """0 is the built-in one; a plugged-in camera is usually 1. An index
    silently coerced to 0 would open the wrong camera and look like the
    setting had no effect."""
    c, eng = client
    assert c.post("/api/cameras", json={"id": "usb", "webcam": 1}).status_code == 201
    assert eng.camera_store.get("usb").source() == 1


def test_a_webcam_camera_survives_a_reload(client, tmp_path):
    """It has to be on disk, not only in the running engine: a demo that
    forgets its camera when the app restarts is worse than no demo."""
    c, _ = client
    c.post("/api/cameras", json={"id": "laptop", "webcam": 0})
    again = CameraStore(tmp_path / "cameras.json").get("laptop")
    assert again.source() == 0
    assert again.safe_url() == "webcam:0"