Nouveaux tests

core/mncheck.py

@@ -100,7 +100,10 @@ class QgsModel():
     
     @classmethod
     def get_geom_name(cls, wkb):
-        return QgsWkbTypes.displayString(wkb)
+        try:
+            return QgsWkbTypes.displayString(wkb)
+        except (ValueError, TypeError):
+            return "<unknown>"
         
     def get_bounding_box(self):
         bb = self.geom.boundingBox()

schemas/mn1_rec.py

@@ -40,9 +40,9 @@ class Artere(QgsModel):
               'AR_FOU_DIS': {'empty': True, 'validator': is_int}, 
               'AR_TYPE_FO': {'type': 'string', 'multiallowed': ['PVC', 'PEHD', 'SOUS-TUBAGE PEHD', 'SOUS-TUBAGE  SOUPLE', 'FACADE', 'AERIEN', 'ENCORBELLEMENT', 'AUTRE']}, 
               'AR_DIAM_FO': {'type': 'string', 'multiallowed': ['10', '14', '18', '25', '28', '32', '40', '45', '60', '80', '150', 'NUL', '']}, 
-              'AR_PRO_FOU': {'type': 'string', 'multiallowed': ['MANCHE NUMERIQUE', 'COLLECTIVITE', 'ORANGE', 'PRIVE', 'ERDF', 'SDEM', 'AUTRE']}, 
+              'AR_PRO_FOU': {'type': 'string', 'multiallowed': ['MANCHE NUMERIQUE', 'COLLECTIVITE', 'ORANGE', 'PRIVE', 'ERDF', 'ENEDIS', 'SDEM', 'AUTRE']}, 
               'AR_PRO_CAB': {'type': 'string', 'empty': False, 'allowed': ['MANCHE NUMERIQUE']}, 
-              'AR_GEST_FO': {'type': 'string', 'multiallowed': ['MANCHE NUMERIQUE', 'MANCHE TELECOM', 'COLLECTIVITE', 'ORANGE', 'MANCHE FIBRE', 'PRIVE', 'ERDF', 'SDEM', 'AUTRE', 'NUL']}, 
+              'AR_GEST_FO': {'type': 'string', 'multiallowed': ['MANCHE NUMERIQUE', 'MANCHE TELECOM', 'COLLECTIVITE', 'ORANGE', 'MANCHE FIBRE', 'PRIVE', 'ERDF', 'ENEDIS', 'SDEM', 'AUTRE', 'NUL']}, 
               'AR_UTIL_FO': {'type': 'string', 'multiallowed': ['MANCHE NUMERIQUE', 'MANCHE TELECOM', 'COLLECTIVITE', 'ORANGE', 'MANCHE FIBRE', 'PRIVE', 'AUTRE', 'SDEM', 'NUL', '']}, 
               'AR_DATE_IN': {'empty': True, 'validator': is_modern_french_date}, 
               'AR_DATE_RE': {'empty': True, 'validator': is_modern_french_date}, 
@@ -129,7 +129,7 @@ class Noeud(QgsModel):
               'NO_PLINOX': {'required': False, 'type': 'string', 'maxlength': 3, 'allowed': ['OUI', 'NON']}, 
               'NO_X': {'empty': True, 'validator': is_float}, 
               'NO_Y': {'empty': True, 'validator': is_float}, 
-              'NO_PRO': {'type': 'string', 'maxlength': 20, 'empty': False, 'allowed': ['MANCHE NUMERIQUE', 'COLLECTIVITE', 'ORANGE', 'ERDF', 'PRIVE', 'ENEDIS', 'SDEM', 'AUTRE', 'NUL']}, 
+              'NO_PRO': {'type': 'string', 'maxlength': 20, 'empty': False, 'allowed': ['MANCHE NUMERIQUE', 'COLLECTIVITE', 'ORANGE', 'ERDF', 'ENEDIS', 'PRIVE', 'SDEM', 'AUTRE', 'NUL']}, 
               'NO_GEST': {'type': 'string', 'maxlength': 20, 'empty': False, 'allowed': ['MANCHE NUMERIQUE', 'MANCHE TELECOM', 'COLLECTIVITE', 'ORANGE', 'ERDF', 'ENEDIS', 'SDEM', 'MANCHE FIBRE', 'PRIVE', 'AUTRE', 'NUL']}, 
               'NO_HAUT': {'empty': True, 'validator': is_float}, 
               'NO_DATE_IN': {'empty': True, 'validator': is_modern_french_date}, 
@@ -391,7 +391,10 @@ class Mn1Checker(BaseChecker):
                 self.log_error(f"L'équipement lié '{cable.CA_EQ_A}' n'existe pas", item=cable)
             if cable.equipement_b is None:
                 self.log_error(f"L'équipement lié '{cable.CA_EQ_B}' n'existe pas", item=cable)
-
+                
+            if cable.CA_STATUT != cable.equipement_b.EQ_STATUT:
+                self.log_error(f"L'équipement B du cable n'a pas le même statut que le cable", item=cable)
+                
     def test_constraints_cables_equipements_b(self):
         """ Application des contraintes: Equipements B
         Vérifie que tous les équipements sont l'équipement B d'au moins un cable """
@@ -516,7 +519,7 @@ class Mn1Checker(BaseChecker):
         cables_full_buffer = Cable.full_buffer(TOLERANCE)
         
         for artere in self.arteres:
-            if any(x in artere.AR_COMMENT.lower() for x in ['racco','client','adductio','attente','bus','sans cable']):
+            if any(x in artere.AR_COMMENT.lower() for x in ['racco','client','adduction','attente','bus','sans cable']):
                 continue
             if not cables_full_buffer.contains(artere.geom):
                 self.log_error("Artère ou portion d'artère sans cable", item=artere)
@@ -524,7 +527,6 @@ class Mn1Checker(BaseChecker):
     def test_arteres_enterrees(self):
         """ Données des artères enterrées
         Contrôle les données des artères enterrées ou en encorbellement  """
-        
         for artere in self.arteres:
             if artere.AR_TYPE_FO in ["FACADE", "AERIEN"]:
                 continue
@@ -548,7 +550,22 @@ class Mn1Checker(BaseChecker):
         for tranchee in self.tranchees:
             if tranchee.TR_STATUT == "REC" and not tranchee.TR_DATE_IN:
                 self.log_error("Date d'installation (TR_DATE_IN) manquante", item=tranchee)
-                
+              
+                        
+    def test_largeur_tranchees(self):
+        """ Tranchées: Dimensions
+        Vérifie que la cohérence des dimensions des tranchées """
+        for tranchee in self.tranchees:
+            if not ("FONCAGE" in tranchee.TR_MOD_POS or 
+                    "FORAGE" in tranchee.TR_MOD_POS or 
+                    "ENCORBELLEMENT" in tranchee.TR_MOD_POS):
+                try:
+                    if float(tranchee.TR_LARG) > 0:
+                        continue
+                except (TypeError, ValueError):
+                    pass
+                self.log_error("La largeur de la tranchée doit être supérieure à 0", item=tranchee)
+    
     def test_dimensions_fourreaux(self):
         """ Dimensions logiques: fourreaux
         Vérifie que les nombres de fourreaux renseignés sont cohérents """
@@ -569,6 +586,30 @@ class Mn1Checker(BaseChecker):
             except (TypeError, ValueError):
                 pass
     
+    def test_prop_gest(self):
+        """ Propriétaires / Gestionnaires
+        Vérifie la cohérence des propriétaires et gestionnaires renseignés """
+        cases = {
+                 "ORANGE": ["ORANGE"],
+                 "ERDF": ["ERDF"],
+                 "ENEDIS": ["ENEDIS"],
+                 "MANCHE NUMERIQUE": ["MANCHE FIBRE", "MANCHE TELECOM"] 
+                 }
+        for artere in self.arteres:
+            for prop, gests in cases.items():
+                if artere.AR_PRO_FOU.upper() == prop and not artere.AR_GEST_FO.upper() in gests:
+                    self.log_error("Propriétaire: {}, gestionnaire(s) possible(s): {} (renseigné: {})".format(prop, ', '.join(gests), artere.AR_GEST_FO), item=artere)
+
+        for equipement in self.equipements:
+            for prop, gests in cases.items():
+                if equipement.EQ_PRO.upper() == prop and not equipement.EQ_GEST.upper() in gests:
+                    self.log_error("Propriétaire: {}, gestionnaire(s) possible(s): {} (renseigné: {})".format(prop, ', '.join(gests), equipement.EQ_GEST), item=equipement)
+
+        for noeud in self.noeuds:
+            for prop, gests in cases.items():
+                if noeud.NO_PRO.upper() == prop and not noeud.NO_GEST.upper() in gests:
+                    self.log_error("Propriétaire: {}, gestionnaire(s) possible(s): {} (renseigné: {})".format(prop, ', '.join(gests), noeud.NO_GEST), item=noeud)
+                    
     def _za_for_pbo(self, pbo):
         # retourne la ZAPBO correspondant à la PBO en parametre, None si aucune
         if hasattr(pbo, 'zapbo') and pbo.zapbo:
@@ -598,8 +639,25 @@ class Mn1Checker(BaseChecker):
             # On se base sur le nom pour trouver la zapbo correspondante
             equipement.zapbo = next((z for z in candidates if equipement.EQ_NOM in z.ID_ZAPBO), None)
             if not equipement.zapbo:
-                self.log_error("Le nom du PBO ne coincide avec le nom d'aucune des ZAPBO qui le contiennent", item=equipement)
-    
+                self.log_error("Le nom de la PBO ne coincide avec le nom d'aucune des ZAPBO qui le contiennent", item=equipement)
+                continue
+            
+            if equipement.EQ_STATUT != equipement.zapbo.STATUT:
+                self.log_error("Le statut de la PBO ne coincide pas avec celui de sa ZAPBO", item=equipement)
+                continue
+            
+    def test_comments(self):
+        """ Commentaires
+        Vérifie la présence de commentaires là où ils sont attendus
+        """
+        for equipement in self.equipements:
+            if equipement.EQ_STATUT != "REC" and not equipement.EQ_COMMENT:
+                self.log_error("L'equipement n'est pas en REC, un commentaire devrait en préciser la raison", item=equipement)
+                
+        for zapbo in self.zapbos:
+            if zapbo.STATUT != "REC" and not zapbo.COMMENTAIR:
+                self.log_error("La Zapbo n'est pas en REC, un commentaire devrait en préciser la raison", item=zapbo)
+            
     def test_zapbos_prises(self):
         """ Topologie: Zapbos / Prises 
         Toutes les zapbo contiennent au moins une prise"""

schemas/mn2_rec.py

@@ -49,14 +49,14 @@ class Artere(QgsModel):
               'AR_TYPE_FO': {'type': 'string', 'empty': False, 'multiallowed': ['PVC', 'PEHD', 'SOUS-TUBAGE PEHD', 'SOUS-TUBAGE  SOUPLE', 'FACADE', 'AERIEN', 'ENCORBELLEMENT', 'AUTRE']}, 
               'AR_TYFO_AI': {'type': 'string', 'empty': False, 'multiallowed': ['PVC', 'PEH', 'TUB', 'FAC', 'ENC', 'APP']}, 
               'AR_DIAM_FO': {'type': 'string', 'multiallowed': ['10', '14', '18', '25', '28', '32', '40', '45', '60', '80', '150', 'NUL', '']}, 
-              'AR_PRO_FOU': {'type': 'string', 'empty': False, 'multiallowed': ['MANCHE NUMERIQUE', 'COLLECTIVITE', 'ORANGE', 'PRIVE', 'ERDF', 'SDEM', 'AUTRE']}, 
+              'AR_PRO_FOU': {'type': 'string', 'empty': False, 'multiallowed': ['MANCHE NUMERIQUE', 'COLLECTIVITE', 'ORANGE', 'PRIVE', 'ERDF', 'ENEDIS', 'SDEM', 'AUTRE']}, 
               'AR_FAB': {'type': 'string', 'empty': True, 'maxlength': 100},
               'AR_REFFAB': {'type': 'string', 'empty': True, 'maxlength': 100},
               'AR_COULEUR': {'type': 'string', 'empty': True, 'maxlength': 20},
               'AR_AIGUIL': {'type': 'string', 'empty': True, 'maxlength': 3, 'allowed': ['OUI', 'NON']},
               'AR_NBCABL': {'empty': False, 'validator': is_positive_int},
               'AR_PRO_CAB': {'type': 'string', 'empty': False, 'allowed': ['MANCHE NUMERIQUE']}, 
-              'AR_GEST_FO': {'type': 'string', 'multiallowed': ['MANCHE NUMERIQUE', 'MANCHE TELECOM', 'COLLECTIVITE', 'ORANGE', 'MANCHE FIBRE', 'PRIVE', 'ERDF', 'SDEM', 'AUTRE', 'NUL']}, 
+              'AR_GEST_FO': {'type': 'string', 'multiallowed': ['MANCHE NUMERIQUE', 'MANCHE TELECOM', 'COLLECTIVITE', 'ORANGE', 'MANCHE FIBRE', 'PRIVE', 'ERDF', 'ENEDIS', 'SDEM', 'AUTRE', 'NUL']}, 
               'AR_UTIL_FO': {'type': 'string', 'multiallowed': ['MANCHE NUMERIQUE', 'MANCHE TELECOM', 'COLLECTIVITE', 'ORANGE', 'MANCHE FIBRE', 'PRIVE', 'AUTRE', 'SDEM', 'NUL', '']}, 
               'AR_DATE_IN': {'empty': True, 'validator': is_modern_french_date}, 
               'AR_DATE_RE': {'empty': True, 'validator': is_modern_french_date}, 
@@ -498,7 +498,10 @@ class Mn2Checker(BaseChecker):
                 self.log_error(f"L'équipement lié '{cable.CA_EQ_A}' n'existe pas", item=cable)
             if cable.equipement_b is None:
                 self.log_error(f"L'équipement lié '{cable.CA_EQ_B}' n'existe pas", item=cable)
-
+                
+            if cable.CA_STATUT != cable.equipement_b.EQ_STATUT:
+                self.log_error(f"L'équipement B du cable n'a pas le même statut que le cable", item=cable)
+                
     def test_constraints_cables_equipements_b(self):
         """ Application des contraintes: Equipements B
         Vérifie que tous les équipements sont l'équipement B d'au moins un cable """
@@ -510,7 +513,6 @@ class Mn2Checker(BaseChecker):
             if not equipement.EQ_NOM in equipements_b:
                 self.log_error(f"L'equipement '{equipement.EQ_NOM}' n'est l'équipement B d'aucun cable et n'est pas un PM", item=equipement)
 
-
     def test_constraints_equipements_noeuds(self):
         """ Application des contraintes: Noeuds / Equipements
         Vérifie que les noeuds attachés aux équipements existent 
@@ -641,6 +643,16 @@ class Mn2Checker(BaseChecker):
                 elif getattr(equipement, field) == "":
                     self.log_error("Le champs doit être renseigné: {}".format(field), item=equipement)
     
+            if equipement.EQ_TYPE == "BPE" and equipement.NO_TYPE_PH == "POTEAU":
+                self.log_error("Une BPE ne peut pas être sur un poteau", item=equipement)
+                
+            if equipement.EQ_TYPE == "PBO" and equipement.NO_TYPE_PH == "POTEAU":
+                try:
+                    if not 2.4 <= float(equipement.EQ_HAUT) <= 4.0:
+                        self.log_error("PBO sur poteau: La hauteur doit être comprise entre 2.40m et 4.00m", item=equipement)
+                except (TypeError, ValueError):
+                    self.log_error("PBO sur poteau: Hauteur manquante ou invalide", item=equipement)
+                
     def test_arteres_enterrees(self):
         """ Données des artères enterrées
         Contrôle les données des artères enterrées ou en encorbellement  """
@@ -661,7 +673,22 @@ class Mn2Checker(BaseChecker):
                         self.log_error("Le champs est obligatoire: {}".format(field), item=artere)
                     elif getattr(artere, field) == "":
                         self.log_error("Le champs doit être renseigné: {}".format(field), item=artere)
-    
+                        
+                        
+    def test_largeur_tranchees(self):
+        """ Tranchées: Dimensions
+        Vérifie que la cohérence des dimensions des tranchées """
+        for tranchee in self.tranchees:
+            if not ("FONCAGE" in tranchee.TR_MOD_POS or 
+                    "FORAGE" in tranchee.TR_MOD_POS or 
+                    "ENCORBELLEMENT" in tranchee.TR_MOD_POS):
+                try:
+                    if float(tranchee.TR_LARG) > 0:
+                        continue
+                except (TypeError, ValueError):
+                    pass
+                self.log_error("La largeur de la tranchée doit être supérieure à 0", item=tranchee)
+                    
     def test_dimensions_fourreaux(self):
         """ Dimensions logiques: fourreaux
         Vérifie que les nombres de fourreaux renseignés sont cohérents """
@@ -675,6 +702,30 @@ class Mn2Checker(BaseChecker):
             except (TypeError, ValueError):
                 pass
 
+    def test_prop_gest(self):
+        """ Propriétaires / Gestionnaires
+        Vérifie la cohérence des propriétaires et gestionnaires renseignés """
+        cases = {
+                 "ORANGE": ["ORANGE"],
+                 "ERDF": ["ERDF"],
+                 "ENEDIS": ["ENEDIS"],
+                 "MANCHE NUMERIQUE": ["MANCHE FIBRE", "MANCHE TELECOM"] 
+                 }
+        for artere in self.arteres:
+            for prop, gests in cases.items():
+                if artere.AR_PRO_FOU.upper() == prop and not artere.AR_GEST_FO.upper() in gests:
+                    self.log_error("Propriétaire: {}, gestionnaire(s) possible(s): {} (renseigné: {})".format(prop, ', '.join(gests), artere.AR_GEST_FO), item=artere)
+
+        for equipement in self.equipements:
+            for prop, gests in cases.items():
+                if equipement.EQ_PRO.upper() == prop and not equipement.EQ_GEST.upper() in gests:
+                    self.log_error("Propriétaire: {}, gestionnaire(s) possible(s): {} (renseigné: {})".format(prop, ', '.join(gests), equipement.EQ_GEST), item=equipement)
+
+        for noeud in self.noeuds:
+            for prop, gests in cases.items():
+                if noeud.NO_PRO.upper() == prop and not noeud.NO_GEST.upper() in gests:
+                    self.log_error("Propriétaire: {}, gestionnaire(s) possible(s): {} (renseigné: {})".format(prop, ', '.join(gests), noeud.NO_GEST), item=noeud)
+                     
     def test_dates_install(self):
         """ Dates d'installation
         Vérifie que les dates d'installation sont renseignées pour les équipements en service """
@@ -720,7 +771,24 @@ class Mn2Checker(BaseChecker):
             equipement.zapbo = next((z for z in candidates if equipement.EQ_CODE in z.ID_ZAPBO), None)
             if not equipement.zapbo:
                 self.log_error("Le nom du PBO ne coincide avec le nom d'aucune des ZAPBO qui le contiennent", item=equipement)
-    
+                continue
+            
+            if equipement.EQ_STATUT != equipement.zapbo.STATUT:
+                self.log_error("Le statut de la PBO ne coincide pas avec celui de sa ZAPBO", item=equipement)
+                continue
+                
+    def test_comments(self):
+        """ Commentaires
+        Vérifie la présence de commentaires là où ils sont attendus
+        """
+        for equipement in self.equipements:
+            if equipement.EQ_STATUT.upper() != 'EN SERVICE' and not equipement.EQ_COMMENT:
+                self.log_error("L'equipement n'est pas en REC, un commentaire devrait en préciser la raison", item=equipement)
+                
+        for zapbo in self.zapbos:
+            if zapbo.STATUT.upper() != 'EN SERVICE' and not zapbo.COMMENTAIR:
+                self.log_error("La Zapbo n'est pas en REC, un commentaire devrait en préciser la raison", item=zapbo)
+                
     def test_zapbos_prises(self):
         """ Topologie: Zapbos / Prises 
         Toutes les zapbo contiennent au moins une prise"""