"""Tests for DMS Phase 11 — OCR & AI integration layer. Covers: * submit_ocr — creates OCRResult in PENDING state * record_ocr_result — transitions to COMPLETED, stores text * fail_ocr — transitions to FAILED, stores error message * OCRError — double-complete, double-fail * get_latest_ocr_result — most recent COMPLETED result * submit_ai_classification — creates AIClassification record * record_ai_classification — stores label, confidence, tags, summary * add_extracted_entities — bulk creates ExtractedEntity records * clear_extracted_entities — soft-deletes all entities for a version * Queries: list_ocr_results, get_ocr_result, list_ai_classifications, get_ai_classification, list_extracted_entities, not-found raises * HTTP API: OCR list/submit/detail/record/fail endpoints * HTTP API: AI classification list/submit/detail/record endpoints * HTTP API: entities list/bulk-add/clear endpoints * Permission gates: 401 unauthenticated, 403 no permission """ from __future__ import annotations import uuid import pytest from simorgh.apps.dms.ai import queries, services from simorgh.apps.dms.ai.iam_permissions import ( PERM_AI_SUBMIT, PERM_AI_VIEW, PERM_ENTITIES_VIEW, PERM_OCR_SUBMIT, PERM_OCR_VIEW, ) from simorgh.apps.dms.ai.models import ( AIClassification, AIClassificationType, EntityType, ExtractedEntity, OCRResult, OCRStatus, ) from simorgh.apps.dms.ai.services import OCRError from simorgh.apps.dms.common.exceptions import AssetNotFound from simorgh.apps.dms.repositories.models import Repository # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest.fixture def alice_membership(alice, tenant_acme, acme_tree, role_admin): """Override conftest alice_membership — alice gets role_admin (all perms).""" from simorgh.apps.memberships.models import Membership m = Membership.objects.create( tenant=tenant_acme, organization_node=acme_tree["root"], role=role_admin, ) m.users.add(alice) return m @pytest.fixture def charlie(db): """A plain non-superuser without membership — for 403 tests.""" from django.contrib.auth import get_user_model User = get_user_model() return User.objects.create_user( "+989000000088", password="x", email="charlie_ai@example.com", ) @pytest.fixture def repo(tenant_acme, acme_tree): return Repository.objects.create( name="AI Test Repo", slug="ai-test-repo", tenant=tenant_acme, organization_node=acme_tree["root"], ) @pytest.fixture def doc(repo, tenant_acme, acme_tree): from simorgh.apps.dms.documents.models import Document return Document.objects.create( title="AI Phase 11 Document", repository=repo, tenant=tenant_acme, organization_node=acme_tree["root"], ) @pytest.fixture def file_asset(tenant_acme, acme_tree): from simorgh.apps.dms.assets.constants import APP_CONTEXT from simorgh.apps.storage.models import FileMetadata, FileUploadStatus return FileMetadata.objects.create( filename="ai_doc.pdf", content_type="application/pdf", size_bytes=2048, path=f"test/ai/{uuid.uuid4()}.pdf", tenant=tenant_acme, organization_node=acme_tree["root"], upload_status=FileUploadStatus.READY, app_context=APP_CONTEXT, ) @pytest.fixture def version(doc, file_asset, tenant_acme, acme_tree): from simorgh.apps.dms.documents.services import add_version return add_version( document=doc, file_asset=file_asset, tenant_id=tenant_acme.pk, organization_node_id=acme_tree["root"].pk, ) @pytest.fixture def ocr_result(version, alice, tenant_acme, acme_tree): return services.submit_ocr( tenant_id=tenant_acme.pk, organization_node_id=acme_tree["root"].pk, document_version_id=version.pk, provider="tesseract", language_code="en", submitted_by=alice, ) @pytest.fixture def completed_ocr(ocr_result, alice): return services.record_ocr_result( ocr_result=ocr_result, full_text="Hello World. This is a test document.", page_count=2, confidence_score=0.95, language_code="en", processing_metadata={"engine_version": "4.1.1"}, updated_by=alice, ) # --------------------------------------------------------------------------- # Service — submit_ocr # --------------------------------------------------------------------------- @pytest.mark.django_db def test_submit_ocr_creates_pending_result(version, alice, tenant_acme, acme_tree): result = services.submit_ocr( tenant_id=tenant_acme.pk, organization_node_id=acme_tree["root"].pk, document_version_id=version.pk, provider="azure_ocr", language_code="fa", submitted_by=alice, ) assert result.pk is not None assert result.status == OCRStatus.PENDING assert result.provider == "azure_ocr" assert result.language_code == "fa" assert result.full_text == "" assert result.is_deleted is False assert result.tenant_id == tenant_acme.pk @pytest.mark.django_db def test_submit_ocr_default_provider(version, alice, tenant_acme, acme_tree): result = services.submit_ocr( tenant_id=tenant_acme.pk, organization_node_id=acme_tree["root"].pk, document_version_id=version.pk, submitted_by=alice, ) assert result.provider == "default" assert result.language_code == "" @pytest.mark.django_db def test_multiple_ocr_runs_allowed(version, alice, tenant_acme, acme_tree): """Multiple OCR runs for the same version/provider are allowed.""" r1 = services.submit_ocr( tenant_id=tenant_acme.pk, organization_node_id=acme_tree["root"].pk, document_version_id=version.pk, provider="tesseract", submitted_by=alice, ) r2 = services.submit_ocr( tenant_id=tenant_acme.pk, organization_node_id=acme_tree["root"].pk, document_version_id=version.pk, provider="tesseract", submitted_by=alice, ) assert r1.pk != r2.pk # --------------------------------------------------------------------------- # Service — record_ocr_result # --------------------------------------------------------------------------- @pytest.mark.django_db def test_record_ocr_result_transitions_to_completed(ocr_result, alice): updated = services.record_ocr_result( ocr_result=ocr_result, full_text="Extracted document text.", page_count=3, confidence_score=0.92, language_code="en", processing_metadata={"engine": "tesseract"}, updated_by=alice, ) assert updated.status == OCRStatus.COMPLETED assert updated.full_text == "Extracted document text." assert updated.page_count == 3 assert float(updated.confidence_score) == pytest.approx(0.92, abs=1e-3) assert updated.language_code == "en" assert updated.processing_metadata == {"engine": "tesseract"} assert updated.completed_at is not None @pytest.mark.django_db def test_record_ocr_result_raises_if_already_completed(completed_ocr, alice): with pytest.raises(OCRError, match="status"): services.record_ocr_result( ocr_result=completed_ocr, full_text="Second pass", updated_by=alice, ) # --------------------------------------------------------------------------- # Service — fail_ocr # --------------------------------------------------------------------------- @pytest.mark.django_db def test_fail_ocr_transitions_to_failed(ocr_result, alice): failed = services.fail_ocr( ocr_result=ocr_result, error_message="Provider timeout", updated_by=alice, ) assert failed.status == OCRStatus.FAILED assert failed.error_message == "Provider timeout" assert failed.completed_at is not None @pytest.mark.django_db def test_fail_ocr_raises_if_already_failed(ocr_result, alice): services.fail_ocr( ocr_result=ocr_result, error_message="First failure", updated_by=alice, ) ocr_result.refresh_from_db() with pytest.raises(OCRError, match="status"): services.fail_ocr( ocr_result=ocr_result, error_message="Second failure", updated_by=alice, ) # --------------------------------------------------------------------------- # Service — submit_ai_classification # --------------------------------------------------------------------------- @pytest.mark.django_db def test_submit_ai_classification_creates_record(version, alice, tenant_acme, acme_tree): cls = services.submit_ai_classification( tenant_id=tenant_acme.pk, organization_node_id=acme_tree["root"].pk, document_version_id=version.pk, classification_type=AIClassificationType.DOCUMENT_TYPE, provider="openai", created_by=alice, ) assert cls.pk is not None assert cls.classification_type == AIClassificationType.DOCUMENT_TYPE assert cls.provider == "openai" assert cls.label == "" assert cls.is_deleted is False @pytest.mark.django_db def test_submit_multiple_classification_types(version, alice, tenant_acme, acme_tree): """Different classification types for the same version are independent.""" c1 = services.submit_ai_classification( tenant_id=tenant_acme.pk, organization_node_id=acme_tree["root"].pk, document_version_id=version.pk, classification_type=AIClassificationType.TOPIC, created_by=alice, ) c2 = services.submit_ai_classification( tenant_id=tenant_acme.pk, organization_node_id=acme_tree["root"].pk, document_version_id=version.pk, classification_type=AIClassificationType.SUMMARY, created_by=alice, ) assert c1.pk != c2.pk assert c1.classification_type != c2.classification_type # --------------------------------------------------------------------------- # Service — record_ai_classification # --------------------------------------------------------------------------- @pytest.mark.django_db def test_record_ai_classification_stores_results(version, alice, tenant_acme, acme_tree): cls = services.submit_ai_classification( tenant_id=tenant_acme.pk, organization_node_id=acme_tree["root"].pk, document_version_id=version.pk, classification_type=AIClassificationType.DOCUMENT_TYPE, created_by=alice, ) updated = services.record_ai_classification( classification=cls, label="contract", confidence_score=0.87, summary="A legal contract document.", tags=["legal", "contract", "signed"], processing_metadata={"model": "gpt-4o"}, updated_by=alice, ) assert updated.label == "contract" assert float(updated.confidence_score) == pytest.approx(0.87, abs=1e-3) assert updated.summary == "A legal contract document." assert updated.tags == ["legal", "contract", "signed"] assert updated.processing_metadata == {"model": "gpt-4o"} @pytest.mark.django_db def test_record_ai_classification_can_overwrite(version, alice, tenant_acme, acme_tree): """Classification can be re-recorded to update results.""" cls = services.submit_ai_classification( tenant_id=tenant_acme.pk, organization_node_id=acme_tree["root"].pk, document_version_id=version.pk, created_by=alice, ) services.record_ai_classification( classification=cls, label="invoice", updated_by=alice, ) services.record_ai_classification( classification=cls, label="contract", confidence_score=0.99, updated_by=alice, ) cls.refresh_from_db() assert cls.label == "contract" # --------------------------------------------------------------------------- # Service — add_extracted_entities # --------------------------------------------------------------------------- @pytest.mark.django_db def test_add_extracted_entities_bulk_creates( version, alice, tenant_acme, acme_tree, completed_ocr ): entities = services.add_extracted_entities( tenant_id=tenant_acme.pk, organization_node_id=acme_tree["root"].pk, document_version_id=version.pk, entities_data=[ {"entity_type": EntityType.PERSON, "value": "John Doe"}, {"entity_type": EntityType.ORGANIZATION, "value": "Acme Corp"}, { "entity_type": EntityType.DATE, "value": "2024-01-15", "normalized_value": "2024-01-15", "confidence_score": 0.98, }, ], ocr_result=completed_ocr, created_by=alice, ) assert len(entities) == 3 types = {e.entity_type for e in entities} assert EntityType.PERSON in types assert EntityType.ORGANIZATION in types assert EntityType.DATE in types @pytest.mark.django_db def test_add_extracted_entities_links_to_ocr_source( version, alice, tenant_acme, acme_tree, completed_ocr ): entities = services.add_extracted_entities( tenant_id=tenant_acme.pk, organization_node_id=acme_tree["root"].pk, document_version_id=version.pk, entities_data=[{"entity_type": EntityType.PERSON, "value": "Jane Doe"}], ocr_result=completed_ocr, created_by=alice, ) assert entities[0].source_id == completed_ocr.pk @pytest.mark.django_db def test_add_extracted_entities_without_source( version, alice, tenant_acme, acme_tree ): entities = services.add_extracted_entities( tenant_id=tenant_acme.pk, organization_node_id=acme_tree["root"].pk, document_version_id=version.pk, entities_data=[{"entity_type": EntityType.AMOUNT, "value": "$1,000"}], created_by=alice, ) assert entities[0].source is None # --------------------------------------------------------------------------- # Service — clear_extracted_entities # --------------------------------------------------------------------------- @pytest.mark.django_db def test_clear_extracted_entities_soft_deletes_all( version, alice, tenant_acme, acme_tree ): services.add_extracted_entities( tenant_id=tenant_acme.pk, organization_node_id=acme_tree["root"].pk, document_version_id=version.pk, entities_data=[ {"entity_type": EntityType.PERSON, "value": "Entity A"}, {"entity_type": EntityType.PERSON, "value": "Entity B"}, ], created_by=alice, ) count = services.clear_extracted_entities( tenant_id=tenant_acme.pk, document_version_id=version.pk, updated_by=alice, ) assert count == 2 remaining = list( queries.list_extracted_entities(tenant_acme.pk, str(version.public_id)) ) assert remaining == [] # --------------------------------------------------------------------------- # Queries — OCRResult # --------------------------------------------------------------------------- @pytest.mark.django_db def test_list_ocr_results(tenant_acme, acme_tree, version, alice): services.submit_ocr( tenant_id=tenant_acme.pk, organization_node_id=acme_tree["root"].pk, document_version_id=version.pk, provider="p1", submitted_by=alice, ) services.submit_ocr( tenant_id=tenant_acme.pk, organization_node_id=acme_tree["root"].pk, document_version_id=version.pk, provider="p2", submitted_by=alice, ) results = list(queries.list_ocr_results(tenant_acme.pk, str(version.public_id))) assert len(results) == 2 providers = {r.provider for r in results} assert "p1" in providers assert "p2" in providers @pytest.mark.django_db def test_get_ocr_result_not_found_raises(tenant_acme): with pytest.raises(AssetNotFound): queries.get_ocr_result(tenant_acme.pk, str(uuid.uuid4())) @pytest.mark.django_db def test_get_latest_ocr_result_returns_newest_completed( version, alice, tenant_acme, acme_tree ): r1 = services.submit_ocr( tenant_id=tenant_acme.pk, organization_node_id=acme_tree["root"].pk, document_version_id=version.pk, provider="tesseract", submitted_by=alice, ) services.record_ocr_result( ocr_result=r1, full_text="First pass", updated_by=alice, ) r2 = services.submit_ocr( tenant_id=tenant_acme.pk, organization_node_id=acme_tree["root"].pk, document_version_id=version.pk, provider="tesseract", submitted_by=alice, ) services.record_ocr_result( ocr_result=r2, full_text="Second pass", updated_by=alice, ) latest = queries.get_latest_ocr_result( tenant_acme.pk, str(version.public_id), provider="tesseract" ) assert latest is not None assert latest.pk == r2.pk @pytest.mark.django_db def test_get_latest_ocr_result_none_if_no_completed( version, alice, tenant_acme, acme_tree, ocr_result ): """PENDING result should not be returned by get_latest_ocr_result.""" latest = queries.get_latest_ocr_result(tenant_acme.pk, str(version.public_id)) assert latest is None # --------------------------------------------------------------------------- # Queries — AIClassification # --------------------------------------------------------------------------- @pytest.mark.django_db def test_list_ai_classifications_filter_by_type(version, alice, tenant_acme, acme_tree): services.submit_ai_classification( tenant_id=tenant_acme.pk, organization_node_id=acme_tree["root"].pk, document_version_id=version.pk, classification_type=AIClassificationType.TOPIC, created_by=alice, ) services.submit_ai_classification( tenant_id=tenant_acme.pk, organization_node_id=acme_tree["root"].pk, document_version_id=version.pk, classification_type=AIClassificationType.SUMMARY, created_by=alice, ) topic_only = list( queries.list_ai_classifications( tenant_acme.pk, str(version.public_id), AIClassificationType.TOPIC ) ) assert len(topic_only) == 1 assert topic_only[0].classification_type == AIClassificationType.TOPIC @pytest.mark.django_db def test_get_ai_classification_not_found_raises(tenant_acme): with pytest.raises(AssetNotFound): queries.get_ai_classification(tenant_acme.pk, str(uuid.uuid4())) # --------------------------------------------------------------------------- # Queries — ExtractedEntity # --------------------------------------------------------------------------- @pytest.mark.django_db def test_list_extracted_entities_filter_by_type(version, alice, tenant_acme, acme_tree): services.add_extracted_entities( tenant_id=tenant_acme.pk, organization_node_id=acme_tree["root"].pk, document_version_id=version.pk, entities_data=[ {"entity_type": EntityType.PERSON, "value": "Alice"}, {"entity_type": EntityType.ORGANIZATION, "value": "Acme Ltd"}, ], created_by=alice, ) persons = list( queries.list_extracted_entities( tenant_acme.pk, str(version.public_id), EntityType.PERSON ) ) assert len(persons) == 1 assert persons[0].value == "Alice" # --------------------------------------------------------------------------- # HTTP API — helpers # --------------------------------------------------------------------------- def _url(path): return f"/api/v1/dms/{path}" # --------------------------------------------------------------------------- # HTTP API — OCR endpoints # --------------------------------------------------------------------------- @pytest.mark.django_db def test_api_submit_ocr(client, alice, alice_membership, tenant_acme, doc, version): client.force_login(alice) resp = client.post( _url(f"documents/{doc.public_id}/versions/{version.public_id}/ocr/"), data={"provider": "tesseract", "language_code": "en"}, content_type="application/json", HTTP_X_TENANT=tenant_acme.slug, ) assert resp.status_code == 201 data = resp.json() assert data["provider"] == "tesseract" assert data["status"] == OCRStatus.PENDING @pytest.mark.django_db def test_api_list_ocr_results(client, alice, alice_membership, tenant_acme, doc, version, ocr_result): client.force_login(alice) resp = client.get( _url(f"documents/{doc.public_id}/versions/{version.public_id}/ocr/"), HTTP_X_TENANT=tenant_acme.slug, ) assert resp.status_code == 200 ids = [r["public_id"] for r in resp.json()] assert str(ocr_result.public_id) in ids @pytest.mark.django_db def test_api_get_ocr_result(client, alice, alice_membership, tenant_acme, doc, version, ocr_result): client.force_login(alice) resp = client.get( _url(f"documents/{doc.public_id}/versions/{version.public_id}/ocr/{ocr_result.public_id}/"), HTTP_X_TENANT=tenant_acme.slug, ) assert resp.status_code == 200 assert resp.json()["public_id"] == str(ocr_result.public_id) @pytest.mark.django_db def test_api_record_ocr_result(client, alice, alice_membership, tenant_acme, doc, version, ocr_result): client.force_login(alice) resp = client.post( _url(f"documents/{doc.public_id}/versions/{version.public_id}/ocr/{ocr_result.public_id}/record/"), data={ "full_text": "Parsed document content.", "page_count": 5, "confidence_score": 0.91, }, content_type="application/json", HTTP_X_TENANT=tenant_acme.slug, ) assert resp.status_code == 200 data = resp.json() assert data["status"] == OCRStatus.COMPLETED assert data["full_text"] == "Parsed document content." assert data["page_count"] == 5 @pytest.mark.django_db def test_api_fail_ocr(client, alice, alice_membership, tenant_acme, doc, version, ocr_result): client.force_login(alice) resp = client.post( _url(f"documents/{doc.public_id}/versions/{version.public_id}/ocr/{ocr_result.public_id}/fail/"), data={"error_message": "Service unavailable"}, content_type="application/json", HTTP_X_TENANT=tenant_acme.slug, ) assert resp.status_code == 200 data = resp.json() assert data["status"] == OCRStatus.FAILED assert data["error_message"] == "Service unavailable" @pytest.mark.django_db def test_api_ocr_not_found_returns_404(client, alice, alice_membership, tenant_acme, doc, version): client.force_login(alice) fake_id = uuid.uuid4() resp = client.get( _url(f"documents/{doc.public_id}/versions/{version.public_id}/ocr/{fake_id}/"), HTTP_X_TENANT=tenant_acme.slug, ) assert resp.status_code == 404 # --------------------------------------------------------------------------- # HTTP API — AI Classification endpoints # --------------------------------------------------------------------------- @pytest.mark.django_db def test_api_submit_classification(client, alice, alice_membership, tenant_acme, doc, version): client.force_login(alice) resp = client.post( _url(f"documents/{doc.public_id}/versions/{version.public_id}/classifications/"), data={"classification_type": "document_type", "provider": "openai"}, content_type="application/json", HTTP_X_TENANT=tenant_acme.slug, ) assert resp.status_code == 201 data = resp.json() assert data["classification_type"] == "document_type" assert data["provider"] == "openai" assert data["label"] == "" @pytest.mark.django_db def test_api_list_classifications(client, alice, alice_membership, tenant_acme, doc, version): services.submit_ai_classification( tenant_id=tenant_acme.pk, organization_node_id=acme_tree["root"].pk if False else None, document_version_id=version.pk, created_by=alice, ) if False else None # Create via API instead client.force_login(alice) client.post( _url(f"documents/{doc.public_id}/versions/{version.public_id}/classifications/"), data={"classification_type": "topic"}, content_type="application/json", HTTP_X_TENANT=tenant_acme.slug, ) resp = client.get( _url(f"documents/{doc.public_id}/versions/{version.public_id}/classifications/"), HTTP_X_TENANT=tenant_acme.slug, ) assert resp.status_code == 200 assert len(resp.json()) >= 1 @pytest.mark.django_db def test_api_get_classification(client, alice, alice_membership, tenant_acme, acme_tree, doc, version): cls = services.submit_ai_classification( tenant_id=tenant_acme.pk, organization_node_id=acme_tree["root"].pk, document_version_id=version.pk, created_by=alice, ) client.force_login(alice) resp = client.get( _url(f"documents/{doc.public_id}/versions/{version.public_id}/classifications/{cls.public_id}/"), HTTP_X_TENANT=tenant_acme.slug, ) assert resp.status_code == 200 assert resp.json()["public_id"] == str(cls.public_id) @pytest.mark.django_db def test_api_record_classification(client, alice, alice_membership, tenant_acme, acme_tree, doc, version): cls = services.submit_ai_classification( tenant_id=tenant_acme.pk, organization_node_id=acme_tree["root"].pk, document_version_id=version.pk, created_by=alice, ) client.force_login(alice) resp = client.put( _url(f"documents/{doc.public_id}/versions/{version.public_id}/classifications/{cls.public_id}/record/"), data={ "label": "invoice", "confidence_score": 0.9, "tags": ["billing", "finance"], "summary": "A vendor invoice document.", }, content_type="application/json", HTTP_X_TENANT=tenant_acme.slug, ) assert resp.status_code == 200 data = resp.json() assert data["label"] == "invoice" assert data["tags"] == ["billing", "finance"] assert data["summary"] == "A vendor invoice document." # --------------------------------------------------------------------------- # HTTP API — Extracted Entities endpoints # --------------------------------------------------------------------------- @pytest.mark.django_db def test_api_bulk_add_entities(client, alice, alice_membership, tenant_acme, doc, version): client.force_login(alice) resp = client.post( _url(f"documents/{doc.public_id}/versions/{version.public_id}/entities/"), data={ "entities": [ {"entity_type": "person", "value": "Mohammed Ali"}, {"entity_type": "organization", "value": "Tehran Inc."}, { "entity_type": "amount", "value": "1,500,000 IRR", "normalized_value": "1500000", "confidence_score": 0.88, }, ] }, content_type="application/json", HTTP_X_TENANT=tenant_acme.slug, ) assert resp.status_code == 201 data = resp.json() assert len(data) == 3 entity_types = {e["entity_type"] for e in data} assert "person" in entity_types assert "organization" in entity_types @pytest.mark.django_db def test_api_list_entities(client, alice, alice_membership, tenant_acme, acme_tree, doc, version): services.add_extracted_entities( tenant_id=tenant_acme.pk, organization_node_id=acme_tree["root"].pk, document_version_id=version.pk, entities_data=[ {"entity_type": EntityType.LOCATION, "value": "Tehran"}, ], created_by=alice, ) client.force_login(alice) resp = client.get( _url(f"documents/{doc.public_id}/versions/{version.public_id}/entities/"), HTTP_X_TENANT=tenant_acme.slug, ) assert resp.status_code == 200 values = [e["value"] for e in resp.json()] assert "Tehran" in values @pytest.mark.django_db def test_api_list_entities_filter_by_type(client, alice, alice_membership, tenant_acme, acme_tree, doc, version): services.add_extracted_entities( tenant_id=tenant_acme.pk, organization_node_id=acme_tree["root"].pk, document_version_id=version.pk, entities_data=[ {"entity_type": EntityType.PERSON, "value": "Ahmad"}, {"entity_type": EntityType.DATE, "value": "1403-01-01"}, ], created_by=alice, ) client.force_login(alice) resp = client.get( _url(f"documents/{doc.public_id}/versions/{version.public_id}/entities/"), {"entity_type": "person"}, HTTP_X_TENANT=tenant_acme.slug, ) assert resp.status_code == 200 assert all(e["entity_type"] == "person" for e in resp.json()) @pytest.mark.django_db def test_api_clear_entities(client, alice, alice_membership, tenant_acme, acme_tree, doc, version): services.add_extracted_entities( tenant_id=tenant_acme.pk, organization_node_id=acme_tree["root"].pk, document_version_id=version.pk, entities_data=[ {"entity_type": EntityType.PERSON, "value": "To Delete"}, ], created_by=alice, ) client.force_login(alice) resp = client.delete( _url(f"documents/{doc.public_id}/versions/{version.public_id}/entities/"), HTTP_X_TENANT=tenant_acme.slug, ) assert resp.status_code == 200 assert resp.json()["cleared"] == 1 # List should now be empty resp2 = client.get( _url(f"documents/{doc.public_id}/versions/{version.public_id}/entities/"), HTTP_X_TENANT=tenant_acme.slug, ) assert resp2.json() == [] # --------------------------------------------------------------------------- # HTTP API — Permission gates # --------------------------------------------------------------------------- @pytest.mark.django_db def test_api_ocr_requires_auth(client, tenant_acme, doc, version): """Unauthenticated requests get 403 from PermissionGateMiddleware.""" resp = client.get( _url(f"documents/{doc.public_id}/versions/{version.public_id}/ocr/"), HTTP_X_TENANT=tenant_acme.slug, ) assert resp.status_code in (401, 403) @pytest.mark.django_db def test_api_submit_ocr_requires_perm(client, charlie, tenant_acme, doc, version): """User without ocr.submit permission gets 403.""" client.force_login(charlie) resp = client.post( _url(f"documents/{doc.public_id}/versions/{version.public_id}/ocr/"), data={"provider": "tesseract"}, content_type="application/json", HTTP_X_TENANT=tenant_acme.slug, ) assert resp.status_code == 403 @pytest.mark.django_db def test_api_view_ocr_requires_perm(client, charlie, tenant_acme, doc, version): """User without ocr.view permission gets 403.""" client.force_login(charlie) resp = client.get( _url(f"documents/{doc.public_id}/versions/{version.public_id}/ocr/"), HTTP_X_TENANT=tenant_acme.slug, ) assert resp.status_code == 403 @pytest.mark.django_db def test_api_submit_classification_requires_perm(client, charlie, tenant_acme, doc, version): """User without ai.submit permission gets 403.""" client.force_login(charlie) resp = client.post( _url(f"documents/{doc.public_id}/versions/{version.public_id}/classifications/"), data={"classification_type": "document_type"}, content_type="application/json", HTTP_X_TENANT=tenant_acme.slug, ) assert resp.status_code == 403 @pytest.mark.django_db def test_api_view_entities_requires_perm(client, charlie, tenant_acme, doc, version): """User without entities.view permission gets 403.""" client.force_login(charlie) resp = client.get( _url(f"documents/{doc.public_id}/versions/{version.public_id}/entities/"), HTTP_X_TENANT=tenant_acme.slug, ) assert resp.status_code == 403