From fa5a7b9549f7b77bd22266b486fa7a5a45e1b746 Mon Sep 17 00:00:00 2001 From: Corin Date: Thu, 30 Jul 2026 14:59:12 +0100 Subject: [PATCH] Migrate alignment integration tests to use fixtures across adapters This provides better reporting for assertion errors and IDE support for test running --- .../test_asynchronous_projections.py | 8 +- tests/component/test_processing.py | 7 +- .../subscribers/stores/state/test_postgres.py | 7 +- .../stores/state/test_postgres.py | 7 +- .../strategies/distributed/test_broker.py | 7 +- .../projection/store/adapters/conftest.py | 62 ++++++++ .../store/adapters/test_adapters.py} | 147 ++++++++---------- .../store/adapters/test_postgres.py | 92 ----------- .../event/store/adapters/test_postgres.py | 19 ++- .../testcases/projection/store/harnesses.py | 81 ++++++++++ .../logicblocks/event/testsupport/db.py | 7 +- .../projection/store/adapters/test_memory.py | 25 --- 12 files changed, 233 insertions(+), 236 deletions(-) create mode 100644 tests/integration/logicblocks/event/projection/store/adapters/conftest.py rename tests/{shared/logicblocks/event/testcases/projection/store/adapters.py => integration/logicblocks/event/projection/store/adapters/test_adapters.py} (89%) delete mode 100644 tests/integration/logicblocks/event/projection/store/adapters/test_postgres.py create mode 100644 tests/shared/logicblocks/event/testcases/projection/store/harnesses.py delete mode 100644 tests/unit/logicblocks/event/projection/store/adapters/test_memory.py diff --git a/tests/component/test_asynchronous_projections.py b/tests/component/test_asynchronous_projections.py index 5fc058aa..57992a06 100644 --- a/tests/component/test_asynchronous_projections.py +++ b/tests/component/test_asynchronous_projections.py @@ -4,7 +4,7 @@ from datetime import timedelta from typing import Any, Callable, Mapping, Self -import pytest_asyncio +import pytest from logicblocks.event.testsupport import ( connection_pool, create_table, @@ -56,7 +56,7 @@ ) -@pytest_asyncio.fixture +@pytest.fixture async def open_connection_pool(): async with connection_pool(connection_settings) as pool: yield pool @@ -65,11 +65,11 @@ async def open_connection_pool(): class TestAsynchronousProjections: connection_pool: AsyncConnectionPool[AsyncConnection] - @pytest_asyncio.fixture(autouse=True) + @pytest.fixture(autouse=True) async def store_connection_pool(self, open_connection_pool): self.connection_pool = open_connection_pool - @pytest_asyncio.fixture(autouse=True) + @pytest.fixture(autouse=True) async def reinitialise_storage(self, open_connection_pool): await drop_table(open_connection_pool, "events") await drop_table(open_connection_pool, "projections") diff --git a/tests/component/test_processing.py b/tests/component/test_processing.py index a3b279b7..441cd9c6 100644 --- a/tests/component/test_processing.py +++ b/tests/component/test_processing.py @@ -5,7 +5,6 @@ from datetime import timedelta import pytest -import pytest_asyncio from logicblocks.event.testsupport import ( connection_pool, create_table, @@ -52,7 +51,7 @@ ) -@pytest_asyncio.fixture +@pytest.fixture async def open_connection_pool(): async with connection_pool(connection_settings) as pool: yield pool @@ -127,11 +126,11 @@ async def cleanup(event_broker_task): class TestEventProcessing: connection_pool: AsyncConnectionPool[AsyncConnection] - @pytest_asyncio.fixture(autouse=True) + @pytest.fixture(autouse=True) async def store_connection_pool(self, open_connection_pool): self.connection_pool = open_connection_pool - @pytest_asyncio.fixture(autouse=True) + @pytest.fixture(autouse=True) async def reinitialise_storage(self, open_connection_pool): await drop_table(open_connection_pool, "events") await drop_table(open_connection_pool, "projections") diff --git a/tests/integration/logicblocks/event/processing/broker/strategies/distributed/subscribers/stores/state/test_postgres.py b/tests/integration/logicblocks/event/processing/broker/strategies/distributed/subscribers/stores/state/test_postgres.py index 75e627a7..70e39396 100644 --- a/tests/integration/logicblocks/event/processing/broker/strategies/distributed/subscribers/stores/state/test_postgres.py +++ b/tests/integration/logicblocks/event/processing/broker/strategies/distributed/subscribers/stores/state/test_postgres.py @@ -2,7 +2,6 @@ import sys import pytest -import pytest_asyncio from logicblocks.event.testcases import ( EventSubscriberStateStoreCases, ) @@ -36,7 +35,7 @@ def read_subscriber_states_query(table: str) -> abc.Query: ) -@pytest_asyncio.fixture +@pytest.fixture async def open_connection_pool(): async with connection_pool(connection_settings) as pool: yield pool @@ -45,11 +44,11 @@ async def open_connection_pool(): class TestPostgresEventSubscriberStateStore(EventSubscriberStateStoreCases): pool: AsyncConnectionPool[AsyncConnection] - @pytest_asyncio.fixture(autouse=True) + @pytest.fixture(autouse=True) async def store_connection_pool(self, open_connection_pool): self.pool = open_connection_pool - @pytest_asyncio.fixture(autouse=True) + @pytest.fixture(autouse=True) async def reinitialise_storage(self, open_connection_pool): await drop_table(open_connection_pool, "subscribers") await create_table(open_connection_pool, "subscribers") diff --git a/tests/integration/logicblocks/event/processing/broker/strategies/distributed/subscriptions/stores/state/test_postgres.py b/tests/integration/logicblocks/event/processing/broker/strategies/distributed/subscriptions/stores/state/test_postgres.py index ed4b2225..3da6331a 100644 --- a/tests/integration/logicblocks/event/processing/broker/strategies/distributed/subscriptions/stores/state/test_postgres.py +++ b/tests/integration/logicblocks/event/processing/broker/strategies/distributed/subscriptions/stores/state/test_postgres.py @@ -2,7 +2,6 @@ import sys import pytest -import pytest_asyncio from logicblocks.event.testcases import ( EventSubscriptionStateStoreCases, ) @@ -40,7 +39,7 @@ def read_subscriber_states_query(table: str) -> abc.Query: ) -@pytest_asyncio.fixture +@pytest.fixture async def open_connection_pool(): async with connection_pool(connection_settings) as pool: yield pool @@ -51,11 +50,11 @@ class TestPostgresEventSubscriptionStateStore( ): pool: AsyncConnectionPool[AsyncConnection] - @pytest_asyncio.fixture(autouse=True) + @pytest.fixture(autouse=True) async def store_connection_pool(self, open_connection_pool): self.pool = open_connection_pool - @pytest_asyncio.fixture(autouse=True) + @pytest.fixture(autouse=True) async def reinitialise_storage(self, open_connection_pool): await drop_table(open_connection_pool, "subscriptions") await create_table(open_connection_pool, "subscriptions") diff --git a/tests/integration/logicblocks/event/processing/broker/strategies/distributed/test_broker.py b/tests/integration/logicblocks/event/processing/broker/strategies/distributed/test_broker.py index e5d8d256..fd9866e8 100644 --- a/tests/integration/logicblocks/event/processing/broker/strategies/distributed/test_broker.py +++ b/tests/integration/logicblocks/event/processing/broker/strategies/distributed/test_broker.py @@ -6,7 +6,6 @@ from datetime import timedelta import pytest -import pytest_asyncio from logicblocks.event.testsupport import ( connection_pool, create_table, @@ -71,7 +70,7 @@ async def process_event( self.events.append(event) -@pytest_asyncio.fixture +@pytest.fixture async def open_connection_pool(): async with connection_pool(connection_settings) as pool: yield pool @@ -351,11 +350,11 @@ async def fail_on_event_processing_timeout(): class TestDistributedEventBrokerManySubscribersAndNodes: - @pytest_asyncio.fixture(autouse=True) + @pytest.fixture(autouse=True) async def store_connection_pool(self, open_connection_pool): self.connection_pool = open_connection_pool - @pytest_asyncio.fixture(autouse=True) + @pytest.fixture(autouse=True) async def reinitialise_storage(self, open_connection_pool): await drop_table(open_connection_pool, "events") await drop_table(open_connection_pool, "subscribers") diff --git a/tests/integration/logicblocks/event/projection/store/adapters/conftest.py b/tests/integration/logicblocks/event/projection/store/adapters/conftest.py new file mode 100644 index 00000000..76fcd931 --- /dev/null +++ b/tests/integration/logicblocks/event/projection/store/adapters/conftest.py @@ -0,0 +1,62 @@ +import os + +import pytest +import pytest_asyncio +from logicblocks.event.testcases.projection.store.harnesses import ( + InMemoryProjectionStorageAdapterHarness, + PostgresProjectionStoreAdapterHarness, + ProjectionStoreAdapterHarness, +) +from logicblocks.event.testsupport import ( + connection_pool, + create_table, + drop_table, +) +from logicblocks.event.testsupport.db import enable_extension +from psycopg import AsyncConnection +from psycopg_pool import AsyncConnectionPool + +from logicblocks.event.persistence.postgres import ConnectionSettings + +connection_settings = ConnectionSettings( + user="admin", + password="super-secret", + host=os.getenv("DB_HOST", "localhost"), + port=int(os.getenv("DB_PORT", "5432")), + dbname="some-database", +) + + +@pytest_asyncio.fixture(scope="session", loop_scope="session") +async def open_connection_pool(): + async with connection_pool(connection_settings) as pool: + yield pool + + +async def _reinitialise_storage( + open_connection_pool: AsyncConnectionPool[AsyncConnection], +): + await enable_extension(open_connection_pool, "pg_trgm") + await drop_table(open_connection_pool, "projections") + await create_table(open_connection_pool, "projections") + + +@pytest.fixture +async def postgres_projection_store_adapter_harness(open_connection_pool): + await _reinitialise_storage(open_connection_pool) + yield PostgresProjectionStoreAdapterHarness(open_connection_pool) + + +@pytest.fixture +def in_memory_projection_storage_adapter_harness(): + return InMemoryProjectionStorageAdapterHarness() + + +@pytest.fixture( + params=[ + in_memory_projection_storage_adapter_harness, + postgres_projection_store_adapter_harness, + ] +) +def harness(request) -> ProjectionStoreAdapterHarness: + return request.getfixturevalue(request.param.__name__) diff --git a/tests/shared/logicblocks/event/testcases/projection/store/adapters.py b/tests/integration/logicblocks/event/projection/store/adapters/test_adapters.py similarity index 89% rename from tests/shared/logicblocks/event/testcases/projection/store/adapters.py rename to tests/integration/logicblocks/event/projection/store/adapters/test_adapters.py index e5e12589..225d8e86 100644 --- a/tests/shared/logicblocks/event/testcases/projection/store/adapters.py +++ b/tests/integration/logicblocks/event/projection/store/adapters/test_adapters.py @@ -1,11 +1,9 @@ -from abc import ABC, abstractmethod from collections.abc import Mapping, Sequence from dataclasses import dataclass, field from typing import Any, Callable, Self, cast import pytest -from logicblocks.event.projection.store import ProjectionStorageAdapter from logicblocks.event.query import ( FilterClause, KeySetPagingClause, @@ -26,7 +24,6 @@ from logicblocks.event.types import ( JsonValue, JsonValueConvertible, - Projection, StreamIdentifier, default_deserialisation_fallback, default_serialisation_fallback, @@ -95,43 +92,27 @@ def default_metadata_factory(self) -> Mapping[str, Any]: return {} -class Base(ABC): - @abstractmethod - def construct_storage_adapter(self) -> ProjectionStorageAdapter: - raise NotImplementedError() - - @abstractmethod - async def clear_storage(self) -> None: - raise NotImplementedError() - - @abstractmethod - async def retrieve_projections( - self, *, adapter: ProjectionStorageAdapter - ) -> Sequence[Projection[JsonValue, JsonValue]]: - raise NotImplementedError() - - -class SaveCases(Base, ABC): - async def test_stores_single_projection_for_later_retrieval(self): +class TestSaveCases: + async def test_stores_single_projection_for_later_retrieval(self, harness): projection_1_id = data.random_projection_id() - adapter = self.construct_storage_adapter() + adapter = harness.construct_storage_adapter() projection = ThingProjectionBuilder().with_id(projection_1_id).build() await adapter.save(projection=projection) - retrieved_projections = await self.retrieve_projections( + retrieved_projections = await harness.retrieve_projections( adapter=adapter ) assert retrieved_projections == [serialise_projection(projection)] - async def test_stores_many_projections_for_later_retrieval(self): + async def test_stores_many_projections_for_later_retrieval(self, harness): projection_1_id = data.random_projection_id() projection_2_id = data.random_projection_id() - adapter = self.construct_storage_adapter() + adapter = harness.construct_storage_adapter() projection_1 = ( ThingProjectionBuilder().with_id(projection_1_id).build() @@ -143,7 +124,7 @@ async def test_stores_many_projections_for_later_retrieval(self): await adapter.save(projection=projection_1) await adapter.save(projection=projection_2) - retrieved_projections = await self.retrieve_projections( + retrieved_projections = await harness.retrieve_projections( adapter=adapter ) @@ -152,7 +133,9 @@ async def test_stores_many_projections_for_later_retrieval(self): serialise_projection(projection_2), ] - async def test_updates_existing_projection_state_and_metadata(self): + async def test_updates_existing_projection_state_and_metadata( + self, harness + ): projection_name = data.random_projection_name() projection_id = data.random_projection_id() @@ -170,7 +153,7 @@ async def test_updates_existing_projection_state_and_metadata(self): projection_v2_state = Thing(value_1=10, value_2="second version") projection_v2_metadata = {"updated_at": "2024-02-02T00:00:00Z"} - adapter = self.construct_storage_adapter() + adapter = harness.construct_storage_adapter() projection_v1 = ( ThingProjectionBuilder() @@ -194,7 +177,7 @@ async def test_updates_existing_projection_state_and_metadata(self): await adapter.save(projection=projection_v1) await adapter.save(projection=provided_projection_v2) - retrieved_projections = await self.retrieve_projections( + retrieved_projections = await harness.retrieve_projections( adapter=adapter ) @@ -213,12 +196,12 @@ async def test_updates_existing_projection_state_and_metadata(self): ] -class FindOneCases(Base, ABC): - async def test_applies_single_filter_on_top_level_field(self): +class TestFindOneCases: + async def test_applies_single_filter_on_top_level_field(self, harness): projection_1_name = data.random_projection_name() projection_2_name = data.random_projection_name() - adapter = self.construct_storage_adapter() + adapter = harness.construct_storage_adapter() projection_1 = ( ThingProjectionBuilder().with_name(projection_1_name).build() @@ -243,10 +226,10 @@ async def test_applies_single_filter_on_top_level_field(self): assert located == projection_1 - async def test_applies_multiple_filters(self): + async def test_applies_multiple_filters(self, harness): projection_name = data.random_event_stream_name() - adapter = self.construct_storage_adapter() + adapter = harness.construct_storage_adapter() projection_1 = ( ThingProjectionBuilder() @@ -280,11 +263,11 @@ async def test_applies_multiple_filters(self): assert located == projection_2 - async def test_applies_single_filter_on_nested_source_field(self): + async def test_applies_single_filter_on_nested_source_field(self, harness): projection_1_name = data.random_projection_name() projection_2_name = data.random_projection_name() - adapter = self.construct_storage_adapter() + adapter = harness.construct_storage_adapter() source = StreamIdentifier( category=data.random_event_category_name(), @@ -326,10 +309,10 @@ async def test_applies_single_filter_on_nested_source_field(self): assert located == projection_1 - async def test_applies_single_filter_on_nested_state_field(self): + async def test_applies_single_filter_on_nested_state_field(self, harness): projection_name = data.random_event_stream_name() - adapter = self.construct_storage_adapter() + adapter = harness.construct_storage_adapter() filter_value = data.random_lowercase_ascii_alphabetics_string(10) projection_1 = ( @@ -361,8 +344,8 @@ async def test_applies_single_filter_on_nested_state_field(self): assert located == projection_2 - async def test_returns_none_when_no_matches(self): - adapter = self.construct_storage_adapter() + async def test_returns_none_when_no_matches(self, harness): + adapter = harness.construct_storage_adapter() located = await adapter.find_one( lookup=Lookup( @@ -379,10 +362,10 @@ async def test_returns_none_when_no_matches(self): assert located is None - async def test_raises_when_multiple_matches(self): + async def test_raises_when_multiple_matches(self, harness): projection_name = data.random_event_stream_name() - adapter = self.construct_storage_adapter() + adapter = harness.construct_storage_adapter() projection_1 = ( ThingProjectionBuilder() @@ -413,12 +396,12 @@ async def test_raises_when_multiple_matches(self): ) -class FindManyCases(Base, ABC): - async def test_applies_single_filter(self): +class TestFindManyCases: + async def test_applies_single_filter(self, harness): projection_1_name = data.random_projection_name() projection_2_name = data.random_projection_name() - adapter = self.construct_storage_adapter() + adapter = harness.construct_storage_adapter() projection_1 = ( ThingProjectionBuilder().with_name(projection_1_name).build() @@ -443,8 +426,8 @@ async def test_applies_single_filter(self): assert located == [projection_1] - async def test_applies_multiple_filters(self): - adapter = self.construct_storage_adapter() + async def test_applies_multiple_filters(self, harness): + adapter = harness.construct_storage_adapter() projection_1 = ( ThingProjectionBuilder() @@ -490,8 +473,8 @@ async def test_applies_multiple_filters(self): assert located == [projection_3, projection_4] - async def test_applies_null_filter(self): - adapter = self.construct_storage_adapter() + async def test_applies_null_filter(self, harness): + adapter = harness.construct_storage_adapter() projection_1 = ( ThingProjectionBuilder() @@ -522,8 +505,8 @@ async def test_applies_null_filter(self): assert located == [projection_1] - async def test_applies_not_filter(self): - adapter = self.construct_storage_adapter() + async def test_applies_not_filter(self, harness): + adapter = harness.construct_storage_adapter() projection_1 = ( ThingProjectionBuilder() @@ -555,8 +538,8 @@ async def test_applies_not_filter(self): assert located == other_projections - async def test_applies_simple_sorting(self): - adapter = self.construct_storage_adapter() + async def test_applies_simple_sorting(self, harness): + adapter = harness.construct_storage_adapter() projection_1 = ( ThingProjectionBuilder() @@ -594,8 +577,8 @@ async def test_applies_simple_sorting(self): assert located == [projection_3, projection_2, projection_1] - async def test_applies_function_based_sorting(self): - adapter = self.construct_storage_adapter() + async def test_applies_function_based_sorting(self, harness): + adapter = harness.construct_storage_adapter() projection_1 = ( ThingProjectionBuilder() @@ -653,8 +636,8 @@ async def test_applies_function_based_sorting(self): projection_1, ] - async def test_applies_paging(self): - adapter = self.construct_storage_adapter() + async def test_applies_paging(self, harness): + adapter = harness.construct_storage_adapter() projection_1 = ( ThingProjectionBuilder() @@ -688,8 +671,8 @@ async def test_applies_paging(self): assert located == [projection_2, projection_3] - async def test_sorts_before_paging(self): - adapter = self.construct_storage_adapter() + async def test_sorts_before_paging(self, harness): + adapter = harness.construct_storage_adapter() projection_1 = ( ThingProjectionBuilder() @@ -728,8 +711,8 @@ async def test_sorts_before_paging(self): assert located == [projection_3, projection_2] - async def test_filters_before_paging(self): - adapter = self.construct_storage_adapter() + async def test_filters_before_paging(self, harness): + adapter = harness.construct_storage_adapter() projection_1 = ( ThingProjectionBuilder() @@ -773,8 +756,8 @@ async def test_filters_before_paging(self): assert located == [projection_2, projection_3] - async def test_filters_sorts_and_pages(self): - adapter = self.construct_storage_adapter() + async def test_filters_sorts_and_pages(self, harness): + adapter = harness.construct_storage_adapter() projection_1 = ( ThingProjectionBuilder() @@ -823,8 +806,8 @@ async def test_filters_sorts_and_pages(self): assert located == [projection_4, projection_2] - async def test_filters_on_value_present_in_list(self): - adapter = self.construct_storage_adapter() + async def test_filters_on_value_present_in_list(self, harness): + adapter = harness.construct_storage_adapter() value_to_filter_1 = data.random_ascii_alphanumerics_string(10) value_to_filter_2 = data.random_ascii_alphanumerics_string(10) @@ -874,8 +857,8 @@ async def test_filters_on_value_present_in_list(self): assert located == [projection_1, projection_2] - async def test_filters_on_list_containing_value(self): - adapter = self.construct_storage_adapter() + async def test_filters_on_list_containing_value(self, harness): + adapter = harness.construct_storage_adapter() value_to_filter = data.random_ascii_alphanumerics_string(10) other_value1 = data.random_ascii_alphanumerics_string(10) @@ -943,16 +926,16 @@ async def test_filters_on_list_containing_value(self): assert located == [projection_2, projection_3] -class CountCases(Base, ABC): - async def test_returns_zero_when_no_projections(self): - adapter = self.construct_storage_adapter() +class TestCountCases: + async def test_returns_zero_when_no_projections(self, harness): + adapter = harness.construct_storage_adapter() total = await adapter.count(search=Search()) assert total == 0 - async def test_returns_count_of_single_projection(self): - adapter = self.construct_storage_adapter() + async def test_returns_count_of_single_projection(self, harness): + adapter = harness.construct_storage_adapter() projection = ThingProjectionBuilder().build() await adapter.save(projection=projection) @@ -961,8 +944,8 @@ async def test_returns_count_of_single_projection(self): assert total == 1 - async def test_returns_count_of_multiple_projections(self): - adapter = self.construct_storage_adapter() + async def test_returns_count_of_multiple_projections(self, harness): + adapter = harness.construct_storage_adapter() projection_1 = ThingProjectionBuilder().build() projection_2 = ThingProjectionBuilder().build() @@ -976,11 +959,13 @@ async def test_returns_count_of_multiple_projections(self): assert total == 3 - async def test_returns_same_count_when_updating_existing_projection(self): + async def test_returns_same_count_when_updating_existing_projection( + self, harness + ): projection_id = data.random_projection_id() projection_name = data.random_projection_name() - adapter = self.construct_storage_adapter() + adapter = harness.construct_storage_adapter() projection_v1 = ( ThingProjectionBuilder() @@ -1004,8 +989,8 @@ async def test_returns_same_count_when_updating_existing_projection(self): assert total == 1 - async def test_returns_count_of_filtered_projections(self): - adapter = self.construct_storage_adapter() + async def test_returns_count_of_filtered_projections(self, harness): + adapter = harness.construct_storage_adapter() projection_1 = ( ThingProjectionBuilder() @@ -1036,9 +1021,3 @@ async def test_returns_count_of_filtered_projections(self): total = await adapter.count(search=search) assert total == 2 - - -class ProjectionStorageAdapterCases( - SaveCases, FindOneCases, FindManyCases, CountCases, ABC -): - pass diff --git a/tests/integration/logicblocks/event/projection/store/adapters/test_postgres.py b/tests/integration/logicblocks/event/projection/store/adapters/test_postgres.py deleted file mode 100644 index d0f7767d..00000000 --- a/tests/integration/logicblocks/event/projection/store/adapters/test_postgres.py +++ /dev/null @@ -1,92 +0,0 @@ -import os -from typing import Sequence - -import pytest_asyncio -from logicblocks.event.testcases.projection.store.adapters import ( - ProjectionStorageAdapterCases, -) -from logicblocks.event.testsupport import ( - clear_table, - connection_pool, - create_table, - drop_table, -) -from logicblocks.event.testsupport.db import ( - enable_extension, -) -from psycopg import AsyncConnection, abc, sql -from psycopg.rows import dict_row -from psycopg_pool import AsyncConnectionPool - -from logicblocks.event.persistence.postgres import ConnectionSettings -from logicblocks.event.projection.store import ProjectionStorageAdapter -from logicblocks.event.projection.store.adapters import ( - PostgresProjectionStorageAdapter, -) -from logicblocks.event.types import JsonValue, Projection, identifier - -connection_settings = ConnectionSettings( - user="admin", - password="super-secret", - host=os.getenv("DB_HOST", "localhost"), - port=int(os.getenv("DB_PORT", "5432")), - dbname="some-database", -) - - -def read_projections_query(table: str) -> abc.Query: - return sql.SQL("SELECT * FROM {0}").format(sql.Identifier(table)) - - -async def read_projections( - pool: AsyncConnectionPool[AsyncConnection], - table: str, -) -> Sequence[Projection[JsonValue, JsonValue]]: - async with pool.connection() as connection: - async with connection.cursor(row_factory=dict_row) as cursor: - results = await cursor.execute(read_projections_query(table)) - return [ - Projection( - id=projection["id"], - name=projection["name"], - source=identifier.event_sequence_identifier( - projection["source"] - ), - state=projection["state"], - metadata=projection["metadata"], - ) - for projection in await results.fetchall() - ] - - -@pytest_asyncio.fixture -async def open_connection_pool(): - async with connection_pool(connection_settings) as pool: - yield pool - - -class TestPostgresProjectionStorageAdapterCommonCases( - ProjectionStorageAdapterCases -): - pool: AsyncConnectionPool[AsyncConnection] - - @pytest_asyncio.fixture(autouse=True) - async def store_connection_pool(self, open_connection_pool): - self.pool = open_connection_pool - - @pytest_asyncio.fixture(autouse=True) - async def reinitialise_storage(self, open_connection_pool): - await enable_extension(open_connection_pool, "pg_trgm") - await drop_table(open_connection_pool, "projections") - await create_table(open_connection_pool, "projections") - - def construct_storage_adapter(self) -> ProjectionStorageAdapter: - return PostgresProjectionStorageAdapter(connection_source=self.pool) - - async def clear_storage(self) -> None: - await clear_table(self.pool, "events") - - async def retrieve_projections( - self, *, adapter: ProjectionStorageAdapter - ) -> Sequence[Projection[JsonValue, JsonValue]]: - return await read_projections(self.pool, "projections") diff --git a/tests/integration/logicblocks/event/store/adapters/test_postgres.py b/tests/integration/logicblocks/event/store/adapters/test_postgres.py index a8a99406..fce5c555 100644 --- a/tests/integration/logicblocks/event/store/adapters/test_postgres.py +++ b/tests/integration/logicblocks/event/store/adapters/test_postgres.py @@ -5,7 +5,6 @@ from collections.abc import AsyncIterator, Sequence import pytest -import pytest_asyncio from logicblocks.event.testcases.store.adapters import ( ConcurrencyParameters, EventStorageAdapterCases, @@ -111,7 +110,7 @@ async def read_iterator_events( return [await anext(iterator) for _ in range(number_of_events)] -@pytest_asyncio.fixture +@pytest.fixture async def open_connection_pool(): async with connection_pool(connection_settings) as pool: yield pool @@ -120,11 +119,11 @@ async def open_connection_pool(): class TestPostgresEventStorageAdapterCommonCases(EventStorageAdapterCases): pool: AsyncConnectionPool[AsyncConnection] - @pytest_asyncio.fixture(autouse=True) + @pytest.fixture(autouse=True) async def store_connection_pool(self, open_connection_pool): self.pool = open_connection_pool - @pytest_asyncio.fixture(autouse=True) + @pytest.fixture(autouse=True) async def reinitialise_storage(self, open_connection_pool): await drop_table(open_connection_pool, "events") await create_table(open_connection_pool, "events") @@ -167,7 +166,7 @@ async def retrieve_events( class TestPostgresStorageAdapterCustomTableName: - @pytest_asyncio.fixture(autouse=True) + @pytest.fixture(autouse=True) async def store_connection_pool(self, open_connection_pool): self.pool = open_connection_pool @@ -227,16 +226,16 @@ async def test_allows_events_table_name_to_be_overridden(self): class TestPostgresStorageAdapterScanPaging: pool: AsyncConnectionPool[AsyncConnection] - @pytest_asyncio.fixture(autouse=True) + @pytest.fixture(autouse=True) async def store_connection_pool(self, open_connection_pool): self.pool = open_connection_pool - @pytest_asyncio.fixture(autouse=True) + @pytest.fixture(autouse=True) async def reinitialise_storage(self, open_connection_pool): await drop_table(open_connection_pool, "events") await create_table(open_connection_pool, "events") - @pytest_asyncio.fixture(autouse=True) + @pytest.fixture(autouse=True) async def shutdown_async_generators(self): yield @@ -621,11 +620,11 @@ async def test_pages_stream_scan_using_overridden_page_size(self): class TestPostgresStorageAdapterQueryConstraints: pool: AsyncConnectionPool[AsyncConnection] - @pytest_asyncio.fixture(autouse=True) + @pytest.fixture(autouse=True) async def store_connection_pool(self, open_connection_pool): self.pool = open_connection_pool - @pytest_asyncio.fixture(autouse=True) + @pytest.fixture(autouse=True) async def reinitialise_storage(self, open_connection_pool): await drop_table(open_connection_pool, "events") await create_table(open_connection_pool, "events") diff --git a/tests/shared/logicblocks/event/testcases/projection/store/harnesses.py b/tests/shared/logicblocks/event/testcases/projection/store/harnesses.py new file mode 100644 index 00000000..1bc065e5 --- /dev/null +++ b/tests/shared/logicblocks/event/testcases/projection/store/harnesses.py @@ -0,0 +1,81 @@ +from abc import ABC, abstractmethod +from dataclasses import dataclass +from typing import Sequence + +from logicblocks.event.testsupport import clear_table +from psycopg import AsyncConnection, sql +from psycopg.rows import dict_row +from psycopg_pool import AsyncConnectionPool + +from logicblocks.event.projection import ( + InMemoryProjectionStorageAdapter, + PostgresProjectionStorageAdapter, + ProjectionStorageAdapter, +) +from logicblocks.event.query import Search +from logicblocks.event.types import JsonValue, Projection, identifier + + +class ProjectionStoreAdapterHarness(ABC): + @abstractmethod + def construct_storage_adapter(self) -> ProjectionStorageAdapter: + raise NotImplementedError() + + @abstractmethod + async def clear_storage(self) -> None: + raise NotImplementedError() + + @abstractmethod + async def retrieve_projections( + self, *, adapter: ProjectionStorageAdapter + ) -> Sequence[Projection[JsonValue, JsonValue]]: + raise NotImplementedError() + + +@dataclass(frozen=True) +class PostgresProjectionStoreAdapterHarness(ProjectionStoreAdapterHarness): + pool: AsyncConnectionPool[AsyncConnection] + + def construct_storage_adapter(self) -> ProjectionStorageAdapter: + return PostgresProjectionStorageAdapter(connection_source=self.pool) + + async def clear_storage(self) -> None: + await clear_table(self.pool, "events") + + async def retrieve_projections( + self, *, adapter: ProjectionStorageAdapter + ) -> Sequence[Projection[JsonValue, JsonValue]]: + async with self.pool.connection() as connection: + async with connection.cursor(row_factory=dict_row) as cursor: + query = sql.SQL("SELECT * FROM {0}").format( + sql.Identifier("projections") + ) + results = await cursor.execute(query) + return [ + Projection( + id=projection["id"], + name=projection["name"], + source=identifier.event_sequence_identifier( + projection["source"] + ), + state=projection["state"], + metadata=projection["metadata"], + ) + for projection in await results.fetchall() + ] + + +@dataclass(frozen=True) +class InMemoryProjectionStorageAdapterHarness(ProjectionStoreAdapterHarness): + def construct_storage_adapter(self) -> ProjectionStorageAdapter: + return InMemoryProjectionStorageAdapter() + + async def clear_storage(self) -> None: + pass + + async def retrieve_projections( + self, + *, + adapter: ProjectionStorageAdapter, + ) -> Sequence[Projection[JsonValue]]: + return await adapter.find_many(search=Search()) diff --git a/tests/shared/logicblocks/event/testsupport/db.py b/tests/shared/logicblocks/event/testsupport/db.py index 2f5e3806..d6e636fd 100644 --- a/tests/shared/logicblocks/event/testsupport/db.py +++ b/tests/shared/logicblocks/event/testsupport/db.py @@ -91,9 +91,6 @@ async def connection_pool(connection_settings: ConnectionSettings): conninfo = connection_settings.to_connection_string() pool = AsyncConnectionPool[AsyncConnection](conninfo, open=False) - await pool.open() - - try: + async with pool: + await pool.wait() yield pool - finally: - await pool.close() diff --git a/tests/unit/logicblocks/event/projection/store/adapters/test_memory.py b/tests/unit/logicblocks/event/projection/store/adapters/test_memory.py deleted file mode 100644 index d48f762c..00000000 --- a/tests/unit/logicblocks/event/projection/store/adapters/test_memory.py +++ /dev/null @@ -1,25 +0,0 @@ -from collections.abc import Sequence - -from logicblocks.event.testcases.projection.store.adapters import ( - ProjectionStorageAdapterCases, -) - -from logicblocks.event.projection import InMemoryProjectionStorageAdapter -from logicblocks.event.projection.store import ProjectionStorageAdapter -from logicblocks.event.query import Search -from logicblocks.event.types import JsonValue, Projection - - -class TestInMemoryProjectionStorageAdapter(ProjectionStorageAdapterCases): - def construct_storage_adapter(self) -> ProjectionStorageAdapter: - return InMemoryProjectionStorageAdapter() - - async def clear_storage(self) -> None: - pass - - async def retrieve_projections( - self, - *, - adapter: ProjectionStorageAdapter, - ) -> Sequence[Projection[JsonValue]]: - return await adapter.find_many(search=Search())