Forráskód Böngészése

restructure les tests pour séparer le chargement des données des tests

omassot 7 éve
szülő
commit
ead425907f
3 módosított fájl, 74 hozzáadás és 89 törlés
  1. 4 1
      core/checking.py
  2. 35 44
      schemas/mn1_rec.py
  3. 35 44
      schemas/mn2_rec.py

+ 4 - 1
core/checking.py

@@ -110,12 +110,15 @@ class BaseChecker():
     def log_critical(self, message, **info):
         self._test_running.log_error(f"[CRITIQUE] {message}", info, critical=True)
     
-    def run(self):
+    def run(self, rxfilter_=""):
         
         tests_results = []
         
         for test in self.tests:
             
+            if rxfilter_ and not re.fullmatch(rxfilter_, test.__name__, re.IGNORECASE):     #@UndefinedVariable
+                continue
+                
             result = TestResult(test)
             
             self._test_running = result

+ 35 - 44
schemas/mn1_rec.py

@@ -187,40 +187,50 @@ models = [Artere, Cable, Equipement, Noeud, Tranchee, Zapbo]
 ####### Validateur
 
 class Mn1Checker(BaseChecker):
+    def __init__(self):
+        super().__init__()
+        
+        self.dataset = {}
+        for model in models:
+            model.layer = next((l for l in QgsProject.instance().mapLayers().values() \
+                          if l.name().lower() == model.layername.lower()), None)
+            self.dataset[model] = [model(f) for f in model.layer.getFeatures()]
+            
+    def setUp(self):
+        
+        self.arteres = self.dataset.get(Artere, [])
+        self.cables = self.dataset.get(Cable, [])
+        self.equipements = self.dataset.get(Equipement, [])
+        self.noeuds = self.dataset.get(Noeud, [])
+        self.tranchees = self.dataset.get(Tranchee, [])
+        self.zapbos = self.dataset.get(Zapbo, [])
+    
+        for artere in self.arteres:
+            artere.noeud_a = next((n for n in self.noeuds if n.NO_NOM == artere.AR_NOEUD_A), None)
+            artere.noeud_b = next((n for n in self.noeuds if n.NO_NOM == artere.AR_NOEUD_B), None)
+    
+        for cable in self.cables:
+            cable.equipement_a = next((e for e in self.equipements if e.EQ_NOM == cable.CA_EQ_A), None)
+            cable.equipement_b = next((e for e in self.equipements if e.EQ_NOM == cable.CA_EQ_B), None)
+    
+        for equipement in self.equipements:
+            equipement.noeud = next((n for n in self.noeuds if n.NO_NOM == equipement.EQ_NOM_NOE), None)
     
     def test_load_layers(self):
         """ Chargement des données
             Contrôle la présence des couches attendues
         """
         
-        self.dataset = {}
-        
         for model in models:
-            layername = model.layername
-            
-            try:
-                layer = next((l for l in QgsProject.instance().mapLayers().values() if l.name().lower() == layername.lower()))
-            except StopIteration:
+            if model.layer is None:
                 self.log_critical("Couche manquante", model=model)
                 continue
             
             if model.pk:
-                if not model.pk.lower() in [f.name().lower() for f in layer.fields()]:
+                if not model.pk.lower() in [f.name().lower() for f in model.layer.fields()]:
                     self.log_critical(f"Clef primaire manquante ({model.pk})", model=model)
                     continue
             
-            model.layer = layer
-            
-            self.dataset[model] = [model(f) for f in layer.getFeatures()]
-        
-        self.arteres = self.dataset.get(Artere, [])
-        self.cables = self.dataset.get(Cable, [])
-        self.equipements = self.dataset.get(Equipement, [])
-        self.noeuds = self.dataset.get(Noeud, [])
-        self.tranchees = self.dataset.get(Tranchee, [])
-        self.zapbos = self.dataset.get(Zapbo, [])
-        
-        
     def test_scr(self):
         """ Contrôle des projections 
         Vérifie que les couches ont le bon sytème de projection
@@ -351,18 +361,10 @@ class Mn1Checker(BaseChecker):
         """ Application des contraintes: Artères / Noeuds
         Vérifie que les noeuds attachés aux artères existent 
         """
-        
         for artere in self.arteres:
-            try:
-                artere.noeud_a = next((n for n in self.noeuds if n.NO_NOM == artere.AR_NOEUD_A))
-            except StopIteration:
-                artere.noeud_a = None
+            if artere.noeud_a is None:
                 self.log_error(f"Le noeud lié '{artere.AR_NOEUD_A}' n'existe pas", item=artere)
-                
-            try:
-                artere.noeud_b = next((n for n in self.noeuds if n.NO_NOM == artere.AR_NOEUD_B))
-            except StopIteration:
-                artere.noeud_b = None
+            if artere.noeud_b is None:
                 self.log_error(f"Le noeud lié '{artere.AR_NOEUD_B}' n'existe pas", item=artere)
 
     def test_constraints_cables_equipements(self):
@@ -370,16 +372,9 @@ class Mn1Checker(BaseChecker):
         Vérifie que les équipements attachés aux cables existent """
         
         for cable in self.cables:
-            try:
-                cable.equipement_a = next((e for e in self.equipements if e.EQ_NOM == cable.CA_EQ_A))
-            except StopIteration:
-                cable.equipement_a = None
+            if cable.equipement_a is None:
                 self.log_error(f"L'équipement lié '{cable.CA_EQ_A}' n'existe pas", item=cable)
-                
-            try:
-                cable.equipement_b = next((e for e in self.equipements if e.EQ_NOM == cable.CA_EQ_B))
-            except StopIteration:
-                cable.equipement_b = None
+            if cable.equipement_b is None:
                 self.log_error(f"L'équipement lié '{cable.CA_EQ_B}' n'existe pas", item=cable)
 
     def test_constraints_cables_equipements_b(self):
@@ -398,13 +393,9 @@ class Mn1Checker(BaseChecker):
         Vérifie que les noeuds attachés aux équipements existent 
         """
         for equipement in self.equipements:
-            try:
-                equipement.noeud = next((n for n in self.noeuds if n.NO_NOM == equipement.EQ_NOM_NOE))
-            except StopIteration:
-                equipement.noeud = None
+            if equipement.noeud is None:
                 self.log_error(f"Le noeud lié '{equipement.EQ_NOM_NOE}' n'existe pas", item=equipement)
 
-
     def test_graphic_duplicates(self):
         """ Recherche de doublons graphiques """
             

+ 35 - 44
schemas/mn2_rec.py

@@ -208,31 +208,16 @@ models = [Artere, Cable, Equipement, Noeud, Tranchee, Zapbo]
 
 
 class Mn2Checker(BaseChecker):
-    
-    def test_load_layers(self):
-        """ Chargement des données
-            Contrôle la présence des couches attendues
-        """
+    def __init__(self):
+        super().__init__()
         
         self.dataset = {}
-        
         for model in models:
-            layername = model.layername
+            model.layer = next((l for l in QgsProject.instance().mapLayers().values() \
+                          if l.name().lower() == model.layername.lower()), None)
+            self.dataset[model] = [model(f) for f in model.layer.getFeatures()]
             
-            try:
-                layer = next((l for l in QgsProject.instance().mapLayers().values() if l.name().lower() == layername.lower()))
-            except StopIteration:
-                self.log_critical("Couche manquante", model=model)
-                continue
-            
-            if model.pk:
-                if not model.pk.lower() in [f.name().lower() for f in layer.fields()]:
-                    self.log_critical(f"Clef primaire manquante ({model.pk})", model=model)
-                    continue
-            
-            model.layer = layer
-            
-            self.dataset[model] = [model(f) for f in layer.getFeatures()]
+    def setUp(self):
         
         self.arteres = self.dataset.get(Artere, [])
         self.cables = self.dataset.get(Cable, [])
@@ -240,7 +225,31 @@ class Mn2Checker(BaseChecker):
         self.noeuds = self.dataset.get(Noeud, [])
         self.tranchees = self.dataset.get(Tranchee, [])
         self.zapbos = self.dataset.get(Zapbo, [])
+    
+        for artere in self.arteres:
+            artere.noeud_a = next((n for n in self.noeuds if n.NO_CODE == artere.AR_NOEUD_A), None)
+            artere.noeud_b = next((n for n in self.noeuds if n.NO_CODE == artere.AR_NOEUD_B), None)
+    
+        for cable in self.cables:
+            cable.equipement_a = next((e for e in self.equipements if e.EQ_CODE == cable.CA_EQ_A), None)
+            cable.equipement_b = next((e for e in self.equipements if e.EQ_CODE == cable.CA_EQ_B), None)
+    
+        for equipement in self.equipements:
+            equipement.noeud = next((n for n in self.noeuds if n.NO_CODE == equipement.EQ_NOM_NOE), None)
         
+    def test_load_layers(self):
+        """ Chargement des données
+            Contrôle la présence des couches attendues
+        """
+        for model in models:
+            if model.layer is None:
+                self.log_critical("Couche manquante", model=model)
+                continue
+            
+            if model.pk:
+                if not model.pk.lower() in [f.name().lower() for f in model.layer.fields()]:
+                    self.log_critical(f"Clef primaire manquante ({model.pk})", model=model)
+                    continue
         
     def test_scr(self):
         """ Contrôle des projections 
@@ -374,18 +383,10 @@ class Mn2Checker(BaseChecker):
         """ Application des contraintes: Artères / Noeuds
         Vérifie que les noeuds attachés aux artères existent 
         """
-        
         for artere in self.arteres:
-            try:
-                artere.noeud_a = next((n for n in self.noeuds if n.NO_CODE == artere.AR_NOEUD_A))
-            except StopIteration:
-                artere.noeud_a = None
+            if artere.noeud_a is None:
                 self.log_error(f"Le noeud lié '{artere.AR_NOEUD_A}' n'existe pas", item=artere)
-                
-            try:
-                artere.noeud_b = next((n for n in self.noeuds if n.NO_CODE == artere.AR_NOEUD_B))
-            except StopIteration:
-                artere.noeud_b = None
+            if artere.noeud_b is None:
                 self.log_error(f"Le noeud lié '{artere.AR_NOEUD_B}' n'existe pas", item=artere)
 
     def test_constraints_cables_equipements(self):
@@ -393,16 +394,9 @@ class Mn2Checker(BaseChecker):
         Vérifie que les équipements attachés aux cables existent """
         
         for cable in self.cables:
-            try:
-                cable.equipement_a = next((e for e in self.equipements if e.EQ_CODE == cable.CA_EQ_A))
-            except StopIteration:
-                cable.equipement_a = None
+            if cable.equipement_a is None:
                 self.log_error(f"L'équipement lié '{cable.CA_EQ_A}' n'existe pas", item=cable)
-                
-            try:
-                cable.equipement_b = next((e for e in self.equipements if e.EQ_CODE == cable.CA_EQ_B))
-            except StopIteration:
-                cable.equipement_b = None
+            if cable.equipement_b is None:
                 self.log_error(f"L'équipement lié '{cable.CA_EQ_B}' n'existe pas", item=cable)
 
     def test_constraints_cables_equipements_b(self):
@@ -422,10 +416,7 @@ class Mn2Checker(BaseChecker):
         Vérifie que les noeuds attachés aux équipements existent 
         """
         for equipement in self.equipements:
-            try:
-                equipement.noeud = next((n for n in self.noeuds if n.NO_CODE == equipement.EQ_NOM_NOE))
-            except StopIteration:
-                equipement.noeud = None
+            if equipement.noeud is None:
                 self.log_error(f"Le noeud lié '{equipement.EQ_NOM_NOE}' n'existe pas", item=equipement)