Improve database connections management

.github/workflows/tester.yml

@@ -17,9 +17,39 @@ env:
   PYTHON_VERSION: 3.7
 
 jobs:
-  tests:
+  tests-unit:
     runs-on: ubuntu-latest
 
+    steps:
+      - name: Get source code
+        uses: actions/checkout@v2
+
+      - name: Set up Python
+        uses: actions/[email protected]
+        with:
+          python-version: ${{ env.PYTHON_VERSION }}
+
+      - uses: actions/[email protected]
+        with:
+          path: ~/.cache/pip
+          key: ${{ runner.os }}-pip-${{ hashFiles('requirements/*.txt') }}
+          restore-keys: |
+            ${{ runner.os }}-pip-
+
+      - name: Install Python requirements
+        run: |
+          python -m pip install -U pip setuptools wheel
+          python -m pip install -U -r requirements/testing.txt
+
+      - name: Run Unit tests
+        run: pytest tests/unit/
+
+      - name: Upload coverage to Codecov
+        uses: codecov/codecov-action@v2
+
+  test-qgis:
+    runs-on: ubuntu-20.04
+
     container:
       image: qgis/qgis:release-3_16
       env:
@@ -46,7 +76,7 @@ jobs:
           python3 -m pip install -U -r requirements/testing.txt
 
       - name: Run Unit tests
-        run: pytest
+        run: pytest tests/qgis/
 
       - name: Upload coverage to Codecov
-        uses: codecov/codecov-action@v2.0.2
+        uses: codecov/codecov-action@v2

docs/development/testing.md

@@ -1,45 +1,50 @@
-# Tests
+# Testing the plugin
+
+Tests are written in 2 separate folders:
+
+- `tests/unit`: testing code which is independent of QGIS API
+- `tests/qgis`: testing code which depends on QGIS API
 
 ## Requirements
 
-QGIS must be installed. Then:
+- QGIS 3.16+
 
 ```bash
+python -m pip install -U pip
 python -m pip install -U -r requirements/testing.txt
 ```
 
-## Run tests
-
-Run all tests:
+## Running tests
 
 ```bash
-pytest
-```
+# run all tests with PyTest and Coverage report
+python -m pytest
 
-Run a specific test module:
+# run only unit tests
+python -m pytest tests/unit
 
-```bash
-python -m unittest tests.test_plg_metadata
-```
+# run only QGIS tests
+python -m pytest tests/qgis
 
-Run a specific test:
+# run a specific test module using standard unittest
+python -m unittest tests.test_plg_metadata
 
-```bash
+# run a specific test function using standard unittest
 python -m unittest tests.test_plg_metadata.TestPluginMetadata.test_version_semver
 ```
 
-## Using Docker
+### Using Docker
 
-Alternatively, you can run unit tests using Docker.
+Build the image:
 
-1. Build the container:
+```bash
+docker build --pull --rm -f "tests/tests_qgis.dockerfile" -t qgis_316:plugin_tester .
+```
 
-    ```bash
-    docker build -f tests/tests_qgis.dockerfile -t qgis-plg-testing-gmlas .
-    ```
+Run tests:
 
-2. Run pytest:
+```bash
+docker run -v "$(pwd):/tmp/plugin" qgis_316:plugin_tester python3 -m pytest
+```
 
-    ```bash
-    docker container run qgis-plg-testing-gmlas pytest
-    ```
+Please note that will use the root rights on some folders.

gml_application_schema_toolbox/constants.py

@@ -0,0 +1,15 @@
+#! python3  # noqa: E265
+
+"""
+    Plugin constants.
+"""
+
+# ############################################################################
+# ########## Globals ###############
+# ##################################
+
+# TODO: add names variants of provider names
+DATABASE_TYPES = (
+    "postgres",
+    "spatialite",
+)

gml_application_schema_toolbox/core/load_gmlas_in_qgis.py

@@ -13,6 +13,8 @@
 #   You should have received a copy of the GNU Library General Public
 #   License along with this library; if not, see <http://www.gnu.org/licenses/>.
 
+from typing import Union
+
 from osgeo import ogr
 from qgis.core import (
     QgsAttributeEditorContainer,
@@ -89,13 +91,26 @@ def createLayerTreeModelLegendNodes(self, layer_tree_layer):
         return [QgsSimpleLegendNode(layer_tree_layer, self.text, self.icon, self)]
 
 
-def import_in_qgis(gmlas_uri, provider: str, schema=None):
+def import_in_qgis(gmlas_uri: str, provider: str, schema: Union[str, None] = None):
     """Imports layers from a GMLAS file in QGIS with relations and editor widgets
 
     @param gmlas_uri connection parameters
-    @param provider name of the QGIS provider that handles gmlas_uri parameters (postgresql or spatialite)
+    @param provider name of the QGIS provider that handles gmlas_uri parameters
     @param schema name of the PostgreSQL schema where tables and metadata tables are
     """
+    PlgLogger.log(
+        message=f"Start importing {gmlas_uri} (provider: {provider}) into QGIS",
+        log_level=4,
+    )
+
+    # get path or URI
+    if provider in ("spatialite", "sqlite"):
+        provider = "sqlite"
+    elif provider in ("postgres", "postgresql"):
+        gmlas_uri = "PG: {}".format(gmlas_uri)
+    else:
+        pass
+
     if schema is not None:
         schema_s = schema + "."
     else:
@@ -107,7 +122,11 @@ def import_in_qgis(gmlas_uri, provider: str, schema=None):
     try:
         ds = drv.Open(gmlas_uri)
     except Exception as err:
-        PlgLogger.log(err, log_level=2)
+        PlgLogger.log(
+            "Import failed - Dataset unproperly loaded. Trace: {}".format(err),
+            log_level=2,
+        )
+        raise err
 
     if ds is None:
         raise RuntimeError("Problem opening {}".format(gmlas_uri))

gml_application_schema_toolbox/core/qgis_urlopener.py

@@ -10,7 +10,6 @@
 from typing import Dict
 
 # project
-from gml_application_schema_toolbox.__about__ import __title__
 from gml_application_schema_toolbox.toolbelt import PlgLogger, PlgOptionsManager
 from gml_application_schema_toolbox.toolbelt.network_manager import (
     NetworkAccessManager,

gml_application_schema_toolbox/extlibs/qgis_processing_postgis.py

@@ -1,3 +1,5 @@
+#! python3  # noqa: E265
+
 """
 ***************************************************************************
     postgis.py
@@ -45,9 +47,9 @@ def __str__(self):
         return "MESSAGE: %s\nQUERY: %s" % (self.message, self.query)
 
 
-def uri_from_name(conn_name):
+def uri_from_name(conn_name) -> str:
     settings = QgsSettings()
-    settings.beginGroup(u"/PostgreSQL/connections/%s" % conn_name)
+    settings.beginGroup("/PostgreSQL/connections/%s" % conn_name)
 
     if not settings.contains("database"):  # non-existent entry?
         raise QgsProcessingException(
@@ -812,7 +814,7 @@ def create_index(self, table, name, column, schema=None):
 
     def create_spatial_index(self, table, schema=None, geom_column="the_geom"):
         table_name = self._table_name(schema, table)
-        idx_name = self._quote(u"sidx_%s_%s" % (table, geom_column))
+        idx_name = self._quote("sidx_%s_%s" % (table, geom_column))
         sql = 'CREATE INDEX "%s" ON %s USING GIST(%s)' % (
             idx_name,
             table_name,
@@ -950,7 +952,7 @@ def _quote(self, identifier):
             return identifier
 
         # It's needed - let's quote it (and double the double-quotes)
-        return u'"%s"' % identifier.replace('"', '""')
+        return '"%s"' % identifier.replace('"', '""')
 
     def _quote_unicode(self, txt):
         """Make the string safe - replace ' with ''."""
@@ -963,26 +965,42 @@ def _table_name(self, schema, table):
         if not schema:
             return self._quote(table)
         else:
-            return u'"%s"."%s"' % (self._quote(schema), self._quote(table))
+            return '"%s"."%s"' % (self._quote(schema), self._quote(table))
 
 
 # For debugging / testing
 if __name__ == "__main__":
 
     db = GeoDB(host="localhost", dbname="gis", user="gisak", passwd="g")
+    # fix_print_with_import
     print(db.list_schemas())
+    # fix_print_with_import
     print("==========")
 
     for row in db.list_geotables():
+        # fix_print_with_import
         print(row)
+    # fix_print_with_import
     print("==========")
 
     for row in db.get_table_indexes("trencin"):
+        # fix_print_with_import
         print(row)
+    # fix_print_with_import
     print("==========")
 
     for row in db.get_table_constraints("trencin"):
+        # fix_print_with_import
         print(row)
+    # fix_print_with_import
     print("==========")
 
+    # fix_print_with_import
     print(db.get_table_rows("trencin"))
+
+    # for fld in db.get_table_metadata('trencin'):
+    # ....print fld
+    # try:
+    # ....db.create_table('trrrr', [('id','serial'), ('test','text')])
+    # except DbError, e:
+    # ....print unicode(e), e.query