nmfs-agents/tests/test_scanner.py

from __future__ import annotations

import pytest
from pathlib import Path
from nmfs_agents.core.scanner import Scanner, ScanResult
from nmfs_agents.config import AgentsConfig, ProjectConfig, SchedulerConfig, ClaudeConfig, FeishuConfig


def _make_config(project_path: Path, mode: str = "auto") -> AgentsConfig:
    return AgentsConfig(
        projects={"test_proj": ProjectConfig(path=project_path, mode=mode)},
        scheduler=SchedulerConfig(),
        claude=ClaudeConfig(),
        feishu=FeishuConfig(),
        devices={},
    )


def test_scanner_detects_known_issues_from_readme(tmp_path):
    proj = tmp_path / "proj"
    proj.mkdir()
    (proj / "README.md").write_text(
        "## 已知问题\n- yolo12n: 零检出\n- Pose FPS 低于 80\n"
    )
    cfg = _make_config(proj)
    scanner = Scanner(cfg)
    results = scanner.scan_project("test_proj", cfg.projects["test_proj"])
    assert len(results) >= 1
    assert any("yolo12n" in r.title or "零检出" in r.title or "Pose" in r.title for r in results)


def test_scanner_detects_todo_in_code(tmp_path):
    proj = tmp_path / "proj"
    (proj / "src").mkdir(parents=True)
    (proj / "src" / "main.py").write_text(
        "def foo():\n    # TODO: fix edge case\n    pass\n"
    )
    cfg = _make_config(proj)
    scanner = Scanner(cfg)
    results = scanner.scan_project("test_proj", cfg.projects["test_proj"])
    assert any("TODO" in r.title or "todo" in r.title.lower() for r in results)


def test_scanner_uses_project_mode(tmp_path):
    proj = tmp_path / "proj"
    proj.mkdir()
    (proj / "README.md").write_text("## 已知问题\n- 问题 A\n")
    cfg = _make_config(proj, mode="confirm")
    scanner = Scanner(cfg)
    results = scanner.scan_project("test_proj", cfg.projects["test_proj"])
    assert all(r.mode == "confirm" for r in results)


def _make_config_with_desc(project_path: Path, description: str) -> AgentsConfig:
    return AgentsConfig(
        projects={"test_proj": ProjectConfig(path=project_path, mode="report",
                                             description=description)},
        scheduler=SchedulerConfig(),
        claude=ClaudeConfig(),
        feishu=FeishuConfig(),
        devices={},
    )


def test_scan_description_bug_keyword(tmp_path):
    proj = tmp_path / "proj"
    proj.mkdir()
    cfg = _make_config_with_desc(proj, "RKNN YOLO 目标检测，yolo12n 板端零检出待修复")
    scanner = Scanner(cfg)
    results = scanner._scan_description("test_proj", cfg.projects["test_proj"])
    assert len(results) == 1
    assert results[0].type == "fix_bug"
    assert results[0].priority == 2
    assert "描述" in results[0].title


def test_scan_description_perf_keyword(tmp_path):
    proj = tmp_path / "proj"
    proj.mkdir()
    cfg = _make_config_with_desc(proj, "MediaPipe 关键点检测，Pose FPS偏低")
    scanner = Scanner(cfg)
    results = scanner._scan_description("test_proj", cfg.projects["test_proj"])
    assert len(results) == 1
    assert results[0].type == "improve_perf"
    assert results[0].priority == 3


def test_scan_description_no_keyword(tmp_path):
    proj = tmp_path / "proj"
    proj.mkdir()
    cfg = _make_config_with_desc(proj, "向量数据库检索功能")
    scanner = Scanner(cfg)
    results = scanner._scan_description("test_proj", cfg.projects["test_proj"])
    assert results == []


def test_scan_description_empty(tmp_path):
    proj = tmp_path / "proj"
    proj.mkdir()
    cfg = _make_config_with_desc(proj, "")
    scanner = Scanner(cfg)
    results = scanner._scan_description("test_proj", cfg.projects["test_proj"])
    assert results == []


def test_scan_periodic_always_returns_one(tmp_path):
    proj = tmp_path / "proj"
    proj.mkdir()
    cfg = _make_config(proj)
    scanner = Scanner(cfg)
    results = scanner._scan_periodic("test_proj", cfg.projects["test_proj"])
    assert len(results) == 1
    assert results[0].type == "code_review"
    assert results[0].priority == 5
    assert "test_proj" in results[0].context


def test_scanner_routes_kernel_role(tmp_path):
    """含 'kernel' 关键词的任务路由到 dev-kernel"""
    from nmfs_agents.core.scanner import Scanner
    from nmfs_agents.config import AgentsConfig, ProjectConfig, SchedulerConfig, ClaudeConfig, FeishuConfig
    proj_path = tmp_path / "proj"
    proj_path.mkdir()
    (proj_path / "README.md").write_text(
        "## 已知问题\n- USB kernel module 崩溃导致系统重启\n"
    )
    cfg = AgentsConfig(
        projects={"proj": ProjectConfig(path=proj_path, mode="auto")},
        scheduler=SchedulerConfig(), claude=ClaudeConfig(api_key="x"),
        feishu=FeishuConfig(), devices={},
    )
    scanner = Scanner(cfg)
    results = scanner.scan_all()
    assert any(r.agent_role == "dev-kernel" for r in results)


def test_scanner_routes_hw_engineer_role(tmp_path):
    """含 '芯片选型' 的任务路由到 hw-engineer"""
    from nmfs_agents.core.scanner import Scanner
    from nmfs_agents.config import AgentsConfig, ProjectConfig, SchedulerConfig, ClaudeConfig, FeishuConfig
    proj_path = tmp_path / "proj"
    proj_path.mkdir()
    (proj_path / "README.md").write_text(
        "## 已知问题\n- 芯片选型方案未定，BOM 成本过高\n"
    )
    cfg = AgentsConfig(
        projects={"proj": ProjectConfig(path=proj_path, mode="auto")},
        scheduler=SchedulerConfig(), claude=ClaudeConfig(api_key="x"),
        feishu=FeishuConfig(), devices={},
    )
    scanner = Scanner(cfg)
    results = scanner.scan_all()
    assert any(r.agent_role == "hw-engineer" for r in results)


def test_scanner_new_feature_from_claude_md(tmp_path):
    """CLAUDE.md 的 ## 规划功能 章节产生 new_feature 任务"""
    from nmfs_agents.core.scanner import Scanner
    from nmfs_agents.config import AgentsConfig, ProjectConfig, SchedulerConfig, ClaudeConfig, FeishuConfig
    proj_path = tmp_path / "proj"
    proj_path.mkdir()
    (proj_path / "CLAUDE.md").write_text(
        "# CLAUDE.md\n\n## 规划功能\n- 添加 WebSocket 实时推流支持\n"
    )
    cfg = AgentsConfig(
        projects={"proj": ProjectConfig(path=proj_path, mode="auto")},
        scheduler=SchedulerConfig(), claude=ClaudeConfig(api_key="x"),
        feishu=FeishuConfig(), devices={},
    )
    scanner = Scanner(cfg)
    results = scanner.scan_all()
    assert any(r.type == "new_feature" for r in results)