from datetime import UTC, datetime from fastapi.testclient import TestClient from zbt.core.config import Settings from zbt.core.passwords import PasswordService from zbt.domains.identity.models import AdminUser from zbt.domains.identity.repository import InMemoryIdentityRepository from zbt.domains.knowledge.index import InMemoryKnowledgeIndex from zbt.domains.knowledge.repository import InMemoryKnowledgeRepository from zbt.main import create_app def test_admin_creates_draft_knowledge_document() -> None: settings = Settings( app_env="test", jwt_access_secret="a" * 32, jwt_refresh_secret="b" * 32, field_encryption_key="c" * 32, ) identities = InMemoryIdentityRepository() identities.save_admin_user( AdminUser( id="01ADMIN0000000000000000001", username="admin", password_hash=PasswordService().hash("zaq1XSW@"), display_name="知识管理员", status="ACTIVE", roles=("SUPER_ADMIN",), permissions=("knowledge:read", "knowledge:write"), data_scope="ALL", ) ) knowledge = InMemoryKnowledgeRepository() application = create_app( settings=settings, identity_repository=identities, knowledge_repository=knowledge, clock=lambda: datetime(2026, 7, 27, 2, 0, tzinfo=UTC), ) with TestClient(application) as client: login = client.post( "/api/v1/admin/auth/login", json={"username": "admin", "password": "zaq1XSW@"}, ) token = login.json()["data"]["tokens"]["access_token"] response = client.post( "/api/v1/admin/knowledge/documents", headers={"Authorization": f"Bearer {token}"}, json={ "title": "安心医疗险保险条款", "document_type": "INSURANCE_TERMS", "source_name": "安心医疗险条款.pdf", "product_code": "MED-BASIC", "content": "本产品等待期为30天，意外伤害导致的医疗费用不受等待期限制。", }, ) assert response.status_code == 201 document = response.json()["data"] assert document["title"] == "安心医疗险保险条款" assert document["document_type"] == "INSURANCE_TERMS" assert document["source_name"] == "安心医疗险条款.pdf" assert document["product_code"] == "MED-BASIC" assert document["status"] == "DRAFT" assert document["version_no"] == 1 assert document["chunk_count"] == 0 assert document["indexed_at"] is None assert document["published_at"] is None def test_admin_cannot_publish_document_before_indexing() -> None: settings = Settings( app_env="test", jwt_access_secret="a" * 32, jwt_refresh_secret="b" * 32, field_encryption_key="c" * 32, ) identities = InMemoryIdentityRepository() identities.save_admin_user( AdminUser( id="01ADMIN0000000000000000001", username="admin", password_hash=PasswordService().hash("zaq1XSW@"), display_name="知识管理员", status="ACTIVE", roles=("SUPER_ADMIN",), permissions=("knowledge:read", "knowledge:write", "knowledge:publish"), data_scope="ALL", ) ) application = create_app( settings=settings, identity_repository=identities, knowledge_repository=InMemoryKnowledgeRepository(), clock=lambda: datetime(2026, 7, 27, 2, 0, tzinfo=UTC), ) with TestClient(application) as client: login = client.post( "/api/v1/admin/auth/login", json={"username": "admin", "password": "zaq1XSW@"}, ) headers = { "Authorization": f"Bearer {login.json()['data']['tokens']['access_token']}" } created = client.post( "/api/v1/admin/knowledge/documents", headers=headers, json={ "title": "安心医疗险保险条款", "document_type": "INSURANCE_TERMS", "source_name": "安心医疗险条款.pdf", "content": "本产品等待期为30天，意外伤害导致的医疗费用不受等待期限制。", }, ) document_id = created.json()["data"]["document_id"] response = client.post( f"/api/v1/admin/knowledge/documents/{document_id}/publish", headers=headers, ) assert response.status_code == 409 assert response.json()["error"]["code"] == "KNOWLEDGE_DOCUMENT_NOT_INDEXED" def test_admin_cannot_publish_indexed_document_without_effective_search_test() -> None: settings = Settings( app_env="test", jwt_access_secret="a" * 32, jwt_refresh_secret="b" * 32, field_encryption_key="c" * 32, ) identities = InMemoryIdentityRepository() identities.save_admin_user( AdminUser( id="01ADMIN0000000000000000001", username="admin", password_hash=PasswordService().hash("zaq1XSW@"), display_name="知识管理员", status="ACTIVE", roles=("SUPER_ADMIN",), permissions=("knowledge:read", "knowledge:write", "knowledge:publish"), data_scope="ALL", ) ) application = create_app( settings=settings, identity_repository=identities, knowledge_repository=InMemoryKnowledgeRepository(), knowledge_index=InMemoryKnowledgeIndex(), clock=lambda: datetime(2026, 7, 27, 2, 0, tzinfo=UTC), ) with TestClient(application) as client: login = client.post( "/api/v1/admin/auth/login", json={"username": "admin", "password": "zaq1XSW@"}, ) headers = { "Authorization": f"Bearer {login.json()['data']['tokens']['access_token']}" } created = client.post( "/api/v1/admin/knowledge/documents", headers=headers, json={ "title": "安心医疗险保险条款", "document_type": "INSURANCE_TERMS", "source_name": "安心医疗险条款.pdf", "content": "本产品等待期为30天，意外伤害不受等待期限制。", }, ) document_id = created.json()["data"]["document_id"] client.post( f"/api/v1/admin/knowledge/documents/{document_id}/index", headers=headers, ) response = client.post( f"/api/v1/admin/knowledge/documents/{document_id}/publish", headers=headers, ) assert response.status_code == 409 assert response.json()["error"]["code"] == "KNOWLEDGE_SEARCH_TEST_REQUIRED" def test_admin_indexes_then_publishes_document() -> None: settings = Settings( app_env="test", jwt_access_secret="a" * 32, jwt_refresh_secret="b" * 32, field_encryption_key="c" * 32, ) identities = InMemoryIdentityRepository() identities.save_admin_user( AdminUser( id="01ADMIN0000000000000000001", username="admin", password_hash=PasswordService().hash("zaq1XSW@"), display_name="知识管理员", status="ACTIVE", roles=("SUPER_ADMIN",), permissions=("knowledge:read", "knowledge:write", "knowledge:publish"), data_scope="ALL", ) ) application = create_app( settings=settings, identity_repository=identities, knowledge_repository=InMemoryKnowledgeRepository(), knowledge_index=InMemoryKnowledgeIndex(), clock=lambda: datetime(2026, 7, 27, 2, 0, tzinfo=UTC), ) with TestClient(application) as client: login = client.post( "/api/v1/admin/auth/login", json={"username": "admin", "password": "zaq1XSW@"}, ) headers = { "Authorization": f"Bearer {login.json()['data']['tokens']['access_token']}" } created = client.post( "/api/v1/admin/knowledge/documents", headers=headers, json={ "title": "安心医疗险保险条款", "document_type": "INSURANCE_TERMS", "source_name": "安心医疗险条款.pdf", "content": ( "等待期规则\n本产品疾病医疗等待期为30天，意外伤害不受等待期限制。\n\n" "报销材料\n申请医疗费用报销协助时，需要提供发票、费用清单和诊断证明。" ), }, ) document_id = created.json()["data"]["document_id"] indexed = client.post( f"/api/v1/admin/knowledge/documents/{document_id}/index", headers=headers, ) search_test = client.post( f"/api/v1/admin/knowledge/documents/{document_id}/search-test", headers=headers, json={"query": "等待期是多少天？", "limit": 3}, ) published = client.post( f"/api/v1/admin/knowledge/documents/{document_id}/publish", headers=headers, ) search_tests = client.get( f"/api/v1/admin/knowledge/documents/{document_id}/search-tests", headers=headers, ) assert indexed.status_code == 200 assert indexed.json()["data"]["status"] == "INDEXED" assert indexed.json()["data"]["chunk_count"] == 2 assert indexed.json()["data"]["indexed_at"] is not None assert search_test.status_code == 200 assert search_test.json()["data"]["test_record"]["passed"] is True assert search_test.json()["data"]["test_record"]["hit_count"] >= 1 assert published.status_code == 200 assert published.json()["data"]["status"] == "PUBLISHED" assert published.json()["data"]["published_at"] is not None assert published.json()["data"]["search_test_count"] == 1 assert published.json()["data"]["last_search_test_passed"] is True assert search_tests.status_code == 200 assert search_tests.json()["data"]["total"] == 1 assert search_tests.json()["data"]["items"][0]["query"] == "等待期是多少天？" def test_admin_lists_and_disables_published_document() -> None: settings = Settings( app_env="test", jwt_access_secret="a" * 32, jwt_refresh_secret="b" * 32, field_encryption_key="c" * 32, ) identities = InMemoryIdentityRepository() identities.save_admin_user( AdminUser( id="01ADMIN0000000000000000001", username="admin", password_hash=PasswordService().hash("zaq1XSW@"), display_name="知识管理员", status="ACTIVE", roles=("SUPER_ADMIN",), permissions=("knowledge:read", "knowledge:write", "knowledge:publish"), data_scope="ALL", ) ) application = create_app( settings=settings, identity_repository=identities, knowledge_repository=InMemoryKnowledgeRepository(), knowledge_index=InMemoryKnowledgeIndex(), clock=lambda: datetime(2026, 7, 27, 2, 0, tzinfo=UTC), ) with TestClient(application) as client: login = client.post( "/api/v1/admin/auth/login", json={"username": "admin", "password": "zaq1XSW@"}, ) headers = { "Authorization": f"Bearer {login.json()['data']['tokens']['access_token']}" } created = client.post( "/api/v1/admin/knowledge/documents", headers=headers, json={ "title": "安心医疗险保险条款", "document_type": "INSURANCE_TERMS", "source_name": "安心医疗险条款.pdf", "content": "本产品疾病医疗等待期为30天，意外伤害不受等待期限制。", }, ) document_id = created.json()["data"]["document_id"] client.post( f"/api/v1/admin/knowledge/documents/{document_id}/index", headers=headers, ) client.post( f"/api/v1/admin/knowledge/documents/{document_id}/search-test", headers=headers, json={"query": "等待期是多少天？"}, ) client.post( f"/api/v1/admin/knowledge/documents/{document_id}/publish", headers=headers, ) disabled = client.post( f"/api/v1/admin/knowledge/documents/{document_id}/disable", headers=headers, ) listing = client.get("/api/v1/admin/knowledge/documents", headers=headers) assert disabled.status_code == 200 assert disabled.json()["data"]["status"] == "INACTIVE" assert listing.status_code == 200 assert listing.json()["data"]["total"] == 1 assert listing.json()["data"]["items"][0]["status"] == "INACTIVE" def test_admin_uploads_utf8_text_as_draft_document() -> None: settings = Settings( app_env="test", jwt_access_secret="a" * 32, jwt_refresh_secret="b" * 32, field_encryption_key="c" * 32, ) identities = InMemoryIdentityRepository() identities.save_admin_user( AdminUser( id="01ADMIN0000000000000000001", username="admin", password_hash=PasswordService().hash("zaq1XSW@"), display_name="知识管理员", status="ACTIVE", roles=("SUPER_ADMIN",), permissions=("knowledge:read", "knowledge:write"), data_scope="ALL", ) ) application = create_app( settings=settings, identity_repository=identities, knowledge_repository=InMemoryKnowledgeRepository(), knowledge_index=InMemoryKnowledgeIndex(), clock=lambda: datetime(2026, 7, 27, 2, 0, tzinfo=UTC), ) with TestClient(application) as client: login = client.post( "/api/v1/admin/auth/login", json={"username": "admin", "password": "zaq1XSW@"}, ) headers = { "Authorization": f"Bearer {login.json()['data']['tokens']['access_token']}" } response = client.post( "/api/v1/admin/knowledge/documents/upload", headers=headers, data={ "title": "医疗费用报销材料说明", "document_type": "SERVICE_RULES", "product_code": "MED-BASIC", }, files={ "file": ( "医疗费用报销材料说明.txt", "申请时需要提供发票、费用清单和诊断证明。".encode(), "text/plain", ) }, ) assert response.status_code == 201 document = response.json()["data"] assert document["title"] == "医疗费用报销材料说明" assert document["source_name"] == "医疗费用报销材料说明.txt" assert document["document_type"] == "SERVICE_RULES" assert document["status"] == "DRAFT" def test_admin_tests_indexed_document_before_publishing() -> None: settings = Settings( app_env="test", jwt_access_secret="a" * 32, jwt_refresh_secret="b" * 32, field_encryption_key="c" * 32, ) identities = InMemoryIdentityRepository() identities.save_admin_user( AdminUser( id="01ADMIN0000000000000000001", username="admin", password_hash=PasswordService().hash("zaq1XSW@"), display_name="知识管理员", status="ACTIVE", roles=("SUPER_ADMIN",), permissions=("knowledge:read", "knowledge:write"), data_scope="ALL", ) ) application = create_app( settings=settings, identity_repository=identities, knowledge_repository=InMemoryKnowledgeRepository(), knowledge_index=InMemoryKnowledgeIndex(), clock=lambda: datetime(2026, 7, 27, 2, 0, tzinfo=UTC), ) with TestClient(application) as client: login = client.post( "/api/v1/admin/auth/login", json={"username": "admin", "password": "zaq1XSW@"}, ) headers = { "Authorization": f"Bearer {login.json()['data']['tokens']['access_token']}" } created = client.post( "/api/v1/admin/knowledge/documents", headers=headers, json={ "title": "安心医疗险保险条款", "document_type": "INSURANCE_TERMS", "source_name": "安心医疗险条款.md", "content": "疾病医疗等待期为30天，意外伤害不受等待期限制。", }, ) document_id = created.json()["data"]["document_id"] client.post( f"/api/v1/admin/knowledge/documents/{document_id}/index", headers=headers, ) response = client.post( f"/api/v1/admin/knowledge/documents/{document_id}/search-test", headers=headers, json={"query": "等待期是多少天？", "limit": 3}, ) assert response.status_code == 200 result = response.json()["data"] assert result["document_id"] == document_id assert result["total"] == 1 assert "30天" in result["items"][0]["content"]