''' @author: olivier.massot, janv. 2019 ''' from core import mncheck import logging from qgis.core import QgsCoordinateReferenceSystem, QgsGeometry, QgsPointXY import re from core.checking import SUCCESS, FAILURE, ERROR from path import Path from test._base import SchemaTest logger = logging.getLogger("mncheck") class Test(SchemaTest): SCHEMA_NAME = "mn1_rec" PROJECT_FILE = Path(__file__).parent / 'projects' / 'mn1_rec' / '1_valid' / '1_valid.qgz' def setUp(self): SchemaTest.setUp(self) self.schema = mncheck.get_schema(self.SCHEMA_NAME) if not self.schema.checkers: raise AttributeError("Aucun testeur trouvé dans le schéma") self.checker = self.schema.checkers[0]() def run_test(self, test_name, dry=False): results = self.checker.run(test_name, dry) if len(results) == 0: raise Exception("No result for test") elif len(results) > 1: raise Exception("More than one result from test") return results[0] def assertSuccess(self, r): if not r.status == SUCCESS: raise AssertionError("Le test a échoué") def assertFailure(self, r): if not r.status == FAILURE: raise AssertionError("Le test n'aurait pas dû réussir") def assertError(self, r): if not r.status == ERROR: raise AssertionError("Le test n'a pas levé d'erreur") def assertErrorLogged(self, result, err_msg): if not any((re.fullmatch(err_msg, err.message) for err in result.errors)): raise AssertionError("Error was not logged: {}".format(err_msg)) def test_load_layers(self): # cas initial: valide r = self.run_test("test_load_layers") self.assertEqual(r.title, 'Chargement des données') self.assertSuccess(r) self.assertEqual(len(r.errors), 0) # une couche manquante et une pk manquante _pk_ini = self.schema.Cable.pk try: self.schema.Artere.layer = None self.schema.Cable.pk = "not_a_qgis_field" r = self.run_test("test_load_layers", True) self.assertFailure(r) self.assertErrorLogged(r, ".*Couche manquante.*") self.assertErrorLogged(r, ".*Clef primaire manquante.*") finally: self.schema.Cable.pk = _pk_ini def test_scr(self): # cas initial: valide r = self.run_test("test_scr") self.assertEqual(r.title, 'Contrôle des projections') self.assertSuccess(r) self.assertEqual(len(r.errors), 0) # mauvaise projection crs = QgsCoordinateReferenceSystem() crs.createFromSrid(4473) self.schema.Artere.layer.setCrs(crs) r = self.run_test("test_scr", True) self.assertFailure(r) self.assertErrorLogged(r, f"Mauvaise projection.*") def test_structure_arteres(self): # cas initial: valide r = self.run_test("test_structure_arteres") self.assertEqual(r.title, 'Structure des données: Artères') self.assertSuccess(r) self.assertEqual(len(r.errors), 0) # valeur non autorisée self.checker.arteres[0].AR_ID_INSE = "invalid" r = self.run_test("test_structure_arteres", True) self.assertFailure(r) def test_structure_cables(self): # cas initial: valide r = self.run_test("test_structure_cables") self.assertEqual(r.title, 'Structure des données: Cables') self.assertSuccess(r) self.assertEqual(len(r.errors), 0) # valeur non autorisée self.checker.cables[0].CA_TYPE = "invalid" r = self.run_test("test_structure_cables", True) self.assertFailure(r) def test_structure_equipements(self): # cas initial: valide r = self.run_test("test_structure_equipements") self.assertEqual(r.title, 'Structure des données: Equipements') self.assertSuccess(r) self.assertEqual(len(r.errors), 0) # valeur non autorisée self.checker.equipements[0].EQ_TYPE = "invalid" r = self.run_test("test_structure_equipements", True) self.assertFailure(r) def test_structure_noeuds(self): # cas initial: valide r = self.run_test("test_structure_noeuds") self.assertEqual(r.title, 'Structure des données: Noeuds') self.assertSuccess(r) self.assertEqual(len(r.errors), 0) # valeur non autorisée self.checker.noeuds[0].NO_ID_INSE = "invalid" r = self.run_test("test_structure_noeuds", True) self.assertFailure(r) def test_structure_tranchees(self): # cas initial: valide r = self.run_test("test_structure_tranchees") self.assertEqual(r.title, 'Structure des données: Tranchées') self.assertSuccess(r) self.assertEqual(len(r.errors), 0) # valeur non autorisée self.checker.tranchees[0].TR_REVET = "invalid" r = self.run_test("test_structure_tranchees", True) self.assertFailure(r) def test_structure_zapbos(self): # cas initial: valide r = self.run_test("test_structure_zapbos") self.assertEqual(r.title, 'Structure des données: Zapbos') self.assertSuccess(r) self.assertEqual(len(r.errors), 0) # valeur non autorisée self.checker.zapbos[0].STATUT = "invalid" r = self.run_test("test_structure_zapbos", True) self.assertFailure(r) def test_geometry_validity(self): # cas initial: valide r = self.run_test("test_geometry_validity") self.assertEqual(r.title, 'Contrôle de la validité des géométries') self.assertSuccess(r) self.assertEqual(len(r.errors), 0) # géométrie invalide self.checker.arteres[0]._feature.setGeometry(QgsGeometry()) r = self.run_test("test_geometry_validity", True) self.assertFailure(r) self.assertErrorLogged(r, "La géométrie de l'objet est invalide") def test_geometry_type(self): # cas initial: valide r = self.run_test("test_geometry_type") self.assertEqual(r.title, 'Contrôle des types de géométries') self.assertSuccess(r) self.assertEqual(len(r.errors), 0) # type de géométrie invalide self.checker.arteres[0]._feature.setGeometry(QgsGeometry()) r = self.run_test("test_geometry_type", True) self.assertFailure(r) self.assertErrorLogged(r, "Type de géométrie invalide.*") def test_bounding_box(self): # cas initial: valide r = self.run_test("test_bounding_box") self.assertEqual(r.title, 'Contrôle des emprises') self.assertSuccess(r) self.assertEqual(len(r.errors), 0) # hors de l'emprise self.checker.noeuds[0]._feature.setGeometry(QgsGeometry.fromPointXY(QgsPointXY(0,0))) r = self.run_test("test_bounding_box", True) self.assertFailure(r) self.assertErrorLogged(r, "Hors de l'emprise autorisée") def test_duplicates(self): # cas initial: valide r = self.run_test("test_duplicates") self.assertEqual(r.title, 'Recherche de doublons') self.assertSuccess(r) self.assertEqual(len(r.errors), 0) # cas de doublon self.checker.noeuds[0].NO_NOM = self.checker.noeuds[1].NO_NOM r = self.run_test("test_duplicates", True) self.assertFailure(r) self.assertErrorLogged(r, "Doublons dans le champs NO_NOM") def test_constraints_arteres_noeuds(self): # cas initial: valide r = self.run_test("test_constraints_arteres_noeuds") self.assertEqual(r.title, 'Application des contraintes: Artères / Noeuds') self.assertSuccess(r) self.assertEqual(len(r.errors), 0) # contrainte invalide self.checker.arteres[0].noeud_a = None r = self.run_test("test_constraints_arteres_noeuds", True) self.assertFailure(r) self.assertErrorLogged(r, "Le noeud lié '.+' n'existe pas") def test_constraints_cables_equipements(self): # cas initial: valide r = self.run_test("test_constraints_cables_equipements") self.assertEqual(r.title, 'Application des contraintes: Equipements / Cables') self.assertSuccess(r) self.assertEqual(len(r.errors), 0) # contrainte invalide self.checker.cables[0].equipement_a = None r = self.run_test("test_constraints_cables_equipements", True) self.assertFailure(r) self.assertErrorLogged(r, "L'équipement lié '.+' n'existe pas*") def test_constraints_cables_equipements_b(self): # cas initial: valide r = self.run_test("test_constraints_cables_equipements_b") self.assertEqual(r.title, 'Application des contraintes: Equipements B') self.assertSuccess(r) self.assertEqual(len(r.errors), 0) # contrainte invalide for cable in self.checker.cables: cable.CA_EQ_B = "" r = self.run_test("test_constraints_cables_equipements_b", True) self.assertFailure(r) self.assertErrorLogged(r, "L'equipement '.+' n'est l'équipement B d'aucun cable.+") def test_constraints_equipements_noeuds(self): # cas initial: valide r = self.run_test("test_constraints_equipements_noeuds") self.assertEqual(r.title, 'Application des contraintes: Noeuds / Equipements') self.assertSuccess(r) self.assertEqual(len(r.errors), 0) # contrainte invalide self.checker.equipements[0].noeud = None r = self.run_test("test_constraints_equipements_noeuds", True) self.assertFailure(r) self.assertErrorLogged(r, "Le noeud lié '.+' n'existe pas") def test_graphic_duplicates(self): # cas initial: valide r = self.run_test("test_graphic_duplicates") self.assertEqual(r.title, 'Recherche de doublons graphiques') self.assertSuccess(r) self.assertEqual(len(r.errors), 0) # doublon graphique self.checker.noeuds[0]._feature.setGeometry(self.checker.noeuds[1]._feature.geometry()) r = self.run_test("test_graphic_duplicates", True) self.assertFailure(r) self.assertErrorLogged(r, "Une entité graphique est dupliquée") def test_positions_noeuds(self): # cas initial: valide r = self.run_test("test_positions_noeuds") self.assertEqual(r.title, 'Topologie: Noeuds / Artères') self.assertSuccess(r) self.assertEqual(len(r.errors), 0) # erreur topo self.checker.noeuds[0]._feature.setGeometry(QgsGeometry.fromPointXY(QgsPointXY(0,0))) r = self.run_test("test_positions_noeuds", True) self.assertFailure(r) self.assertErrorLogged(r, "Pas de noeud aux coordonnées attendues.*") def test_positions_equipements(self): # cas initial: valide r = self.run_test("test_positions_equipements") self.assertEqual(r.title, 'Topologie: Equipements / Cables') self.assertSuccess(r) self.assertEqual(len(r.errors), 0) # erreur topo self.checker.cables[0].equipement_b.noeud._feature.setGeometry(QgsGeometry.fromPointXY(QgsPointXY(0,0))) r = self.run_test("test_positions_equipements", True) self.assertFailure(r) self.assertErrorLogged(r, "Pas d'équipement aux coordonnées attendues.*") def test_tranchee_artere(self): # cas initial: valide r = self.run_test("test_tranchee_artere") self.assertEqual(r.title, 'Topologie: Tranchées / Artères') self.assertSuccess(r) self.assertEqual(len(r.errors), 0) # erreur topo new_geom = self.checker.tranchees[0]._feature.geometry() new_geom.translate(10000, 10000) self.checker.tranchees[0]._feature.setGeometry(new_geom) r = self.run_test("test_tranchee_artere", True) self.assertFailure(r) self.assertErrorLogged(r, "Tranchée ou portion de tranchée sans artère") def test_cable_artere(self): # cas initial: valide r = self.run_test("test_cable_artere") self.assertEqual(r.title, 'Topologie: Cables / Artères') self.assertSuccess(r) self.assertEqual(len(r.errors), 0) # erreur topo self.checker.cables[0].CA_COMMENT = "" new_geom = self.checker.cables[0]._feature.geometry() new_geom.translate(10000, 10000) self.checker.cables[0]._feature.setGeometry(new_geom) r = self.run_test("test_cable_artere", True) self.assertFailure(r) self.assertErrorLogged(r, "Cable ou portion de cable sans artère") def test_artere_cable(self): # cas initial: valide r = self.run_test("test_artere_cable") self.assertEqual(r.title, 'Topologie: Artères / Cables') self.assertSuccess(r) self.assertEqual(len(r.errors), 0) # erreur topo self.checker.arteres[0].AR_COMMENT = "" new_geom = self.checker.arteres[0]._feature.geometry() new_geom.translate(10000, 10000) self.checker.arteres[0]._feature.setGeometry(new_geom) r = self.run_test("test_artere_cable", True) self.assertFailure(r) self.assertErrorLogged(r, "Artère ou portion d'artère sans cable") def test_arteres_enterrees(self): # cas initial: valide r = self.run_test("test_arteres_enterrees") self.assertEqual(r.title, 'Données des artères enterrées') self.assertSuccess(r) self.assertEqual(len(r.errors), 0) self.checker.arteres[0].AR_TYPE_FO = "PVC" del self.checker.arteres[0].__dict__["AR_NB_FOUR"] self.checker.arteres[0].AR_FOU_DIS = "" r = self.run_test("test_arteres_enterrees", True) self.assertFailure(r) self.assertErrorLogged(r, "Le champs est obligatoire: AR_NB_FOUR") self.assertErrorLogged(r, "Le champs doit être renseigné: AR_FOU_DIS") def test_dates_install(self): """ Dates d'installation Vérifie que les dates d'installation sont renseignées pour les équipements en service """ # cas initial: valide r = self.run_test("test_dates_install") self.assertEqual(r.title, "Dates d'installation") self.assertSuccess(r) self.assertEqual(len(r.errors), 0) self.checker.cables[0].CA_STATUT = "REC" self.checker.cables[0].CA_DATE_IN = "" r = self.run_test("test_dates_install", True) self.assertFailure(r) self.assertErrorLogged(r, "Date d'installation \(CA_DATE_IN\) manquante") def test_largeur_tranchees(self): # cas initial: valide r = self.run_test("test_largeur_tranchees") self.assertEqual(r.title, "Tranchées: Dimensions") self.assertSuccess(r) self.assertEqual(len(r.errors), 0) self.checker.tranchees[0].TR_MOD_POS = "TRANCHEE" self.checker.tranchees[0].TR_LARG = 0 r = self.run_test("test_largeur_tranchees", True) self.assertFailure(r) self.assertErrorLogged(r, "La largeur de la tranchée doit être supérieure à 0") def test_prop_gest(self): # cas initial: valide r = self.run_test("test_prop_gest") self.assertEqual(r.title, "Propriétaires / Gestionnaires") self.assertSuccess(r) self.assertEqual(len(r.errors), 0) self.checker.arteres[0].AR_PRO_FOU = "ORANGE" self.checker.arteres[0].AR_GEST_FO = "ERDF" r = self.run_test("test_prop_gest", True) self.assertFailure(r) self.assertErrorLogged(r, "Propriétaire: ORANGE, gestionnaire\(s\) possible\(s\): ORANGE \(renseigné: ERDF\)") def test_comments(self): # cas initial: valide r = self.run_test("test_comments") self.assertEqual(r.title, "Présence de commentaires") self.assertSuccess(r) self.assertEqual(len(r.errors), 0) self.checker.equipements[0].EQ_STATUT = "APD" self.checker.equipements[0].EQ_COMMENT = "" r = self.run_test("test_comments", True) self.assertFailure(r) self.assertErrorLogged(r, "L'equipement n'est pas en REC, un commentaire devrait en préciser la raison") def test_dimensions_fourreaux(self): # cas initial: valide r = self.run_test("test_dimensions_fourreaux") self.assertEqual(r.title, 'Dimensions logiques: fourreaux') self.assertSuccess(r) self.assertEqual(len(r.errors), 0) # nombre de fourreaux incohérent self.checker.arteres[0].AR_FOU_DIS = 1000 r = self.run_test("test_dimensions_fourreaux", True) self.assertFailure(r) self.assertErrorLogged(r, "Le nombre de fourreaux disponibles \(AR_FOU_DIS\) doit être inférieur au nombre total \(AR_NB_FOUR\)") self.checker.arteres[0].AR_FOU_DIS = 0 # nombre de fourreaux incohérent self.checker.cables[0].CA_NB_FO_U = 1000 r = self.run_test("test_dimensions_fourreaux", True) self.assertFailure(r) self.assertErrorLogged(r, "Le nombre de fourreaux utilisés \(CA_NB_FO_U\) doit être inférieur au nombre total \(CA_NB_FO\)") self.checker.cables[0].CA_NB_FO_U = 0 # nombre de fourreaux incohérent self.checker.cables[0].CA_NB_FO_D = 1000 r = self.run_test("test_dimensions_fourreaux", True) self.assertFailure(r) self.assertErrorLogged(r, "Le nombre de fourreaux disponibles \(CA_NB_FO_D\) doit être inférieur au nombre total \(CA_NB_FO\)") def test_pbos(self): # cas initial: valide r = self.run_test("test_pbos") self.assertEqual(r.title, 'Topologie: PBO / ZAPBO') self.assertSuccess(r) self.assertEqual(len(r.errors), 0) # noms equipement / zapbo incoherents eq = next(e for e in self.checker.equipements if e.EQ_TYPE == "PBO") eq.EQ_NOM = "babidiboo" r = self.run_test("test_pbos", True) self.assertFailure(r) self.assertErrorLogged(r, "Le nom de la PBO ne coincide avec le nom d'aucune des ZAPBO qui le contiennent") # equipement hors zapbo _ini_geom = eq.noeud._feature.geometry() eq.noeud._feature.setGeometry(QgsGeometry.fromPointXY(QgsPointXY(0,0))) r = self.run_test("test_pbos", True) self.assertFailure(r) self.assertErrorLogged(r, "Le PBO n'est contenu dans aucune ZAPBO") def test_zapbos_prises(self): # cas initial: valide r = self.run_test("test_zapbos_prises") self.assertEqual(r.title, 'Topologie: Zapbos / Prises') self.assertSuccess(r) self.assertEqual(len(r.errors), 0) # pas de prise dans la zapbo new_geom = self.checker.zapbos[0]._feature.geometry() new_geom.translate(1000,1000) self.checker.zapbos[0]._feature.setGeometry(new_geom) r = self.run_test("test_zapbos_prises", True) self.assertFailure(r) self.assertErrorLogged(r, "La Zapbo ne contient aucune prise") # couche non chargée self.schema.Prise.layer = None r = self.run_test("test_zapbos_prises", True) self.assertError(r) self.assertErrorLogged(r, "La couche des prises n'est pas chargée") def test_pbo_dimension(self): # cas initial: valide r = self.run_test("test_pbo_dimension") self.assertEqual(r.title, 'Dimensionnement des PBO') self.assertSuccess(r) self.assertEqual(len(r.errors), 0) # couche non chargée self.schema.Prise.layer = None r = self.run_test("test_pbo_dimension", True) self.assertError(r) self.assertErrorLogged(r, "La couche des prises n'est pas chargée")