Ajoute tests pour qgis_sync_wincan et corrections

qgis_sync_wincan.py

@@ -30,7 +30,6 @@ from core.pde import WincanDb, ITV_DIR, QGisPoint
 
 logger = logging.getLogger("qgis_sync_wincan")
 logconf.start("qgis_sync_wincan", logging.DEBUG)
-debug = False
 
 IMPORT_DEPUIS = 24  # Ne cherche des données à importer que sur les X derniers mois (mettre à 0 pour ignorer)
 
@@ -39,7 +38,7 @@ IMPORT_DEPUIS = 24  # Ne cherche des données à importer que sur les X derniers
 ##-----------------------------------------------
 
 WincanDb._path = Path(r"\\h2o\local\4-transversal\BDD\mdb_test\Wincan\parc_2007\DB\PARC_2007.mdb")
-debug = True
+ITV_DIR = Path(__file__).parent / "resources" / "test_qgis_sync_wincan"
 logger.handlers = [h for h in logger.handlers if (type(h) == logging.StreamHandler)]
 logger.warning("<<<<<<<<<<<<<<   Mode TEST   >>>>>>>>>>>>>>>>>")
 
@@ -67,11 +66,10 @@ for chantier_dir in a_importer:
     logger.info("Traitement du répertoire: {}".format(chantier_dir.name))
 
     # check the existence of the chantier in WincanDb, and check if the chantier has been treated in Controles (SI_Spare1 = 1)
-    query = wincan_db.exec_(Sql.format("""SELECT SI_T.SI_AutoNumber
-                                     FROM SI_T
-                                     WHERE (((SI_T.SI_Spare1) is not null
-                                     AND (SI_T.SI_JobNumber) Like {:text}))""", chantier_dir.name))
-    if not query.first():
+    if not wincan_db.exists(Sql.format("""SELECT SI_T.SI_AutoNumber
+                                         FROM SI_T
+                                         WHERE (((SI_T.SI_Spare1) is not null
+                                         AND (SI_T.SI_JobNumber) Like {:text}))""", chantier_dir.name)):
         logger.error("Le chantier n'existe pas dans wincan, ou SI_Spare1 est null")
         continue
 
@@ -111,21 +109,22 @@ for chantier_dir in a_importer:
     # Vérifie l'existence des regards dans WincanDb
     wincan_regards_name = set([])  # Le set garantit l'unicité des items
 
-    wincan_regards_name |= wincan_db.readall(Sql.format("""SELECT S_T.S_EndNode
+    wincan_regards_name |= {r.nom for r in wincan_db.read_all(Sql.format("""SELECT S_T.S_EndNode as nom
                                                             FROM S_T INNER JOIN SI_T ON S_T.S_ID = SI_T.SI_Section_ID
                                                             WHERE (S_T.S_EndNode Not Like 'T%' And S_T.S_EndNode Not Like 'BP%'
-                                                            AND SI_T.SI_JobNumber Like {:text})""", chantier_dir.name))
+                                                            AND SI_T.SI_JobNumber Like {:text})""", chantier_dir.name))}
 
-    wincan_regards_name |= wincan_db.readall(Sql.format("""SELECT S_T.S_StartNode
+    wincan_regards_name |= {r.nom for r in wincan_db.read_all(Sql.format("""SELECT S_T.S_StartNode as nom
                                                             FROM S_T INNER JOIN SI_T ON S_T.S_ID = SI_T.SI_Section_ID
                                                             WHERE (S_T.S_StartNode Not Like 'T%' And S_T.S_StartNode Not Like 'BP%'
-                                                            AND SI_T.SI_JobNumber Like {:text})""", chantier_dir.name))
+                                                            AND SI_T.SI_JobNumber Like {:text})""", chantier_dir.name))}
 
     for regard_name in shp_regards_name - wincan_regards_name:
-        logger.error("Regards manquant dans Wincan ({})".format(regard_name))
+        logger.error("Le regards suivant n'existe pas dans Wincan ({})".format(regard_name))
     for regard_name in wincan_regards_name - shp_regards_name:
-        logger.error("Regards manquant dans le shapefile ({})".format(regard_name))
+        logger.error("Le regards suivant est absent du fichier shapefile ({})".format(regard_name))
     if shp_regards_name != wincan_regards_name:
+        logger.info("Mise à jour annulée")
         continue
 
     # # Update the coordinates of the regards in WincanDb
@@ -150,11 +149,10 @@ for chantier_dir in a_importer:
     logger.info("> Mise à jour de la base Wincan")
 
     # rename the directory to mark it as imported ('I_')
-    new_path = r"{}\I_{}".format(chantier_dir.parent, chantier_dir.name)
+    new_path = chantier_dir.parent / "I_{}".format(chantier_dir.name)
     logger.debug("> Renomme {} en {}".format(chantier_dir.name, new_path.name))
 
-    if not debug:
-        try:
-            chantier_dir.rename(new_path)
-        except:
-            logger.error("Impossible de renommer le dossier")
+    try:
+        chantier_dir.rename(new_path)
+    except:
+        logger.error("Impossible de renommer le dossier")

resources/test_qgis_sync_wincan.py

@@ -0,0 +1,33 @@
+'''
+Created on 6 juin 2018
+
+@author: olivier.massot
+'''
+from path import Path
+
+from core.pde import CSigDb, WincanDb
+
+
+if __name__ == "__main__":
+
+    ITV_DIR = Path(__file__).parent / "test_qgis_sync_wincan"
+
+    CSigDb.server = "TR-POSTGIS-02"
+    CSigDb.pwd = "Am5VOMkdFHU7WwrfVOs9"
+    csig_db = CSigDb(autocommit=False)
+
+    WincanDb._path = Path(r"\\h2o\local\4-transversal\BDD\mdb_test\Wincan\parc_2007\DB\PARC_2007.mdb")
+    wincan_db = WincanDb(autocommit=False)
+
+    for d in ITV_DIR.dirs():
+        if d.name[:2] == "I_":
+            d.rename(ITV_DIR / d.name[2:])
+
+        chantier = csig_db.first("SELECT id FROM t_chantiers WHERE id_type_chantier=1 AND nom ~* '{}\s.*'".format(d.name))
+        if chantier:
+            q = wincan_db.execute("""UPDATE S_T INNER JOIN SI_T ON S_T.S_ID = SI_T.SI_Section_ID
+                                    SET S_T.S_StartNodeCoord_X = Null, S_T.S_StartNodeCoord_Y = Null, S_T.S_EndNodeCoord_X = Null, S_T.S_EndNodeCoord_Y = Null
+                                    WHERE SI_T.SI_InspName Like '{}%';
+                                    """.format(d.name))
+
+    wincan_db.commit()