tests/test_app.py

"""Streamlit 主应用装配逻辑的单元测试（任务 20）。

覆盖需求 15.1 / 15.3 / 15.4 / 15.5 / 17.1 / 17.2 / 17.6 与 5.1 / 5.4：

- :func:`build_services`：构建 Services 容器，可选依赖缺失时优雅降级而非崩溃（需求 6.5）。
- :func:`discover_skills` / :func:`merge_skill_i18n`：发现 Skill 并把各 Skill 的 i18n
  资源并入 I18nService（需求 1.1 / 17.6）。
- :func:`visible_skills_for_role` / :func:`can_view_audit`：按角色过滤可见 Skill 与
  审计区可见性（需求 15.3 / 15.5）。
- :func:`set_session_language`：写入 Services.lang 并联动 I18nService（需求 17.1/17.2）。
- :func:`authenticate`：加盐哈希校验、角色归一；明文密码不入逻辑（需求 5.1/5.4）。
- :func:`dispatch`：固定流水线分发并渲染 ReportSections（需求 2.2 / 15.1）。

测试不依赖 Streamlit（仅测装配纯函数）；导入路径由 tests/conftest.py 统一设置。
``app`` 以顶层模块导入（platform/ 目录在 sys.path 上）。
"""

from __future__ import annotations

import inspect
import sys

import pytest

# 底座以顶层包导入；app 同样为 platform/ 下的顶层模块。
import app  # noqa: E402
from kernel.registry import SkillRegistry  # noqa: E402
from kernel.services import Services  # noqa: E402
from kernel.skill_base import (  # noqa: E402
    ComputeResult,
    ExtractedData,
    InputKind,
    PharmaSkill,
    RawInput,
    ReportSections,
    SkillMeta,
)
from services.auth_service import AuthService  # noqa: E402
from services.i18n_service import I18nService  # noqa: E402


# ---------------------------------------------------------------------------
# 桩 Skill / 桩服务（自包含、不触网）
# ---------------------------------------------------------------------------

class _StubSkill(PharmaSkill):
    """可配置 id / 顺序 / i18n 的桩 Skill，三层返回固定占位结果。"""

    def __init__(self, skill_id: str, *, order: int = 100, i18n=None, kinds=None):
        self.meta = SkillMeta(
            id=skill_id,
            display_name=f"{skill_id}-name",
            description=f"{skill_id}-desc",
            version="1.0.0",
            input_kinds=set(kinds or {InputKind.TEXT}),
            icon="🧪",
            order=order,
        )
        if i18n is not None:
            self.i18n = i18n

    def render_inputs(self, st_ctx):  # pragma: no cover - UI 渲染不在单测覆盖
        return RawInput(goal="hello")

    def extract(self, raw, svc):
        return ExtractedData(payload={"goal": raw.goal}, method="passthrough")

    def compute(self, data):
        return ComputeResult(summary={"goal": (data.payload or {}).get("goal", "")})

    def explain(self, result, svc):
        return ReportSections(sections={"answer": "ok"}, html="<p>ok</p>")


class _StubAudit:
    """记录调用的桩审计服务。"""

    def __init__(self):
        self.analyses = []
        self.logins = []

    def record_analysis(self, user, skill_id, *, step="execute", **kw):
        self.analyses.append((user, skill_id, step))

    def record_login(self, user, *, success=True, **kw):
        self.logins.append((user, success))


def _registry(*skills: PharmaSkill) -> SkillRegistry:
    reg = SkillRegistry()
    for s in skills:
        reg.register(s)
    return reg


# ===========================================================================
# build_services（需求 7 / 6.5）
# ===========================================================================

def test_build_services_populates_all_fields():
    """无密钥环境下也应成功构建容器，各字段非 None（LLM 仍可构建，仅无可用提供商）。"""
    svc = app.build_services(env={})
    assert isinstance(svc, Services)
    for field in ("i18n", "usage", "llm", "audit", "report", "chart", "file", "prompt_guard", "auth"):
        assert getattr(svc, field) is not None, f"{field} 应被成功构建"
    assert svc.lang == "zh"
    # 构建告警列表存在（本环境通常为空）。
    assert isinstance(getattr(svc, "_build_warnings"), list)


def test_build_services_degrades_when_a_service_factory_fails(monkeypatch):
    """单个服务构建失败时降级为 None 并记入告警，不影响其它服务（需求 6.5）。"""
    def _boom():
        raise RuntimeError("simulated missing dependency")

    monkeypatch.setattr(app, "_build_report", _boom)
    svc = app.build_services(env={})
    assert svc.report is None
    assert any(name == "report" for name, _ in svc._build_warnings)
    # 其它服务不受影响。
    assert svc.i18n is not None and svc.auth is not None


def test_build_services_passes_skill_ids_to_auth():
    """传入 skill_ids 后，AuthService.visible_skills 应据此返回（需求 15.5）。"""
    svc = app.build_services(env={}, skill_ids=["stability", "compatibility"])
    visible = svc.auth.visible_skills("researcher")
    assert visible == {"stability", "compatibility"}


# ===========================================================================
# discover_skills / merge_skill_i18n（需求 1.1 / 17.6）
# ===========================================================================

def test_discover_skills_finds_firstparty_skills():
    """扫描真实 skills 包应发现首发 Skill（需求 1.1 / 15.3：统一导航可达）。"""
    reg = app.discover_skills("skills")
    ids = {s.meta.id for s in reg.all()}
    assert {"general_qa", "example_stub"} <= ids


def test_merge_skill_i18n_namespaces_unprefixed_catalog():
    """未加前缀的 Skill 目录以 namespace 合并；已加前缀的直接合并（需求 17.6）。"""
    svc = Services(i18n=I18nService())
    unprefixed = _StubSkill("stab", i18n={"name": {"zh": "稳定", "en": "Stab"}})
    prefixed = _StubSkill(
        "qa", i18n={"qa.title": {"zh": "问答", "en": "QA"}}
    )
    reg = _registry(unprefixed, prefixed)

    merged = app.merge_skill_i18n(svc, reg)
    assert merged == 2
    # 未加前缀 → 自动加 "stab." 前缀。
    assert svc.i18n.has("stab.name")
    assert svc.i18n.t("stab.name", "en") == "Stab"
    # 已加前缀 → 不重复前缀（不应出现 qa.qa.title）。
    assert svc.i18n.has("qa.title")
    assert not svc.i18n.has("qa.qa.title")


def test_merge_skill_i18n_safe_without_i18n():
    """i18n 不可用时安全返回 0，不抛异常。"""
    svc = Services(i18n=None)
    reg = _registry(_StubSkill("x", i18n={"name": {"zh": "X", "en": "X"}}))
    assert app.merge_skill_i18n(svc, reg) == 0


def test_bootstrap_wires_everything():
    """bootstrap 一次性发现 + 构建 + 合并，返回可用的 svc/registry。"""
    svc, reg = app.bootstrap(env={}, skills_source="skills")
    assert isinstance(svc, Services) and isinstance(reg, SkillRegistry)
    ids = {s.meta.id for s in reg.all()}
    assert "general_qa" in ids
    # Skill 的 i18n 已并入：general_qa 自带前缀键应可取用。
    assert svc.i18n.has("general_qa.title")
    # registry 反向持有 svc（失败可落审计）。
    assert reg.svc is svc


# ===========================================================================
# 角色可见性（需求 15.3 / 15.5）
# ===========================================================================

def test_visible_skills_for_role_filters_by_auth():
    """researcher 可见全部分析 Skill；auditor 只读不展示可执行 Skill（需求 15.5）。"""
    reg = _registry(
        _StubSkill("stability", order=10),
        _StubSkill("compatibility", order=20),
    )
    auth = AuthService(all_skill_ids=["stability", "compatibility"])

    researcher_view = app.visible_skills_for_role(reg, auth, "researcher")
    assert {s.meta.id for s in researcher_view} == {"stability", "compatibility"}
    # 导航顺序按 meta.order。
    assert [s.meta.id for s in researcher_view] == ["stability", "compatibility"]

    auditor_view = app.visible_skills_for_role(reg, auth, "auditor")
    assert auditor_view == []


def test_visible_skills_fallback_when_auth_missing():
    """auth 不可用时回退展示全部已注册 Skill（避免空界面）。"""
    reg = _registry(_StubSkill("a"), _StubSkill("b"))
    view = app.visible_skills_for_role(reg, None, "researcher")
    assert {s.meta.id for s in view} == {"a", "b"}


def test_can_view_audit_role_isolation():
    """审计区仅 admin / auditor 可见，researcher 不可见（需求 15.5）。"""
    auth = AuthService()
    assert app.can_view_audit(auth, "admin") is True
    assert app.can_view_audit(auth, "auditor") is True
    assert app.can_view_audit(auth, "researcher") is False
    # auth 不可用时默认不可见（安全默认）。
    assert app.can_view_audit(None, "admin") is False


def test_skill_nav_label_uses_i18n_then_fallbacks():
    """导航标签优先取 i18n（nav.<id> / <id>.name），缺失回退 display_name（需求 17.4）。"""
    i18n = I18nService()
    i18n.merge({"nav.stability": {"zh": "稳定性预测", "en": "Stability"}})
    skill = _StubSkill("stability")
    label = app.skill_nav_label(skill, i18n)
    assert "稳定性预测" in label

    # 无 i18n 命中时回退 display_name。
    skill2 = _StubSkill("unknown_skill")
    label2 = app.skill_nav_label(skill2, i18n)
    assert "unknown_skill-name" in label2


# ===========================================================================
# 语言状态（需求 17.1 / 17.2）
# ===========================================================================

def test_set_session_language_writes_services_lang():
    """设置语言应同时写入 Services.lang 与 I18nService.lang（需求 17.2）。"""
    svc = Services(i18n=I18nService())
    assert app.set_session_language(svc, "en") == "en"
    assert svc.lang == "en"
    assert svc.i18n.lang == "en"
    # 切回中文。
    assert app.set_session_language(svc, "zh") == "zh"
    assert svc.lang == "zh"


def test_set_session_language_unknown_falls_back_to_default():
    """未知语言回退默认语言，不报错（需求 17.5）。"""
    svc = Services(i18n=I18nService())
    assert app.set_session_language(svc, "fr") == "zh"
    assert svc.lang == "zh"


def test_set_session_language_without_i18n():
    """i18n 不可用时仍能写入 Services.lang。"""
    svc = Services(i18n=None)
    assert app.set_session_language(svc, "en") == "en"
    assert svc.lang == "en"


# ===========================================================================
# 认证（需求 5.1 / 5.4 / 15.5）
# ===========================================================================

def test_authenticate_success_with_salted_hash():
    """加盐哈希校验通过并归一角色（需求 5.4 / 15.5）。"""
    auth = AuthService()
    stored = auth.hash_password("s3cret")
    users = {"admin": {"password_hash": stored, "role": "admin"}}

    result = app.authenticate(
        "admin", "s3cret", user_lookup=users.get, auth=auth
    )
    assert result == {"user": "admin", "role": "admin"}


def test_authenticate_wrong_password_returns_none():
    auth = AuthService()
    users = {"u": {"password_hash": auth.hash_password("right"), "role": "researcher"}}
    assert app.authenticate("u", "wrong", user_lookup=users.get, auth=auth) is None


def test_authenticate_unknown_user_returns_none():
    auth = AuthService()
    assert app.authenticate("ghost", "x", user_lookup=lambda e: None, auth=auth) is None


def test_authenticate_empty_credentials_returns_none():
    assert app.authenticate("", "", user_lookup=lambda e: {}, auth=AuthService()) is None


def test_authenticate_normalizes_legacy_role_to_researcher():
    """历史 'user' 角色归一为 researcher（最小权限默认）。"""
    auth = AuthService()
    users = {"u": {"password_hash": auth.hash_password("pw"), "role": "user"}}
    result = app.authenticate("u", "pw", user_lookup=users.get, auth=auth)
    assert result["role"] == "researcher"


# ===========================================================================
# 分发（需求 2.2 / 15.1）
# ===========================================================================

def test_dispatch_runs_pipeline_and_returns_sections():
    """dispatch 走固定流水线并返回 ReportSections，同时记录分析审计（需求 7.4）。"""
    skill = _StubSkill("general_qa")
    audit = _StubAudit()
    svc = Services(audit=audit)
    sections = app.dispatch(skill, RawInput(goal="hi"), svc, user="tester")

    assert isinstance(sections, ReportSections)
    assert sections.sections["answer"] == "ok"
    # 分析执行被审计（用户 + skill_id）。
    assert audit.analyses == [("tester", "general_qa", "execute")]


def test_dispatch_audit_failure_does_not_break():
    """审计记录失败不应中断分发（best-effort）。"""
    class _BadAudit:
        def record_analysis(self, *a, **k):
            raise RuntimeError("audit down")

    skill = _StubSkill("x")
    svc = Services(audit=_BadAudit())
    sections = app.dispatch(skill, RawInput(goal="hi"), svc)
    assert isinstance(sections, ReportSections)


# ===========================================================================
# 契约：app 的装配逻辑不在 compute 注入服务（需求 2.1 协同）
# ===========================================================================

def test_dispatch_does_not_pass_svc_to_compute():
    """通过 Pipeline 调用时 compute 仅接收 data（无 svc）——签名层面保证。"""
    sig = inspect.signature(_StubSkill.compute)
    assert list(sig.parameters) == ["self", "data"]


if __name__ == "__main__":  # pragma: no cover
    sys.exit(pytest.main([__file__, "-v"]))