use testcontainer for dbsync postgres test

this runs the test in a throwaway postgres docker container rather
than needing to run the tests on an actual persistent postgres server

Author: austinweisgrau

requirements-dev.txt

@@ -7,4 +7,5 @@ pytest-mock==3.14.0
 pytest==8.3.3
 requests-mock==1.12.1
 ruff==0.6.9
-testfixtures==8.3.0
+testfixtures==8.3.0
+testcontainers==4.9.0

test/test_databases/conftest.py

@@ -0,0 +1,19 @@
+import pytest
+import os
+from testcontainers.postgres import PostgresContainer
+
+
+@pytest.fixture(scope="class")
+def postgres_container():
+    """Initialize a Postgres container
+
+    Sets environment variables so that parsons.Postgres(port=None)
+    connects to this container automatically.
+    """
+    with PostgresContainer("postgres:9.5") as postgres:
+        os.environ["PGUSER"] = "test"
+        os.environ["PGPASSWORD"] = "test"
+        os.environ["PGHOST"] = "localhost"
+        os.environ["PGDATABASE"] = "test"
+        os.environ["PGPORT"] = postgres.get_exposed_port(5432)
+        yield

test/test_databases/test_dbsync.py

@@ -1,7 +1,9 @@
 from parsons import Postgres, DBSync, Table, Redshift
+import pytest
 from parsons.databases.database_connector import DatabaseConnector
 from abc import ABC
 from test.test_databases.fakes import FakeDatabase
+
 from test.utils import assert_matching_tables
 from typing import Type
 import unittest
@@ -41,7 +43,9 @@ def setUp(self):
             f"{self.temp_schema}.source_table" if self.temp_schema else "source_table"
         )
         self.destination_table = (
-            f"{self.temp_schema}.destination_table" if self.temp_schema else "destination_table"
+            f"{self.temp_schema}.destination_table"
+            if self.temp_schema
+            else "destination_table"
         )
 
         # Create source table
@@ -64,7 +68,9 @@ def tearDown(self):
 
     def assert_matching_tables(self) -> None:
         source = self.source_db.query(f"SELECT * FROM {self.source_table}")
-        destination = self.destination_db.query(f"SELECT * FROM {self.destination_table}")
+        destination = self.destination_db.query(
+            f"SELECT * FROM {self.destination_table}"
+        )
         assert_matching_tables(source, destination)
 
     def table_sync_full(self, if_exists: str, **kwargs):
@@ -100,15 +106,19 @@ def test_table_sync_full_empty_table(self):
     def test_table_sync_full_chunk(self):
         # Test chunking in full sync.
         self.db_sync.chunk_size = 10
-        self.db_sync.table_sync_full(self.source_table, self.destination_table, if_exists="drop")
+        self.db_sync.table_sync_full(
+            self.source_table, self.destination_table, if_exists="drop"
+        )
         self.assert_matching_tables()
 
     def test_table_sync_incremental(self):
         # Test that incremental sync
 
         self.destination_db.copy(self.table1, self.destination_table)
         self.source_db.copy(self.table2, self.source_table, if_exists="append")
-        self.db_sync.table_sync_incremental(self.source_table, self.destination_table, "pk")
+        self.db_sync.table_sync_incremental(
+            self.source_table, self.destination_table, "pk"
+        )
         self.assert_matching_tables()
 
     def test_table_sync_incremental_chunk(self):
@@ -117,13 +127,17 @@ def test_table_sync_incremental_chunk(self):
         self.db_sync.chunk_size = 10
         self.destination_db.copy(self.table1, self.destination_table)
         self.source_db.copy(self.table2, self.source_table, if_exists="append")
-        self.db_sync.table_sync_incremental(self.source_table, self.destination_table, "pk")
+        self.db_sync.table_sync_incremental(
+            self.source_table, self.destination_table, "pk"
+        )
 
         self.assert_matching_tables()
 
     def test_table_sync_incremental_create_destination_table(self):
         # Test that an incremental sync works if the destination table does not exist.
-        self.db_sync.table_sync_incremental(self.source_table, self.destination_table, "pk")
+        self.db_sync.table_sync_incremental(
+            self.source_table, self.destination_table, "pk"
+        )
         self.assert_matching_tables()
 
     def test_table_sync_incremental_empty_table(self):
@@ -182,9 +196,7 @@ def test_table_sync_full_write_chunk(self):
         )
 
 
-# These tests interact directly with the Postgres database. In order to run, set the
-# env to LIVE_TEST='TRUE'.
-@unittest.skipIf(not os.environ.get("LIVE_TEST"), "Skipping because not running live test")
+@pytest.mark.usefixtures("postgres_container")
 class TestPostgresDBSync(TestDBSync):
     db = Postgres
     setup_sql = f"""
@@ -198,6 +210,8 @@ class TestPostgresDBSync(TestDBSync):
 
 # These tests interact directly with the Postgres database. In order to run, set the
 # env to LIVE_TEST='TRUE'.
-@unittest.skipIf(not os.environ.get("LIVE_TEST"), "Skipping because not running live test")
+@unittest.skipIf(
+    not os.environ.get("LIVE_TEST"), "Skipping because not running live test"
+)
 class TestRedshiftDBSync(TestPostgresDBSync):
     db = Redshift