[source-postgres]Non CDC CAT test for postgres

airbyte-integrations/bases/connector-acceptance-test/connector_acceptance_test/tests/test_incremental.py

@@ -285,10 +285,9 @@ async def test_state_with_abnormally_large_values(
         self, inputs: IncrementalConfig, connector_config, configured_catalog, future_state, docker_runner: ConnectorRunner
     ):
         configured_catalog = incremental_only_catalog(configured_catalog)
-        output = await docker_runner.call_read_with_state(config=connector_config, catalog=configured_catalog, state=future_state)
+        output = await docker_runner.call_read_with_state(config=connector_config, catalog=configured_catalog, state=future_state, enable_caching=False)
         records = filter_output(output, type_=Type.RECORD)
         states = filter_output(output, type_=Type.STATE)
-
         assert (
             not records
         ), f"The sync should produce no records when run with the state with abnormally large values {records[0].record.stream}"
@@ -318,6 +317,10 @@ async def test_state_with_abnormally_large_values(
             if state["stream"]["stream_descriptor"]["name"] in cursor_fields_per_stream
         }
 
+        print(f"actual state: {states}")
+
+        print(f"cursor_fields_per_stream: {cursor_fields_per_stream}")
+
         assert all(future_state_cursor_values_per_stream.values()), "Future state must be set up for all given streams"
 
         expected_cursor_value_schema_per_stream = {
@@ -407,7 +410,10 @@ def _get_cursor_values_from_states_by_cursor(states: Union[list, dict], cursor_f
 
             if isinstance(current_node, dict):
                 for key, value in current_node.items():
-                    if key == cursor_field:
+                    # DB sources use a hardcoded field `cursor` to denote cursor value.
+                    if key == "cursor":
+                        values.append(value)
+                    elif key == cursor_field:
                         values.append(value)
                     nodes_to_visit.append(value)
             elif isinstance(current_node, list):

airbyte-integrations/connectors/source-postgres/acceptance-test-config.yml

@@ -28,10 +28,51 @@ acceptance_tests:
   discovery:
     tests:
       - config_path: "secrets/config.json"
-      - config_path: "secrets/config_cdc.json"
         backward_compatibility_tests_config:
           disable_for_version: "2.1.1"
+  basic_read:
+    tests:
+      - config_path: "secrets/config.json"
+        configured_catalog_path: "integration_tests/configured_catalog_copy.json"
+        expect_records:
+          path: "integration_tests/expected_records.txt"
+        client_container_config:
+          client_container_dockerfile_path: "integration_tests/Dockerfile"
+          setup_command:
+            - "python"
+            - "./hook.py"
+            - "setup"
+          teardown_command:
+            - "python"
+            - "./hook.py"
+            - "teardown"
   full_refresh:
     tests:
       - config_path: "secrets/config.json"
-      - config_path: "secrets/config_cdc.json"
+        configured_catalog_path: "integration_tests/configured_catalog_copy.json"
+        client_container_config:
+          client_container_dockerfile_path: "integration_tests/Dockerfile"
+          setup_command:
+            - "python"
+            - "./hook.py"
+            - "setup"
+          teardown_command:
+            - "python"
+            - "./hook.py"
+            - "teardown"
+  incremental:
+    tests:
+      - config_path: "secrets/config.json"
+        configured_catalog_path: "integration_tests/incremental_configured_catalog_copy.json"
+        client_container_config:
+          client_container_dockerfile_path: "integration_tests/Dockerfile"
+          setup_command:
+            - "python"
+            - "./hook.py"
+            - "setup"
+          teardown_command:
+            - "python"
+            - "./hook.py"
+            - "teardown"
+        future_state:
+          future_state_path: "integration_tests/abnormal_state_copy.json"

airbyte-integrations/connectors/source-postgres/integration_tests/Dockerfile

@@ -0,0 +1,23 @@
+# Use a base image with PostgreSQL client
+FROM python:3.9
+
+RUN apt-get update && apt-get install -y \
+    libpq-dev \
+    gcc \
+    && rm -rf /var/lib/apt/lists/* \
+
+WORKDIR /usr/src/app
+
+# Copy the script to the container
+COPY seed/hook.py .
+COPY seed/requirement.txt .
+
+RUN ls .
+
+RUN pip install --no-cache-dir -r ./requirement.txt
+
+EXPOSE 80
+
+# Give execution rights to the script and pre-generate all scripts.
+RUN chmod +x ./hook.py
+RUN python ./hook.py prepare

airbyte-integrations/connectors/source-postgres/integration_tests/abnormal_state.json

@@ -3,15 +3,17 @@
     "type": "STREAM",
     "stream": {
       "stream_state": {
+        "version": 2,
+        "state_type": "cursor_based",
         "stream_name": "id_and_name",
-        "stream_namespace": "public",
         "cursor_field": ["id"],
         "cursor": "4",
-        "cursor_record_count": 1
+        "cursor_record_count": 1,
+        "stream_namespace": "%s"
       },
       "stream_descriptor": {
         "name": "id_and_name",
-        "namespace": "public"
+        "namespace": "%s"
       }
     }
   }

airbyte-integrations/connectors/source-postgres/integration_tests/abnormal_state_copy.json

@@ -0,0 +1,20 @@
+[
+  {
+    "type": "STREAM",
+    "stream": {
+      "stream_state": {
+        "version": 2,
+        "state_type": "cursor_based",
+        "stream_name": "id_and_name",
+        "cursor_field": ["id"],
+        "cursor": "4",
+        "cursor_record_count": 1,
+        "stream_namespace": "20240712-56hlSijO"
+      },
+      "stream_descriptor": {
+        "name": "id_and_name",
+        "namespace": "20240712-56hlSijO"
+      }
+    }
+  }
+]

airbyte-integrations/connectors/source-postgres/integration_tests/configured_catalog.json

@@ -0,0 +1,28 @@
+{
+  "streams": [
+    {
+      "stream": {
+        "name": "id_and_name",
+        "json_schema": {
+          "type": "object",
+          "properties": {
+            "name": {
+              "type": "string"
+            },
+            "id": {
+              "type": "string"
+            }
+          }
+        },
+        "supported_sync_modes": ["full_refresh", "incremental"],
+        "default_cursor_field": [],
+        "source_defined_primary_key": [],
+        "namespace": "%s"
+      },
+      "sync_mode": "full_refresh",
+      "destination_sync_mode": "append",
+      "cursor_field": ["id"],
+      "user_defined_primary_key": ["id"]
+    }
+  ]
+}

airbyte-integrations/connectors/source-postgres/integration_tests/configured_catalog_copy.json

@@ -0,0 +1,28 @@
+{
+  "streams": [
+    {
+      "stream": {
+        "name": "id_and_name",
+        "json_schema": {
+          "type": "object",
+          "properties": {
+            "name": {
+              "type": "string"
+            },
+            "id": {
+              "type": "string"
+            }
+          }
+        },
+        "supported_sync_modes": ["full_refresh", "incremental"],
+        "default_cursor_field": [],
+        "source_defined_primary_key": [],
+        "namespace": "20240712-56hlSijO"
+      },
+      "sync_mode": "full_refresh",
+      "destination_sync_mode": "append",
+      "cursor_field": ["id"],
+      "user_defined_primary_key": ["id"]
+    }
+  ]
+}

airbyte-integrations/connectors/source-postgres/integration_tests/configured_catalog_view.json

@@ -0,0 +1,28 @@
+{
+  "streams": [
+    {
+      "stream": {
+        "name": "id_and_name_view",
+        "json_schema": {
+          "type": "object",
+          "properties": {
+            "name": {
+              "type": "string"
+            },
+            "id": {
+              "type": "string"
+            }
+          }
+        },
+        "supported_sync_modes": ["full_refresh"],
+        "default_cursor_field": [],
+        "source_defined_primary_key": [],
+        "namespace": "%s"
+      },
+      "sync_mode": "incremental",
+      "destination_sync_mode": "append",
+      "cursor_field": ["id"],
+      "user_defined_primary_key": ["id"]
+    }
+  ]
+}

airbyte-integrations/connectors/source-postgres/integration_tests/configured_catalog_view_copy.json

@@ -0,0 +1,28 @@
+{
+  "streams": [
+    {
+      "stream": {
+        "name": "id_and_name_view",
+        "json_schema": {
+          "type": "object",
+          "properties": {
+            "name": {
+              "type": "string"
+            },
+            "id": {
+              "type": "string"
+            }
+          }
+        },
+        "supported_sync_modes": ["full_refresh"],
+        "default_cursor_field": [],
+        "source_defined_primary_key": [],
+        "namespace": "20240712-56hlSijO"
+      },
+      "sync_mode": "incremental",
+      "destination_sync_mode": "append",
+      "cursor_field": ["id"],
+      "user_defined_primary_key": ["id"]
+    }
+  ]
+}

airbyte-integrations/connectors/source-postgres/integration_tests/expected_records.txt

@@ -1,3 +1,3 @@
-{"stream": "id_and_name", "data": {"id": 1, "name": "picard"}, "emitted_at": 999999}
-{"stream": "id_and_name", "data": {"id": 2, "name": "crusher"}, "emitted_at": 999999}
-{"stream": "id_and_name", "data": {"id": 3, "name": "vash"}, "emitted_at": 999999}
+{"stream": "id_and_name", "data": {"id": "1", "name": "one"}, "emitted_at": 999999}
+{"stream": "id_and_name", "data": {"id": "2", "name": "two"}, "emitted_at": 999999}
+{"stream": "id_and_name", "data": {"id": "3", "name": "three"}, "emitted_at": 999999}

airbyte-integrations/connectors/source-postgres/integration_tests/expected_records_view.txt

@@ -0,0 +1,5 @@
+{"stream": "id_and_name_view", "data": {"id": 1, "name": "picard"}, "emitted_at": 999999}
+{"stream": "id_and_name_view", "data": {"id": 2, "name": "crusher"}, "emitted_at": 999999}
+{"stream": "id_and_name_view", "data": {"id": 3, "name": "vash"}, "emitted_at": 999999}
+{"stream": "id_and_name_view", "data": {"id": 4, "name": "benoit"}, "emitted_at": 999999}
+{"stream": "id_and_name_view", "data": {"id": 5, "name": "benoit2"}, "emitted_at": 999999}

airbyte-integrations/connectors/source-postgres/integration_tests/incremental_configured_catalog.json

@@ -10,15 +10,14 @@
               "type": "string"
             },
             "id": {
-              "type": "number",
-              "airbyte_type": "integer"
+              "type": "string"
             }
           }
         },
         "supported_sync_modes": ["full_refresh", "incremental"],
         "default_cursor_field": [],
         "source_defined_primary_key": [],
-        "namespace": "public"
+        "namespace": "%s"
       },
       "sync_mode": "incremental",
       "destination_sync_mode": "append",

airbyte-integrations/connectors/source-postgres/integration_tests/incremental_configured_catalog_copy.json

@@ -0,0 +1,28 @@
+{
+  "streams": [
+    {
+      "stream": {
+        "name": "id_and_name",
+        "json_schema": {
+          "type": "object",
+          "properties": {
+            "name": {
+              "type": "string"
+            },
+            "id": {
+              "type": "string"
+            }
+          }
+        },
+        "supported_sync_modes": ["full_refresh", "incremental"],
+        "default_cursor_field": [],
+        "source_defined_primary_key": [],
+        "namespace": "20240712-56hlSijO"
+      },
+      "sync_mode": "incremental",
+      "destination_sync_mode": "append",
+      "cursor_field": ["id"],
+      "user_defined_primary_key": ["id"]
+    }
+  ]
+}