test: Slightly refactor fixed config values into tests/settings.py

.gitignore

@@ -1,3 +1,4 @@
+.idea
 .vscode/**
 # Secrets and internal config files
 **/.secrets/*

tests/settings.py

@@ -0,0 +1,5 @@
+# PostgreSQL default schema name.
+DB_SCHEMA_NAME = "public"
+
+# SQLAlchemy connection URL.
+DB_SQLALCHEMY_URL = "postgresql://postgres:postgres@localhost:5432/postgres"

tests/test_core.py

@@ -11,8 +11,9 @@
 from singer_sdk.testing.runners import TapTestRunner
 from sqlalchemy import Column, DateTime, Integer, MetaData, Numeric, String, Table, text
 from sqlalchemy.dialects.postgresql import BIGINT, DATE, JSON, JSONB, TIME, TIMESTAMP
-from test_replication_key import TABLE_NAME, TapTestReplicationKey
-from test_selected_columns_only import (
+from tests.settings import DB_SCHEMA_NAME, DB_SQLALCHEMY_URL
+from tests.test_replication_key import TABLE_NAME, TapTestReplicationKey
+from tests.test_selected_columns_only import (
     TABLE_NAME_SELECTED_COLUMNS_ONLY,
     TapTestSelectedColumnsOnly,
 )
@@ -21,7 +22,7 @@
 
 SAMPLE_CONFIG = {
     "start_date": pendulum.datetime(2022, 11, 1).to_iso8601_string(),
-    "sqlalchemy_url": "postgresql://postgres:postgres@localhost:5432/postgres",
+    "sqlalchemy_url": DB_SQLALCHEMY_URL,
 }
 
 NO_SQLALCHEMY_CONFIG = {
@@ -159,7 +160,7 @@ def test_temporal_datatypes():
         conn.execute(insert)
     tap = TapPostgres(config=SAMPLE_CONFIG)
     tap_catalog = json.loads(tap.catalog_json_text)
-    altered_table_name = f"public-{table_name}"
+    altered_table_name = f"{DB_SCHEMA_NAME}-{table_name}"
     for stream in tap_catalog["streams"]:
         if stream.get("stream") and altered_table_name not in stream["stream"]:
             for metadata in stream["metadata"]:
@@ -217,7 +218,7 @@ def test_jsonb_json():
         conn.execute(insert)
     tap = TapPostgres(config=SAMPLE_CONFIG)
     tap_catalog = json.loads(tap.catalog_json_text)
-    altered_table_name = f"public-{table_name}"
+    altered_table_name = f"{DB_SCHEMA_NAME}-{table_name}"
     for stream in tap_catalog["streams"]:
         if stream.get("stream") and altered_table_name not in stream["stream"]:
             for metadata in stream["metadata"]:
@@ -268,7 +269,7 @@ def test_decimal():
         conn.execute(insert)
     tap = TapPostgres(config=SAMPLE_CONFIG)
     tap_catalog = json.loads(tap.catalog_json_text)
-    altered_table_name = f"public-{table_name}"
+    altered_table_name = f"{DB_SCHEMA_NAME}-{table_name}"
     for stream in tap_catalog["streams"]:
         if stream.get("stream") and altered_table_name not in stream["stream"]:
             for metadata in stream["metadata"]:
@@ -354,7 +355,7 @@ def test_invalid_python_dates():
     tap = TapPostgres(config=SAMPLE_CONFIG)
     # Alter config and then check the data comes through as a string
     tap_catalog = json.loads(tap.catalog_json_text)
-    altered_table_name = f"public-{table_name}"
+    altered_table_name = f"{DB_SCHEMA_NAME}-{table_name}"
     for stream in tap_catalog["streams"]:
         if stream.get("stream") and altered_table_name not in stream["stream"]:
             for metadata in stream["metadata"]:
@@ -376,7 +377,7 @@ def test_invalid_python_dates():
     copied_config["dates_as_string"] = True
     tap = TapPostgres(config=copied_config)
     tap_catalog = json.loads(tap.catalog_json_text)
-    altered_table_name = f"public-{table_name}"
+    altered_table_name = f"{DB_SCHEMA_NAME}-{table_name}"
     for stream in tap_catalog["streams"]:
         if stream.get("stream") and altered_table_name not in stream["stream"]:
             for metadata in stream["metadata"]:

tests/test_replication_key.py

@@ -9,13 +9,14 @@
 from singer_sdk.testing.templates import TapTestTemplate
 from sqlalchemy import Column, MetaData, String, Table
 from sqlalchemy.dialects.postgresql import TIMESTAMP
+from tests.settings import DB_SCHEMA_NAME, DB_SQLALCHEMY_URL
 
 from tap_postgres.tap import TapPostgres
 
 TABLE_NAME = "test_replication_key"
 SAMPLE_CONFIG = {
     "start_date": pendulum.datetime(2022, 11, 1).to_iso8601_string(),
-    "sqlalchemy_url": "postgresql://postgres:postgres@localhost:5432/postgres",
+    "sqlalchemy_url": DB_SQLALCHEMY_URL,
 }
 
 
@@ -80,7 +81,7 @@ def test_null_replication_key_with_start_date():
         conn.execute(insert)
     tap = TapPostgres(config=SAMPLE_CONFIG)
     tap_catalog = json.loads(tap.catalog_json_text)
-    altered_table_name = f"public-{table_name}"
+    altered_table_name = f"{DB_SCHEMA_NAME}-{table_name}"
     for stream in tap_catalog["streams"]:
         if stream.get("stream") and altered_table_name not in stream["stream"]:
             for metadata in stream["metadata"]:
@@ -137,7 +138,7 @@ def test_null_replication_key_without_start_date():
         conn.execute(insert)
     tap = TapPostgres(config=modified_config)
     tap_catalog = json.loads(tap.catalog_json_text)
-    altered_table_name = f"public-{table_name}"
+    altered_table_name = f"{DB_SCHEMA_NAME}-{table_name}"
     for stream in tap_catalog["streams"]:
         if stream.get("stream") and altered_table_name not in stream["stream"]:
             for metadata in stream["metadata"]:

tests/test_selected_columns_only.py

@@ -2,12 +2,13 @@
 import json
 
 from singer_sdk.testing.templates import TapTestTemplate
+from tests.settings import DB_SQLALCHEMY_URL
 
 from tap_postgres.tap import TapPostgres
 
 TABLE_NAME_SELECTED_COLUMNS_ONLY = "test_selected_columns_only"
 SAMPLE_CONFIG = {
-    "sqlalchemy_url": "postgresql://postgres:postgres@localhost:5432/postgres",
+    "sqlalchemy_url": DB_SQLALCHEMY_URL,
 }