generated from john/python-template
V3 fix document delete issue
This commit is contained in:
@@ -128,6 +128,7 @@ class DocumentService(ServiceBase):
|
||||
options=(
|
||||
selectinload(Document.jobs), # pyright: ignore[reportArgumentType]
|
||||
selectinload(Document.sources), # pyright: ignore[reportArgumentType]
|
||||
selectinload(Document.document_people), # pyright: ignore[reportArgumentType]
|
||||
),
|
||||
)
|
||||
if existing is None:
|
||||
@@ -151,6 +152,9 @@ class DocumentService(ServiceBase):
|
||||
suggestion="Remove related Sources and Jobs first, then retry deletion.",
|
||||
)
|
||||
|
||||
for link in list(existing.document_people):
|
||||
await _session.delete(link)
|
||||
|
||||
await _session.delete(existing)
|
||||
await self._finalize(session=_session, caller_session=session)
|
||||
|
||||
|
||||
@@ -450,8 +450,4 @@ def _parse_uuid(value: str | None) -> UUID | None:
|
||||
|
||||
|
||||
def _latest_prompt_name(job: Job) -> str | None:
|
||||
ordered = sorted(job.job_sources, key=lambda item: item.executed_at, reverse=True)
|
||||
for job_source in ordered:
|
||||
if job_source.job and job_source.job.prompt_name:
|
||||
return job_source.job.prompt_name
|
||||
return None
|
||||
return job.prompt_name
|
||||
@@ -41,6 +41,7 @@ async def default_settings():
|
||||
"""Provide default settings for tests."""
|
||||
settings = get_settings(database_url="sqlite:///:memory:")
|
||||
db_url = get_database_url(settings)
|
||||
await dispose_session_factory(db_url)
|
||||
engine = get_engine(database_url=db_url)
|
||||
|
||||
# Cached in-memory engines persist across tests; reset schema per test for isolation.
|
||||
|
||||
@@ -114,6 +114,39 @@ async def test_delete_document_succeeds_when_unlinked(default_session_factory, t
|
||||
await service.read_document_detail(document.id)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_document_removes_person_links(default_session_factory, tmp_path):
|
||||
service = DocumentService(session_factory=default_session_factory)
|
||||
service.settings.upload_dir = tmp_path
|
||||
|
||||
document = await service.create_document(
|
||||
Document(
|
||||
id=uuid4(),
|
||||
name="person-linked-delete",
|
||||
document_type="memo",
|
||||
)
|
||||
)
|
||||
person = await service.create_person(Person(full_name="Linked Person"))
|
||||
await service.create_document_person(
|
||||
DocumentPerson(
|
||||
document_id=document.id,
|
||||
person_id=person.id,
|
||||
role=DocumentPersonRole.AUTHOR,
|
||||
)
|
||||
)
|
||||
|
||||
document_dir = service.settings.upload_dir / "documents" / str(document.id)
|
||||
document_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
await service.delete_document(document)
|
||||
|
||||
assert not document_dir.exists()
|
||||
assert await service.list_document_people(document_id=document.id) == []
|
||||
|
||||
with pytest.raises(DocumentError):
|
||||
await service.read_document_detail(document.id)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_document_removes_populated_storage_tree(default_session_factory, tmp_path):
|
||||
service = DocumentService(session_factory=default_session_factory)
|
||||
|
||||
Reference in New Issue
Block a user