chuanqi/pipeline_bot/tests/test_db_units.py

"""db 层纯逻辑单测(不触网/不触库的部分)。"""
from __future__ import annotations

import pytest

from pbot.db.scanner import parse_open_times
from pbot.db.scheduler import parse_progress_marks
from pbot.db.seed import TASKS


class TestParseOpenTimes:
    """serverlist 明文 JSON → {serverid: open_time}。"""

    def test_parses_nested_list(self) -> None:
        text = 'x({"serverData": [{"serverid": 14628306, "open_time": 1791043200, "state": 1}]})'
        out = parse_open_times(text)
        assert out == {14628306: 1791043200}

    def test_no_json_raises(self) -> None:
        from pbot.errors import PbotError
        with pytest.raises(PbotError):
            parse_open_times("plain text no json")

    def test_skips_missing_open_time(self) -> None:
        text = '{"a": [{"serverid": 1, "open_time": 100}, {"serverid": 2}]}'
        out = parse_open_times(text)
        assert out == {1: 100}


class TestParseProgressMarks:
    """打勾文本 → (done, total)。"""

    def test_counts_done_and_fail(self) -> None:
        text = (
            "# header\n\n"
            "- 2026-10-04 00:15:33 ☑ **T1 开启喝药保护** — ok\n"
            "- 2026-10-04 00:22:10 ☑ **T2 开区挖矿(20分钟窗口)** — 851锄\n"
            "- 2026-10-04 00:30:00 ☒ **T3 比奇大城杀怪20** — 失败\n"
        )
        done, total, name = parse_progress_marks(text)
        assert (done, total) == (2, 3)
        assert name == ""

    def test_empty_text(self) -> None:
        done, total, _ = parse_progress_marks("")
        assert (done, total) == (0, 0)


class TestSeedTasks:
    """种子任务清单与 run_pipeline 定案一致性。"""

    def test_order_t7_tail(self) -> None:
        codes = [str(t["code"]) for t in TASKS]
        assert codes.index("T07") == len(codes) - 2  # 倒数第二(队尾兜底, T10 最后)
        assert codes[-1] == "T10"

    def test_soft_set_matches_pipeline(self) -> None:
        soft = {str(t["code"]) for t in TASKS if t["soft"]}
        assert soft == {"T02", "T07", "T08", "T10"}

    def test_watch_only_t07(self) -> None:
        watch = [str(t["code"]) for t in TASKS if t["watch"]]
        assert watch == ["T07"]

    def test_modules_exist(self) -> None:
        import importlib
        for t in TASKS:
            mod = importlib.import_module(str(t["module"]))
            assert hasattr(mod, "run_task"), f"{t['module']} 缺 run_task"