feat: DO-gated manual batch mode (auto default, office-only switches)

tests/test_do_batch.py

@@ -0,0 +1,283 @@
+"""Smoke tests for DO-gated manual batch helpers (src/do_batch.py). Stdlib only."""
+
+import json
+from datetime import datetime, timedelta
+
+import pytest
+
+from src.do_batch import (
+    BATCH_MODES,
+    OCR_ENGINES,
+    classify_unit_token,
+    empty_do_settings,
+    group_dos_by_plate,
+    is_valid_batch_mode,
+    is_valid_ocr_engine,
+    net_counts,
+    normalize_batch_mode,
+    retention_cutoff_date,
+    should_discard_batch,
+    should_purge_counting_date,
+    start_gate_errors,
+)
+from src.do_ocr import parse_fields_from_text
+
+
+def test_normalize_batch_mode_defaults_auto():
+    assert normalize_batch_mode(None) == "auto"
+    assert normalize_batch_mode("bogus") == "auto"
+    assert normalize_batch_mode("do_manual") == "do_manual"
+    assert normalize_batch_mode("MANUAL") == "manual"
+    assert normalize_batch_mode("auto", "manual") == "auto"
+    for m in BATCH_MODES:
+        assert is_valid_batch_mode(m)
+    assert not is_valid_batch_mode("DO_ONLY")
+
+
+def test_net_counts_and_discard():
+    assert net_counts(10, 2, 5, 1) == (8, 4)
+    assert should_discard_batch(0, 0) is True
+    assert should_discard_batch(1, 0) is False
+    assert should_discard_batch(0, 1) is False
+    assert should_discard_batch(-1, 0) is False
+
+
+def test_classify_unit_tokens():
+    assert classify_unit_token("KRG") == "sack"
+    assert classify_unit_token("karung") == "sack"
+    assert classify_unit_token("DUS") == "box"
+    assert classify_unit_token("kardus") == "box"
+    assert classify_unit_token("BOX") == "box"
+    assert classify_unit_token("CTN") == "box"
+    assert classify_unit_token("kg") == "unknown"
+
+
+def test_group_dos_same_plate():
+    dos = [
+        {"no_do": "DO-1", "plate": "b 1234 xyz"},
+        {"no_do": "DO-2", "plate": "B 1234 XYZ"},
+    ]
+    plate, out = group_dos_by_plate(dos)
+    assert plate == "B 1234 XYZ"
+    assert [d["no_do"] for d in out] == ["DO-1", "DO-2"]
+
+
+def test_group_dos_mixed_plates_raises():
+    dos = [{"no_do": "A", "plate": "B 1"}, {"no_do": "B", "plate": "B 2"}]
+    with pytest.raises(ValueError, match="mixed_plates"):
+        group_dos_by_plate(dos)
+
+
+def test_group_dos_empty_plates_ok():
+    dos = [{"no_do": "A", "plate": ""}, {"no_do": "B", "plate": ""}]
+    plate, out = group_dos_by_plate(dos)
+    assert plate == ""
+    assert len(out) == 2
+
+
+def test_start_gate_errors_matrix():
+    good = {"no_do": "DO-1", "plate": "B 1234 XYZ"}
+    # auto wrong mode
+    assert "wrong_mode" in start_gate_errors("auto", [good])
+    assert "wrong_mode" in start_gate_errors("manual", [good])
+    # empty when do_manual + ok
+    assert start_gate_errors("do_manual", [good]) == []
+    # missing DO
+    assert "missing_do" in start_gate_errors("do_manual", [], require_do=True)
+    # missing no_do
+    assert "missing_no_do" in start_gate_errors(
+        "do_manual", [{"no_do": "", "plate": "B 1"}])
+    # plate optional: empty plate ok
+    assert start_gate_errors(
+        "do_manual", [{"no_do": "D", "plate": ""}], require_plate=False) == []
+    # plate required
+    assert "missing_plate" in start_gate_errors(
+        "do_manual", [{"no_do": "D", "plate": ""}], require_plate=True)
+    # mixed
+    assert "mixed_plates" in start_gate_errors(
+        "do_manual",
+        [{"no_do": "A", "plate": "B 1"}, {"no_do": "B", "plate": "B 2"}],
+    )
+    # batch active
+    assert "batch_active" in start_gate_errors(
+        "do_manual", [good], batch_active=True)
+
+
+def test_retention_cutoff_and_purge():
+    today = datetime(2026, 9, 24)
+    cutoff = retention_cutoff_date(7, today)
+    assert cutoff == "2026-09-17"
+    assert should_purge_counting_date("2026-09-16", cutoff) is True
+    assert should_purge_counting_date("2026-09-17", cutoff) is False  # equal kept
+    assert should_purge_counting_date("2026-09-18", cutoff) is False
+
+
+def test_ocr_engine_validation():
+    for e in OCR_ENGINES:
+        assert is_valid_ocr_engine(e)
+    assert not is_valid_ocr_engine("easyocr")
+
+
+def test_empty_do_settings_defaults():
+    s = empty_do_settings({"require_plate": False, "require_do": True, "ocr_engine": "tesseract"})
+    assert s["ocr_engine"] == "tesseract"
+    assert s["require_do"] is True
+    bad = empty_do_settings({"ocr_engine": "nope"})
+    assert bad["ocr_engine"] == "tesseract"
+
+
+def test_parse_fields_from_text():
+    text = "No. DO: DO-001\nPlat: B 1234 XYZ\n80 KRG\n10 DUS\n"
+    fields = parse_fields_from_text(text)
+    assert fields["no_do"] == "DO-001"
+    assert fields["plate"].startswith("B 1234")
+    assert fields["expected_sack"] == 80
+    assert fields["expected_box"] == 10
+
+
+def test_extract_do_fields_none_engine(tmp_path):
+    from src.do_ocr import extract_do_fields
+
+    # no real image needed for none
+    out = extract_do_fields(str(tmp_path / "x.jpg"), "none")
+    assert out["ocr_ok"] is False
+    assert out["ocr_error"] == "disabled"
+    assert out["engine"] == "none"
+
+
+# ---------------------------------------------------------------------------
+# Flask API auth tests — skipped when flask/openpyxl not installed (CI)
+# ---------------------------------------------------------------------------
+
+@pytest.fixture()
+def dash_client(tmp_path, monkeypatch):
+    pytest.importorskip("flask")
+    pytest.importorskip("openpyxl")
+    import counter_dashboard as cd
+
+    # isolate state files into tmp
+    mode_path = tmp_path / "batch_mode.json"
+    state_path = tmp_path / "current_batch.json"
+    settings_path = tmp_path / "do_settings.json"
+    monkeypatch.setattr(cd, "BATCH_MODE_PATH", str(mode_path))
+    monkeypatch.setattr(cd, "CURRENT_BATCH_PATH", str(state_path))
+    monkeypatch.setattr(cd, "DO_SETTINGS_PATH", str(settings_path))
+    client = cd.app.test_client()
+    return client, mode_path, state_path, settings_path
+
+
+def _office(client, **kw):
+    return client.post("/api/batch/mode", json=kw,
+                       headers={"Host": f"localhost:{cd_office_port()}"})
+
+
+def cd_office_port():
+    import counter_dashboard as cd
+    return cd.OFFICE_PORT
+
+
+def test_mode_post_operator_403(dash_client):
+    client, mode_path, _, _ = dash_client
+    res = client.post("/api/batch/mode", json={"mode": "do_manual"},
+                      headers={"Host": "localhost:5000"})
+    assert res.status_code == 403
+
+
+def test_mode_post_office_200_and_invalid_400(dash_client):
+    client, mode_path, _, _ = dash_client
+    office = {"Host": f"localhost:{cd_office_port()}"}
+    res = client.post("/api/batch/mode", json={"mode": "do_manual"}, headers=office)
+    assert res.status_code == 200
+    assert res.get_json()["mode"] == "do_manual"
+    assert mode_path.exists()
+    assert json.loads(mode_path.read_text())["mode"] == "do_manual"
+    res = client.post("/api/batch/mode", json={"mode": "nope"}, headers=office)
+    assert res.status_code == 400
+
+
+def test_mode_post_409_when_batch_active(dash_client):
+    client, _, state_path, _ = dash_client
+    office = {"Host": f"localhost:{cd_office_port()}"}
+    state_path.write_text(json.dumps({
+        "counting_date": "2026-09-24", "batch_number": 1,
+        "count": 1, "start_time": "x",
+    }))
+    res = client.post("/api/batch/mode", json={"mode": "auto"}, headers=office)
+    assert res.status_code == 409
+
+
+def test_model_mode_post_operator_403(dash_client):
+    client, _, _, _ = dash_client
+    res = client.post("/api/batch/mode", json={"model_mode": "B"},
+                      headers={"Host": "localhost:5000"})
+    assert res.status_code == 403
+
+
+def test_ocr_engine_post_both_ports(dash_client):
+    client, _, _, settings_path = dash_client
+    res = client.post("/api/do/settings", json={"ocr_engine": "paddle"},
+                      headers={"Host": "localhost:5000"})
+    assert res.status_code == 200
+    assert res.get_json()["ocr_engine"] == "paddle"
+    res = client.post("/api/do/settings", json={"ocr_engine": "tesseract"},
+                      headers={"Host": f"localhost:{cd_office_port()}"})
+    assert res.status_code == 200
+    res = client.post("/api/do/settings", json={"ocr_engine": "bogus"},
+                      headers={"Host": "localhost:5000"})
+    assert res.status_code == 400
+
+
+def test_require_plate_post_operator_403(dash_client):
+    client, _, _, _ = dash_client
+    res = client.post("/api/do/settings", json={"require_plate": True},
+                      headers={"Host": "localhost:5000"})
+    assert res.status_code == 403
+    office = {"Host": f"localhost:{cd_office_port()}"}
+    res = client.post("/api/do/settings", json={"require_plate": True}, headers=office)
+    assert res.status_code == 200
+    assert res.get_json()["require_plate"] is True
+
+
+def test_start_stop_auto_409(dash_client):
+    client, mode_path, _, _ = dash_client
+    office = {"Host": f"localhost:{cd_office_port()}"}
+    client.post("/api/batch/mode", json={"mode": "auto"}, headers=office)
+    res = client.post("/api/batch/start", headers={"Host": "localhost:5000"})
+    assert res.status_code == 409
+    res = client.post("/api/batch/stop", headers={"Host": "localhost:5000"})
+    assert res.status_code == 409
+
+
+def test_stop_discard_both_nets_zero(dash_client):
+    client, mode_path, state_path, _ = dash_client
+    office = {"Host": f"localhost:{cd_office_port()}"}
+    client.post("/api/batch/mode", json={"mode": "manual"}, headers=office)
+    state_path.write_text(json.dumps({
+        "counting_date": "2026-09-24", "batch_number": 99,
+        "count": 0, "box_count": 0, "box_unloading": 0, "unloading": 0,
+        "start_time": "2026-09-24T00:00:00",
+        "model_mode": "C",
+    }))
+    res = client.post("/api/batch/stop", headers={"Host": "localhost:5000"})
+    body = res.get_json()
+    assert res.status_code == 200
+    assert body["discarded"] is True
+    assert body["net_sack"] == 0 and body["net_box"] == 0
+    assert not state_path.exists()
+
+
+def test_stop_keeps_box_only_batch(dash_client):
+    client, mode_path, state_path, _ = dash_client
+    office = {"Host": f"localhost:{cd_office_port()}"}
+    client.post("/api/batch/mode", json={"mode": "manual"}, headers=office)
+    state_path.write_text(json.dumps({
+        "counting_date": "2026-09-24", "batch_number": 98,
+        "count": 0, "box_count": 5, "box_unloading": 1, "unloading": 0,
+        "start_time": "2026-09-24T00:00:00",
+        "model_mode": "C",
+    }))
+    res = client.post("/api/batch/stop", headers={"Host": "localhost:5000"})
+    body = res.get_json()
+    assert res.status_code == 200
+    assert body["discarded"] is False
+    assert body["net_box"] == 4