re-xxm/tests/test_fans_csv.py

# -*- coding: utf-8 -*-
"""粉丝历史 CSV 回填测试（离线）: uv run python tests/test_fans_csv.py"""

import os
import sys
import tempfile
from pathlib import Path

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
sys.stdout.reconfigure(encoding="utf-8")

from reaction_board.database import Database  # noqa: E402
from reaction_board.repository import Repository  # noqa: E402
from tools.import_fans_csv import (  # noqa: E402
    _database_path,
    import_fans_series,
    parse_fans_csv,
)

CSV_TEXT = """\ufeff时间,粉丝,7日变化,1日变化
2026-01-03 10:00,1003,0,1
2026-01-01 09:00,1001,0,0
2026-01-02 08:30,1002,0,1
2026-01-02 23:30,1005,0,4
2026-01-04,1007,0,1
"""


def test_parse_csv():
    series = parse_fans_csv(CSV_TEXT)
    assert series["first_date"] == "2026-01-01"
    assert series["last_date"] == "2026-01-04"
    assert [point["date"] for point in series["points"]] == [
        "2026-01-01", "2026-01-02", "2026-01-03", "2026-01-04",
    ]
    # 同一天两行，保留时间较晚的一条
    assert [point["follower"] for point in series["points"]] == [1001, 1005, 1003, 1007]
    assert series["unique_days"] == 4
    assert series["total_rows"] == 5
    assert series["gaps"] == 0
    assert any("多行" in warning for warning in series["warnings"])

    for bad in ("", "时间,粉丝\n2026-01-01\n", "date,follower\nx,1\n", "时间,粉丝\n"):
        try:
            parse_fans_csv(bad)
        except ValueError:
            pass
        else:
            raise AssertionError(f"非法 CSV 应报错: {bad!r}")


def test_parse_rejects_bad_values():
    for bad in (
        "时间,粉丝\n2026-01-01,nope\n",
        "时间,粉丝\nnot-a-date,100\n",
        "时间,粉丝\n2026-01-01,-1\n",
    ):
        try:
            parse_fans_csv(bad)
        except ValueError:
            pass
        else:
            raise AssertionError(f"非法行应报错: {bad!r}")


def test_import_default_force_and_replace():
    series = parse_fans_csv(CSV_TEXT)
    with tempfile.TemporaryDirectory() as tmp:
        db = Database(Path(tmp) / "app.db")
        db.initialize()
        try:
            repo = Repository(db)
            repo.upsert_up_stat("2026-01-01", 37754047, follower=1, following=9, ok=True, captured_at=1)
            repo.upsert_up_stat("2026-01-02", 37754047, follower=None, following=None, ok=False, captured_at=1)
            repo.upsert_up_stat("2026-01-03", 99999, follower=1, following=9, ok=True, captured_at=1)

            dry = import_fans_series(db, series, dry_run=True)
            assert dry["inserted"] == 2
            assert dry["updated"] == 1
            assert dry["skipped"] == 1
            assert repo.get_up_stat("2026-01-02", 37754047)["ok"] == 0

            summary = import_fans_series(db, series)
            assert summary["inserted"] == 2
            assert summary["updated"] == 1
            assert summary["skipped"] == 1
            assert repo.get_up_stat("2026-01-01", 37754047)["follower"] == 1
            assert repo.get_up_stat("2026-01-02", 37754047)["follower"] == 1005
            assert repo.get_up_stat("2026-01-02", 37754047)["ok"] == 1
            assert repo.get_up_stat("2026-01-04", 37754047)["follower"] == 1007
            # 另一个 mid 不受影响
            assert repo.get_up_stat("2026-01-03", 99999)["follower"] == 1

            forced = import_fans_series(db, series, force=True)
            assert forced["updated"] == 4
            assert repo.get_up_stat("2026-01-01", 37754047)["follower"] == 1001

            replaced = import_fans_series(db, series, replace=True)
            assert replaced["deleted"] == 4
            assert replaced["inserted"] == 4
            assert replaced["updated"] == 0
            assert repo.get_up_stat("2026-01-01", 37754047)["follower"] == 1001
            assert repo.get_up_stat("2026-01-03", 99999)["follower"] == 1
        finally:
            db.close()


def test_database_path_resolution():
    assert _database_path("missing.json", "x.db") == Path("x.db")
    with tempfile.TemporaryDirectory() as tmp:
        from tests.web_fixtures import write_config

        config = write_config(tmp)
        assert _database_path(str(config), None) == Path(tmp) / "app.db"


def test_replace_guard_for_newer_rows():
    series = parse_fans_csv(CSV_TEXT)
    with tempfile.TemporaryDirectory() as tmp:
        db = Database(Path(tmp) / "app.db")
        db.initialize()
        try:
            repo = Repository(db)
            repo.upsert_up_stat("2026-02-01", 37754047, follower=2000, following=None, ok=True, captured_at=1)
            try:
                import_fans_series(db, series, replace=True)
            except ValueError as exc:
                assert "allow-lossy-replace" in str(exc)
            else:
                raise AssertionError("存在更新记录时应拒绝 --replace")
            result = import_fans_series(
                db, series, replace=True, allow_lossy_replace=True, dry_run=True
            )
            assert result["deleted"] == 1
            assert repo.get_up_stat("2026-02-01", 37754047)["follower"] == 2000
        finally:
            db.close()


def test_replace_dry_run_counts_only():
    series = parse_fans_csv(CSV_TEXT)
    with tempfile.TemporaryDirectory() as tmp:
        db = Database(Path(tmp) / "app.db")
        db.initialize()
        try:
            repo = Repository(db)
            repo.upsert_up_stat("2026-01-01", 37754047, follower=1, following=None, ok=True, captured_at=1)
            result = import_fans_series(db, series, replace=True, dry_run=True)
            assert result["deleted"] == 1
            assert result["inserted"] == 4
            assert repo.get_up_stat("2026-01-01", 37754047)["follower"] == 1
        finally:
            db.close()


def main_test():
    test_parse_csv()
    test_parse_rejects_bad_values()
    test_import_default_force_and_replace()
    test_database_path_resolution()
    test_replace_guard_for_newer_rows()
    test_replace_dry_run_counts_only()
    print("OK: 粉丝 CSV 解析/导入/替换/幂等测试全部通过")


if __name__ == "__main__":
    main_test()