120 lines
6.6 KiB
Python
120 lines
6.6 KiB
Python
|
from datetime import date, datetime, time
|
||
|
|
|
||
|
|
from app_new.extensions import db
|
||
|
|
from app_new.core.models.college import Building, Room, RoomSchedule, Zone
|
||
|
|
from app_new.core.models.cleaning import CommercialProduct, ProductGeneric, ProductDocument
|
||
|
|
from app_new.core.models.equipment import Equipment, EquipmentCategory, EquipmentDocument
|
||
|
|
from app_new.core.models.maintenance import Intervention, InterventionDocument
|
||
|
|
from app_new.core.models.rbac import Permission, RolePermission
|
||
|
|
from app_new.core.services.room_planning import active_room_schedules, room_is_available, sync_pronote_schedules
|
||
|
|
|
||
|
|
|
||
|
|
def test_document_center_aggregates_sources_and_download_handles_missing_file(authenticated_client, app):
|
||
|
|
with app.app_context():
|
||
|
|
building = Building(name="Phase1 bâtiment")
|
||
|
|
db.session.add(building)
|
||
|
|
db.session.flush()
|
||
|
|
zone = Zone(name="Phase1 zone", building_id=building.id)
|
||
|
|
room = Room(name="Phase1 salle", building_id=building.id, zone_id=zone.id)
|
||
|
|
category = EquipmentCategory(name="Phase1 catégorie")
|
||
|
|
db.session.add_all([zone, room, category])
|
||
|
|
db.session.flush()
|
||
|
|
equipment = Equipment(name="Phase1 équipement", room_id=room.id, category_id=category.id)
|
||
|
|
intervention = Intervention(title="Phase1 intervention", room_id=room.id, equipment=equipment)
|
||
|
|
generic = ProductGeneric(name="Phase1 produit")
|
||
|
|
db.session.add_all([equipment, intervention, generic])
|
||
|
|
db.session.flush()
|
||
|
|
commercial = CommercialProduct(generic_product_id=generic.id, commercial_name="Phase1 référence")
|
||
|
|
db.session.add(commercial)
|
||
|
|
db.session.flush()
|
||
|
|
intervention_doc = InterventionDocument(
|
||
|
|
intervention_id=intervention.id, filename="intervention.pdf",
|
||
|
|
filepath="/tmp/phase1-missing.pdf", document_type="rapport",
|
||
|
|
)
|
||
|
|
equipment_doc = EquipmentDocument(
|
||
|
|
equipment_id=equipment.id, filename="notice.pdf",
|
||
|
|
filepath="/tmp/phase1-missing-equipment.pdf", document_type="notice",
|
||
|
|
)
|
||
|
|
product_doc = ProductDocument(
|
||
|
|
commercial_product_id=commercial.id, title="FDS Phase1",
|
||
|
|
filename="fds.pdf", filepath="/tmp/phase1-missing-fds.pdf", document_type="fds",
|
||
|
|
)
|
||
|
|
db.session.add_all([intervention_doc, equipment_doc, product_doc])
|
||
|
|
db.session.commit()
|
||
|
|
intervention_id, equipment_id, product_id = intervention_doc.id, equipment_doc.id, product_doc.id
|
||
|
|
response = authenticated_client.get("/documents/")
|
||
|
|
assert response.status_code == 200
|
||
|
|
body = response.get_data(as_text=True)
|
||
|
|
assert "intervention.pdf" in body
|
||
|
|
assert "notice.pdf" in body
|
||
|
|
assert "fds.pdf" in body
|
||
|
|
assert authenticated_client.get("/documents/fds").status_code == 200
|
||
|
|
assert authenticated_client.get(f"/documents/download/intervention/{intervention_id}").status_code == 302
|
||
|
|
assert authenticated_client.get(f"/documents/download/equipment/{equipment_id}").status_code == 302
|
||
|
|
assert authenticated_client.get(f"/documents/download/product/{product_id}").status_code == 302
|
||
|
|
|
||
|
|
|
||
|
|
def test_room_schedule_replaces_only_unprotected_manual_slot(app):
|
||
|
|
with app.app_context():
|
||
|
|
building = Building(name="Planning phase1 building")
|
||
|
|
db.session.add(building)
|
||
|
|
db.session.flush()
|
||
|
|
room = Room(name="Planning phase1 room", building_id=building.id)
|
||
|
|
db.session.add(room)
|
||
|
|
db.session.flush()
|
||
|
|
week = date(2026, 9, 7)
|
||
|
|
manual = RoomSchedule(room_id=room.id, week_start=week, day_of_week=0,
|
||
|
|
start_time=time(8), end_time=time(9), subject="Maths",
|
||
|
|
source="manual", protected_from_sync=False)
|
||
|
|
protected = RoomSchedule(room_id=room.id, week_start=week, day_of_week=0,
|
||
|
|
start_time=time(10), end_time=time(11), subject="Réunion",
|
||
|
|
source="manual", protected_from_sync=True)
|
||
|
|
pronote = RoomSchedule(room_id=room.id, week_start=week, day_of_week=0,
|
||
|
|
start_time=time(8, 30), end_time=time(9, 30), subject="Maths",
|
||
|
|
source="pronote", external_id="course-1")
|
||
|
|
conflicting = RoomSchedule(room_id=room.id, week_start=week, day_of_week=0,
|
||
|
|
start_time=time(10, 30), end_time=time(11, 30), subject="Cours",
|
||
|
|
source="pronote", external_id="course-2")
|
||
|
|
db.session.add_all([manual, protected, pronote, conflicting])
|
||
|
|
db.session.commit()
|
||
|
|
rows = active_room_schedules(room.id, week)
|
||
|
|
ids = {row.id for row in rows}
|
||
|
|
assert manual.id not in ids
|
||
|
|
assert protected.id in ids
|
||
|
|
assert pronote.id in ids
|
||
|
|
assert conflicting.id in ids
|
||
|
|
assert room_is_available(room.id, datetime.combine(week, time(9, 30)), datetime.combine(week, time(10)))
|
||
|
|
assert not room_is_available(room.id, datetime.combine(week, time(10, 30)), datetime.combine(week, time(10, 45)))
|
||
|
|
|
||
|
|
|
||
|
|
def test_pronote_sync_supersedes_initial_manual_and_preserves_protected(app):
|
||
|
|
with app.app_context():
|
||
|
|
building = Building(name="Planning sync building")
|
||
|
|
db.session.add(building)
|
||
|
|
db.session.flush()
|
||
|
|
room = Room(name="Planning sync room", building_id=building.id)
|
||
|
|
db.session.add(room)
|
||
|
|
db.session.flush()
|
||
|
|
week = date(2026, 9, 7)
|
||
|
|
initial = RoomSchedule(room_id=room.id, week_start=week, day_of_week=0,
|
||
|
|
start_time=time(8), end_time=time(9), subject="Maths",
|
||
|
|
source="manual", protected_from_sync=False)
|
||
|
|
protected = RoomSchedule(room_id=room.id, week_start=week, day_of_week=0,
|
||
|
|
start_time=time(10), end_time=time(11), subject="Réunion",
|
||
|
|
source="manual", protected_from_sync=True)
|
||
|
|
db.session.add_all([initial, protected])
|
||
|
|
db.session.commit()
|
||
|
|
sync_pronote_schedules(room.id, [{
|
||
|
|
"external_id": "sync-course", "week_start": week, "day_of_week": 0,
|
||
|
|
"start_time": time(8, 30), "end_time": time(9, 30), "subject": "Maths",
|
||
|
|
}, {
|
||
|
|
"external_id": "sync-conflict", "week_start": week, "day_of_week": 0,
|
||
|
|
"start_time": time(10, 30), "end_time": time(11, 30), "subject": "Cours",
|
||
|
|
}])
|
||
|
|
db.session.refresh(initial)
|
||
|
|
db.session.refresh(protected)
|
||
|
|
rows = RoomSchedule.query.filter_by(room_id=room.id).all()
|
||
|
|
assert initial.resolution_status == "superseded"
|
||
|
|
assert protected.protected_from_sync is True
|
||
|
|
assert any(row.external_id == "sync-conflict" and row.conflict_note for row in rows)
|