137 lines
4.7 KiB
Python
137 lines
4.7 KiB
Python
import pytest
|
|
from nanoid import generate as generate_nanoid
|
|
from sqlalchemy import func, update
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from src import models
|
|
from src.crud.representation import RepresentationManager
|
|
|
|
|
|
class TestRepresentationManagerSoftDelete:
|
|
"""Tests that RepresentationManager query methods exclude soft-deleted documents."""
|
|
|
|
async def _setup(
|
|
self,
|
|
db_session: AsyncSession,
|
|
test_workspace: models.Workspace,
|
|
test_peer: models.Peer,
|
|
) -> tuple[models.Peer, models.Session, models.Collection, RepresentationManager]:
|
|
"""Create peers, session, collection, and a RepresentationManager."""
|
|
test_peer2 = models.Peer(
|
|
name=str(generate_nanoid()), workspace_name=test_workspace.name
|
|
)
|
|
db_session.add(test_peer2)
|
|
await db_session.flush()
|
|
|
|
test_session = models.Session(
|
|
name=str(generate_nanoid()), workspace_name=test_workspace.name
|
|
)
|
|
db_session.add(test_session)
|
|
await db_session.flush()
|
|
|
|
collection = models.Collection(
|
|
workspace_name=test_workspace.name,
|
|
observer=test_peer.name,
|
|
observed=test_peer2.name,
|
|
)
|
|
db_session.add(collection)
|
|
await db_session.flush()
|
|
|
|
manager = RepresentationManager(
|
|
test_workspace.name,
|
|
observer=test_peer.name,
|
|
observed=test_peer2.name,
|
|
)
|
|
|
|
return test_peer2, test_session, collection, manager
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_query_documents_recent_excludes_soft_deleted(
|
|
self,
|
|
db_session: AsyncSession,
|
|
sample_data: tuple[models.Workspace, models.Peer],
|
|
):
|
|
"""Soft-deleted documents must not appear in the recent-documents query."""
|
|
test_workspace, test_peer = sample_data
|
|
test_peer2, test_session, _, manager = await self._setup(
|
|
db_session, test_workspace, test_peer
|
|
)
|
|
|
|
# Create two documents
|
|
doc_live = models.Document(
|
|
workspace_name=test_workspace.name,
|
|
observer=test_peer.name,
|
|
observed=test_peer2.name,
|
|
content="Live observation",
|
|
session_name=test_session.name,
|
|
)
|
|
doc_deleted = models.Document(
|
|
workspace_name=test_workspace.name,
|
|
observer=test_peer.name,
|
|
observed=test_peer2.name,
|
|
content="Deleted observation",
|
|
session_name=test_session.name,
|
|
)
|
|
db_session.add_all([doc_live, doc_deleted])
|
|
await db_session.flush()
|
|
|
|
# Soft-delete one
|
|
await db_session.execute(
|
|
update(models.Document)
|
|
.where(models.Document.id == doc_deleted.id)
|
|
.values(deleted_at=func.now())
|
|
)
|
|
await db_session.commit()
|
|
|
|
results = await manager._query_documents_recent(db_session, top_k=10) # pyright: ignore[reportPrivateUsage]
|
|
|
|
result_ids = [doc.id for doc in results]
|
|
assert doc_live.id in result_ids
|
|
assert doc_deleted.id not in result_ids
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_query_documents_most_derived_excludes_soft_deleted(
|
|
self,
|
|
db_session: AsyncSession,
|
|
sample_data: tuple[models.Workspace, models.Peer],
|
|
):
|
|
"""Soft-deleted documents must not appear in the most-derived query."""
|
|
test_workspace, test_peer = sample_data
|
|
test_peer2, test_session, _, manager = await self._setup(
|
|
db_session, test_workspace, test_peer
|
|
)
|
|
|
|
# Create two documents with different times_derived
|
|
doc_live = models.Document(
|
|
workspace_name=test_workspace.name,
|
|
observer=test_peer.name,
|
|
observed=test_peer2.name,
|
|
content="Live observation",
|
|
session_name=test_session.name,
|
|
times_derived=5,
|
|
)
|
|
doc_deleted = models.Document(
|
|
workspace_name=test_workspace.name,
|
|
observer=test_peer.name,
|
|
observed=test_peer2.name,
|
|
content="Deleted high-derived observation",
|
|
session_name=test_session.name,
|
|
times_derived=100,
|
|
)
|
|
db_session.add_all([doc_live, doc_deleted])
|
|
await db_session.flush()
|
|
|
|
# Soft-delete the high-derived one
|
|
await db_session.execute(
|
|
update(models.Document)
|
|
.where(models.Document.id == doc_deleted.id)
|
|
.values(deleted_at=func.now())
|
|
)
|
|
await db_session.commit()
|
|
|
|
results = await manager._query_documents_most_derived(db_session, top_k=10) # pyright: ignore[reportPrivateUsage]
|
|
|
|
result_ids = [doc.id for doc in results]
|
|
assert doc_live.id in result_ids
|
|
assert doc_deleted.id not in result_ids
|