ScoDoc/tests/unit/yaml_setup.py

##############################################################################
# ScoDoc
# Copyright (c) 1999 - 2023 Emmanuel Viennet.  All rights reserved.
# See LICENSE
##############################################################################

"""
Met en place une base pour les tests unitaires, à partir d'une description
YAML qui peut donner la formation, son ref. compétences, les formsemestres,
les étudiants et leurs notes et décisions de jury.

Le traitement est effectué dans l'ordre suivant:

setup_from_yaml()

    - setup_but_formation():
        - import de la formation (le test utilise une seule formation)
    - associe_ues_et_parcours():
        - crée les associations formation <-> référentiel de compétence
    - setup_formsemestres()
        - crée les formsemestres décrits dans le YAML
        avec tous les modules du semestre ou des parcours si indiqués
        et une évaluation dans chaque moduleimpl.
    - inscrit_les_etudiants()
        - inscrit et place dans les groupes de parcours
    - note_les_modules()
        - saisie de toutes les notes indiquées dans le YAML

check_deca_fields()
    - vérifie les champs du deca (nb UE, compétences, ...) mais pas les décisions de jury.

formsemestre_validation_auto_but(only_adm=False)
    - enregistre toutes les décisions "par défaut" proposées (pas seulement les ADM)

test_but_jury()
    - compare décisions attendues indiquées dans le YAML avec celles de ScoDoc
     et enregistre immédiatement APRES la décision manuelle indiquée par `decision_jury`
     dans le YAML.


Les tests unitaires associés sont généralement lents (construction de la base),
et donc marqués par `@pytest.mark.slow`.
"""

import os
from pathlib import Path
import re

import yaml
from flask import current_app, g

from app import db

from app.auth.models import User
from app.but.import_refcomp import orebut_import_refcomp
from app.but.jury_but import (
    DecisionsProposeesAnnee,
    DecisionsProposeesRCUE,
    DecisionsProposeesUE,
)
from app.models import (
    ApcNiveau,
    ApcParcours,
    ApcReferentielCompetences,
    Evaluation,
    Formation,
    FormSemestre,
    Identite,
    Module,
    ModuleImpl,
    ModuleUECoef,
    UniteEns,
)

from app.scodoc import sco_formations
from app.scodoc import sco_formsemestre_inscriptions
from app.scodoc import sco_groups
from app.scodoc import sco_saisie_notes
from app.scodoc import sco_utils as scu

from tests.conftest import RESOURCES_DIR


def setup_but_formation(formation_infos: dict) -> Formation:
    """Importe la formation, qui est lue à partir du fichier XML
    formation_infos["filename"].
    La formation peut être quelconque, on vérifie juste qu'elle est bien créée.
    """
    # Lecture fichier XML local:
    with open(
        os.path.join(RESOURCES_DIR, "formations", formation_infos["filename"]),
        encoding="utf-8",
    ) as f:
        doc = f.read()

    # --- Création de la formation
    formation_id, _, _ = sco_formations.formation_import_xml(doc)
    formation: Formation = Formation.query.get(formation_id)
    assert formation
    return formation


def setup_formation_referentiel(formation: Formation, refcomp_infos: dict):
    """Associe la formation au référentiel de compétences"""
    if not refcomp_infos:
        return
    refcomp_filename = refcomp_infos["filename"]
    refcomp_specialite = refcomp_infos["specialite"]
    # --- Chargement Référentiel
    if (
        ApcReferentielCompetences.query.filter_by(
            scodoc_orig_filename=refcomp_filename, dept_id=g.scodoc_dept_id
        ).first()
        is None
    ):
        # pas déjà chargé
        filename = (
            Path(current_app.config["SCODOC_DIR"])
            / "ressources/referentiels/but2022/competences"
            / refcomp_filename
        )
        with open(filename, encoding="utf-8") as f:
            xml_data = f.read()
            referentiel_competence = orebut_import_refcomp(
                xml_data, dept_id=g.scodoc_dept_id, orig_filename=Path(filename).name
            )
            assert referentiel_competence

    # --- Association au référentiel de compétences
    referentiel_competence = ApcReferentielCompetences.query.filter_by(
        specialite=refcomp_specialite
    ).first()  # le recherche à nouveau (test)
    assert referentiel_competence
    formation.referentiel_competence_id = referentiel_competence.id
    db.session.add(formation)


def associe_ues_et_parcours(formation: Formation, formation_infos: dict):
    """Associe les UE et modules de la formation aux parcours du ref. comp."""
    referentiel_competence = formation.referentiel_competence
    if not referentiel_competence:
        return
    # --- Association des UEs aux parcours niveaux de compétences
    for ue_acronyme, ue_infos in formation_infos["ues"].items():
        ue: UniteEns = formation.ues.filter_by(acronyme=ue_acronyme).first()
        assert ue is not None  # l'UE doit exister dans la formation avec cet acronyme
        # Parcours:
        if ue_infos.get("parcours", False):
            parcour = referentiel_competence.parcours.filter_by(
                code=ue_infos["parcours"]
            ).first()
            assert parcour is not None  # le parcours indiqué pour cette UE doit exister
            ue.set_parcour(parcour)

        # Niveaux compétences:
        competence = referentiel_competence.competences.filter_by(
            titre=ue_infos["competence"]
        ).first()
        assert competence is not None  # La compétence de titre indiqué doit exister
        niveau: ApcNiveau = competence.niveaux.filter_by(
            annee=ue_infos["annee"]
        ).first()
        assert niveau is not None  # le niveau de l'année indiquée doit exister
        ue.set_niveau_competence(niveau)

    db.session.commit()
    associe_modules_et_parcours(formation, formation_infos)


def associe_modules_et_parcours(formation: Formation, formation_infos: dict):
    """Associe les modules à des parcours, grâce au champ modules_parcours"""
    for code_parcours, codes_modules in formation_infos.get(
        "modules_parcours", {}
    ).items():
        parcour = formation.referentiel_competence.parcours.filter_by(
            code=code_parcours
        ).first()
        assert parcour is not None  # code parcours doit exister dans le ref. comp.
        for code_module in codes_modules:
            for module in [
                module
                for module in formation.modules
                if re.match(code_module, module.code)
            ]:
                module.parcours.append(parcour)
                db.session.add(module)
    db.session.commit()


def _un_semestre(
    formation: Formation,
    parcours: list[ApcParcours],
    semestre_id: int,
    titre: str,
    date_debut: str,
    date_fin: str,
) -> FormSemestre:
    "Création d'un formsemestre"
    formsemestre = FormSemestre(
        formation=formation,
        parcours=parcours,
        dept_id=g.scodoc_dept_id,
        titre=titre,
        semestre_id=semestre_id,
        date_debut=date_debut,
        date_fin=date_fin,
    )
    # set responsable (list)
    a_user = User.query.first()
    formsemestre.responsables = [a_user]
    db.session.add(formsemestre)
    # Ajoute tous les modules du semestre sans parcours OU avec l'un des parcours indiqués
    sem_parcours_ids = {p.id for p in parcours}
    modules = [
        m
        for m in formsemestre.formation.modules.filter_by(semestre_id=semestre_id)
        if (not m.parcours) or ({p.id for p in m.parcours} & sem_parcours_ids)
    ]
    for module in modules:
        modimpl = ModuleImpl(module=module, responsable_id=a_user.id)
        db.session.add(modimpl)
        formsemestre.modimpls.append(modimpl)

    # Crée une évaluation dans chaque module
    create_evaluations(formsemestre)

    # Partition par défaut:
    db.session.commit()
    partition_id = sco_groups.partition_create(
        formsemestre.id, default=True, redirect=False
    )
    _ = sco_groups.create_group(partition_id, default=True)
    # Partition de parcours:
    formsemestre.setup_parcours_groups()

    return formsemestre


def create_evaluations(formsemestre: FormSemestre):
    """Crée une évaluation dans chaque module du semestre"""
    for modimpl in formsemestre.modimpls:
        evaluation = Evaluation(
            moduleimpl=modimpl,
            jour=formsemestre.date_debut,
            description=f"Exam {modimpl.module.titre}",
            coefficient=1.0,
            note_max=20.0,
            numero=1,
        )
        db.session.add(evaluation)
        # Affecte les mêmes poids que les coefs APC:
        ue_coef_dict = modimpl.module.get_ue_coef_dict()  # { ue_id : coef }
        evaluation.set_ue_poids_dict(ue_coef_dict)


def note_les_modules(doc: dict):
    """Saisie les notes des étudiants"""
    a_user = User.query.first()
    for nom, infos in doc["Etudiants"].items():
        etud: Identite = Identite.query.filter_by(nom=nom).first()
        assert etud is not None
        for titre, sem_infos in infos["formsemestres"].items():
            formsemestre: FormSemestre = FormSemestre.query.filter_by(
                titre=titre
            ).first()
            assert formsemestre is not None
            for code_module, note in sem_infos.get("notes_modules", {}).items():
                modimpl = (
                    formsemestre.modimpls.join(Module)
                    .filter_by(code=code_module)
                    .first()
                )
                # le sem. doit avoir un module du code indiqué:
                if modimpl is None:
                    raise ValueError(f"module de code '{code_module}' introuvable")
                for evaluation in modimpl.evaluations:
                    # s'il y a plusieurs evals, affecte la même note à chacune
                    note_value, invalid = sco_saisie_notes.convert_note_from_string(
                        str(note),
                        20.0,
                        note_min=scu.NOTES_MIN,
                        etudid=etud.id,
                        absents=[],
                        tosuppress=[],
                        invalids=[],
                    )
                    assert not invalid  # valeur note invalide
                    assert isinstance(note_value, float)
                    sco_saisie_notes.notes_add(
                        a_user,
                        evaluation.id,
                        [(etud.id, note_value)],
                        comment="note_les_modules",
                    )


def setup_formsemestres(formation: Formation, doc: str):
    """Création des formsemestres pour tester les parcours BUT"""
    for titre, infos in doc["FormSemestres"].items():
        parcours = []
        for code_parcour in infos.get("codes_parcours", []):
            parcour = formation.referentiel_competence.parcours.filter_by(
                code=code_parcour
            ).first()
            assert parcour is not None
            parcours.append(parcour)
        _ = _un_semestre(
            formation,
            parcours,
            infos["idx"],
            titre,
            infos["date_debut"],
            infos["date_fin"],
        )

    db.session.flush()
    assert FormSemestre.query.count() == len(doc["FormSemestres"])


def inscrit_les_etudiants(formation: Formation, doc: dict):
    """Inscrit les étudiants dans chacun de leurs formsemestres"""
    for nom, infos in doc["Etudiants"].items():
        etud = Identite.create_etud(
            dept_id=g.scodoc_dept_id,
            nom=nom,
            prenom=infos.get("prenom", "prénom"),
            civilite=infos.get("civilite", "X"),
        )
        db.session.add(etud)
        db.session.commit()
        # L'inscrire à ses formsemestres
        for titre, sem_infos in infos["formsemestres"].items():
            formsemestre: FormSemestre = FormSemestre.query.filter_by(
                titre=titre
            ).first()
            assert formsemestre is not None
            partition_parcours = formsemestre.partitions.filter_by(
                partition_name=scu.PARTITION_PARCOURS
            ).first()
            if partition_parcours is not None and "parcours" in sem_infos:
                group = partition_parcours.groups.filter_by(
                    group_name=sem_infos["parcours"]
                ).first()
                assert group is not None  # le groupe de parcours doit exister
                group_ids = [group.id]
            else:
                group_ids = []
            sco_formsemestre_inscriptions.do_formsemestre_inscription_with_modules(
                formsemestre.id,
                etud.id,
                group_ids=group_ids,
                etat=scu.INSCRIT,
                method="tests/unit/inscrit_les_etudiants",
            )
    # Met à jour les inscriptions:
    for formsemestre in formation.formsemestres:
        formsemestre.update_inscriptions_parcours_from_groups()


def setup_from_yaml(filename: str) -> dict:
    """Lit le fichier yaml et construit l'ensemble des objets"""
    with open(filename, encoding="utf-8") as f:
        doc = yaml.safe_load(f.read())

    formation = setup_but_formation(doc["Formation"])
    setup_formation_referentiel(formation, doc.get("ReferentielCompetences", {}))
    associe_ues_et_parcours(formation, doc["Formation"])
    setup_formsemestres(formation, doc)
    inscrit_les_etudiants(formation, doc)
    note_les_modules(doc)
    return doc


def _check_codes_jury(codes: list[str], codes_att: list[str]):
    """Vérifie (assert) la liste des codes
    l'ordre n'a pas d'importance ici.
    Si codes_att contient un "...", on se contente de vérifie que
    les codes de codes_att sont tous présents dans codes.
    """
    codes_set = set(codes)
    codes_att_set = set(codes_att)
    if "..." in codes_att_set:
        codes_att_set.remove("...")
        assert codes_att_set.issubset(codes_set)
    else:
        assert codes_att_set == codes_set


def _check_decisions_ues(
    decisions_ues: dict[int, DecisionsProposeesUE], decisions_ues_att: dict[str:dict]
):
    """Vérifie les décisions d'UE
    puis enregistre décision manuelle si indiquée dans le YAML.
    """
    for acronyme, dec_ue_att in decisions_ues_att.items():
        # retrouve l'UE
        ues_d = [
            dec_ue
            for dec_ue in decisions_ues.values()
            if dec_ue.ue.acronyme == acronyme
        ]
        assert len(ues_d) == 1  # une et une seule UE avec l'acronyme indiqué
        dec_ue = ues_d[0]
        if "codes" in dec_ue_att:
            _check_codes_jury(dec_ue.codes, dec_ue_att["codes"])

        for attr in ("moy_ue", "moy_ue_with_cap", "explanation", "code_valide"):
            if attr in dec_ue_att:
                if getattr(dec_ue, attr) != dec_ue_att[attr]:
                    raise ValueError(
                        f"""Erreur: décision d'UE: {dec_ue.ue.acronyme
                        } : champs {attr}={getattr(dec_ue, attr)} != attendu {dec_ue_att[attr]}"""
                    )
        # Force décision de jury:
        code_manuel = dec_ue_att.get("decision_jury")
        if code_manuel is not None:
            assert code_manuel in dec_ue.codes
            dec_ue.record(code_manuel)


def _check_decisions_rcues(
    decisions_rcues: list[DecisionsProposeesRCUE], decisions_rcues_att: dict
):
    "Vérifie les décisions d'RCUEs"
    for acronyme, dec_rcue_att in decisions_rcues_att.items():
        # retrouve la décision RCUE à partir de l'acronyme de la 1ère UE
        rcues_d = [
            dec_rcue
            for dec_rcue in decisions_rcues
            if dec_rcue.rcue.ue_1.acronyme == acronyme
        ]
        assert len(rcues_d) == 1  # un et un seul RCUE avec l'UE d'acronyme indiqué
        dec_rcue = rcues_d[0]
        if "codes" in dec_rcue_att:
            _check_codes_jury(dec_rcue.codes, dec_rcue_att["codes"])
        for attr in ("explanation", "code_valide"):
            if attr in dec_rcue_att:
                assert getattr(dec_rcue, attr) == dec_rcue_att[attr]
        # Descend dans le RCUE:
        if "rcue" in dec_rcue_att:
            if "moy_rcue" in dec_rcue_att["rcue"]:
                assert (
                    abs(dec_rcue.rcue.moy_rcue - dec_rcue_att["rcue"]["moy_rcue"])
                    < scu.NOTES_PRECISION
                )
            if "est_compensable" in dec_rcue_att["rcue"]:
                assert (
                    dec_rcue.rcue.est_compensable()
                    == dec_rcue_att["rcue"]["est_compensable"]
                )
        # Force décision de jury:
        code_manuel = dec_rcue_att.get("decision_jury")
        if code_manuel is not None:
            assert code_manuel in dec_rcue.codes
            dec_rcue.record(code_manuel)


def compare_decisions_annee(deca: DecisionsProposeesAnnee, deca_att: dict):
    """Vérifie que les résultats de jury calculés pour l'année, les RCUEs et les UEs
    sont ceux attendus,
    puis enregistre les décisions manuelles indiquées dans le YAML.

    deca est le résultat calculé par ScoDoc
    deca_att est un dict lu du YAML
    """
    if "codes" in deca_att:
        _check_codes_jury(deca.codes, deca_att["codes"])

    for attr in ("passage_de_droit", "code_valide", "nb_competences"):
        if attr in deca_att:
            assert getattr(deca, attr) == deca_att[attr]

    if "decisions_ues" in deca_att:
        _check_decisions_ues(deca.decisions_ues, deca_att["decisions_ues"])

    if "nb_rcues_annee" in deca_att:
        assert deca_att["nb_rcues_annee"] == len(deca.rcues_annee)

    if "decisions_rcues" in deca_att:
        _check_decisions_rcues(
            deca.decisions_rcue_by_niveau.values(), deca_att["decisions_rcues"]
        )
    # Force décision de jury:
    code_manuel = deca_att.get("decision_jury")
    if code_manuel is not None:
        assert code_manuel in deca.codes
        deca.record(code_manuel)
        assert deca.recorded


def check_deca_fields(formsemestre: FormSemestre, etud: Identite = None):
    """Vérifie les champs principaux (inscription, nb UE, nb compétences)
    de l'instance de DecisionsProposeesAnnee.
    Ne vérifie pas les décisions de jury proprement dites.
    Si etud n'est pas spécifié, prend le premier inscrit trouvé dans le semestre.
    """
    etud = etud or formsemestre.etuds.first()
    assert etud  # il faut au moins un étudiant dans le semestre
    deca = DecisionsProposeesAnnee(etud, formsemestre)
    assert deca.validation is None  # pas encore de validation enregistrée
    assert False is deca.recorded
    assert deca.code_valide is None
    if formsemestre.semestre_id % 2:
        assert deca.formsemestre_impair == formsemestre
        assert formsemestre.query_ues_parcours_etud(etud.id).all() == deca.ues_impair
    else:
        assert deca.formsemestre_pair == formsemestre
        assert formsemestre.query_ues_parcours_etud(etud.id).all() == deca.ues_pair
    assert deca.inscription_etat == scu.INSCRIT
    assert deca.inscription_etat_impair == scu.INSCRIT
    assert (deca.parcour is None) or (
        deca.parcour.id in {p.id for p in formsemestre.parcours}
    )

    nb_ues = (
        len(deca.formsemestre_pair.query_ues_parcours_etud(etud.id).all())
        if deca.formsemestre_pair
        else 0
    )
    nb_ues += (
        len(deca.formsemestre_impair.query_ues_parcours_etud(etud.id).all())
        if deca.formsemestre_impair
        else 0
    )
    assert len(deca.decisions_ues) == nb_ues

    nb_ues_un_sem = (
        len(deca.formsemestre_impair.query_ues_parcours_etud(etud.id).all())
        if deca.formsemestre_impair
        else len(deca.formsemestre_pair.query_ues_parcours_etud(etud.id).all())
    )
    assert len(deca.niveaux_competences) == nb_ues_un_sem
    assert deca.nb_competences == nb_ues_un_sem


def test_but_jury(formsemestre: FormSemestre, doc: dict):
    """Test jurys BUT
    Vérifie les champs de DecisionsProposeesAnnee et UEs
    """
    for etud in formsemestre.etuds:
        deca = DecisionsProposeesAnnee(etud, formsemestre)
        doc_formsemestre = doc["Etudiants"][etud.nom]["formsemestres"][
            formsemestre.titre
        ]
        assert doc_formsemestre
        if "attendu" in doc_formsemestre:
            if "deca" in doc_formsemestre["attendu"]:
                deca_att = doc_formsemestre["attendu"]["deca"]
                compare_decisions_annee(deca, deca_att)