Share connection as fixture

data/h3_data_importer/tests/test_h3_import.py

@@ -11,7 +11,7 @@
 
 @pytest.fixture(scope="session", autouse=True)
 def load_env():
-    project_root = Path(__file__).resolve().parents[4]
+    project_root = Path(__file__).resolve().parents[3]
     load_dotenv(project_root / ".env")
 
 
@@ -22,38 +22,42 @@ def test_env_variables():
         pytest.fail("API_POSTGRES_DATABASE is not set -> .env not found")
 
 
-def test_db_connection():
+@pytest.fixture()
+def conn():
+    conn = psycopg.connect(conninfo=get_connection_info())
+    yield conn
+    conn.close()
+
+
+def test_db_connection(conn):
     try:
-        with psycopg.connect(conninfo=get_connection_info()) as conn:
-            with conn.cursor() as cur:
-                cur.execute("SELECT 1")
-                result = cur.fetchone()
-                assert result[0] == 1, "The query result is not 1"
+        with conn.cursor() as cur:
+            cur.execute("SELECT 1")
+            result = cur.fetchone()
+            assert result[0] == 1, "The query result is not 1"
     except Exception as e:
         pytest.fail(f"DB connection Error: {e}")
 
 
-def test_indicator_coefficient_has_any_link_to_material():
-    with psycopg.connect(conninfo=get_connection_info()) as conn:
-        with conn.cursor() as cur:
-            cur.execute(
-                """
-            select "materialId" from indicator_coefficient
+def test_indicator_coefficient_has_any_link_to_material(conn):
+    with conn.cursor() as cur:
+        cur.execute(
             """
-            )
-            result = cur.fetchall()
-            assert len(result) > 0, "The query result is empty"
+        select "materialId" from indicator_coefficient
+        """
+        )
+        result = cur.fetchall()
+        assert len(result) > 0, "The query result is empty"
 
 
-def test_material_is_linked_in_indicator_coefficients():
-    with psycopg.connect(conninfo=get_connection_info()) as conn:
-        with conn.cursor() as cur:
-            cur.execute(
-                """
-            select ic.id from indicator_coefficient ic
-            join material on "materialId" = material.id
-            where material."hsCodeId" = '0407'
+def test_material_is_linked_in_indicator_coefficients(conn):
+    with conn.cursor() as cur:
+        cur.execute(
             """
-            )
-            result = cur.fetchall()
-            assert len(result) > 0, "The query result is empty"
+        select ic.id from indicator_coefficient ic
+        join material on "materialId" = material.id
+        where material."hsCodeId" = '0407'
+        """
+        )
+        result = cur.fetchall()
+        assert len(result) > 0, "The query result is empty"