tests/test_post_sync_use_case.py

from datetime import UTC, datetime

import pytest

from app.modules.posts.application.use_cases.sync_post_embeddings import SyncPostEmbeddingsUseCase
from app.modules.posts.domain.sync import PostChangeRecord, PostSourceRecord


class SourceStub:
    def __init__(self, changes):
        self.changes = changes

    async def iter_snapshot(self, page_limit=None, max_pages=None):
        if False:
            yield None

    async def iter_changes(self, after_id, page_limit=None, max_pages=None):
        for item in self.changes:
            yield item


class RepositoryStub:
    def __init__(self):
        self.checkpoint = 0
        self.upserted = []
        self.deactivated = []

    async def fetch_content_hashes(self, ids):
        return {}

    async def upsert_embeddings(self, records):
        self.upserted.extend(records)

    async def deactivate_embedding(self, post_id, source_version):
        self.deactivated.append((post_id, source_version))

    async def get_checkpoint(self, consumer):
        return self.checkpoint

    async def save_checkpoint(self, consumer, event_id):
        self.checkpoint = event_id


class EmbeddingStub:
    def embed_text(self, text):
        return [1.0, 0.0]

    def embed_batch(self, texts):
        return [[1.0, 0.0] for _ in texts]


@pytest.mark.asyncio
async def test_incremental_sync_upserts_and_deactivates_in_order() -> None:
    post = PostSourceRecord(
        "p1", "cafe", {}, "hash", True, "user", "u1", datetime.now(UTC), 1
    )
    source = SourceStub(
        [
            PostChangeRecord(1, "p1", "upsert", 1, post),
            PostChangeRecord(2, "p2", "delete", 2, None),
        ]
    )
    repository = RepositoryStub()
    use_case = SyncPostEmbeddingsUseCase(
        source, repository, EmbeddingStub(), "model", "v1", 2
    )
    result = await use_case.sync_incremental(batch_size=10)
    assert [item.source.id for item in repository.upserted] == ["p1"]
    assert repository.deactivated == [("p2", 2)]
    assert repository.checkpoint == 2
    assert result.errors == []