''' @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 = "mn3_apd" PROJECT_FILE = Path(__file__).parent / 'projects' / 'mn3_apd' / '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)): print(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.SiteTelecom.pk try: self.schema.SiteClient.layer = None self.schema.SiteTelecom.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.SiteTelecom.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.SiteTelecom.layer.setCrs(crs) r = self.run_test("test_scr", True) self.assertFailure(r) self.assertErrorLogged(r, f"Mauvaise projection.*") def test_structure_sites_telecom(self): # cas initial: valide r = self.run_test("test_structure_sites_telecom") self.assertEqual(r.title, 'Structure des données: Sites Telecom') self.assertSuccess(r) self.assertEqual(len(r.errors), 0) # valeur non autorisée self.checker.sites_telecom[0].ST_NBPRISE = "abcd" r = self.run_test("test_structure_sites_telecom", True) self.assertFailure(r) def test_structure_sites_client(self): # cas initial: valide r = self.run_test("test_structure_sites_client") self.assertEqual(r.title, 'Structure des données: Sites Clients') self.assertSuccess(r) self.assertEqual(len(r.errors), 0) # valeur non autorisée self.checker.sites_client[0].SC_TYPFON = "invalid" r = self.run_test("test_structure_sites_client", 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_TYPFON = "invalid" r = self.run_test("test_structure_cables", 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].ZP_ISOLE = "2" r = self.run_test("test_structure_zapbos", True) self.assertFailure(r) def test_structure_zasros(self): # cas initial: valide r = self.run_test("test_structure_zasros") self.assertEqual(r.title, 'Structure des données: Zasros') self.assertSuccess(r) self.assertEqual(len(r.errors), 0) # valeur non autorisée self.checker.zasros[0].ZS_NBPRISE = "invalid" r = self.run_test("test_structure_zasros", True) self.assertFailure(r) def test_structure_adductions(self): # cas initial: valide r = self.run_test("test_structure_adductions") self.assertEqual(r.title, 'Structure des données: Adduction') self.assertSuccess(r) self.assertEqual(len(r.errors), 0) # valeur non autorisée self.checker.adductions[0].AD_LONG = "invalid" r = self.run_test("test_structure_adductions", 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): pass def test_dimension_zasro(self): pass def test_affaiblissement(self): pass def test_capacite_modulo(self): pass def test_longueur_racco(self): pass def test_zapbos_prises(self): pass