gmao/tests/integration/test_phase1_documents_planning.py

from datetime import date, datetime, time
import os

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, normalize_pronote_lessons, 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
        from flask import current_app
        intervention_doc.filepath = os.path.join(current_app.config["UPLOAD_FOLDER"], "..", "..", "etc", "passwd")
        db.session.commit()
        assert authenticated_client.get(f"/documents/download/intervention/{intervention_id}").status_code == 302
        assert authenticated_client.get("/documents/?source=intervention&document_type=rapport").status_code == 200
        assert "intervention.pdf" in authenticated_client.get("/documents/?q=Phase1%20intervention").get_data(as_text=True)
        assert "notice.pdf" not in authenticated_client.get("/documents/?source=intervention").get_data(as_text=True)
        assert authenticated_client.get("/documents/download/unknown/1").status_code == 403


def test_document_center_requires_source_permission(app, monkeypatch):
    from app_new.documents import service
    monkeypatch.setattr(service, "has_permission", lambda permission, user=None: permission == "documents.view")
    with app.app_context():
        assert service.list_documents(user=object()) == []


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)


def test_pronote_sync_change_removal_ambiguity_and_freshness(app):
    with app.app_context():
        building = Building(name="Phase15 edge building")
        db.session.add(building)
        db.session.flush()
        room = Room(name="Phase15 edge room", building_id=building.id)
        db.session.add(room)
        db.session.flush()
        week = date(2026, 9, 7)
        # Deux manuels compatibles rendent le rapprochement ambigu.
        manuals = [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,
        ) for _ in range(2)]
        db.session.add_all(manuals)
        db.session.commit()
        sync_at = datetime(2026, 9, 1, 8, 15)
        sync_pronote_schedules(room.id, [{
            "external_id": "edge-course", "week_start": week, "day_of_week": 0,
            "start_time": time(8, 30), "end_time": time(9, 30), "subject": "Maths",
        }], synced_at=sync_at)
        row = RoomSchedule.query.filter_by(room_id=room.id, external_id="edge-course").one()
        assert row.last_synced_at == sync_at
        assert row.conflict_note and "ambiguë" in row.conflict_note
        assert all(item.resolution_status == "active" for item in manuals)

        # Une synchronisation réussie sans ce créneau le désactive sans DELETE.
        sync_pronote_schedules(room.id, [], synced_at=datetime(2026, 9, 2, 8, 15))
        db.session.refresh(row)
        assert row.resolution_status == "disabled"
        assert db.session.get(RoomSchedule, row.id) is not None


def test_normalize_pronote_lessons_is_stable_and_skips_invalid_entries():
    week = date(2026, 9, 7)
    lessons = [{
        "day_of_week": 0, "start_time": "08:00", "end_time": "09:00",
        "subject": "Maths", "teacher": "Mme Test", "class_name": "6e A",
    }, {"day_of_week": 0, "start_time": "invalide", "end_time": "09:00"}]
    first = normalize_pronote_lessons(lessons, week)
    second = normalize_pronote_lessons(lessons, week)
    assert len(first) == 1
    assert first == second
    assert len(first[0]["external_id"]) == 64