Updated the postgresql test case

alex-au-922 · Dec 2, 2023 · 76778c0 · 76778c0
1 parent 1290ec4
commit 76778c0
Show file tree

Hide file tree

Showing 20 changed files with 887 additions and 48 deletions.
diff --git a/.env b/.env
@@ -32,5 +32,5 @@ CONSUMER_LOG_DIR=./logs/producer
 CONSUMER_LOG_RETENTION=7
 CONSUMER_LOG_ROTATION=midnight
 
-CSV_PARSER_RECOGNIZED_DATETIME_FORMATS="%Y-%m-%dT%H:%M:%S.%fZ"
+CSV_PARSER_RECOGNIZED_DATETIME_FORMATS="%Y-%m-%dT%H:%M:%S.%f%z"
 CSV_PARSER_DELIMITER=","
diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
@@ -9,12 +9,12 @@ jobs:
     outputs:
       postgres-version-tag: ${{ steps.load-dotenv.outputs.POSTGRES_VERSION_TAG }}
       postgres-port: ${{ steps.load-dotenv.outputs.POSTGRES_PORT }}
-      postgres-user: ${{ steps.load-dotenv.outputs.POSTGRES_USER }}
+      postgres-username: ${{ steps.load-dotenv.outputs.POSTGRES_USERNAME }}
       postgres-password: ${{ steps.load-dotenv.outputs.POSTGRES_PASSWORD }}
       postgres-database: ${{ steps.load-dotenv.outputs.POSTGRES_DATABASE }}
       rabbitmq-version-tag: ${{ steps.load-dotenv.outputs.RABBITMQ_VERSION_TAG }}
       rabbitmq-port: ${{ steps.load-dotenv.outputs.RABBITMQ_PORT }}
-      rabbitmq-user: ${{ steps.load-dotenv.outputs.RABBITMQ_USER }}
+      rabbitmq-username: ${{ steps.load-dotenv.outputs.RABBITMQ_USERNAME }}
       rabbitmq-password: ${{ steps.load-dotenv.outputs.RABBITMQ_PASSWORD }}
       queue-name: ${{ steps.load-dotenv.outputs.QUEUE_NAME }}
     steps:
@@ -28,12 +28,12 @@ jobs:
           set +o allexport
           echo "POSTGRES_VERSION_TAG=$POSTGRES_VERSION_TAG" >> $GITHUB_OUTPUT
           echo "POSTGRES_PORT=$POSTGRES_PORT" >> $GITHUB_OUTPUT
-          echo "POSTGRES_USER=$POSTGRES_USER" >> $GITHUB_OUTPUT
+          echo "POSTGRES_USERNAME=$POSTGRES_USERNAME" >> $GITHUB_OUTPUT
           echo "POSTGRES_PASSWORD=$POSTGRES_PASSWORD" >> $GITHUB_OUTPUT
           echo "POSTGRES_DATABASE=$POSTGRES_DATABASE" >> $GITHUB_OUTPUT
           echo "RABBITMQ_VERSION_TAG=$RABBITMQ_VERSION_TAG" >> $GITHUB_OUTPUT
           echo "RABBITMQ_PORT=$RABBITMQ_PORT" >> $GITHUB_OUTPUT
-          echo "RABBITMQ_USER=$RABBITMQ_USER" >> $GITHUB_OUTPUT
+          echo "RABBITMQ_USERNAME=$RABBITMQ_USERNAME" >> $GITHUB_OUTPUT
           echo "RABBITMQ_PASSWORD=$RABBITMQ_PASSWORD" >> $GITHUB_OUTPUT
           echo "QUEUE_NAME=$QUEUE_NAME" >> $GITHUB_OUTPUT
   test-producer:
@@ -52,7 +52,7 @@ jobs:
       rabbitmq:
         image: rabbitmq:${{ needs.load-dotenv.outputs.rabbitmq-version-tag }}
         env:
-          RABBITMQ_DEFAULT_USER: ${{ needs.load-dotenv.outputs.rabbitmq-user }}
+          RABBITMQ_DEFAULT_USER: ${{ needs.load-dotenv.outputs.rabbitmq-username }}
           RABBITMQ_DEFAULT_PASS: ${{ needs.load-dotenv.outputs.rabbitmq-password }}
         options: >-
           --health-cmd "rabbitmq-diagnostics -q check_running"
@@ -86,14 +86,9 @@ jobs:
         run: |
           coverage run -m pytest -v producer/tests
         env:
-          POSTGRES_HOST: localhost
-          POSTGRES_PORT: ${{ needs.load-dotenv.outputs.postgres-port }}
-          POSTGRES_USER: ${{ needs.load-dotenv.outputs.postgres-user }}
-          POSTGRES_PASSWORD: ${{ needs.load-dotenv.outputs.postgres-password }}
-          POSTGRES_DATABASE: ${{ needs.load-dotenv.outputs.postgres-database }}
           RABBITMQ_HOST: localhost
           RABBITMQ_PORT: ${{ needs.load-dotenv.outputs.rabbitmq-port }}
-          RABBITMQ_USER: ${{ needs.load-dotenv.outputs.rabbitmq-user }}
+          RABBITMQ_USERNAME: ${{ needs.load-dotenv.outputs.rabbitmq-username }}
           RABBITMQ_PASSWORD: ${{ needs.load-dotenv.outputs.rabbitmq-password }}
           QUEUE_NAME: ${{ needs.load-dotenv.outputs.queue-name }}
       - name: Output coverage file
@@ -117,7 +112,7 @@ jobs:
       rabbitmq:
         image: rabbitmq:${{ needs.load-dotenv.outputs.rabbitmq-version-tag }}
         env:
-          RABBITMQ_DEFAULT_USER: ${{ needs.load-dotenv.outputs.rabbitmq-user }}
+          RABBITMQ_DEFAULT_USER: ${{ needs.load-dotenv.outputs.rabbitmq-username }}
           RABBITMQ_DEFAULT_PASS: ${{ needs.load-dotenv.outputs.rabbitmq-password }}
         options: >-
           --health-cmd "rabbitmq-diagnostics -q check_running"
@@ -129,9 +124,9 @@ jobs:
       postgres:
         image: postgres:${{ needs.load-dotenv.outputs.postgres-version-tag }}
         env:
-          POSTGRES_USER: ${{ needs.load-dotenv.outputs.postgres-user }}
+          POSTGRES_USER: ${{ needs.load-dotenv.outputs.postgres-username }}
           POSTGRES_PASSWORD: ${{ needs.load-dotenv.outputs.postgres-password }}
-          POSTGRES_DB: ${{ needs.load-dotenv.outputs.postgres-database }}
+          POSTGRES_DATABASE: ${{ needs.load-dotenv.outputs.postgres-database }}
         options: >-
           --health-cmd pg_isready
           --health-interval 5s
@@ -164,12 +159,12 @@ jobs:
         env:
           POSTGRES_HOST: localhost
           POSTGRES_PORT: ${{ needs.load-dotenv.outputs.postgres-port }}
-          POSTGRES_USER: ${{ needs.load-dotenv.outputs.postgres-user }}
+          POSTGRES_USERNAME: ${{ needs.load-dotenv.outputs.postgres-username }}
           POSTGRES_PASSWORD: ${{ needs.load-dotenv.outputs.postgres-password }}
           POSTGRES_DATABASE: ${{ needs.load-dotenv.outputs.postgres-database }}
           RABBITMQ_HOST: localhost
           RABBITMQ_PORT: ${{ needs.load-dotenv.outputs.rabbitmq-port }}
-          RABBITMQ_USER: ${{ needs.load-dotenv.outputs.rabbitmq-user }}
+          RABBITMQ_USERNAME: ${{ needs.load-dotenv.outputs.rabbitmq-username }}
           RABBITMQ_PASSWORD: ${{ needs.load-dotenv.outputs.rabbitmq-password }}
           QUEUE_NAME: ${{ needs.load-dotenv.outputs.queue-name }}
       - name: Output coverage file

diff --git a/Makefile b/Makefile
@@ -25,7 +25,7 @@ test_producer:
 	export POSTGRES_PORT=$(POSTGRES_PORT) && \
 	export POSTGRES_USERNAME=$(POSTGRES_USERNAME) && \
 	export POSTGRES_PASSWORD=$(POSTGRES_PASSWORD) && \
-	export POSTGRES_DATABASE=$(POSTGRES_DB) && \
+	export POSTGRES_DATABASE=$(POSTGRES_DATABASE) && \
 	export RABBITMQ_HOST=localhost && \
 	export RABBITMQ_PORT=$(RABBITMQ_PORT) && \
 	export RABBITMQ_USERNAME=$(RABBITMQ_USERNAME) && \
@@ -37,13 +37,13 @@ test_consumer:
 	export POSTGRES_PORT=$(POSTGRES_PORT) && \
 	export POSTGRES_USERNAME=$(POSTGRES_USERNAME) && \
 	export POSTGRES_PASSWORD=$(POSTGRES_PASSWORD) && \
-	export POSTGRES_DATABASE=$(POSTGRES_DB) && \
+	export POSTGRES_DATABASE=$(POSTGRES_DATABASE) && \
 	export RABBITMQ_HOST=localhost && \
 	export RABBITMQ_PORT=$(RABBITMQ_PORT) && \
 	export RABBITMQ_USERNAME=$(RABBITMQ_USERNAME) && \
 	export RABBITMQ_PASSWORD=$(RABBITMQ_PASSWORD) && \
 	export QUEUE_NAME=$(QUEUE_NAME) && \
-	COVERAGE_FILE=.coverage_consumer coverage run -m pytest -vx consumer/tests
+	COVERAGE_FILE=.coverage_consumer coverage run -m pytest -vxs consumer/tests
 coverage_report:
 	coverage combine .coverage_producer .coverage_consumer && \
 	coverage report -m --omit="*/tests/*"

diff --git a/consumer/src/adapters/fetch_filenames/rabbitmq.py b/consumer/src/adapters/fetch_filenames/rabbitmq.py
@@ -78,8 +78,7 @@ def close(self) -> bool:
         try:
             if self._conn is not None:
                 self._conn.close()
-                return True
-            return False
+            return True
         except Exception as e:
             logging.exception(e)
             return False
diff --git a/consumer/src/adapters/upsert_iot_records/postgres.py b/consumer/src/adapters/upsert_iot_records/postgres.py
@@ -64,7 +64,7 @@ def _get_sql_stmt(self) -> str:
                 sensor_id,
                 value
             ) VALUES (
-                %(datetime)s,
+                %(record_time)s,
                 %(sensor_id)s,
                 %(value)s
             ) ON CONFLICT (record_time, sensor_id) DO UPDATE SET
@@ -133,3 +133,13 @@ def _upsert_batch(self, iot_records: Sequence[IOTRecord]) -> list[bool]:
                 self._reset_conn()
                 successes.extend([False] * len(batch))
         return successes
+
+    @override
+    def close(self) -> bool:
+        try:
+            if self._conn is not None:
+                self._conn.close()
+            return True
+        except Exception as e:
+            logging.exception(e)
+            return False
diff --git a/consumer/src/deployments/scripts/main.py b/consumer/src/deployments/scripts/main.py
@@ -25,8 +25,7 @@
 upsert_iot_records_client = PostgresUpsertIOTRecordsClient(
     host=PostgresConfig.HOST,
     port=PostgresConfig.PORT,
-    username=PostgresConfig.USERNAME,
-    password=PostgresConfig.PASSWORD,
+    credentials_service=lambda: (PostgresConfig.USERNAME, PostgresConfig.PASSWORD),
     database=PostgresConfig.DATABASE,
     batch_upsert_size=PostgresConfig.BATCH_UPSERT_SIZE,
 )

diff --git a/consumer/tests/test_adapters/test_fetch_filenames/test_rabbitmq/conftest.py b/consumer/tests/test_adapters/test_fetch_filenames/test_rabbitmq/conftest.py
@@ -0,0 +1,30 @@
+from src.adapters.fetch_filenames.rabbitmq import RabbitMQFetchFilenamesClient
+from src.deployments.scripts.config import RabbitMQConfig
+import pika
+import pytest
+from pytest import MonkeyPatch
+
+
+@pytest.fixture(scope="function")
+def rabbitmq_fetch_filenames_client() -> RabbitMQFetchFilenamesClient:
+    return RabbitMQFetchFilenamesClient(
+        host=RabbitMQConfig.HOST,
+        port=RabbitMQConfig.PORT,
+        credentials_service=lambda: (RabbitMQConfig.USERNAME, RabbitMQConfig.PASSWORD),
+        queue=RabbitMQConfig.QUEUE,
+        polling_timeout=RabbitMQConfig.POLLING_TIMEOUT,
+    )
+
+
+@pytest.fixture(scope="function")
+def raw_rabbitmq_pika_conn_config() -> tuple[pika.BaseConnection, str]:
+    pika_conn = pika.BlockingConnection(
+        pika.ConnectionParameters(
+            host=RabbitMQConfig.HOST,
+            port=RabbitMQConfig.PORT,
+            credentials=pika.PlainCredentials(
+                RabbitMQConfig.USERNAME, RabbitMQConfig.PASSWORD
+            ),
+        )
+    )
+    return pika_conn, RabbitMQConfig.QUEUE
diff --git a/consumer/tests/test_adapters/test_fetch_filenames/test_rabbitmq/test_helloworld.py b/consumer/tests/test_adapters/test_fetch_filenames/test_rabbitmq/test_helloworld.py
diff --git a/consumer/tests/test_adapters/test_upsert_iot_records/test_postgres/conftest.py b/consumer/tests/test_adapters/test_upsert_iot_records/test_postgres/conftest.py
@@ -0,0 +1,52 @@
+from src.adapters.upsert_iot_records.postgres import PostgresUpsertIOTRecordsClient
+from src.deployments.scripts.config import PostgresConfig
+import psycopg2
+import pytest
+
+
+@pytest.fixture(scope="function")
+def postgres_upsert_iot_records_client() -> PostgresUpsertIOTRecordsClient:
+    return PostgresUpsertIOTRecordsClient(
+        host=PostgresConfig.HOST,
+        port=PostgresConfig.PORT,
+        credentials_service=lambda: (PostgresConfig.USERNAME, PostgresConfig.PASSWORD),
+        database=PostgresConfig.DATABASE,
+        batch_upsert_size=PostgresConfig.BATCH_UPSERT_SIZE,
+    )
+
+
+@pytest.fixture(scope="function")
+def raw_postgres_psycopg2_conn_config() -> psycopg2.extensions.connection:
+    with psycopg2.connect(
+        host=PostgresConfig.HOST,
+        port=PostgresConfig.PORT,
+        user=PostgresConfig.USERNAME,
+        password=PostgresConfig.PASSWORD,
+        database=PostgresConfig.DATABASE,
+    ) as conn:
+        yield conn
+
+
+@pytest.fixture(scope="function", autouse=True)
+def setup_teardown_postgres_tables(
+    raw_postgres_psycopg2_conn_config: psycopg2.extensions.connection,
+) -> None:
+    with raw_postgres_psycopg2_conn_config.cursor() as cursor:
+        try:
+            cursor.execute(
+                """
+                TRUNCATE TABLE records;
+                """
+            )
+            raw_postgres_psycopg2_conn_config.commit()
+            yield
+        except Exception as e:
+            raw_postgres_psycopg2_conn_config.rollback()
+            raise e
+        finally:
+            cursor.execute(
+                """
+                TRUNCATE TABLE records;
+                """
+            )
+            raw_postgres_psycopg2_conn_config.commit()
diff --git a/consumer/tests/test_adapters/test_upsert_iot_records/test_postgres/test_close_conn_failed.py b/consumer/tests/test_adapters/test_upsert_iot_records/test_postgres/test_close_conn_failed.py
@@ -0,0 +1,19 @@
+from pytest import LogCaptureFixture
+from src.adapters.upsert_iot_records.postgres import PostgresUpsertIOTRecordsClient
+from .utils import random_iot_records, MockedPostgresConnection
+import pytest
+
+
+def test_close_conn_failed(
+    postgres_upsert_iot_records_client: PostgresUpsertIOTRecordsClient,
+    caplog: LogCaptureFixture,
+):
+    postgres_upsert_iot_records_client.upsert(random_iot_records()[0])
+
+    assert postgres_upsert_iot_records_client._conn is not None
+
+    postgres_upsert_iot_records_client._conn = MockedPostgresConnection()
+
+    with caplog.at_level("ERROR"):
+        assert not postgres_upsert_iot_records_client.close()
+        assert "Failed to close!" in caplog.text
diff --git a/...r/tests/test_adapters/test_upsert_iot_records/test_postgres/test_close_conn_successful.py b/...r/tests/test_adapters/test_upsert_iot_records/test_postgres/test_close_conn_successful.py
@@ -0,0 +1,16 @@
+from src.adapters.upsert_iot_records.postgres import PostgresUpsertIOTRecordsClient
+from .utils import random_iot_records
+
+
+def test_close_conn_successful(
+    postgres_upsert_iot_records_client: PostgresUpsertIOTRecordsClient,
+):
+    postgres_upsert_iot_records_client.upsert(random_iot_records()[0])
+    assert postgres_upsert_iot_records_client._conn is not None
+    assert postgres_upsert_iot_records_client.close()
+
+
+def test_none_conn_close_successful(
+    postgres_upsert_iot_records_client: PostgresUpsertIOTRecordsClient,
+):
+    assert postgres_upsert_iot_records_client.close()