"""Tests for Admin Graph Management API endpoints. Covers all 26 endpoints under /api/v1/admin/graph: - Stats & Health (2) - Entity Type CRUD (4) - Relationship Type CRUD (4) - Entity CRUD (5) - Duplicate detection & Merge (2) - Rebuild (3) - Relationships (3) - Placeholders (3) """ import pytest from httpx import AsyncClient # ── Statistics & Health ────────────────────────────────────────────────────── @pytest.mark.asyncio class TestGraphStats: """GET /api/v1/admin/graph/stats — graph-level statistics.""" async def test_stats_requires_auth(self, client: AsyncClient, api_prefix: str): resp = await client.get(f"{api_prefix}/admin/graph/stats") assert resp.status_code in (401, 403) async def test_stats_returns_200(self, client: AsyncClient, api_prefix: str, auth_headers: dict): resp = await client.get(f"{api_prefix}/admin/graph/stats", headers=auth_headers) assert resp.status_code == 200 body = resp.json() assert "total_nodes" in body assert "total_relationships" in body assert "orphan_nodes" in body assert "node_counts" in body assert "rel_counts" in body assert isinstance(body["node_counts"], list) assert isinstance(body["rel_counts"], list) @pytest.mark.asyncio class TestGraphHealth: """GET /api/v1/admin/graph/health — data quality health metrics.""" async def test_health_requires_auth(self, client: AsyncClient, api_prefix: str): resp = await client.get(f"{api_prefix}/admin/graph/health") assert resp.status_code in (401, 403) async def test_health_returns_200(self, client: AsyncClient, api_prefix: str, auth_headers: dict): resp = await client.get(f"{api_prefix}/admin/graph/health", headers=auth_headers) assert resp.status_code == 200 body = resp.json() assert "no_relation_ratio" in body assert "duplicate_candidates" in body assert "missing_in_graph" in body assert "total_es_docs" in body assert isinstance(body["no_relation_ratio"], (int, float)) # ── Entity Type Definitions ────────────────────────────────────────────────── @pytest.mark.asyncio class TestEntityTypes: """CRUD for entity type definitions: /api/v1/admin/graph/entity-types.""" async def test_list_entity_types_requires_auth(self, client: AsyncClient, api_prefix: str): resp = await client.get(f"{api_prefix}/admin/graph/entity-types") assert resp.status_code in (401, 403) async def test_list_entity_types(self, client: AsyncClient, api_prefix: str, auth_headers: dict): resp = await client.get(f"{api_prefix}/admin/graph/entity-types", headers=auth_headers) assert resp.status_code == 200 body = resp.json() assert "items" in body assert isinstance(body["items"], list) async def test_create_entity_type(self, client: AsyncClient, api_prefix: str, auth_headers: dict): resp = await client.post( f"{api_prefix}/admin/graph/entity-types", json={ "name": "TestEntityType_XYZ", "description": "测试用实体类型", "icon": "mdi-test", "color": "#FF5722", "properties_schema": {"name": {"type": "string"}}, }, headers=auth_headers, ) assert resp.status_code == 201 body = resp.json() assert body["name"] == "TestEntityType_XYZ" assert body["description"] == "测试用实体类型" # Clean up: delete the created entity type await client.delete( f"{api_prefix}/admin/graph/entity-types/TestEntityType_XYZ", headers=auth_headers, ) async def test_create_entity_type_requires_auth(self, client: AsyncClient, api_prefix: str): resp = await client.post( f"{api_prefix}/admin/graph/entity-types", json={"name": "NoAuth", "description": ""}, ) assert resp.status_code in (401, 403) async def test_create_duplicate_entity_type_409(self, client: AsyncClient, api_prefix: str, auth_headers: dict): """Creating a duplicate entity type should return 409 Conflict.""" # Create first await client.post( f"{api_prefix}/admin/graph/entity-types", json={"name": "DupTest_XYZ", "description": "first"}, headers=auth_headers, ) # Attempt duplicate resp = await client.post( f"{api_prefix}/admin/graph/entity-types", json={"name": "DupTest_XYZ", "description": "duplicate"}, headers=auth_headers, ) assert resp.status_code == 409 # Clean up await client.delete( f"{api_prefix}/admin/graph/entity-types/DupTest_XYZ", headers=auth_headers, ) async def test_update_entity_type(self, client: AsyncClient, api_prefix: str, auth_headers: dict): # Create await client.post( f"{api_prefix}/admin/graph/entity-types", json={"name": "UpdTest_XYZ", "description": "original"}, headers=auth_headers, ) # Update resp = await client.put( f"{api_prefix}/admin/graph/entity-types/UpdTest_XYZ", json={"description": "updated", "color": "#00BCD4"}, headers=auth_headers, ) assert resp.status_code == 200 # Clean up await client.delete( f"{api_prefix}/admin/graph/entity-types/UpdTest_XYZ", headers=auth_headers, ) async def test_update_entity_type_requires_auth(self, client: AsyncClient, api_prefix: str): resp = await client.put( f"{api_prefix}/admin/graph/entity-types/SomeType", json={"description": "nope"}, ) assert resp.status_code in (401, 403) async def test_delete_entity_type(self, client: AsyncClient, api_prefix: str, auth_headers: dict): # Create await client.post( f"{api_prefix}/admin/graph/entity-types", json={"name": "DelTest_XYZ", "description": "to delete"}, headers=auth_headers, ) # Delete resp = await client.delete( f"{api_prefix}/admin/graph/entity-types/DelTest_XYZ", headers=auth_headers, ) assert resp.status_code == 200 async def test_delete_entity_type_requires_auth(self, client: AsyncClient, api_prefix: str): resp = await client.delete(f"{api_prefix}/admin/graph/entity-types/SomeType") assert resp.status_code in (401, 403) # ── Relationship Type Definitions ──────────────────────────────────────────── @pytest.mark.asyncio class TestRelTypes: """CRUD for relationship type definitions: /api/v1/admin/graph/relationship-types.""" async def test_list_rel_types_requires_auth(self, client: AsyncClient, api_prefix: str): resp = await client.get(f"{api_prefix}/admin/graph/relationship-types") assert resp.status_code in (401, 403) async def test_list_rel_types(self, client: AsyncClient, api_prefix: str, auth_headers: dict): resp = await client.get(f"{api_prefix}/admin/graph/relationship-types", headers=auth_headers) assert resp.status_code == 200 body = resp.json() assert "items" in body assert isinstance(body["items"], list) async def test_create_rel_type(self, client: AsyncClient, api_prefix: str, auth_headers: dict): resp = await client.post( f"{api_prefix}/admin/graph/relationship-types", json={ "name": "TEST_REL_XYZ", "description": "测试关系类型", "source_labels": ["Organization"], "target_labels": ["Person"], }, headers=auth_headers, ) assert resp.status_code == 201 body = resp.json() assert body["name"] == "TEST_REL_XYZ" # Clean up await client.delete( f"{api_prefix}/admin/graph/relationship-types/TEST_REL_XYZ", headers=auth_headers, ) async def test_create_rel_type_requires_auth(self, client: AsyncClient, api_prefix: str): resp = await client.post( f"{api_prefix}/admin/graph/relationship-types", json={"name": "NoAuth", "description": ""}, ) assert resp.status_code in (401, 403) async def test_create_duplicate_rel_type_409(self, client: AsyncClient, api_prefix: str, auth_headers: dict): await client.post( f"{api_prefix}/admin/graph/relationship-types", json={"name": "DupRel_XYZ", "description": "first"}, headers=auth_headers, ) resp = await client.post( f"{api_prefix}/admin/graph/relationship-types", json={"name": "DupRel_XYZ", "description": "dup"}, headers=auth_headers, ) assert resp.status_code == 409 await client.delete( f"{api_prefix}/admin/graph/relationship-types/DupRel_XYZ", headers=auth_headers, ) async def test_update_rel_type(self, client: AsyncClient, api_prefix: str, auth_headers: dict): await client.post( f"{api_prefix}/admin/graph/relationship-types", json={"name": "UpdRel_XYZ", "description": "original"}, headers=auth_headers, ) resp = await client.put( f"{api_prefix}/admin/graph/relationship-types/UpdRel_XYZ", json={"description": "updated"}, headers=auth_headers, ) assert resp.status_code == 200 await client.delete( f"{api_prefix}/admin/graph/relationship-types/UpdRel_XYZ", headers=auth_headers, ) async def test_update_rel_type_requires_auth(self, client: AsyncClient, api_prefix: str): resp = await client.put( f"{api_prefix}/admin/graph/relationship-types/SomeType", json={"description": "nope"}, ) assert resp.status_code in (401, 403) async def test_delete_rel_type(self, client: AsyncClient, api_prefix: str, auth_headers: dict): await client.post( f"{api_prefix}/admin/graph/relationship-types", json={"name": "DelRel_XYZ", "description": "to delete"}, headers=auth_headers, ) resp = await client.delete( f"{api_prefix}/admin/graph/relationship-types/DelRel_XYZ", headers=auth_headers, ) assert resp.status_code == 200 async def test_delete_rel_type_requires_auth(self, client: AsyncClient, api_prefix: str): resp = await client.delete(f"{api_prefix}/admin/graph/relationship-types/SomeType") assert resp.status_code in (401, 403) # ── Entity CRUD ────────────────────────────────────────────────────────────── @pytest.mark.asyncio class TestEntityCRUD: """Entity management: /api/v1/admin/graph/entities.""" async def test_list_entities_requires_auth(self, client: AsyncClient, api_prefix: str): resp = await client.get(f"{api_prefix}/admin/graph/entities") assert resp.status_code in (401, 403) async def test_list_entities(self, client: AsyncClient, api_prefix: str, auth_headers: dict): resp = await client.get(f"{api_prefix}/admin/graph/entities", headers=auth_headers) assert resp.status_code == 200 body = resp.json() assert "items" in body assert "total" in body assert "page" in body assert "page_size" in body assert isinstance(body["items"], list) async def test_list_entities_with_filters(self, client: AsyncClient, api_prefix: str, auth_headers: dict): """Test entity listing with label filter and pagination.""" resp = await client.get( f"{api_prefix}/admin/graph/entities", params={"label": "Organization", "page": 1, "page_size": 5, "sort_by": "connection_count"}, headers=auth_headers, ) assert resp.status_code == 200 body = resp.json() assert body["page"] == 1 assert body["page_size"] == 5 async def test_list_entities_name_search(self, client: AsyncClient, api_prefix: str, auth_headers: dict): """Test entity listing with name search.""" resp = await client.get( f"{api_prefix}/admin/graph/entities", params={"name": "广东"}, headers=auth_headers, ) assert resp.status_code == 200 async def test_create_entity(self, client: AsyncClient, api_prefix: str, auth_headers: dict): resp = await client.post( f"{api_prefix}/admin/graph/entities", json={ "label": "Organization", "properties": {"name": "测试实体_XYZ_CREATE"}, }, headers=auth_headers, ) assert resp.status_code == 201 body = resp.json() assert "id" in body assert "labels" in body entity_id = body["id"] # Clean up await client.delete( f"{api_prefix}/admin/graph/entities/{entity_id}", headers=auth_headers, ) async def test_create_entity_requires_auth(self, client: AsyncClient, api_prefix: str): resp = await client.post( f"{api_prefix}/admin/graph/entities", json={"label": "Organization", "properties": {"name": "NoAuth"}}, ) assert resp.status_code in (401, 403) async def test_get_entity_detail(self, client: AsyncClient, api_prefix: str, auth_headers: dict): """Create an entity and get its detail.""" # Create create_resp = await client.post( f"{api_prefix}/admin/graph/entities", json={ "label": "Organization", "properties": {"name": "详情测试_XYZ"}, }, headers=auth_headers, ) assert create_resp.status_code == 201 entity_id = create_resp.json()["id"] # Get detail resp = await client.get( f"{api_prefix}/admin/graph/entities/{entity_id}", headers=auth_headers, ) assert resp.status_code == 200 body = resp.json() assert body["id"] == entity_id assert "labels" in body assert "properties" in body assert "neighbors" in body assert "related_docs" in body # Clean up await client.delete( f"{api_prefix}/admin/graph/entities/{entity_id}", headers=auth_headers, ) async def test_get_entity_detail_requires_auth(self, client: AsyncClient, api_prefix: str): resp = await client.get(f"{api_prefix}/admin/graph/entities/nonexistent_id") assert resp.status_code in (401, 403) async def test_get_entity_not_found(self, client: AsyncClient, api_prefix: str, auth_headers: dict): resp = await client.get( f"{api_prefix}/admin/graph/entities/nonexistent_entity_99999", headers=auth_headers, ) assert resp.status_code == 404 async def test_update_entity(self, client: AsyncClient, api_prefix: str, auth_headers: dict): # Create create_resp = await client.post( f"{api_prefix}/admin/graph/entities", json={ "label": "Person", "properties": {"name": "更新测试_XYZ"}, }, headers=auth_headers, ) assert create_resp.status_code == 201 entity_id = create_resp.json()["id"] # Update resp = await client.put( f"{api_prefix}/admin/graph/entities/{entity_id}", json={"properties": {"name": "更新后_XYZ", "alias": "test"}}, headers=auth_headers, ) assert resp.status_code == 200 # Clean up await client.delete( f"{api_prefix}/admin/graph/entities/{entity_id}", headers=auth_headers, ) async def test_update_entity_requires_auth(self, client: AsyncClient, api_prefix: str): resp = await client.put( f"{api_prefix}/admin/graph/entities/some_id", json={"properties": {"name": "nope"}}, ) assert resp.status_code in (401, 403) async def test_update_entity_not_found(self, client: AsyncClient, api_prefix: str, auth_headers: dict): resp = await client.put( f"{api_prefix}/admin/graph/entities/nonexistent_entity_99999", json={"properties": {"name": "ghost"}}, headers=auth_headers, ) assert resp.status_code == 404 async def test_delete_entity(self, client: AsyncClient, api_prefix: str, auth_headers: dict): # Create create_resp = await client.post( f"{api_prefix}/admin/graph/entities", json={ "label": "Organization", "properties": {"name": "删除测试_XYZ"}, }, headers=auth_headers, ) assert create_resp.status_code == 201 entity_id = create_resp.json()["id"] # Delete resp = await client.delete( f"{api_prefix}/admin/graph/entities/{entity_id}", headers=auth_headers, ) assert resp.status_code == 200 body = resp.json() assert body["id"] == entity_id assert "deleted_relationships" in body async def test_delete_entity_requires_auth(self, client: AsyncClient, api_prefix: str): resp = await client.delete(f"{api_prefix}/admin/graph/entities/some_id") assert resp.status_code in (401, 403) # ── Duplicate Detection & Entity Merge ─────────────────────────────────────── @pytest.mark.asyncio class TestDuplicatesAndMerge: """Duplicate detection & entity merge endpoints.""" async def test_duplicates_requires_auth(self, client: AsyncClient, api_prefix: str): resp = await client.get(f"{api_prefix}/admin/graph/duplicates") assert resp.status_code in (401, 403) async def test_detect_duplicates(self, client: AsyncClient, api_prefix: str, auth_headers: dict): resp = await client.get( f"{api_prefix}/admin/graph/duplicates", headers=auth_headers, ) assert resp.status_code == 200 body = resp.json() assert "candidates" in body assert "total" in body assert isinstance(body["candidates"], list) async def test_detect_duplicates_with_params(self, client: AsyncClient, api_prefix: str, auth_headers: dict): """Test duplicate detection with label filter and threshold.""" resp = await client.get( f"{api_prefix}/admin/graph/duplicates", params={"label": "Organization", "threshold": 0.9, "limit": 10}, headers=auth_headers, ) assert resp.status_code == 200 body = resp.json() assert body["total"] <= 10 async def test_merge_entities_requires_auth(self, client: AsyncClient, api_prefix: str): resp = await client.post( f"{api_prefix}/admin/graph/entities/merge", json={"primary_id": "a", "secondary_id": "b"}, ) assert resp.status_code in (401, 403) async def test_merge_entities(self, client: AsyncClient, api_prefix: str, auth_headers: dict): """Create two entities, merge them, verify result.""" # Create primary r1 = await client.post( f"{api_prefix}/admin/graph/entities", json={"label": "Organization", "properties": {"name": "合并主体_XYZ"}}, headers=auth_headers, ) assert r1.status_code == 201 primary_id = r1.json()["id"] # Create secondary r2 = await client.post( f"{api_prefix}/admin/graph/entities", json={"label": "Organization", "properties": {"name": "合并副体_XYZ"}}, headers=auth_headers, ) assert r2.status_code == 201 secondary_id = r2.json()["id"] # Merge resp = await client.post( f"{api_prefix}/admin/graph/entities/merge", json={ "primary_id": primary_id, "secondary_id": secondary_id, "add_alias": True, }, headers=auth_headers, ) assert resp.status_code == 200 body = resp.json() assert body["primary_id"] == primary_id assert "migrated_relationships" in body assert "aliases" in body # Clean up primary await client.delete( f"{api_prefix}/admin/graph/entities/{primary_id}", headers=auth_headers, ) async def test_merge_invalid_ids(self, client: AsyncClient, api_prefix: str, auth_headers: dict): resp = await client.post( f"{api_prefix}/admin/graph/entities/merge", json={ "primary_id": "nonexistent_aaa", "secondary_id": "nonexistent_bbb", }, headers=auth_headers, ) assert resp.status_code == 400 # ── Rebuild ────────────────────────────────────────────────────────────────── @pytest.mark.asyncio class TestRebuild: """Graph rebuild endpoints: rebuild, rebuild-all, rebuild-status.""" async def test_rebuild_requires_auth(self, client: AsyncClient, api_prefix: str): resp = await client.post( f"{api_prefix}/admin/graph/rebuild", json={"doc_ids": ["test"]}, ) assert resp.status_code in (401, 403) async def test_rebuild_doc_graphs(self, client: AsyncClient, api_prefix: str, auth_headers: dict): resp = await client.post( f"{api_prefix}/admin/graph/rebuild", json={"doc_ids": ["test-doc-rebuild-xyz"]}, headers=auth_headers, ) assert resp.status_code == 200 body = resp.json() assert "task_id" in body assert body["doc_count"] == 1 async def test_rebuild_empty_doc_ids_rejected(self, client: AsyncClient, api_prefix: str, auth_headers: dict): """Empty doc_ids list should be rejected (422).""" resp = await client.post( f"{api_prefix}/admin/graph/rebuild", json={"doc_ids": []}, headers=auth_headers, ) assert resp.status_code == 422 async def test_rebuild_all_requires_auth(self, client: AsyncClient, api_prefix: str): resp = await client.post(f"{api_prefix}/admin/graph/rebuild-all") assert resp.status_code in (401, 403) async def test_rebuild_all(self, client: AsyncClient, api_prefix: str, auth_headers: dict): resp = await client.post( f"{api_prefix}/admin/graph/rebuild-all", headers=auth_headers, ) assert resp.status_code == 200 body = resp.json() assert "task_id" in body assert "total_docs" in body async def test_rebuild_status_requires_auth(self, client: AsyncClient, api_prefix: str): resp = await client.get(f"{api_prefix}/admin/graph/rebuild-status/fake-task-id") assert resp.status_code in (401, 403) async def test_rebuild_status(self, client: AsyncClient, api_prefix: str, auth_headers: dict): """Check status of a non-existent task (should return PENDING).""" resp = await client.get( f"{api_prefix}/admin/graph/rebuild-status/nonexistent-task-xyz", headers=auth_headers, ) assert resp.status_code == 200 body = resp.json() assert body["task_id"] == "nonexistent-task-xyz" assert body["status"] == "PENDING" # ── Relationships ──────────────────────────────────────────────────────────── @pytest.mark.asyncio class TestRelationshipManagement: """Relationship management: /api/v1/admin/graph/relationships.""" async def test_list_relationships_requires_auth(self, client: AsyncClient, api_prefix: str): resp = await client.get(f"{api_prefix}/admin/graph/relationships") assert resp.status_code in (401, 403) async def test_list_relationships(self, client: AsyncClient, api_prefix: str, auth_headers: dict): resp = await client.get( f"{api_prefix}/admin/graph/relationships", headers=auth_headers, ) assert resp.status_code == 200 body = resp.json() assert "items" in body assert "total" in body assert "page" in body assert "page_size" in body async def test_list_relationships_with_filters(self, client: AsyncClient, api_prefix: str, auth_headers: dict): resp = await client.get( f"{api_prefix}/admin/graph/relationships", params={"page": 1, "page_size": 5}, headers=auth_headers, ) assert resp.status_code == 200 async def test_create_relationship_requires_auth(self, client: AsyncClient, api_prefix: str): resp = await client.post( f"{api_prefix}/admin/graph/relationships", json={"source_id": "a", "target_id": "b", "type": "RELATED"}, ) assert resp.status_code in (401, 403) async def test_create_and_delete_relationship(self, client: AsyncClient, api_prefix: str, auth_headers: dict): """Create two entities with a custom rel type, link them, then clean up.""" # Create a custom relationship type first (default types may not match our entities) await client.post( f"{api_prefix}/admin/graph/relationship-types", json={ "name": "TEST_LINK_XYZ", "description": "test relationship", "source_labels": ["Subject"], "target_labels": ["Subject"], }, headers=auth_headers, ) # Create source entity r1 = await client.post( f"{api_prefix}/admin/graph/entities", json={"label": "Subject", "properties": {"name": "关系测试源_XYZ"}}, headers=auth_headers, ) assert r1.status_code == 201 source_id = r1.json()["id"] # Create target entity r2 = await client.post( f"{api_prefix}/admin/graph/entities", json={"label": "Subject", "properties": {"name": "关系测试目标_XYZ"}}, headers=auth_headers, ) assert r2.status_code == 201 target_id = r2.json()["id"] # Create relationship using our custom type resp = await client.post( f"{api_prefix}/admin/graph/relationships", json={ "source_id": source_id, "target_id": target_id, "type": "TEST_LINK_XYZ", "properties": {"since": "2024"}, }, headers=auth_headers, ) assert resp.status_code == 201 body = resp.json() assert "id" in body assert body["type"] == "TEST_LINK_XYZ" assert body["source_id"] == source_id assert body["target_id"] == target_id rel_id = body["id"] # Delete relationship del_resp = await client.delete( f"{api_prefix}/admin/graph/relationships/{rel_id}", headers=auth_headers, ) assert del_resp.status_code == 200 assert del_resp.json()["deleted"] is True # Clean up entities and relationship type await client.delete(f"{api_prefix}/admin/graph/entities/{source_id}", headers=auth_headers) await client.delete(f"{api_prefix}/admin/graph/entities/{target_id}", headers=auth_headers) await client.delete(f"{api_prefix}/admin/graph/relationship-types/TEST_LINK_XYZ", headers=auth_headers) async def test_delete_relationship_requires_auth(self, client: AsyncClient, api_prefix: str): resp = await client.delete(f"{api_prefix}/admin/graph/relationships/some-rel-id") assert resp.status_code in (401, 403) async def test_delete_relationship_not_found(self, client: AsyncClient, api_prefix: str, auth_headers: dict): resp = await client.delete( f"{api_prefix}/admin/graph/relationships/nonexistent_rel_99999", headers=auth_headers, ) assert resp.status_code == 404 async def test_create_relationship_invalid_ids(self, client: AsyncClient, api_prefix: str, auth_headers: dict): """Creating a relationship with invalid entity IDs should fail.""" resp = await client.post( f"{api_prefix}/admin/graph/relationships", json={ "source_id": "fake_source_xyz", "target_id": "fake_target_xyz", "type": "MENTIONS", }, headers=auth_headers, ) assert resp.status_code == 400 # ── Placeholders ───────────────────────────────────────────────────────────── @pytest.mark.asyncio class TestPlaceholders: """Placeholder management: /api/v1/admin/graph/placeholders.""" async def test_list_placeholders_requires_auth(self, client: AsyncClient, api_prefix: str): resp = await client.get(f"{api_prefix}/admin/graph/placeholders") assert resp.status_code in (401, 403) async def test_list_placeholders(self, client: AsyncClient, api_prefix: str, auth_headers: dict): resp = await client.get( f"{api_prefix}/admin/graph/placeholders", headers=auth_headers, ) assert resp.status_code == 200 body = resp.json() assert "items" in body assert "total" in body assert isinstance(body["items"], list) async def test_link_placeholder_requires_auth(self, client: AsyncClient, api_prefix: str): resp = await client.post( f"{api_prefix}/admin/graph/placeholders/some-id/link", json={"real_doc_id": "doc123"}, ) assert resp.status_code in (401, 403) async def test_link_placeholder_invalid(self, client: AsyncClient, api_prefix: str, auth_headers: dict): """Linking a non-existent placeholder should return 400.""" resp = await client.post( f"{api_prefix}/admin/graph/placeholders/nonexistent_placeholder_xyz/link", json={"real_doc_id": "doc123"}, headers=auth_headers, ) assert resp.status_code == 400 async def test_delete_placeholder_requires_auth(self, client: AsyncClient, api_prefix: str): resp = await client.delete(f"{api_prefix}/admin/graph/placeholders/some-id") assert resp.status_code in (401, 403) async def test_delete_placeholder_not_found(self, client: AsyncClient, api_prefix: str, auth_headers: dict): resp = await client.delete( f"{api_prefix}/admin/graph/placeholders/nonexistent_placeholder_99999", headers=auth_headers, ) assert resp.status_code == 404