letteka
f19ccf349b
Reviewed-on: #29 Co-authored-by: Andrew Kettel <andrew.kettel@gmail.com> Co-committed-by: Andrew Kettel <andrew.kettel@gmail.com>
166 lines
5.5 KiB
Python
166 lines
5.5 KiB
Python
from collections.abc import Generator
|
|
from typing import Any
|
|
|
|
import pytest
|
|
from flask.testing import FlaskClient
|
|
|
|
from src.app import app, db
|
|
|
|
|
|
@pytest.fixture
|
|
def client() -> Generator[FlaskClient, Any, Any]:
|
|
app.config["TESTING"] = True
|
|
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///:memory:"
|
|
with app.app_context():
|
|
db.create_all()
|
|
yield app.test_client()
|
|
db.drop_all()
|
|
|
|
|
|
# ── Stream tests ─────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_heartbeat_status_code(client: FlaskClient) -> None:
|
|
response = client.get("/heartbeat")
|
|
assert response.status_code == 200
|
|
|
|
|
|
def test_camera_start(client: FlaskClient) -> None:
|
|
response = client.post("/camera/start")
|
|
assert response.status_code == 200
|
|
assert response.get_json()["status"] in ("started", "already_running")
|
|
|
|
|
|
def test_camera_stop(client: FlaskClient) -> None:
|
|
client.post("/camera/start")
|
|
response = client.post("/camera/stop")
|
|
assert response.status_code == 200
|
|
assert response.get_json()["status"] in ("stopped", "already_stopped")
|
|
|
|
|
|
def test_double_start_is_idempotent(client: FlaskClient) -> None:
|
|
client.post("/camera/start")
|
|
res = client.post("/camera/start")
|
|
assert res.get_json()["status"] == "already_running"
|
|
|
|
|
|
def test_camera_status(client: FlaskClient) -> None:
|
|
response = client.get("/camera/status")
|
|
assert response.status_code == 200
|
|
data = response.get_json()
|
|
assert "running" in data
|
|
assert "ready" in data
|
|
assert "updated_at" in data
|
|
# recording fields must always be present
|
|
assert "recording" in data
|
|
assert "recording_started_at" in data
|
|
|
|
|
|
def test_camera_status_includes_recording_false_by_default(client: FlaskClient) -> None:
|
|
res = client.get("/camera/status")
|
|
data = res.get_json()
|
|
assert data["recording"] is False
|
|
assert data["recording_started_at"] is None
|
|
|
|
|
|
def test_camera_log(client: FlaskClient) -> None:
|
|
client.post("/camera/start")
|
|
client.post("/camera/stop")
|
|
res = client.get("/camera/log")
|
|
assert res.status_code == 200
|
|
events = res.get_json()
|
|
assert len(events) == 2
|
|
assert events[0]["action"] == "stop" # most recent first
|
|
assert events[1]["action"] == "start"
|
|
assert "ip_address" in events[0]
|
|
assert "timestamp" in events[0]
|
|
|
|
|
|
def test_hls_404_when_not_running(client: FlaskClient) -> None:
|
|
response = client.get("/camera/hls/stream.m3u8")
|
|
assert response.status_code == 404
|
|
|
|
|
|
def test_index_page(client: FlaskClient) -> None:
|
|
response = client.get("/")
|
|
assert response.status_code == 200
|
|
assert b"Pi Camera" in response.data
|
|
|
|
|
|
# ── Recording tests ───────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_record_start_returns_200(client: FlaskClient) -> None:
|
|
res = client.post("/camera/record/start")
|
|
assert res.status_code == 200
|
|
data = res.get_json()
|
|
assert data["status"] in ("recording", "already_recording")
|
|
|
|
|
|
def test_record_stop_when_not_recording(client: FlaskClient) -> None:
|
|
res = client.post("/camera/record/stop")
|
|
assert res.status_code == 200
|
|
assert res.get_json()["status"] == "not_recording"
|
|
|
|
|
|
def test_record_start_then_stop(client: FlaskClient) -> None:
|
|
start = client.post("/camera/record/start")
|
|
assert start.get_json()["status"] == "recording"
|
|
|
|
stop = client.post("/camera/record/stop")
|
|
assert stop.status_code == 200
|
|
assert stop.get_json()["status"] == "stopped"
|
|
|
|
|
|
def test_double_record_start_is_idempotent(client: FlaskClient) -> None:
|
|
client.post("/camera/record/start")
|
|
res = client.post("/camera/record/start")
|
|
assert res.get_json()["status"] == "already_recording"
|
|
# clean up
|
|
client.post("/camera/record/stop")
|
|
|
|
|
|
def test_status_reflects_recording_state(client: FlaskClient) -> None:
|
|
client.post("/camera/record/start")
|
|
status = client.get("/camera/status").get_json()
|
|
assert status["recording"] is True
|
|
assert status["recording_started_at"] is not None
|
|
|
|
client.post("/camera/record/stop")
|
|
status = client.get("/camera/status").get_json()
|
|
assert status["recording"] is False
|
|
assert status["recording_started_at"] is None
|
|
|
|
|
|
def test_stream_and_record_simultaneously(client: FlaskClient) -> None:
|
|
"""Stream and record can be active at the same time without interference."""
|
|
client.post("/camera/start")
|
|
rec = client.post("/camera/record/start")
|
|
assert rec.get_json()["status"] in ("recording", "already_recording")
|
|
|
|
status = client.get("/camera/status").get_json()
|
|
assert status["running"] is True
|
|
assert status["recording"] is True
|
|
|
|
# stopping stream does not affect recording state report from camera object
|
|
client.post("/camera/stop")
|
|
# recording was stopped as part of camera.stop() — that is expected behaviour
|
|
# (the camera itself cleans up on stop)
|
|
|
|
|
|
def test_list_recordings_empty(client: FlaskClient) -> None:
|
|
res = client.get("/camera/recordings")
|
|
assert res.status_code == 200
|
|
assert res.get_json() == []
|
|
|
|
|
|
def test_download_recording_invalid_filename(client: FlaskClient) -> None:
|
|
# filename pattern not matching should 404
|
|
res = client.get("/camera/recordings/../../etc/passwd")
|
|
assert res.status_code in (404, 400)
|
|
|
|
|
|
def test_download_recording_wrong_prefix(client: FlaskClient) -> None:
|
|
res = client.get("/camera/recordings/evil.mp4")
|
|
assert res.status_code == 404
|