From b1513f46eeab860430b908fd268c830e36945fd7 Mon Sep 17 00:00:00 2001 From: lin121291 <4jp33f9e@gmail.com> Date: Sat, 14 Mar 2026 12:52:23 +0800 Subject: [PATCH] [tests] Introduce pytest fixtures for database management Move database creation from class body to pytest fixtures. Eliminates file pollution and improves test isolation. --- tests/conftest.py | 277 +++++++++++++++++++++++++++++ tests/test_continuous_check.py | 180 ++----------------- tests/test_route_postprocessing.py | 154 +++++----------- 3 files changed, 340 insertions(+), 271 deletions(-) create mode 100644 tests/conftest.py diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 00000000..30bef2c1 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,277 @@ +""" +Pytest configuration and shared fixtures for WeatherRoutingTool tests. + +This module provides reusable fixtures for test setup and teardown, +replacing ad-hoc helper patterns with proper pytest fixtures. +""" + +import pytest +import geopandas as gpd +import sqlalchemy as db +from shapely.geometry import LineString, Point, box + + +@pytest.fixture(scope="class") +def test_nodes_gdf(): + """ + Fixture providing test nodes GeoDataFrame for seamark/constraint tests. + + :return: GeoDataFrame with test node data + :rtype: gpd.GeoDataFrame + """ + return gpd.GeoDataFrame( + columns=["tags", "geometry"], + data=[ + [{"waterway": "lock_gate", "seamark:type": "gate"}, Point(5, 15)], + [{"seamark:type": "harbour"}, Point(9.91950, 57.06081)], + [{"seamark:type": "buoy_cardinal"}, Point(12.01631, 48.92595)], + [{"seamark:type": "separation_boundary"}, Point(12.01631, 48.92595)], + [{"seamark:type": "separation_crossing"}, Point(12.01631, 48.92595)], + [{"seamark:type": "separation_lane"}, Point(12.01631, 48.92595)], + [{"seamark:type": "separation_roundabout"}, Point(12.01631, 48.92595)], + [{"seamark:type": "separation_zone"}, Point(12.01631, 48.92595)], + [{"seamark:type": "restricted_area"}, Point(12.01631, 48.92595)], + ], + ) + + +@pytest.fixture(scope="class") +def test_ways_gdf(): + """ + Fixture providing test ways GeoDataFrame for seamark/constraint tests. + + :return: GeoDataFrame with test ways data + :rtype: gpd.GeoDataFrame + """ + return gpd.GeoDataFrame( + columns=["tags", "geometry"], + data=[ + [ + {"seamark:type": "separation_boundary"}, + LineString([(3.2738333, 51.8765), (3.154833, 51.853667)]), + ], + [ + {"seamark:type": "separation_crossing"}, + LineString( + [ + (6.3732417, 54.192091), + (6.3593333, 54.1919199), + (6.3310833, 54.1905871), + (6.3182992, 54.189601), + ] + ), + ], + [ + {"seamark:type": "separation_lane"}, + LineString([(24.6575999, 59.6085725), (24.7026512, 59.5505585)]), + ], + [ + {"seamark:type": "separation_roundabout"}, + LineString( + [ + (27.9974563, 43.186327), + (27.998524, 43.1864565), + (27.9995173, 43.186412), + (28.0012373, 43.1859232), + (28.0020059, 43.1854689), + (28.0025203, 43.1850186), + (28.0029253, 43.1845006), + (28.0032216, 43.1838693), + (27.9947856, 43.1813859), + (27.9944414, 43.1819034), + (27.9941705, 43.1826993), + (27.9941723, 43.1835194), + (27.9944142, 43.1842511), + (27.9947709, 43.1848037), + (27.9953623, 43.1853841), + (27.9961109, 43.1858589), + (27.9974563, 43.186327), + ] + ), + ], + [ + {"seamark:type": "separation_zone"}, + LineString( + [ + (-1.9830398, 49.2927514), + (-1.9830233, 49.2925889), + (-1.9828257, 49.2924815), + (-1.9827145, 49.2925089), + (-1.9828543, 49.2927771), + (-1.9830398, 49.2927514), + ] + ), + ], + [ + {"seamark:type": "restricted_area"}, + LineString( + [ + (12.3569916, 47.9186626), + (12.3567217, 47.9188108), + (12.3564934, 47.9189565), + (12.3564734, 47.9191199), + (12.3565413, 47.9192803), + (12.3568636, 47.919524), + (12.3571719, 47.9196858), + (12.3575482, 47.9197593), + (12.3579399, 47.9198024), + (12.3587152, 47.9200541), + (12.3594448, 47.9203064), + (12.3596907, 47.9203917), + (12.3599993, 47.9204654), + (12.3604107, 47.9205391), + (12.3608174, 47.9205554), + (12.3610279, 47.9205224), + (12.3612053, 47.9204511), + (12.3614394, 47.9201326), + (12.3616484, 47.9198195), + (12.3616249, 47.9196335), + (12.361631, 47.9194503), + (12.3616174, 47.9193071), + (12.36156, 47.9192435), + (12.3614394, 47.9191936), + (12.3611173, 47.9191633), + (12.3609535, 47.9190676), + (12.3607335, 47.9189749), + (12.3604259, 47.918891), + (12.3595763, 47.9187613), + (12.3587674, 47.9185358), + (12.3584371, 47.9183784), + (12.3582044, 47.9182997), + (12.3579056, 47.918306), + (12.3576587, 47.9183381), + (12.3573105, 47.9184692), + (12.3569916, 47.9186626), + ] + ), + ], + ], + ) + + +@pytest.fixture(scope="class") +def test_land_polygons_gdf(): + """ + Fixture providing test land polygons GeoDataFrame for constraint tests. + + :return: GeoDataFrame with test land polygon data + :rtype: gpd.GeoDataFrame + """ + return gpd.GeoDataFrame( + geometry=[box(4.056342603541809, 49.06378892560051, 8.748316591073674, 51.19862259935186)] + ) + + +@pytest.fixture(scope="class") +def continuous_check_database(tmp_path_factory, test_nodes_gdf, test_ways_gdf, test_land_polygons_gdf): + """ + Fixture providing a temporary SQLite database for ContinuousCheck tests. + + Creates database in temporary directory with nodes, ways, and land_polygons layers. + Database is automatically cleaned up after test class completes. + + :param tmp_path_factory: pytest fixture for creating temporary directories + :param test_nodes_gdf: GeoDataFrame with test nodes + :type test_nodes_gdf: gpd.GeoDataFrame + :param test_ways_gdf: GeoDataFrame with test ways + :type test_ways_gdf: gpd.GeoDataFrame + :param test_land_polygons_gdf: GeoDataFrame with test land polygons + :type test_land_polygons_gdf: gpd.GeoDataFrame + :yields: Database engine connected to temporary database + :rtype: sqlalchemy.engine.Engine + """ + # Create temporary directory for this test class + tmp_dir = tmp_path_factory.mktemp("continuous_check_db") + db_path = tmp_dir / "gdfDB.sqlite" + + # Create database engine + engine = db.create_engine(f"sqlite:///{db_path}") + + # Write GeoDataFrames to database + test_nodes_gdf.to_file(str(db_path), driver="SQLite", layer="nodes", overwrite=True) + test_ways_gdf.to_file(str(db_path), driver="SQLite", layer="ways", overwrite=True) + test_land_polygons_gdf.to_file(str(db_path), driver="SQLite", layer="land_polygons", overwrite=True) + + yield engine + + # Cleanup (engine disposal) + engine.dispose() + # tmp_path_factory automatically cleans up the temporary directory + + +@pytest.fixture(scope="class") +def test_seamark_gdf(): + """ + Fixture providing test seamark GeoDataFrame for route postprocessing tests. + + :return: GeoDataFrame with test seamark data + :rtype: gpd.GeoDataFrame + """ + return gpd.GeoDataFrame( + columns=["tags", "geometry"], + data=[ + [ + {"seamark:type": "separation_boundary"}, + LineString([(12, 0), (11, 3)]), + ], + [ + {"seamark:type": "separation_boundary"}, + LineString([(11, 3), (5, 9)]), + ], + [ + {"seamark:type": "separation_boundary"}, + LineString([(5, 9), (0, 6)]), + ], + [ + {"seamark:type": "separation_line"}, + LineString([(11, 3), (8, 6), (5, 9)]), + ], + [ + {"seamark:type": "separation_line"}, + LineString([(17, 7), (14, 10), (10, 14)]), + ], + [ + {"seamark:type": "separation_zone"}, + LineString([(15, 6), (9, 12), (7, 10), (13, 4), (15, 6)]), + ], + [ + {"seamark:type": "separation_lane"}, + LineString([(6, 10), (9, 7), (12, 4)]), + ], + [ + {"seamark:type": "separation_lane"}, + LineString([(16, 6), (13, 9), (9, 13)]), + ], + ], + ) + + +@pytest.fixture(scope="class") +def route_postprocessing_database(tmp_path_factory, test_seamark_gdf): + """ + Fixture providing a temporary SQLite database for RoutePostprocessing tests. + + Creates database in temporary directory with seamark layer. + Database is automatically cleaned up after test class completes. + + :param tmp_path_factory: pytest fixture for creating temporary directories + :param test_seamark_gdf: GeoDataFrame with test seamark data + :type test_seamark_gdf: gpd.GeoDataFrame + :yields: Database engine connected to temporary database + :rtype: sqlalchemy.engine.Engine + """ + # Create temporary directory for this test class + tmp_dir = tmp_path_factory.mktemp("route_postprocessing_db") + db_path = tmp_dir / "gdfDB.sqlite" + + # Create database engine + engine = db.create_engine(f"sqlite:///{db_path}") + + # Write GeoDataFrame to database + test_seamark_gdf.to_file(str(db_path), driver="SQLite", layer="seamark", overwrite=True) + + yield engine + + # Cleanup (engine disposal) + engine.dispose() + # tmp_path_factory automatically cleans up the temporary directory diff --git a/tests/test_continuous_check.py b/tests/test_continuous_check.py index 28cfaeb4..1c1e734a 100644 --- a/tests/test_continuous_check.py +++ b/tests/test_continuous_check.py @@ -1,7 +1,7 @@ import geopandas as gpd import numpy as np import pandas as pd -import sqlalchemy as db +import pytest from shapely.geometry import LineString, Point, box from shapely.strtree import STRtree @@ -10,150 +10,10 @@ import tests.basic_test_func as basic_test_func -# Create dummy GeoDataFrames -test_nodes_gdf = gpd.GeoDataFrame( - columns=["tags", "geometry"], - data=[ - [{"waterway": "lock_gate", "seamark:type": "gate"}, Point(5, 15)], - [{"seamark:type": "harbour"}, Point(9.91950, 57.06081)], - [{"seamark:type": "buoy_cardinal"}, Point(12.01631, 48.92595)], - [{"seamark:type": "separation_boundary"}, Point(12.01631, 48.92595)], - [{"seamark:type": "separation_crossing"}, Point(12.01631, 48.92595)], - [{"seamark:type": "separation_lane"}, Point(12.01631, 48.92595)], - [{"seamark:type": "separation_roundabout"}, Point(12.01631, 48.92595)], - [{"seamark:type": "separation_zone"}, Point(12.01631, 48.92595)], - [{"seamark:type": "restricted_area"}, Point(12.01631, 48.92595)], - ], -) - -test_ways_gdf = gpd.GeoDataFrame( - columns=["tags", "geometry"], - data=[ - [ - {"seamark:type": "separation_boundary"}, - LineString([(3.2738333, 51.8765), (3.154833, 51.853667)]), - ], - [ - {"seamark:type": "separation_crossing"}, - LineString( - [ - (6.3732417, 54.192091), - (6.3593333, 54.1919199), - (6.3310833, 54.1905871), - (6.3182992, 54.189601), - ] - ), - ], - [ - {"seamark:type": "separation_lane"}, - LineString([(24.6575999, 59.6085725), (24.7026512, 59.5505585)]), - ], - [ - {"seamark:type": "separation_roundabout"}, - LineString( - [ - (27.9974563, 43.186327), - (27.998524, 43.1864565), - (27.9995173, 43.186412), - (28.0012373, 43.1859232), - (28.0020059, 43.1854689), - (28.0025203, 43.1850186), - (28.0029253, 43.1845006), - (28.0032216, 43.1838693), - (27.9947856, 43.1813859), - (27.9944414, 43.1819034), - (27.9941705, 43.1826993), - (27.9941723, 43.1835194), - (27.9944142, 43.1842511), - (27.9947709, 43.1848037), - (27.9953623, 43.1853841), - (27.9961109, 43.1858589), - (27.9974563, 43.186327), - ] - ), - ], - [ - {"seamark:type": "separation_zone"}, - LineString( - [ - (-1.9830398, 49.2927514), - (-1.9830233, 49.2925889), - (-1.9828257, 49.2924815), - (-1.9827145, 49.2925089), - (-1.9828543, 49.2927771), - (-1.9830398, 49.2927514), - ] - ), - ], - [ - {"seamark:type": "restricted_area"}, - LineString( - [ - (12.3569916, 47.9186626), - (12.3567217, 47.9188108), - (12.3564934, 47.9189565), - (12.3564734, 47.9191199), - (12.3565413, 47.9192803), - (12.3568636, 47.919524), - (12.3571719, 47.9196858), - (12.3575482, 47.9197593), - (12.3579399, 47.9198024), - (12.3587152, 47.9200541), - (12.3594448, 47.9203064), - (12.3596907, 47.9203917), - (12.3599993, 47.9204654), - (12.3604107, 47.9205391), - (12.3608174, 47.9205554), - (12.3610279, 47.9205224), - (12.3612053, 47.9204511), - (12.3614394, 47.9201326), - (12.3616484, 47.9198195), - (12.3616249, 47.9196335), - (12.361631, 47.9194503), - (12.3616174, 47.9193071), - (12.36156, 47.9192435), - (12.3614394, 47.9191936), - (12.3611173, 47.9191633), - (12.3609535, 47.9190676), - (12.3607335, 47.9189749), - (12.3604259, 47.918891), - (12.3595763, 47.9187613), - (12.3587674, 47.9185358), - (12.3584371, 47.9183784), - (12.3582044, 47.9182997), - (12.3579056, 47.918306), - (12.3576587, 47.9183381), - (12.3573105, 47.9184692), - (12.3569916, 47.9186626), - ] - ), - ], - ], -) - -test_land_polygons_gdf = gpd.GeoDataFrame(geometry=[box(4.056342603541809, 49.06378892560051, - 8.748316591073674, 51.19862259935186)]) - -# Create engine using SQLite -engine = db.create_engine("sqlite:///gdfDB.sqlite") - - +@pytest.mark.usefixtures("continuous_check_database") class TestContinuousCheck: - # write geodataframe into spatialite database - test_nodes_gdf.to_file( - f'{"gdfDB.sqlite"}', driver="SQLite", layer="nodes", overwrite=True - ) - - test_ways_gdf.to_file( - f'{"gdfDB.sqlite"}', driver="SQLite", layer="ways", overwrite=True - ) - test_land_polygons_gdf.to_file( - f'{"gdfDB.sqlite"}', driver="SQLite", layer="land_polygons", - overwrite=True - ) - - def test_set_map_bbox(self): + def test_set_map_bbox(self, continuous_check_database): lat1 = 1 lat2 = 5 lon1 = 2 @@ -163,14 +23,14 @@ def test_set_map_bbox(self): test_bbox = box(2, 1, 7, 5) map_bbx = Map(lat1, lon1, lat2, lon2) - with engine.connect() as conn: + with continuous_check_database.connect() as conn: continuouscheck_obj = ContinuousCheck(db_engine=conn.connection) continuous_bbox_wkt = continuouscheck_obj.set_map_bbox(map_bbx) assert continuous_bbox_wkt == test_bbox.wkt - def test_query_nodes(self): - with engine.connect() as conn: + def test_query_nodes(self, continuous_check_database): + with continuous_check_database.connect() as conn: seamark_obj = basic_test_func.create_dummy_SeamarkCrossing_object( db_engine=conn.connection) gdf = seamark_obj.query_nodes(conn.connection, @@ -188,8 +48,8 @@ def test_query_nodes(self): assert isinstance(geom, Point), "Point Instantiation Error" print("point type checked") - def test_query_ways(self): - with engine.connect() as conn: + def test_query_ways(self, continuous_check_database): + with continuous_check_database.connect() as conn: seamark_obj = basic_test_func.create_dummy_SeamarkCrossing_object( db_engine=conn.connection) gdf = seamark_obj.query_ways(conn.connection, @@ -207,11 +67,11 @@ def test_query_ways(self): assert isinstance(geom, LineString), "LineString Instantiation Error" print("Linestring type checked") - def test_concat_nodes_ways(self): + def test_concat_nodes_ways(self, continuous_check_database): """ Test for checking if table with ways and nodes includes geometries (Point, LineString) """ - with engine.connect() as conn: + with continuous_check_database.connect() as conn: seamark_obj = basic_test_func.create_dummy_SeamarkCrossing_object( db_engine=conn.connection) concat_all = seamark_obj.concat_nodes_ways(db_engine=conn.connection, @@ -239,8 +99,8 @@ def test_concat_nodes_ways(self): type_list = [type(geometry) for geometry in concat_all["geom"]] assert set(type_list).intersection([Point, LineString]), "Geometry type error" - def test_set_STRETree(self): - with engine.connect() as conn: + def test_set_STRETree(self, continuous_check_database): + with continuous_check_database.connect() as conn: seamark_obj = basic_test_func.create_dummy_SeamarkCrossing_object( db_engine=conn.connection) test_query = ["SELECT *, geometry as geom FROM nodes", @@ -269,7 +129,7 @@ def test_set_STRETree(self): assert isinstance(concat_tree, type(test_concat_tree)) assert not (concat_tree.geometries.size == 0), "ndarray is empty." - def test_check_crossing(self): + def test_check_crossing(self, continuous_check_database): lat_start = np.array((54.192091, 54.1919199, 54.1905871, 54.189601, 1)) lon_start = np.array((6.3732417, 6.3593333, 6.3310833, 6.3182992, 1)) lat_end = np.array((48.92595, 48.02595, 48.12595, 48.22595, 48.42595, 2)) @@ -277,7 +137,7 @@ def test_check_crossing(self): test_crossing_list = [True, True, True, True, False] - with engine.connect() as conn: + with continuous_check_database.connect() as conn: seamark_obj = basic_test_func.create_dummy_SeamarkCrossing_object(db_engine=conn.connection) test_query = ["SELECT *, geometry as geom FROM nodes", "SELECT *, geometry AS geom FROM ways"] @@ -294,8 +154,8 @@ def test_check_crossing(self): assert isinstance(check_list[i], bool) assert test_crossing_list == check_list - def test_set_landpolygon_STRTree(self): - with engine.connect() as conn: + def test_set_landpolygon_STRTree(self, continuous_check_database, test_land_polygons_gdf): + with continuous_check_database.connect() as conn: landpolygoncrossing_obj = basic_test_func.create_dummy_landpolygonsCrossing_object(conn.connection) test_query = "SELECT *,geometry as geom from land_polygons" landpolygon_tree = landpolygoncrossing_obj.set_landpolygon_STRTree( @@ -309,14 +169,14 @@ def test_set_landpolygon_STRTree(self): assert isinstance(landpolygon_tree, type(test_concat_tree)) assert not (landpolygon_tree.geometries.size == 0), "ndarray is empty." - def test_check_land_crossing(self): + def test_check_land_crossing(self, continuous_check_database, test_land_polygons_gdf): lat_start = np.array((47, 49.5, 48, 48, 47)) lon_start = np.array((4.5, 6, 7, 9, 5)) lat_end = np.array((52, 50.7, 50.5, 50, 50.2)) lon_end = np.array((4.5, 7.5, 10, 10, 5)) test_list = [True, True, True, False, True] - with engine.connect() as conn: + with continuous_check_database.connect() as conn: landpolygoncrossing_obj = basic_test_func.create_dummy_landpolygonsCrossing_object( conn.connection) landpolygoncrossing_obj.land_polygon_STRTree = STRtree(test_land_polygons_gdf["geometry"]) @@ -327,7 +187,3 @@ def test_check_land_crossing(self): assert test_list == check_list for i in range(len(check_list)): assert isinstance(check_list[i], bool) - - -# Closing engine -engine.dispose() diff --git a/tests/test_route_postprocessing.py b/tests/test_route_postprocessing.py index fbf64600..3bc0a59d 100644 --- a/tests/test_route_postprocessing.py +++ b/tests/test_route_postprocessing.py @@ -3,7 +3,7 @@ import geopandas as gpd import numpy as np import pandas as pd -import sqlalchemy as db +import pytest from astropy import units as u from shapely.geometry import box, LineString, Point @@ -13,72 +13,11 @@ from WeatherRoutingTool.ship.shipparams import ShipParams from WeatherRoutingTool.ship.ship_config import ShipConfig -test_seamark_gdf = gpd.GeoDataFrame( - columns=["tags", "geometry"], - data=[ - [ - {"seamark:type": "separation_boundary"}, - LineString([(12, 0), (11, 3)]), - ], - [ - {"seamark:type": "separation_boundary"}, - LineString([(11, 3), (5, 9)]), - ], - [ - {"seamark:type": "separation_boundary"}, - LineString([(5, 9), (0, 6)]), - ], - [ - {"seamark:type": "separation_line"}, - LineString([(11, 3), (8, 6), (5, 9)]), - ], - [ - {"seamark:type": "separation_line"}, - LineString([(17, 7), (14, 10), (10, 14)]), - ], - [ - {"seamark:type": "separation_zone"}, - LineString( - [ - (15, 6), - (9, 12), - (7, 10), - (13, 4), - (15, 6) - ] - ), - ], - [ - {"seamark:type": "separation_lane"}, - LineString( - [ - (6, 10), - (9, 7), - (12, 4) - - ] - ), - ], - [ - {"seamark:type": "separation_lane"}, - LineString( - [ - (16, 6), - (13, 9), - (9, 13) - ] - ), - ] - ], -) -# Create engine using SQLite -engine = db.create_engine("sqlite:///gdfDB.sqlite") - +@pytest.mark.usefixtures("route_postprocessing_database") class TestRoutePostprocessing: - test_seamark_gdf.to_file(f'{"gdfDB.sqlite"}', driver="SQLite", layer="seamark", overwrite=True) - def generate_test_route_postprocessing_obj(self): + def generate_test_route_postprocessing_obj(self, route_postprocessing_database): lats = np.array([40, 50, 60, 70]) lons = np.array([4, 5, 6, 7]) dist = np.array([100, 200, 150]) * u.meter @@ -87,12 +26,13 @@ def generate_test_route_postprocessing_obj(self): datetime(2022, 12, 19) + timedelta(hours=2), datetime(2022, 12, 19) + timedelta(hours=3)]) dummy = np.array([0, 0, 0, 0]) + speed_array = np.array([6, 6, 6, 6]) sp = ShipParams( fuel_rate=dummy * u.kg / u.second, power=dummy * u.Watt, rpm=dummy * u.Hz, - speed=dummy * u.m / u.s, + speed=speed_array * u.m / u.s, r_calm=dummy * u.N, r_wind=dummy * u.N, r_waves=dummy * u.N, @@ -128,17 +68,17 @@ def generate_test_route_postprocessing_obj(self): ship_params_per_step=sp ) boat = basic_test_func.create_dummy_Direct_Power_Ship("simpleship") - with engine.connect() as conn: + with route_postprocessing_database.connect() as conn: postprocessed_route = RoutePostprocessing(rp, boat, db_engine=conn.connection) return postprocessed_route - def test_create_route_segments(self): + def test_create_route_segments(self, route_postprocessing_database): test_gdf = gpd.GeoDataFrame(columns=["timestamp", "geometry"], data=[[datetime(2024, 5, 17, 8), LineString([(1, 2), (2, 4)])], [datetime(2024, 5, 17, 9), LineString([(2, 4), (3, 6)])], [datetime(2024, 5, 17, 10), LineString([(3, 6), (4, 8)])]]) - rpp = self.generate_test_route_postprocessing_obj() + rpp = self.generate_test_route_postprocessing_obj(route_postprocessing_database) rpp.lats_per_step = [2, 4, 6, 8] rpp.lons_per_step = [1, 2, 3, 4] rpp.starttime_per_step = [datetime(2024, 5, 17, 8), datetime(2024, 5, 17, 9), datetime(2024, 5, 17, 10)] @@ -146,9 +86,9 @@ def test_create_route_segments(self): route_gdf = rpp.create_route_segments() pd.testing.assert_frame_equal(route_gdf, test_gdf) - def test_get_route_box(self): + def test_get_route_box(self, route_postprocessing_database): test_bbox = box((1 - 0.5), (2 - 0.5), (4 + 0.5), (8 + 0.5)) - rpp = self.generate_test_route_postprocessing_obj() + rpp = self.generate_test_route_postprocessing_obj(route_postprocessing_database) rpp.lats_per_step = [2, 4, 6, 8] rpp.lons_per_step = [1, 2, 3, 4] @@ -156,16 +96,16 @@ def test_get_route_box(self): assert isinstance(bbox, type(test_bbox)) assert bbox == test_bbox - def test_query_data(self): + def test_query_data(self, route_postprocessing_database, test_seamark_gdf): test_query = "SELECT *, geometry as geom From seamark" - rpp = self.generate_test_route_postprocessing_obj() - with engine.connect() as conn: + rpp = self.generate_test_route_postprocessing_obj(route_postprocessing_database) + with route_postprocessing_database.connect() as conn: gdf_seamark = rpp.query_data(test_query, engine=conn.connection) assert isinstance(gdf_seamark, type(test_seamark_gdf)) type_list = [type(geometry) for geometry in gdf_seamark["geom"]] assert set(type_list).intersection([LineString]), "Geometry type error" - def test_find_seamark_intersections(self): + def test_find_seamark_intersections(self, route_postprocessing_database, test_seamark_gdf): test_list = [1, 3] route_gdf = gpd.GeoDataFrame( columns=["timestamp", "geometry"], @@ -175,11 +115,11 @@ def test_find_seamark_intersections(self): [datetime(2024, 5, 17, 11), LineString([(4, 5), (2, 11)])] ] ) - rpp = self.generate_test_route_postprocessing_obj() + rpp = self.generate_test_route_postprocessing_obj(route_postprocessing_database) intersect_indx_list = rpp.find_seamark_intersections(route_gdf, test_seamark_gdf) assert intersect_indx_list == test_list - def test_is_start_or_finish_node_in_separation_zone(self): + def test_is_start_or_finish_node_in_separation_zone(self, route_postprocessing_database, test_seamark_gdf): test_val = True route_gdf = gpd.GeoDataFrame( columns=["timestamp", "geometry"], @@ -189,12 +129,12 @@ def test_is_start_or_finish_node_in_separation_zone(self): test_seamark_gdf.set_geometry('geometry', inplace=True) test_seamark_gdf.rename_geometry('geom', inplace=True) - rpp = self.generate_test_route_postprocessing_obj() + rpp = self.generate_test_route_postprocessing_obj(route_postprocessing_database) is_true = rpp.is_start_or_finish_node_in_separation_zone(route_gdf, test_seamark_gdf) assert is_true == test_val - def test_find_first_node_of_route_seg(self): + def test_find_first_node_of_route_seg(self, route_postprocessing_database): test_fisrt_node = Point(16, 1) route_gdf = gpd.GeoDataFrame( columns=["timestamp", "geometry"], @@ -206,12 +146,12 @@ def test_find_first_node_of_route_seg(self): [datetime(2024, 5, 17, 12), LineString([(4, 5), (2, 11)])] ] ) - rpp = self.generate_test_route_postprocessing_obj() + rpp = self.generate_test_route_postprocessing_obj(route_postprocessing_database) first_node = rpp.find_first_node_of_route_seg(route_gdf) assert first_node == test_fisrt_node - def test_find_last_node_of_route_seg(self): + def test_find_last_node_of_route_seg(self, route_postprocessing_database): test_last_node = Point(2, 11) route_gdf = gpd.GeoDataFrame( columns=["timestamp", "geometry"], @@ -222,12 +162,12 @@ def test_find_last_node_of_route_seg(self): [datetime(2024, 5, 17, 12), LineString([(4, 5), (2, 11)])] ] ) - rpp = self.generate_test_route_postprocessing_obj() + rpp = self.generate_test_route_postprocessing_obj(route_postprocessing_database) last_node = rpp.find_last_node_of_route_seg(route_gdf) assert last_node == test_last_node - def test_find_separation_lane_to_follow(self): + def test_find_separation_lane_to_follow(self, route_postprocessing_database): seperation_lane_dict = {'tags': {"seamark:type": "separation_lane"}, 'geom': LineString([(16, 6), (13, 9), (9, 13)])} test_separation_lane = pd.Series(seperation_lane_dict) @@ -238,12 +178,12 @@ def test_find_separation_lane_to_follow(self): [{"seamark:type": "separation_lane"}, LineString([(6, 10), (9, 7), (12, 4)])]]) test_last_node = Point(16, 1) - rpp = self.generate_test_route_postprocessing_obj() + rpp = self.generate_test_route_postprocessing_obj(route_postprocessing_database) separation_lane = rpp.find_seperation_lane_to_follow(test_last_node, separation_lanes) pd.testing.assert_series_equal(separation_lane, test_separation_lane, check_names=False) - def test_connect_route_segments(self): + def test_connect_route_segments(self, route_postprocessing_database): test_connected_seg = gpd.GeoDataFrame( columns=["timestamp", "geometry"], data=[[datetime(2024, 5, 17, 9), LineString([(16, 1), (13, 2)])], @@ -269,7 +209,7 @@ def test_connect_route_segments(self): [datetime(2024, 5, 17, 10), LineString([(7, 17), (9, 20)])] ] ) - rpp = self.generate_test_route_postprocessing_obj() + rpp = self.generate_test_route_postprocessing_obj(route_postprocessing_database) connected_df = rpp.connect_route_segments(first_route_gdf, separation_lanes_series, last_route_gdf) @@ -277,7 +217,7 @@ def test_connect_route_segments(self): assert isinstance(connected_df, gpd.GeoDataFrame) assert not connected_df.empty - def test_create_first_connecting_seg(self): + def test_create_first_connecting_seg(self, route_postprocessing_database): test_connected_seg = LineString([(12, 2), (16, 6)]) first_route_gdf = gpd.GeoDataFrame( @@ -290,12 +230,12 @@ def test_create_first_connecting_seg(self): [(16, 6), (13, 9), (9, 13)])} separation_lanes_series = pd.Series(separation_lanes_dict) - rpp = self.generate_test_route_postprocessing_obj() + rpp = self.generate_test_route_postprocessing_obj(route_postprocessing_database) first_connecting_seg = rpp.create_first_connecting_seg(first_route_gdf, separation_lanes_series) assert first_connecting_seg == test_connected_seg - def test_create_first_connecting_seg_from_node(self): + def test_create_first_connecting_seg_from_node(self, route_postprocessing_database): test_connected_seg = LineString([(15, 5), (16, 6)]) first_route_gdf = gpd.GeoDataFrame( columns=["timestamp", "geometry"], @@ -308,12 +248,12 @@ def test_create_first_connecting_seg_from_node(self): [(16, 6), (13, 9), (9, 13)])} separation_lanes_series = pd.Series(separation_lanes_dict) - rpp = self.generate_test_route_postprocessing_obj() + rpp = self.generate_test_route_postprocessing_obj(route_postprocessing_database) seg_from_node = rpp.create_first_connecting_seg_from_node(first_route_gdf, separation_lanes_series) assert seg_from_node == test_connected_seg - def test_create_last_connecting_line(self): + def test_create_last_connecting_line(self, route_postprocessing_database): test_connecting_seg = LineString([(9, 13), (8, 15)]) separation_lanes_dict = {'tags': {"seamark:type": "separation_lane"}, @@ -327,12 +267,12 @@ def test_create_last_connecting_line(self): ] ) - rpp = self.generate_test_route_postprocessing_obj() + rpp = self.generate_test_route_postprocessing_obj(route_postprocessing_database) last_connecting_seg = rpp.create_last_connecting_line(last_route_gdf, separation_lanes_series) assert last_connecting_seg == test_connecting_seg - def test_create_last_connecting_line_from_node(self): + def test_create_last_connecting_line_from_node(self, route_postprocessing_database): test_last_connecting_seg = LineString([(9, 13), (9, 20)]) last_route_gdf = gpd.GeoDataFrame( @@ -345,19 +285,18 @@ def test_create_last_connecting_line_from_node(self): [(16, 6), (13, 9), (9, 13)])} separation_lanes_series = pd.Series(separation_lanes_dict) - rpp = self.generate_test_route_postprocessing_obj() + rpp = self.generate_test_route_postprocessing_obj(route_postprocessing_database) last_connecting_seg = rpp.create_last_connecting_line_from_node(last_route_gdf, separation_lanes_series) assert last_connecting_seg == test_last_connecting_seg - def test_recalculate_starttime_per_node(self): + def test_recalculate_starttime_per_node(self, route_postprocessing_database): final_route = gpd.GeoDataFrame( columns=["timestamp", "geometry"], data=[[datetime(2024, 5, 17, 9), LineString([(16, 1), (13, 2)])], [datetime(2024, 5, 17, 10), LineString([(13, 2), (12, 2)])]]) - rpp = self.generate_test_route_postprocessing_obj() - rpp.ship_speed = np.full(len(rpp.ship_speed), 6) * rpp.ship_speed.unit + rpp = self.generate_test_route_postprocessing_obj(route_postprocessing_database) time_list = rpp.recalculate_starttime_per_node(final_route) # test_list = [(datetime(2024, 5, 17, 9), datetime(2024, 5, 18, 1, 16, 57), datetime(2024, 5, 18, 6, 26, 7))] # delta time 16:16:57, 05:09:10 @@ -369,19 +308,19 @@ def test_recalculate_starttime_per_node(self): for timestamp in time_list: assert isinstance(timestamp, datetime) - def test_find_point_from_perpendicular_angle(self): + def test_find_point_from_perpendicular_angle(self, route_postprocessing_database): test_point = Point(1, 1) start_node = Point(0, 2) segment = LineString([(0, 0), (2, 2)]) - rpp = self.generate_test_route_postprocessing_obj() + rpp = self.generate_test_route_postprocessing_obj(route_postprocessing_database) x, y = rpp.find_point_from_perpendicular_angle(start_node, segment) rpp_point = Point(x, y) assert test_point == rpp_point - def test_check_valid_crossing(self): + def test_check_valid_crossing(self, route_postprocessing_database): test_valid_crossing = True separation_line_1 = LineString([(0, 2), (2, 0)]) @@ -393,25 +332,25 @@ def test_check_valid_crossing(self): last_node_of_fisrt_routing_seg = Point(0, 0) fisrt_node_of_last_routing_seg = Point(4, 4) - rpp = self.generate_test_route_postprocessing_obj() + rpp = self.generate_test_route_postprocessing_obj(route_postprocessing_database) valid_crossing, segment = rpp.check_valid_crossing(separation_lane_gdf, last_node_of_fisrt_routing_seg, fisrt_node_of_last_routing_seg) assert test_valid_crossing == valid_crossing - def test_calculate_slope(self): + def test_calculate_slope(self, route_postprocessing_database): test_slope = 1 x1, y1, x2, y2 = 1, 1, 2, 2 - rpp = self.generate_test_route_postprocessing_obj() + rpp = self.generate_test_route_postprocessing_obj(route_postprocessing_database) slope = rpp.calculate_slope(x1, y1, x2, y2) assert test_slope == slope - def test_calculate_angle_from_slope(self): + def test_calculate_angle_from_slope(self, route_postprocessing_database): test_angle = 90.0 s1 = 2 s2 = (-1) / 2 - rpp = self.generate_test_route_postprocessing_obj() + rpp = self.generate_test_route_postprocessing_obj(route_postprocessing_database) angle1 = rpp.calculate_angle_from_slope(s1, s2) assert test_angle == angle1 @@ -423,7 +362,7 @@ def test_calculate_angle_from_slope(self): assert test_angle == angle2 - def test_calculate_angle_of_current_crossing(self): + def test_calculate_angle_of_current_crossing(self, route_postprocessing_database): test_angle = 90.0 start_node = Point(0, 0) end_node = Point(4, 4) @@ -437,10 +376,7 @@ def test_calculate_angle_of_current_crossing(self): separation_lane_gdf['id'] = [0, 1] intersecting_route_seg_geom = LineString([(0, 2), (2, 0)]) - rpp = self.generate_test_route_postprocessing_obj() + rpp = self.generate_test_route_postprocessing_obj(route_postprocessing_database) angle, segment = rpp.calculate_angle_of_current_crossing(start_node, end_node, separation_lane_gdf, intersecting_route_seg_geom) assert test_angle == angle - - -engine.dispose