Skip to content

castlecraft_engineer.testing.projection_store

castlecraft_engineer.testing.projection_store

BaseProjectionStoreTest

Bases: Generic[TEventId_InMemory]

Source code in src/castlecraft_engineer/testing/projection_store.py 
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
class BaseProjectionStoreTest(Generic[TEventId_InMemory]):
    @pytest_asyncio.fixture
    @abstractmethod
    async def projection_store(
        self,
    ) -> AsyncGenerator[ProjectionStore[TEventId_InMemory], None]:
        """Yields a clean instance of the ProjectionStore."""
        raise NotImplementedError

    @pytest.fixture
    def sample_projection_id(self) -> ProjectionId:
        return f"test_projection_{uuid.uuid4()}"

    @pytest.mark.asyncio
    async def test_save_and_get_projection_state(
        self,
        projection_store: ProjectionStore[TEventId_InMemory],
        sample_projection_id: ProjectionId,
    ):
        event_id = uuid.uuid4()
        event_ts = datetime.datetime.now(datetime.timezone.utc)
        state = ProjectionState(
            projection_id=sample_projection_id,
            last_processed_event_id=event_id,
            last_processed_event_timestamp=event_ts,
        )
        await projection_store.save_projection_state(state)

        retrieved_state = await projection_store.get_projection_state(
            sample_projection_id
        )
        assert retrieved_state is not None
        assert retrieved_state.projection_id == sample_projection_id
        assert retrieved_state.last_processed_event_id == event_id
        assert retrieved_state.last_processed_event_timestamp == event_ts

    @pytest.mark.asyncio
    async def test_get_non_existent_projection_state(
        self,
        projection_store: ProjectionStore[TEventId_InMemory],
        sample_projection_id: ProjectionId,
    ):
        retrieved_state = await projection_store.get_projection_state(
            sample_projection_id
        )
        assert retrieved_state is None

    @pytest.mark.asyncio
    async def test_update_projection_state(
        self,
        projection_store: ProjectionStore[TEventId_InMemory],
        sample_projection_id: ProjectionId,
    ):
        initial_event_id = uuid.uuid4()
        initial_state = ProjectionState(
            projection_id=sample_projection_id, last_processed_event_id=initial_event_id
        )
        await projection_store.save_projection_state(initial_state)

        updated_event_id = uuid.uuid4()
        updated_ts = datetime.datetime.now(datetime.timezone.utc)

        # Simulate retrieving and updating
        current_state = await projection_store.get_projection_state(
            sample_projection_id
        )
        assert current_state is not None
        current_state.update_progress(
            event_id=updated_event_id, event_timestamp=updated_ts
        )
        await projection_store.save_projection_state(current_state)

        final_state = await projection_store.get_projection_state(sample_projection_id)
        assert final_state is not None
        assert final_state.last_processed_event_id == updated_event_id
        assert final_state.last_processed_event_timestamp == updated_ts
        assert final_state.last_updated_at > initial_state.last_updated_at

    @pytest.mark.asyncio
    async def test_clear_projection_state(
        self,
        projection_store: ProjectionStore[TEventId_InMemory],
        sample_projection_id: ProjectionId,
    ):
        state = ProjectionState(
            projection_id=sample_projection_id, last_processed_event_id=uuid.uuid4()
        )
        await projection_store.save_projection_state(state)
        assert (
            await projection_store.get_projection_state(sample_projection_id)
            is not None
        )

        await projection_store.clear_projection_state(sample_projection_id)
        assert await projection_store.get_projection_state(sample_projection_id) is None

projection_store() abstractmethod async

Yields a clean instance of the ProjectionStore.

Source code in src/castlecraft_engineer/testing/projection_store.py 
53
54
55
56
57
58
59
@pytest_asyncio.fixture
@abstractmethod
async def projection_store(
    self,
) -> AsyncGenerator[ProjectionStore[TEventId_InMemory], None]:
    """Yields a clean instance of the ProjectionStore."""
    raise NotImplementedError

InMemoryProjectionStore

Bases: ProjectionStore[TEventId_InMemory]

An in-memory implementation of the ProjectionStore for testing purposes.

Source code in src/castlecraft_engineer/testing/projection_store.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
class InMemoryProjectionStore(ProjectionStore[TEventId_InMemory]):
    """
    An in-memory implementation of the ProjectionStore for testing purposes.
    """

    def __init__(self) -> None:
        self._projection_states: Dict[ProjectionId, ProjectionState] = {}
        # Global lock for simplicity
        self._lock: asyncio.Lock = asyncio.Lock()

    async def get_projection_state(
        self, projection_id: ProjectionId
    ) -> Optional[ProjectionState]:
        async with self._lock:
            state = self._projection_states.get(projection_id)
            if state:
                return dataclasses.replace(state)  # Return a copy
            return None

    async def save_projection_state(self, projection_state: ProjectionState) -> None:
        async with self._lock:
            self._projection_states[projection_state.projection_id] = projection_state

    async def clear_projection_state(self, projection_id: ProjectionId) -> None:
        async with self._lock:
            if projection_id in self._projection_states:
                del self._projection_states[projection_id]

    async def clear_all_for_testing(self) -> None:
        """Clears all projection states from the store for testing."""
        async with self._lock:
            self._projection_states.clear()

clear_all_for_testing() async

Clears all projection states from the store for testing.

Source code in src/castlecraft_engineer/testing/projection_store.py 
46
47
48
49
async def clear_all_for_testing(self) -> None:
    """Clears all projection states from the store for testing."""
    async with self._lock:
        self._projection_states.clear()