Initial Plast Track MVP

backend/tests/test_events.py

@@ -0,0 +1,159 @@
+from datetime import datetime, timedelta, timezone
+
+import pytest
+from fastapi import HTTPException
+from sqlalchemy import create_engine, select
+from sqlalchemy.orm import Session, sessionmaker
+
+from app.database import Base
+from app.main import start_production_order
+from app.models import Cycle, Downtime, Machine, MachineEvent, ProductionOrder
+from app.schemas import MachineEventIn
+from app.services.events import _is_matching_edge, detect_missing_cycles, handle_machine_event, normalize_signal_value, parse_machine_event
+
+
+def test_parse_machine_event() -> None:
+    payload = """
+    {
+      "machine_id": "INJ-01",
+      "event_type": "digital_input_changed",
+      "timestamp": "2026-06-14T10:32:15Z",
+      "input_name": "cycle_signal",
+      "input_value": 1
+    }
+    """
+
+    event = parse_machine_event(payload)
+
+    assert event.machine_id == "INJ-01"
+    assert event.event_type == "digital_input_changed"
+    assert event.input_name == "cycle_signal"
+    assert event.input_value == 1
+
+
+def test_normalize_signal_value() -> None:
+    assert normalize_signal_value(True) is True
+    assert normalize_signal_value(0) is False
+    assert normalize_signal_value("high") is True
+    assert normalize_signal_value("OFF") is False
+    assert normalize_signal_value("unknown") is None
+
+
+def test_edge_matching() -> None:
+    assert _is_matching_edge(False, True, "rising") is True
+    assert _is_matching_edge(True, False, "falling") is True
+    assert _is_matching_edge(False, False, "rising") is False
+
+
+@pytest.fixture()
+def db_session() -> Session:
+    engine = create_engine("sqlite:///:memory:")
+    Base.metadata.create_all(engine)
+    TestingSessionLocal = sessionmaker(bind=engine)
+    session = TestingSessionLocal()
+    try:
+        yield session
+    finally:
+        session.close()
+        Base.metadata.drop_all(engine)
+
+
+def seed_running_machine(session: Session) -> Machine:
+    machine = Machine(
+        code="INJ-T1",
+        name="INJ-T1",
+        brand="Test",
+        model="Press",
+        powered_on=True,
+        cycle_signal_edge="rising",
+        debounce_ms=300,
+        min_cycle_time_sec=5,
+        max_cycle_time_sec=300,
+        stop_detection_delay_sec=180,
+    )
+    session.add(machine)
+    session.flush()
+    session.add(
+        ProductionOrder(
+            order_number="OF-T1",
+            machine_id=machine.id,
+            article_ref="ART-T1",
+            article_name="Test Part",
+            planned_qty=100,
+            cavities=1,
+            theoretical_cycle_time_sec=10,
+            status="running",
+            started_at=datetime(2026, 6, 14, 8, 0, tzinfo=timezone.utc),
+        )
+    )
+    session.commit()
+    return machine
+
+
+def test_cycle_signal_debounce_suppresses_noisy_edge(db_session: Session) -> None:
+    machine = seed_running_machine(db_session)
+    first_edge = datetime(2026, 6, 14, 8, 0, 0, tzinfo=timezone.utc)
+    noisy_edge = first_edge + timedelta(milliseconds=100)
+    db_session.add(
+        MachineEvent(
+            machine_id=machine.id,
+            event_type="digital_input_changed",
+            timestamp=first_edge,
+            payload={"input_name": "cycle_signal", "input_value": False},
+        )
+    )
+    db_session.commit()
+
+    result = handle_machine_event(
+        db_session,
+        MachineEventIn(
+            machine_id=machine.code,
+            event_type="digital_input_changed",
+            timestamp=noisy_edge,
+            input_name="cycle_signal",
+            input_value=True,
+        ),
+    )
+
+    cycle_count = db_session.execute(select(Cycle)).scalars().all()
+    db_session.refresh(machine)
+    assert cycle_count == []
+    assert machine.last_cycle_at is None
+    assert result["should_refresh"] is False
+
+
+def test_detect_missing_cycles_opens_unqualified_downtime(db_session: Session) -> None:
+    machine = seed_running_machine(db_session)
+    machine.last_cycle_at = datetime(2026, 6, 14, 8, 0, tzinfo=timezone.utc)
+    db_session.commit()
+
+    changes = detect_missing_cycles(db_session, now=datetime(2026, 6, 14, 8, 4, tzinfo=timezone.utc))
+
+    downtime = db_session.execute(select(Downtime).where(Downtime.machine_id == machine.id)).scalar_one()
+    db_session.refresh(machine)
+    assert changes == [{"machine_code": machine.code, "event_type": "machine_stopped", "status": "arret_non_qualifie"}]
+    assert downtime.reason_code is None
+    assert downtime.start_time.replace(tzinfo=timezone.utc) == datetime(2026, 6, 14, 8, 3, tzinfo=timezone.utc)
+    assert machine.status == "arret_non_qualifie"
+
+
+def test_start_order_rejects_second_running_order_on_machine(db_session: Session) -> None:
+    machine = seed_running_machine(db_session)
+    planned_order = ProductionOrder(
+        order_number="OF-T2",
+        machine_id=machine.id,
+        article_ref="ART-T2",
+        article_name="Second Part",
+        planned_qty=50,
+        cavities=1,
+        theoretical_cycle_time_sec=12,
+        status="planned",
+    )
+    db_session.add(planned_order)
+    db_session.commit()
+
+    with pytest.raises(HTTPException) as exc:
+        start_production_order(planned_order.id, db_session)
+
+    assert exc.value.status_code == 400
+    assert exc.value.detail == "Another order is already running on this machine"