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 run_simulator_scenario, start_production_order from app.models import Cycle, Downtime, Machine, MachineEvent, ProductionOrder from app.schemas import MachineEventIn, SimulatorScenarioCreate 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" def test_simulator_cycle_burst_uses_production_event_logic(db_session: Session) -> None: machine = seed_running_machine(db_session) result = run_simulator_scenario( SimulatorScenarioCreate( machine_id=machine.code, scenario="cycle_burst", cycle_time_sec=12, count=3, ), db_session, ) cycles = db_session.execute(select(Cycle).where(Cycle.machine_id == machine.id)).scalars().all() assert result["scenario"] == "cycle_burst" assert len(result["events"]) == 3 assert len(cycles) == 3 assert {cycle.cycle_time_sec for cycle in cycles} == {12} def test_simulator_debounce_noise_is_ignored(db_session: Session) -> None: machine = seed_running_machine(db_session) result = run_simulator_scenario( SimulatorScenarioCreate( machine_id=machine.code, scenario="debounce_noise", cycle_time_sec=12, ), db_session, ) cycles = db_session.execute(select(Cycle).where(Cycle.machine_id == machine.id)).scalars().all() assert result["scenario"] == "debounce_noise" assert len(result["events"]) == 2 assert cycles == [] def test_simulator_passive_edge_uses_requested_real_cycle_time(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() result = run_simulator_scenario( SimulatorScenarioCreate( machine_id=machine.code, scenario="cycle_signal_edge", cycle_time_sec=12, ), db_session, ) cycle = db_session.execute(select(Cycle).where(Cycle.machine_id == machine.id)).scalar_one() assert result["scenario"] == "cycle_signal_edge" assert len(result["events"]) == 2 assert cycle.cycle_time_sec == 12