隔離 PG sync 整合測試

tests/test_pg_sync.py

@@ -1,5 +1,5 @@
 #!/usr/bin/env python3
-"""測試完整 76 欄位同步"""
+"""Opt-in integration smoke for the legacy SQLite → PostgreSQL sync script."""
 
 import os
 import sqlite3
@@ -8,9 +8,18 @@ import pytest
 
 psycopg2 = pytest.importorskip("psycopg2")
 
+pytestmark = pytest.mark.skipif(
+    os.environ.get("RUN_PG_SYNC_INTEGRATION") != "1",
+    reason="PG sync integration touches a local PostgreSQL database; set RUN_PG_SYNC_INTEGRATION=1 to run.",
+)
+
 
 def test_pg_sync():
+    if not os.environ.get("POSTGRES_PASSWORD"):
+        pytest.skip("POSTGRES_PASSWORD is required for PG sync integration")
+
     sqlite_conn = sqlite3.connect("data/momo_database.db")
+    test_table = "test_full76"
     try:
         cursor = sqlite_conn.cursor()
         cursor.execute("PRAGMA table_info(realtime_sales_monthly)")
@@ -30,14 +39,14 @@ def test_pg_sync():
         )
         try:
             pg_cursor = pg_conn.cursor()
-            pg_cursor.execute("DROP TABLE IF EXISTS test_full76 CASCADE")
+            pg_cursor.execute(f'DROP TABLE IF EXISTS "{test_table}" CASCADE')
             cols_def = ", ".join([f'"{c}" TEXT' for c in columns])
-            pg_cursor.execute(f"CREATE TABLE test_full76 (id SERIAL PRIMARY KEY, {cols_def})")
+            pg_cursor.execute(f'CREATE TABLE "{test_table}" (id SERIAL PRIMARY KEY, {cols_def})')
             pg_conn.commit()
 
             cols_sql = ", ".join([f'"{c}"' for c in columns])
             placeholders = ", ".join(["%s"] * len(columns))
-            sql = f"INSERT INTO test_full76 ({cols_sql}) VALUES ({placeholders})"
+            sql = f'INSERT INTO "{test_table}" ({cols_sql}) VALUES ({placeholders})'
 
             success_count = 0
             for row in rows:
@@ -49,7 +58,7 @@ def test_pg_sync():
 
             pg_conn.commit()
             assert success_count >= 0
-            pg_cursor.execute("DROP TABLE test_full76 CASCADE")
+            pg_cursor.execute(f'DROP TABLE IF EXISTS "{test_table}" CASCADE')
             pg_conn.commit()
         finally:
             pg_conn.close()