feat: add admin permissions voice input analytics and feedback

ai_knowledge_base_v2/apps/backend/tests/test_admin_permissions.py

@@ -0,0 +1,24 @@
+from __future__ import annotations
+
+import json
+
+import pytest
+from fastapi import HTTPException
+
+from app.models.admin import Admin, Role
+from app.services.admin_permission_service import ALL_PERMISSION_CODES, permissions_for, require_permission
+
+
+def test_super_admin_has_all_permissions() -> None:
+    admin = Admin(id=1, username="root", password="unused", name="root", status=1, is_super_admin=1)
+    assert permissions_for(admin) == ALL_PERMISSION_CODES
+
+
+def test_role_permissions_are_restricted_to_catalog() -> None:
+    role = Role(code="operator", name="operator", permissions=json.dumps(["users.view", "unknown.permission"]))
+    admin = Admin(id=2, username="operator", password="hash", name="operator", status=1, is_super_admin=0, role=role)
+    assert permissions_for(admin) == {"users.view"}
+    require_permission(admin, "users.view")
+    with pytest.raises(HTTPException) as exc:
+        require_permission(admin, "users.delete")
+    assert exc.value.status_code == 403

ai_knowledge_base_v2/apps/backend/tests/test_message_feedback.py

@@ -0,0 +1,35 @@
+from __future__ import annotations
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+from sqlalchemy.pool import StaticPool
+
+from app.api.feedback import FeedbackCreate, create_feedback, delete_feedback, feedback_detail, feedback_list
+from app.models import Base
+from app.models.admin import Admin
+from app.models.chat import ChatMessage, ChatSession
+from app.models.user import User
+
+
+def test_feedback_binds_target_answer_and_exposes_context() -> None:
+    engine = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool)
+    Base.metadata.create_all(engine)
+    with Session(engine) as db:
+        user = User(id=1, phone="13800000000", name="测试用户")
+        admin = Admin(id=1, username="admin", password="hash", name="管理员", is_super_admin=1, must_change_password=0)
+        db.add_all([user, admin]); db.flush()
+        session = ChatSession(id=1, user_id=user.id, title="测试会话", message_count=2)
+        db.add(session); db.flush()
+        question = ChatMessage(id=1, session_id=session.id, user_id=user.id, role="user", content="问题", message_status="FINISHED")
+        answer = ChatMessage(id=2, session_id=session.id, user_id=user.id, role="assistant", content="回答", message_status="FINISHED")
+        db.add_all([question, answer]); db.commit()
+
+        created = create_feedback(FeedbackCreate(messageId=answer.id, content="这条回答不准确"), db=db, user=user)["data"]
+        listed = feedback_list(readStatus="unread", page=1, pageSize=20, db=db, _admin=admin)["data"]
+        assert listed["total"] == 1
+        detail = feedback_detail(created["id"], db=db, admin=admin)["data"]
+        assert detail["messageId"] == answer.id
+        assert [item["content"] for item in detail["messages"]] == ["问题", "回答"]
+        assert detail["messages"][-1]["isTarget"] is True
+        delete_feedback(created["id"], db=db, admin=admin)
+        assert feedback_list(readStatus="all", page=1, pageSize=20, db=db, _admin=admin)["data"]["total"] == 0

ai_knowledge_base_v2/apps/backend/tests/test_request_traffic.py

@@ -0,0 +1,21 @@
+from __future__ import annotations
+
+from datetime import datetime
+from zoneinfo import ZoneInfo
+
+from app.services.request_traffic_service import _format_period, _period_sequence
+
+
+def test_peak_traffic_period_labels_cover_all_grains() -> None:
+    assert _format_period("minute", "202608111405") == "2026-08-11 14:05"
+    assert _format_period("hour", "2026081114") == "2026-08-11 14:00"
+    assert _format_period("day", "20260811") == "2026-08-11"
+    assert _format_period("week", "2026-W33") == "2026 年第 33 周"
+
+
+def test_period_sequence_includes_empty_time_buckets() -> None:
+    timezone = ZoneInfo("Asia/Shanghai")
+    start = datetime(2026, 8, 11, 14, 58, tzinfo=timezone)
+    end = datetime(2026, 8, 11, 15, 1, tzinfo=timezone)
+    assert _period_sequence("minute", start, end) == ["202608111458", "202608111459", "202608111500", "202608111501"]
+    assert _period_sequence("hour", start, end) == ["2026081114", "2026081115"]

ai_knowledge_base_v2/apps/backend/tests/test_voice_input.py

@@ -0,0 +1,70 @@
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+import pytest
+from fastapi import HTTPException
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+from sqlalchemy.pool import StaticPool
+
+from app.models import Base
+from app.models.ai_config import SystemConfig
+from app.services.voice_input_service import VoiceInputService, load_voice_config
+
+
+def database() -> Session:
+    engine = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool)
+    Base.metadata.create_all(engine)
+    return Session(engine)
+
+
+def settings(**overrides):
+    values = {
+        "voice_input_enabled": False,
+        "voice_max_duration_seconds": 60,
+        "aliyun_nls_app_key": "app-key",
+        "aliyun_nls_endpoint": "https://nls.example/asr",
+        "aliyun_sms_access_key_id": "access-id",
+        "aliyun_sms_access_key_secret": "access-secret",
+    }
+    values.update(overrides)
+    return SimpleNamespace(**values)
+
+
+def test_public_config_is_hidden_by_default_and_duration_is_clamped(monkeypatch):
+    monkeypatch.setattr("app.services.voice_input_service.get_settings", lambda: settings())
+    with database() as db:
+        db.add_all([
+            SystemConfig(config_key="voice_input_enabled", config_value="true"),
+            SystemConfig(config_key="voice_max_duration_seconds", config_value="999"),
+        ])
+        db.commit()
+        assert VoiceInputService.public_config(db) == {"enabled": True, "maxDurationSeconds": 60}
+
+
+def test_transcription_endpoint_is_rejected_when_switch_is_off(monkeypatch):
+    monkeypatch.setattr("app.services.voice_input_service.get_settings", lambda: settings())
+    with database() as db, pytest.raises(HTTPException) as exc:
+        VoiceInputService.transcribe(db, 1, b"audio", "audio/webm")
+    assert exc.value.status_code == 404
+
+
+def test_transcription_returns_provider_text(monkeypatch):
+    monkeypatch.setattr("app.services.voice_input_service.get_settings", lambda: settings(voice_input_enabled=True))
+    monkeypatch.setattr("app.services.voice_input_service._normalize_audio", lambda *_: b"wav")
+    monkeypatch.setattr("app.services.voice_input_service._wav_duration", lambda *_: 8.4)
+    monkeypatch.setattr("app.services.voice_input_service._create_aliyun_token", lambda *_: "token")
+    monkeypatch.setattr("app.services.voice_input_service.SecurityStateService.enforce_limit", lambda *_, **__: None)
+
+    class Response:
+        status_code = 200
+
+        @staticmethod
+        def json():
+            return {"status": 20000000, "result": "这是转写结果。"}
+
+    monkeypatch.setattr("app.services.voice_input_service.httpx.post", lambda *_, **__: Response())
+    with database() as db:
+        result = VoiceInputService.transcribe(db, 1, b"audio", "audio/webm")
+    assert result == {"text": "这是转写结果。", "durationSeconds": 8.4}