283 lines
9.5 KiB
Python
283 lines
9.5 KiB
Python
"""Background clip-job endpoints (counter_dashboard /api/batch-clip/*).
|
|
|
|
Skipped when flask/openpyxl missing (CI). Worker runs through a monkeypatched
|
|
`counter_dashboard._clip_job_worker` seam (POST resolves the module global at
|
|
call time, so patching replaces the real ffmpeg worker).
|
|
"""
|
|
|
|
import os
|
|
import sqlite3
|
|
import threading
|
|
import time
|
|
|
|
import pytest
|
|
|
|
DATE = "2026-09-24"
|
|
BATCH = 7
|
|
START = "2026-09-24T14:25:09"
|
|
END = "2026-09-24T14:30:00"
|
|
KEY = f"{DATE}/{BATCH}"
|
|
|
|
|
|
@pytest.fixture()
|
|
def dash(tmp_path, monkeypatch):
|
|
pytest.importorskip("flask")
|
|
pytest.importorskip("openpyxl")
|
|
import counter_dashboard as cd
|
|
|
|
# isolate state files into tmp (mirrors tests/test_do_batch.py dash_client)
|
|
monkeypatch.setattr(cd, "BATCH_MODE_PATH", str(tmp_path / "batch_mode.json"))
|
|
monkeypatch.setattr(cd, "CURRENT_BATCH_PATH", str(tmp_path / "current_batch.json"))
|
|
monkeypatch.setattr(cd, "DO_SETTINGS_PATH", str(tmp_path / "do_settings.json"))
|
|
monkeypatch.setattr(cd, "DB_PATH", str(tmp_path / "clip_jobs.db"))
|
|
with cd._CLIP_JOBS_LOCK:
|
|
cd._CLIP_JOBS.clear()
|
|
cd._ensure_db()
|
|
client = cd.app.test_client()
|
|
yield client, cd, tmp_path
|
|
with cd._CLIP_JOBS_LOCK:
|
|
cd._CLIP_JOBS.clear()
|
|
|
|
|
|
def _office():
|
|
import counter_dashboard as cd
|
|
return {"Host": f"localhost:{cd.OFFICE_PORT}"}
|
|
|
|
|
|
def _insert_batch(cd, date=DATE, batch=BATCH, start=START, end=END):
|
|
conn = sqlite3.connect(cd.DB_PATH)
|
|
conn.execute(
|
|
"INSERT INTO batches (counting_date, batch_number, camera_name,"
|
|
" object_label, count, start_time, end_time) VALUES (?,?,?,?,?,?,?)",
|
|
(date, batch, "cam-1", "sack", 12, start, end),
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
|
|
def _job(cd, key=KEY):
|
|
with cd._CLIP_JOBS_LOCK:
|
|
return dict(cd._CLIP_JOBS.get(key) or {})
|
|
|
|
|
|
def _fake_worker(cd, calls, behavior="ready", release=None):
|
|
"""Stand-in for _clip_job_worker: ready | error | block(then ready)."""
|
|
def fake(key, date, batch_number, start_iso, end_iso, motion_url,
|
|
camera_id, out_path, pad, align):
|
|
calls.append(key)
|
|
if behavior == "block":
|
|
assert release is not None
|
|
release.wait(5)
|
|
with cd._CLIP_JOBS_LOCK:
|
|
job = cd._CLIP_JOBS.get(key)
|
|
if job is None:
|
|
return
|
|
if behavior == "error":
|
|
cd._unlink_quiet(out_path)
|
|
status, error = "error", "ffmpeg gagal: boom"
|
|
else:
|
|
with open(out_path, "wb") as fh:
|
|
fh.write(b"\x00\x00\x00\x18ftypmp42")
|
|
status, error = "ready", ""
|
|
with cd._CLIP_JOBS_LOCK:
|
|
job = cd._CLIP_JOBS.get(key)
|
|
if job is not None:
|
|
job["status"] = status
|
|
job["error"] = error
|
|
return fake
|
|
|
|
|
|
def _wait_status(client, want, key=KEY, timeout=5.0):
|
|
date, batch = key.split("/")
|
|
url = f"/api/batch-clip/{date}/{batch}/status"
|
|
deadline = time.time() + timeout
|
|
body = None
|
|
while time.time() < deadline:
|
|
res = client.get(url, headers=_office())
|
|
body = res.get_json()
|
|
if res.status_code == 200 and body.get("status") == want:
|
|
return body
|
|
time.sleep(0.02)
|
|
raise AssertionError(f"status never became {want!r}: {body}")
|
|
|
|
|
|
def _post(client):
|
|
return client.post(f"/api/batch-clip/{DATE}/{BATCH}", headers=_office())
|
|
|
|
|
|
def test_post_then_status_ready_then_file(dash, monkeypatch):
|
|
client, cd, tmp = dash
|
|
monkeypatch.setenv("MOTIONEYE_URL", "http://example.invalid")
|
|
_insert_batch(cd)
|
|
calls = []
|
|
monkeypatch.setattr(cd, "_clip_job_worker", _fake_worker(cd, calls))
|
|
|
|
res = _post(client)
|
|
assert res.status_code == 200
|
|
body = res.get_json()
|
|
assert body["success"] is True
|
|
assert body["status"] == "running"
|
|
assert body["job"] == KEY
|
|
|
|
st = _wait_status(client, "ready")
|
|
assert st["success"] is True
|
|
assert st["job"] == KEY
|
|
assert st["status"] == "ready"
|
|
assert st["error"] == ""
|
|
assert isinstance(st["elapsed"], int)
|
|
|
|
res = client.get(f"/api/batch-clip/{DATE}/{BATCH}/file", headers=_office())
|
|
assert res.status_code == 200
|
|
assert res.mimetype == "video/mp4"
|
|
assert "filename=b7-2026-09-24-14-25-09.mp4" in res.headers.get("Content-Disposition", "")
|
|
# temp kept for re-download (no cleanup after send)
|
|
entry = _job(cd)
|
|
assert entry["status"] == "ready"
|
|
assert os.path.isfile(entry["path"])
|
|
|
|
|
|
def test_duplicate_post_single_worker(dash, monkeypatch):
|
|
client, cd, tmp = dash
|
|
monkeypatch.setenv("MOTIONEYE_URL", "http://example.invalid")
|
|
_insert_batch(cd)
|
|
release = threading.Event()
|
|
calls = []
|
|
monkeypatch.setattr(cd, "_clip_job_worker",
|
|
_fake_worker(cd, calls, behavior="block", release=release))
|
|
|
|
res1 = _post(client)
|
|
assert res1.status_code == 200 and res1.get_json()["status"] == "running"
|
|
res2 = _post(client)
|
|
assert res2.status_code == 200
|
|
assert res2.get_json()["job"] == KEY
|
|
|
|
deadline = time.time() + 5
|
|
while not calls and time.time() < deadline:
|
|
time.sleep(0.02)
|
|
assert calls == [KEY] # worker started exactly once
|
|
|
|
res = client.get(f"/api/batch-clip/{DATE}/{BATCH}/file", headers=_office())
|
|
assert res.status_code == 409
|
|
assert res.get_json()["error"] == "masih diproses"
|
|
|
|
release.set()
|
|
_wait_status(client, "ready")
|
|
assert calls == [KEY] # still one invocation after completion
|
|
|
|
|
|
def test_pending_lists_job(dash, monkeypatch):
|
|
client, cd, tmp = dash
|
|
monkeypatch.setenv("MOTIONEYE_URL", "http://example.invalid")
|
|
_insert_batch(cd)
|
|
monkeypatch.setattr(cd, "_clip_job_worker", _fake_worker(cd, []))
|
|
_post(client)
|
|
_wait_status(client, "ready")
|
|
|
|
res = client.get("/api/batch-clip/pending", headers=_office())
|
|
assert res.status_code == 200
|
|
body = res.get_json()
|
|
assert body["success"] is True
|
|
assert len(body["jobs"]) == 1
|
|
job = body["jobs"][0]
|
|
assert job["date"] == DATE
|
|
assert job["batch_number"] == BATCH
|
|
assert job["start_time"] == START
|
|
assert job["status"] == "ready"
|
|
assert job["error"] == ""
|
|
|
|
|
|
def test_error_status_file_404_and_retry(dash, monkeypatch):
|
|
client, cd, tmp = dash
|
|
monkeypatch.setenv("MOTIONEYE_URL", "http://example.invalid")
|
|
_insert_batch(cd)
|
|
calls = []
|
|
monkeypatch.setattr(cd, "_clip_job_worker", _fake_worker(cd, calls, behavior="error"))
|
|
|
|
_post(client)
|
|
st = _wait_status(client, "error")
|
|
assert st["error"] == "ffmpeg gagal: boom"
|
|
|
|
res = client.get(f"/api/batch-clip/{DATE}/{BATCH}/file", headers=_office())
|
|
assert res.status_code == 404
|
|
assert res.get_json()["error"] == "ffmpeg gagal: boom"
|
|
|
|
res = _post(client) # retry replaces the errored job
|
|
assert res.status_code == 200
|
|
assert res.get_json()["status"] == "running"
|
|
_wait_status(client, "error")
|
|
assert calls == [KEY, KEY]
|
|
|
|
|
|
def test_error_worker_leaves_no_temp_file(dash, monkeypatch):
|
|
client, cd, tmp = dash
|
|
monkeypatch.setenv("MOTIONEYE_URL", "http://example.invalid")
|
|
_insert_batch(cd)
|
|
monkeypatch.setattr(cd, "_clip_job_worker", _fake_worker(cd, [], behavior="error"))
|
|
|
|
_post(client)
|
|
_wait_status(client, "error")
|
|
assert list((tmp / "tmp").glob("clip_*.mp4")) == []
|
|
|
|
|
|
def test_operator_host_403_all_endpoints(dash, monkeypatch):
|
|
client, cd, tmp = dash
|
|
monkeypatch.setenv("MOTIONEYE_URL", "http://example.invalid")
|
|
_insert_batch(cd)
|
|
operator = {"Host": "localhost:5000"}
|
|
|
|
assert client.post(f"/api/batch-clip/{DATE}/{BATCH}", headers=operator).status_code == 403
|
|
assert client.get(f"/api/batch-clip/{DATE}/{BATCH}/status", headers=operator).status_code == 403
|
|
assert client.get(f"/api/batch-clip/{DATE}/{BATCH}/file", headers=operator).status_code == 403
|
|
assert client.get("/api/batch-clip/pending", headers=operator).status_code == 403
|
|
|
|
|
|
def test_unknown_batch_404_and_missing_motioneye(dash, monkeypatch):
|
|
client, cd, tmp = dash
|
|
monkeypatch.setenv("MOTIONEYE_URL", "http://example.invalid")
|
|
_insert_batch(cd)
|
|
|
|
res = client.post(f"/api/batch-clip/{DATE}/999", headers=_office())
|
|
assert res.status_code == 404
|
|
assert "Batch tidak ditemukan" in res.get_json()["error"]
|
|
|
|
res = client.get(f"/api/batch-clip/{DATE}/999/status", headers=_office())
|
|
assert res.status_code == 404
|
|
assert res.get_json()["error"] == "not found"
|
|
|
|
monkeypatch.setenv("MOTIONEYE_URL", "")
|
|
res = _post(client)
|
|
assert res.status_code == 404
|
|
assert "belum dikonfigurasi" in res.get_json()["error"]
|
|
|
|
|
|
def test_sweep_drops_stale_entries_keeps_recent_error(dash):
|
|
client, cd, tmp = dash
|
|
tmp_dir = tmp / "tmp"
|
|
tmp_dir.mkdir()
|
|
old = tmp_dir / "clip_old.mp4"
|
|
old.write_bytes(b"x")
|
|
stale = time.time() - 7200
|
|
os.utime(old, (stale, stale))
|
|
fresh = tmp_dir / "clip_fresh.mp4"
|
|
fresh.write_bytes(b"x")
|
|
|
|
def entry(path, status, created):
|
|
return {"status": status, "date": DATE, "batch_number": BATCH,
|
|
"start_time": START, "error": "boom" if status == "error" else "",
|
|
"path": str(path), "created": created, "started": created}
|
|
|
|
with cd._CLIP_JOBS_LOCK:
|
|
cd._CLIP_JOBS.clear()
|
|
cd._CLIP_JOBS[f"{DATE}/7"] = entry(old, "ready", stale)
|
|
cd._CLIP_JOBS[f"{DATE}/8"] = entry(tmp_dir / "clip_gone.mp4", "error", time.time())
|
|
|
|
cd._clip_sweep()
|
|
|
|
with cd._CLIP_JOBS_LOCK:
|
|
keys = set(cd._CLIP_JOBS)
|
|
assert f"{DATE}/7" not in keys # >1 h → entry + temp dropped
|
|
assert f"{DATE}/8" in keys # recent error stays visible for retry
|
|
assert not old.exists()
|
|
assert fresh.exists() # fresh tmp untouched
|
|
|