Pipeline integration testing

.gitignore

@@ -9,4 +9,8 @@ resources/germany.osm.pbf
 resources/monaco.osm.pbf
 resources/performance_report.json
 python/performance/performance_report.json
-pgpass.conf
+pgpass.conf
+
+# poetry environment exclusions
+poetry.lock
+pyproject.toml

SQL/after_import.sql

@@ -1,65 +1,69 @@
-\echo 'Altering column nodes.id to integer...'
+DO $$
+BEGIN
+RAISE NOTICE 'Altering column nodes.id to integer...';
 
 ALTER TABLE nodes ALTER COLUMN id TYPE integer;
 
-\echo 'Altering column ways.id to integer...'
+RAISE NOTICE 'Altering column ways.id to integer...';
 
 ALTER TABLE ways ALTER COLUMN id TYPE integer;
 
-\echo 'Altering column relations.id to integer...'
+RAISE NOTICE 'Altering column relations.id to integer...';
 
 ALTER TABLE relations ALTER COLUMN id TYPE integer;
 
-\echo 'Altering column nodes_ways.id to integer...'
+RAISE NOTICE 'Altering column nodes_ways.id to integer...';
 
 ALTER TABLE nodes_ways ALTER COLUMN way_id TYPE integer;
 
 -- PRIMARY KEYS
-\echo 'Adding PRIMARY KEY constraints to table nodes...'
+RAISE NOTICE 'Adding PRIMARY KEY constraints to table nodes...';
 ALTER TABLE nodes ADD CONSTRAINT pk_nodes PRIMARY KEY (id);
 
-\echo 'Adding PRIMARY KEY constraints to table nodes_ways'
+RAISE NOTICE 'Adding PRIMARY KEY constraints to table nodes_ways';
 ALTER TABLE nodes_ways ADD CONSTRAINT nodes_ways_pk PRIMARY KEY (id);
 
-\echo 'Adding PRIMARY KEY constraints to table ways...'
+RAISE NOTICE 'Adding PRIMARY KEY constraints to table ways...';
 ALTER TABLE ways ADD CONSTRAINT pk_ways PRIMARY KEY (id);
 
-\echo 'Adding PRIMARY KEY constraints to table relations...'
+RAISE NOTICE 'Adding PRIMARY KEY constraints to table relations...';
 ALTER TABLE relations ADD CONSTRAINT pk_relations PRIMARY KEY (id);
 
 
 -- FOREIGN KEYS
 -- ways table foreign keys
-\echo 'Adding FOREIGN KEY constraints to table ways...'
+RAISE NOTICE 'Adding FOREIGN KEY constraints to table ways...';
 
 ALTER TABLE ways ADD CONSTRAINT fk_ways_from FOREIGN KEY ("from") REFERENCES nodes(id);
 ALTER TABLE ways ADD CONSTRAINT fk_ways_to FOREIGN KEY ("to") REFERENCES nodes(id);
 ALTER TABLE ways ADD CONSTRAINT fk_ways_area FOREIGN KEY (area) REFERENCES areas(id);
 
 -- nodes table foreign key
-\echo 'Adding FOREIGN KEY constraints to table nodes...'
+RAISE NOTICE 'Adding FOREIGN KEY constraints to table nodes...';
 
 ALTER TABLE nodes ADD CONSTRAINT fk_nodes_area FOREIGN KEY (area) REFERENCES areas(id);
 
 -- edges table foreign keys
-\echo 'Adding FOREIGN KEY constraints to table edges...'
+RAISE NOTICE 'Adding FOREIGN KEY constraints to table edges...';
 
 ALTER TABLE edges ADD CONSTRAINT fk_edges_from FOREIGN KEY ("from") REFERENCES nodes(id);
 ALTER TABLE edges ADD CONSTRAINT fk_edges_to FOREIGN KEY ("to") REFERENCES nodes(id);
 
 -- trip_locations table foreign keys
-\echo 'Adding FOREIGN KEY constraints to table trip_location...'
+RAISE NOTICE 'Adding FOREIGN KEY constraints to table trip_location...';
 
 ALTER TABLE trip_locations ADD CONSTRAINT fk_trip_locations_destination FOREIGN KEY (destination) REFERENCES nodes(id);
 ALTER TABLE trip_locations ADD CONSTRAINT fk_trip_locations_origin FOREIGN KEY (origin) REFERENCES nodes(id);
 
 -- nodes_ways table foreign key
-\echo 'Adding FOREIGN KEY constraints to table nodes_ways...'
+RAISE NOTICE 'Adding FOREIGN KEY constraints to table nodes_ways...';
 
 ALTER TABLE nodes_ways ADD CONSTRAINT fk_nodes_ways_area FOREIGN KEY (area) REFERENCES areas(id);
 
 -- nodes_ways_speeds table foreign keys
-\echo 'Adding FOREIGN KEY constraints to table nodes_ways_speeds...'
+RAISE NOTICE 'Adding FOREIGN KEY constraints to table nodes_ways_speeds...';
 
 ALTER TABLE nodes_ways_speeds ADD CONSTRAINT fk_nodes_ways_speeds_from FOREIGN KEY (from_node_ways_id) REFERENCES nodes_ways(id);
 ALTER TABLE nodes_ways_speeds ADD CONSTRAINT fk_nodes_ways_speeds_to FOREIGN KEY (to_node_ways_id) REFERENCES nodes_ways(id);
+END
+$$;

SQL/main.sql

@@ -154,7 +154,7 @@ CREATE TABLE IF NOT EXISTS public.demand (
 --
 
 CREATE TABLE IF NOT EXISTS public.nodes (
-    id bigint NOT NULL,
+    id integer NOT NULL,
     geom public.geometry(Point,4326) NOT NULL,
     area integer,
     contracted boolean DEFAULT false NOT NULL
@@ -522,12 +522,12 @@ CREATE TABLE IF NOT EXISTS public.trip_times (
 --
 
 CREATE TABLE IF NOT EXISTS public.ways (
-    id bigint NOT NULL,
+    id integer NOT NULL,
     tags public.hstore,
     geom public.geometry(Geometry,4326) NOT NULL,
     area integer,
-    "from" bigint NOT NULL,
-    "to" bigint NOT NULL,
+    "from" integer NOT NULL,
+    "to" integer NOT NULL,
     oneway boolean NOT NULL
 );
 

SQL/testing_extension.sql

@@ -262,25 +262,25 @@ BEGIN
     -- drop every sequence in test_scheme_name
     FOR tmp IN (SELECT sequence_name FROM information_schema.sequences WHERE sequence_schema = test_scheme_name)
         LOOP
-            EXECUTE format('DROP SEQUENCE %I.%I', test_scheme_name, tmp);
+            EXECUTE format('DROP SEQUENCE %I.%I CASCADE', test_scheme_name, tmp);
         END LOOP;
 
     -- drop every view in test_scheme_name
     FOR table_name_i IN (SELECT table_name FROM information_schema.views WHERE table_schema = test_scheme_name)
         LOOP
-            EXECUTE format('DROP VIEW %I.%I', test_scheme_name, table_name_i);
+            EXECUTE format('DROP VIEW %I.%I CASCADE', test_scheme_name, table_name_i);
         END LOOP;
 
     -- drop every table in test_scheme_name
     FOR table_name_i IN (SELECT table_name FROM information_schema.tables WHERE table_schema = test_scheme_name)
         LOOP
-            EXECUTE format('DROP TABLE %I.%I', test_scheme_name, table_name_i);
+            EXECUTE format('DROP TABLE %I.%I CASCADE', test_scheme_name, table_name_i);
         END LOOP;
 
     -- drop routines in test_scheme_name
     FOR table_name_i IN (SELECT routine_name FROM information_schema.routines WHERE routine_schema = test_scheme_name)
         LOOP
-            EXECUTE format('DROP FUNCTION %I.%I', test_scheme_name, table_name_i);
+            EXECUTE format('DROP FUNCTION %I.%I CASCADE', test_scheme_name, table_name_i);
         END LOOP;
 
     -- update search path

python/roadgraphtool/db.py

@@ -1,17 +1,22 @@
 import atexit
 import logging
-import psycopg2
-import sshtunnel
-import sqlalchemy
+from pathlib import Path
+
+import geopandas as gpd
 import pandas as pd
+import psycopg2
 import psycopg2.errors
-import geopandas as gpd
-from pathlib import Path
+import sqlalchemy
+import sshtunnel
 from sqlalchemy.engine import Row
 
 from roadgraphtool.credentials_config import CREDENTIALS
 
-logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s', datefmt='%H:%M:%S')
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s %(levelname)s %(message)s",
+    datefmt="%H:%M:%S",
+)
 
 
 def connect_db_if_required(db_function):
@@ -21,7 +26,7 @@ def connect_db_if_required(db_function):
 
     def wrapper(*args, **kwargs):
         db = args[0]
-        if hasattr(db, 'server'):
+        if hasattr(db, "server"):
             db.start_or_restart_ssh_connection_if_needed()
         if not db.is_connected():
             db.set_up_db_connections()
@@ -39,14 +44,15 @@ class __Database:
     Import as:
     from db import db
     """
+
     config = CREDENTIALS
 
     def __init__(self):
         # If private key specified, assume ssh connection and try to set it up
         self.db_server_port = self.config.db_server_port
         self.db_name = self.config.db_name
         self.ssh_tunnel_local_port = 1113
-        if hasattr(self.config, 'server'):
+        if hasattr(self.config, "server"):
             self.server = self.config.server
             self.ssh_server = None
             self.host = self.config.host
@@ -58,7 +64,9 @@ def __init__(self):
         self._sql_alchemy_engine_str = None
 
     def is_connected(self):
-        return (self._psycopg2_connection is not None) and (self._sqlalchemy_engine is not None)
+        return (self._psycopg2_connection is not None) and (
+            self._sqlalchemy_engine is not None
+        )
 
     def set_up_db_connections(self):
         # psycopg2 connection object
@@ -75,22 +83,24 @@ def set_ssh_to_db_server_and_set_port(self):
             ssh_pkey=self.config.private_key_path,
             ssh_username=self.config.server_username,
             ssh_private_key_password=self.config.private_key_phrase,
-            remote_bind_address=('localhost', self.db_server_port),
-            local_bind_address=('localhost', self.ssh_tunnel_local_port)
+            remote_bind_address=("localhost", self.db_server_port),
+            local_bind_address=("localhost", self.ssh_tunnel_local_port),
         )
         try:
             self.ssh_server = sshtunnel.open_tunnel(self.server, **ssh_kwargs)
         except sshtunnel.paramiko.SSHException as e:
             # sshtunnel dependency paramiko may attempt to use ssh-agent and crashes if it fails
             logging.warning(f"sshtunnel.paramiko.SSHException: '{e}'")
-            self.ssh_server = sshtunnel.open_tunnel(self.server, **ssh_kwargs, allow_agent=False)
+            self.ssh_server = sshtunnel.open_tunnel(
+                self.server, **ssh_kwargs, allow_agent=False
+            )
 
         self.ssh_server.start()
         logging.info(
             "SSH tunnel established from %s to %s/%s",
             self.ssh_server.local_bind_address,
             self.ssh_server.ssh_host,
-            self.db_server_port
+            self.db_server_port,
         )
 
         self.db_server_port = self.ssh_server.local_bind_port
@@ -110,12 +120,15 @@ def start_or_restart_ssh_connection_if_needed(self):
                     self.ssh_server.restart()
 
     def get_sql_alchemy_engine_str(self):
-        sql_alchemy_engine_str = 'postgresql+psycopg2://{user}:{password}@{host}:{port}/{dbname}'.format(
-            user=self.config.username,
-            password=self.config.db_password,
-            host=self.host,
-            port=self.db_server_port,
-            dbname=self.db_name)
+        sql_alchemy_engine_str = (
+            "postgresql+psycopg2://{user}:{password}@{host}:{port}/{dbname}".format(
+                user=self.config.username,
+                password=self.config.db_password,
+                host=self.host,
+                port=self.db_server_port,
+                dbname=self.db_name,
+            )
+        )
 
         return sql_alchemy_engine_str
 
@@ -129,7 +142,7 @@ def get_new_psycopg2_connection(self):
                 password=self.config.db_password,
                 host=self.config.db_host,
                 port=self.db_server_port,
-                dbname=self.db_name
+                dbname=self.db_name,
             )
 
             atexit.register(psycopg2_connection.close)
@@ -148,29 +161,40 @@ def execute_sql(self, query, *args, schema='public', use_transactions=True) -> N
             if not use_transactions:
                 connection.execution_options(isolation_level="AUTOCOMMIT")
             with connection.begin():
-                connection.execute(sqlalchemy.text(f"SET search_path TO {schema};"))
+                search_path = connection.execute(sqlalchemy.text("SHOW search_path;")).all()[0][0]
+                connection.execute(sqlalchemy.text(f"SET search_path TO {','.join([schema, search_path])};"))
                 connection.execute(sqlalchemy.text(query), *args)
-                connection.execute(sqlalchemy.text(f"SET search_path TO public;"))
+                connection.execute(sqlalchemy.text(f"SET search_path TO {search_path};"))
 
     @connect_db_if_required
-    def execute_sql_and_fetch_all_rows(self, query, *args) -> list[Row]:
+    def execute_sql_and_fetch_all_rows(self, query, *args, schema=None) -> list[Row]:
         with self._sqlalchemy_engine.connect() as conn:
-            result = conn.execute(sqlalchemy.text(query), *args).all()
+            result = None
+            if schema:
+                search_path = conn.execute(sqlalchemy.text("SHOW search_path;")).all()[0][0]
+                conn.execute(sqlalchemy.text(f"SET search_path TO {','.join([schema, search_path])};"))
+                result = conn.execute(sqlalchemy.text(query), *args).all()
+                conn.execute(sqlalchemy.text(f"SET search_path TO {search_path};"))
+            else:
+                result = conn.execute(sqlalchemy.text(query), *args).all()
             return result
 
     @connect_db_if_required
-    def execute_script(self, script_path: Path):
+    def execute_script(self, script_path: Path) -> int:
         with open(script_path) as f:
             script = f.read()
             cursor = self._psycopg2_connection.cursor()
+            retcode = 0
             try:
                 cursor.execute(script)
                 self._psycopg2_connection.commit()
             except Exception as e:
                 logging.error(f"Error executing script {script_path}: {e}")
                 self._psycopg2_connection.rollback()
+                retcode = 1
             finally:
                 cursor.close()
+                return retcode
 
     def set_schema(self, schema: str):
         """
@@ -215,7 +239,9 @@ def execute_query_to_pandas(self, sql: str, **kwargs) -> pd.DataFrame:
         return data
 
     @connect_db_if_required
-    def dataframe_to_db_table(self, df: pd.DataFrame, table_name: str, **kwargs) -> None:
+    def dataframe_to_db_table(
+        self, df: pd.DataFrame, table_name: str, **kwargs
+    ) -> None:
         """
         Save DataFrame to a new table in the database
 
@@ -224,7 +250,9 @@ def dataframe_to_db_table(self, df: pd.DataFrame, table_name: str, **kwargs) ->
 
         https://pandas.pydata.org/pandas-docs/stable/reference/api/pandas.DataFrame.to_sql.html
         """
-        df.to_sql(table_name, con=self._sqlalchemy_engine, if_exists='append', index=False)
+        df.to_sql(
+            table_name, con=self._sqlalchemy_engine, if_exists="append", index=False
+        )
 
     @connect_db_if_required
     def db_table_to_pandas(self, table_name: str, **kwargs) -> pd.DataFrame: