gmao/tests/integration/test_maintenance_engine.py

from datetime import date, time, timedelta
from uuid import uuid4

from app_new.extensions import db
from app_new.core.models.college import Building, Room
from app_new.core.models.equipment import Equipment, EquipmentCategory
from app_new.core.models.maintenance import Lot, LotTask
from app_new.core.models.planning import (
    CollegeClosure, ClosureWorkDay, Meter, ScheduledTask, Training, TrainingParticipant,
)
from app_new.core.services.maintenance_engine import generate_due_tasks
from app_new.scheduler.engine import get_working_hours
from app_new.core.models.user import User


def test_exceptional_vacation_day_has_its_specific_hours(app):
    with app.app_context():
        worked = date.today() + timedelta(days=20)
        closure = CollegeClosure(name=f"Vacances {uuid4().hex}", start_date=worked, end_date=worked, work_hours_type="none")
        db.session.add(closure); db.session.flush()
        db.session.add(ClosureWorkDay(closure_id=closure.id, work_date=worked, start_time=time(7, 30), end_time=time(14)))
        db.session.commit()
        assert get_working_hours(worked)[:2] == (time(7, 30), time(14))


def test_lot_task_engine_uses_individual_units_and_prevents_duplicates(app):
    with app.app_context():
        suffix = uuid4().hex[:7]
        building = Building(name=f"B {suffix}"); db.session.add(building); db.session.flush()
        room = Room(name="Local", building_id=building.id); category = EquipmentCategory(name=f"Cat {suffix}")
        lot = Lot(name=f"Lot {suffix}", category=category); db.session.add_all([room, category, lot]); db.session.flush()
        root = Equipment(name=f"Groupe {suffix}", lot=lot, category=category, is_group=True, tracked_individually=True)
        db.session.add(root); db.session.flush()
        unit = Equipment(name=f"Unité {suffix}", parent_id=root.id, room_id=room.id, tracked_individually=True)
        task = LotTask(lot=lot, tache="Contrôle", trigger_type="event", trigger_event="gel", is_active=True)
        db.session.add_all([unit, task]); db.session.commit()
        assert generate_due_tasks(task, event="orage") == []
        assert len(generate_due_tasks(task, event="gel")) == 1
        assert generate_due_tasks(task, event="gel") == []
        scheduled = ScheduledTask.query.filter_by(lot_task_id=task.id).one()
        assert scheduled.equipment_id == unit.id


def test_calendar_task_waits_for_interval_after_completion(app):
    with app.app_context():
        suffix = uuid4().hex[:7]
        building = Building(name=f"B cal {suffix}"); db.session.add(building); db.session.flush()
        room = Room(name="Local cal", building_id=building.id)
        category = EquipmentCategory(name=f"Cat cal {suffix}")
        lot = Lot(name=f"Lot cal {suffix}", category=category)
        equipment = Equipment(name=f"Eq cal {suffix}", lot=lot, category=category, room=room)
        task = LotTask(lot=lot, tache="Contrôle périodique", trigger_type="calendar",
                       jours_entre_interventions=30, advance_days=7, is_active=True)
        db.session.add_all([room, category, lot, equipment, task]); db.session.flush()
        completed = ScheduledTask(lot_task_id=task.id, equipment_id=equipment.id,
                                  scheduled_date=date.today(), status='completed')
        db.session.add(completed); db.session.commit()
        assert generate_due_tasks(task, today=date.today()) == []
        created = generate_due_tasks(task, today=date.today() + timedelta(days=23))
        assert len(created) == 1


def test_meter_trigger_uses_delta_since_last_maintenance(app):
    with app.app_context():
        suffix = uuid4().hex[:7]
        building = Building(name=f"B meter {suffix}"); db.session.add(building); db.session.flush()
        room = Room(name="Local meter", building_id=building.id)
        category = EquipmentCategory(name=f"Cat meter {suffix}")
        lot = Lot(name=f"Lot meter {suffix}", category=category)
        equipment = Equipment(name=f"Eq meter {suffix}", lot=lot, category=category, room=room)
        task = LotTask(lot=lot, tache="Compteur", trigger_type="meter", meter_threshold=100, is_active=True)
        db.session.add_all([room, category, lot, equipment, task]); db.session.flush()
        meter = Meter(equipment_id=equipment.id, name='Copies', current_value=1050,
                      initial_value=0, last_maintenance_value=1000, is_active=True)
        db.session.add(meter); db.session.commit()
        assert generate_due_tasks(task) == []
        meter.current_value = 1100; db.session.commit()
        assert len(generate_due_tasks(task)) == 1


def test_confirmed_training_blocks_maintenance_day(app):
    with app.app_context():
        suffix = uuid4().hex[:7]
        user = User(username=f'tech_{suffix}', email=f'{suffix}@test.local', full_name='Tech', role='technicien')
        user.set_password('test-password-strong')
        training = Training(name='Formation bloquante', start_date=date.today(), end_date=date.today())
        db.session.add_all([user, training]); db.session.flush()
        db.session.add(TrainingParticipant(training_id=training.id, user_id=user.id, is_confirmed=True))
        db.session.commit()
        assert get_working_hours(date.today(), user_id=user.id) is None