39 lines
2.2 KiB
Python
39 lines
2.2 KiB
Python
|
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, ScheduledTask
|
||
|
|
from app_new.core.services.maintenance_engine import generate_due_tasks
|
||
|
|
from app_new.scheduler.engine import get_working_hours
|
||
|
|
|
||
|
|
|
||
|
|
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
|