90 lines
5.2 KiB
Python
90 lines
5.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, 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
|