ScoDoc/tests/unit/yaml_setup_but.py

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

"""Mise en place pour tests unitaires à partir de descriptions YAML:
     fonctions spécifiques au BUT
"""
from pathlib import Path
import re

from flask import current_app, g

from app import db
from app.but.import_refcomp import orebut_import_refcomp
from app.but.jury_but import (
    DecisionsProposeesAnnee,
    DecisionsProposeesRCUE,
    DecisionsProposeesUE,
)

from app.models import (
    ApcNiveau,
    ApcReferentielCompetences,
    Formation,
    FormSemestre,
    Identite,
    UniteEns,
)
from app.scodoc import sco_utils as scu
from app.scodoc import sco_pv_dict


def setup_formation_referentiel(
    refcomp_infos: dict, formation: Formation = None
) -> ApcReferentielCompetences:
    """Si il y a un référentiel de compétences, indiqué dans le YAML,
    le charge au besoin et l'associe à la formation.
    """
    if not refcomp_infos:
        return None
    assert formation is None or formation.is_apc()  # si ref. comp., doit être APC
    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
    if formation:
        formation.referentiel_competence_id = referentiel_competence.id
        db.session.add(formation)
    return referentiel_competence


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):
            # On peut spécifier un seul parcours (cas le plus fréquent) ou une liste
            if isinstance(ue_infos["parcours"], list):
                parcours = [
                    referentiel_competence.parcours.filter_by(code=code_parcour).first()
                    for code_parcour in ue_infos["parcours"]
                ]
                assert (
                    None not in parcours
                )  # les parcours indiqués pour cette UE doivent exister
            else:
                parcours = referentiel_competence.parcours.filter_by(
                    code=ue_infos["parcours"]
                ).all()
                assert (
                    len(parcours) == 1
                )  # le parcours indiqué pour cette UE doit exister
            ue.set_parcours(parcours)

        # Niveaux compétences:
        if ue_infos.get("competence"):
            competence = referentiel_competence.acronymes_ues_to_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)
            ]:
                if not parcour in module.parcours:
                    module.parcours.append(parcour)
                    db.session.add(module)
    db.session.commit()


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érifier 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 but_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 ("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]}"""
                    )
        for attr in ("moy_ue", "moy_ue_with_cap"):
            if attr in dec_ue_att:
                assert (
                    abs(getattr(dec_ue, attr) - dec_ue_att[attr]) < scu.NOTES_PRECISION
                )
        # 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 but_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 but_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 deca_att[attr] != "" else None
            )

    if "decisions_ues" in deca_att:
        but_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.get_decisions_rcues_annee())

    if "decisions_rcues" in deca_att:
        but_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)
        deca.record_autorisation_inscription(code_manuel)
        db.session.commit()
        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 False is deca.recorded
    parcour = deca.parcour
    formation: Formation = formsemestre.formation
    ues = (
        formation.query_ues_parcour(parcour)
        .filter(UniteEns.semestre_idx == formsemestre.semestre_id)
        .all()
    )
    assert len(ues) == len(deca.decisions_rcue_by_niveau.values())
    if formsemestre.semestre_id % 2:
        assert deca.formsemestre_impair == formsemestre
    else:
        assert deca.formsemestre_pair == formsemestre
    assert deca.inscription_etat == scu.INSCRIT
    assert (deca.parcour is None) or (
        deca.parcour.id in {p.id for p in formsemestre.parcours}
    )

    nb_ues = (
        len(
            formation.query_ues_parcour(parcour)
            .filter(UniteEns.semestre_idx == deca.formsemestre_pair.semestre_id)
            .all()
        )
        if deca.formsemestre_pair
        else 0
    )
    nb_ues += (
        len(
            formation.query_ues_parcour(parcour)
            .filter(UniteEns.semestre_idx == deca.formsemestre_impair.semestre_id)
            .all()
        )
        if deca.formsemestre_impair
        else 0
    )
    assert nb_ues > 0

    assert len(deca.niveaux_competences) == len(ues)
    assert deca.nb_competences == len(ues)


def but_test_jury(formsemestre: FormSemestre, doc: dict):
    """Test jurys BUT
    Vérifie les champs de DecisionsProposeesAnnee et UEs
    """
    dpv = None
    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"]
                but_compare_decisions_annee(deca, deca_att)
            if "autorisations_inscription" in doc_formsemestre["attendu"]:
                if dpv is None:  # lazy load
                    dpv = sco_pv_dict.dict_pvjury(formsemestre.id)
                check_autorisations_inscription(
                    etud, dpv, doc_formsemestre["attendu"]["autorisations_inscription"]
                )


def check_autorisations_inscription(
    etud: Identite, dpv: dict, autorisations_inscription_att: list[int]
):
    """Vérifie que les autorisations d'inscription"""
    dec_etud = dpv["decisions_dict"][etud.id]
    autorisations_inscription = {d["semestre_id"] for d in dec_etud["autorisations"]}
    assert autorisations_inscription == set(autorisations_inscription_att)