"""S7：账号检测保留旧的手动输入，新增「按账号选择」。

本文件守住的核心是**两套来源并存**，而不是替换：

| | 手动输入（旧） | 选择账号（新） |
| --- | --- | --- |
| 来源 | 粘贴文本 | 库里账号的凭据 |
| `account_id` | **无** | 有 |
| 结果落账号时间线 | **否**（保持旧行为） | 是 |

最容易做错的两件事：① 把旧路径一起改成发事件（那会凭空造出没有归属的事件）；
② 新路径忘了按可见性过滤（客服就能检测别人名下的账号）。
"""

from datetime import UTC, datetime

import pytest

from tests.support.db import get_test_engine, make_test_session
from web.backend.models import Account, AccountEvent, User
from web.backend.services import account_check_service as svc

NOW = datetime(2026, 10, 4, 12, 0, 0, tzinfo=UTC)


class _Base:
    def setup_method(self):
        self.engine = get_test_engine()
        self.session = make_test_session(self.engine)
        self.session.query(AccountEvent).delete()
        self.session.commit()
        self.user = User(username="chk-admin", password_hash="h", role="super_admin")
        self.session.add(self.user)
        self.session.commit()

    def teardown_method(self):
        self.session.close()

    def account(self, name: str, **kwargs) -> Account:
        acc = Account(
            username=name,
            password=f"pw-{name}",
            email=f"{name}@e.com",
            email_password=f"ep-{name}",
            **kwargs,
        )
        self.session.add(acc)
        self.session.commit()
        return acc


class TestLegacyTextPathUnchanged(_Base):
    """**旧逻辑一字未动** —— 这是本次明确要求保留的部分。"""

    def test_parses_without_account_id(self):
        inputs = svc.parse_account_check_lines("u1----p1----e1----ep1")
        assert len(inputs) == 1
        assert inputs[0].username == "u1"
        assert inputs[0].password == "p1"
        # 关键：没有归属 → 后续不会往任何账号写事件
        assert inputs[0].account_id is None

    def test_parse_supports_both_separators(self):
        inputs = svc.parse_account_check_lines("u1----p1----e1----ep1\nu2|p2|e2|ep2")
        assert [i.username for i in inputs] == ["u1", "u2"]

    def test_export_line_unchanged(self):
        inputs = svc.parse_account_check_lines("u1----p1----e1----ep1")
        assert inputs[0].export_line() == "u1----p1----e1----ep1"


class TestSelectAccountsPath(_Base):
    def test_builds_inputs_with_account_id(self):
        a = self.account("sa-1")
        b = self.account("sa-2")
        rows = self.session.query(Account).filter(Account.id.in_([a.id, b.id])).all()

        inputs = svc.build_inputs_from_accounts(rows, [a.id, b.id])
        assert [i.account_id for i in inputs] == [a.id, b.id]
        # 凭据从库里取，不用人工粘贴
        assert inputs[0].username == "sa-1"
        assert inputs[0].password == "pw-sa-1"
        assert inputs[0].email == "sa-1@e.com"

    def test_order_follows_requested_ids(self):
        a = self.account("ord-1")
        b = self.account("ord-2")
        rows = self.session.query(Account).filter(Account.id.in_([a.id, b.id])).all()
        inputs = svc.build_inputs_from_accounts(rows, [b.id, a.id])
        assert [i.account_id for i in inputs] == [b.id, a.id]

    def test_unknown_id_is_skipped_not_crashing(self):
        a = self.account("sk-1")
        rows = self.session.query(Account).filter(Account.id == a.id).all()
        inputs = svc.build_inputs_from_accounts(rows, [a.id, 10_000_000])
        assert [i.account_id for i in inputs] == [a.id]


class TestResultGoesToAccountTimeline(_Base):
    """只有「按账号选择」的条目才写事件。"""

    def _runner(self):
        runner = svc.AccountCheckRunner.__new__(svc.AccountCheckRunner)
        runner.batch = svc.AccountCheckBatch(
            batch_id="batch-x",
            created_by=self.user.id,
            concurrency=1,
            max_login_retries=0,
            max_total_time=60,
            use_proxy=False,
            items=[],
        )
        import threading

        runner._lock = threading.Lock()
        runner._stop = threading.Event()
        return runner

    def test_account_linked_item_emits_event(self):
        a = self.account("evt-1")
        runner = self._runner()
        runner.batch.items = [
            svc.AccountCheckItemState(
                line=1,
                username="evt-1",
                email="evt-1@e.com",
                export_text="evt-1----pw----e----ep",
                account_id=a.id,
            )
        ]
        # 终态（带 finished_at）→ 写事件
        runner._set_item(
            0,
            status="account_verified",
            message="账号已认证",
            finished_at=NOW,
        )
        # 事件由 runner 自己开的会话提交；本测试会话在 REPEATABLE READ 下
        # 沿用旧快照看不到它 —— 先结束自己的事务再查。
        self.session.commit()

        events = (
            self.session.query(AccountEvent)
            .filter(
                AccountEvent.account_id == a.id,
                AccountEvent.event_type == svc.EVENT_ACCOUNT_CHECK,
            )
            .all()
        )
        assert len(events) == 1
        assert events[0].detail["status"] == "account_verified"
        assert events[0].detail["batch_id"] == "batch-x"

    def test_text_item_emits_nothing(self):
        """手动粘贴的条目没有归属 —— 绝不能凭空造出没有账号的事件。"""
        self.account("evt-2")
        runner = self._runner()
        runner.batch.items = [
            svc.AccountCheckItemState(
                line=1,
                username="pasted",
                email="pasted@e.com",
                export_text="pasted----pw----e----ep",
                account_id=None,
            )
        ]
        runner._set_item(0, status="account_verified", message="ok", finished_at=NOW)

        assert self.session.query(AccountEvent).count() == 0

    def test_non_terminal_update_emits_nothing(self):
        """中间态（running 等，不带 finished_at）不写事件，否则时间线会被刷屏。"""
        a = self.account("evt-3")
        runner = self._runner()
        runner.batch.items = [
            svc.AccountCheckItemState(
                line=1,
                username="evt-3",
                email="e@e.com",
                export_text="x",
                account_id=a.id,
            )
        ]
        # ⚠️ 必须**照真实调用**传 `finished_at=None`：`_run_one` 就是这么传的。
        # 第一版用例漏了它，于是「中间态也发事件」这个 bug 逃过了用例，
        # 最后是靠跑一次真实检测、看到事件 title 是「running」才发现的。
        runner._set_item(
            0, status="running", message="检测中", started_at=NOW, finished_at=None
        )
        # 事件由 runner 自己的会话提交；本会话要结束事务才能看到新快照，
        # 否则「没查到」可能只是快照旧了 —— 那会让这条断言假绿。
        self.session.commit()
        assert self.session.query(AccountEvent).count() == 0

    def test_same_batch_does_not_duplicate(self):
        """同一次检测重试/重放不会写两条（幂等键带 account_id 与 batch_id）。"""
        a = self.account("evt-4")
        runner = self._runner()
        runner.batch.items = [
            svc.AccountCheckItemState(
                line=1,
                username="evt-4",
                email="e@e.com",
                export_text="x",
                account_id=a.id,
            )
        ]
        runner._set_item(0, status="account_verified", message="ok", finished_at=NOW)
        runner._set_item(0, status="account_verified", message="ok", finished_at=NOW)
        self.session.commit()

        assert (
            self.session.query(AccountEvent)
            .filter(AccountEvent.event_type == svc.EVENT_ACCOUNT_CHECK)
            .count()
            == 1
        )


class TestRequestValidation:
    """两种来源二选一：都给或都不给都该被拒。"""

    @staticmethod
    def _call(**kwargs):
        from fastapi import HTTPException

        from web.backend.routers.account_check import create_account_check_batch
        from web.backend.schemas import AccountCheckBatchRequest

        return (
            create_account_check_batch,
            AccountCheckBatchRequest,
            HTTPException,
            kwargs,
        )

    def test_both_sources_rejected(self):
        create, Request, HTTPException, _ = self._call()
        with pytest.raises(HTTPException) as exc:
            create(
                Request(text="u----p----e----ep", account_ids=[1]),
                db=None,
                current=User(username="x", password_hash="h", role="super_admin"),
            )
        assert exc.value.status_code == 400
        assert "二选一" in str(exc.value.detail)

    def test_neither_source_rejected(self):
        create, Request, HTTPException, _ = self._call()
        with pytest.raises(HTTPException) as exc:
            create(
                Request(text="   ", account_ids=[]),
                db=None,
                current=User(username="x", password_hash="h", role="super_admin"),
            )
        assert exc.value.status_code == 400


class TestFilterBasedSelection(_Base):
    """「按筛选全选」：与账号列表同一套条件，列表筛出什么就检测什么。"""

    def _resolve(self, **kwargs):
        """复现路由里的解析逻辑（用同一套 _filter_accounts_query）。"""
        from web.backend.routers.accounts import (
            _filter_accounts_query,
            _visible_accounts_query,
        )

        query = _filter_accounts_query(
            self.session,
            _visible_accounts_query(self.session, self.user),
            self.user,
            search=kwargs.get("search", ""),
            tag=kwargs.get("tag", ""),
            status=kwargs.get("status", ""),
            ck_status=kwargs.get("ck_status", ""),
            bind_status="",
        )
        return [a.username for a in query.order_by(Account.id.asc()).all()]

    def test_filters_by_tag(self):
        self.account("tag-a", tag="青龙")
        self.account("tag-b", tag="白虎")
        assert self._resolve(tag="青龙") == ["tag-a"]

    def test_filters_by_status_and_ck(self):
        self.account("st-imp")
        self.account("st-bad", cookie="ck", ck_check_status="invalid")
        assert self._resolve(ck_status="invalid") == ["st-bad"]
        assert "st-imp" in self._resolve(status="imported")

    def test_no_filter_returns_all_visible(self):
        self.account("all-1")
        self.account("all-2")
        assert {"all-1", "all-2"} <= set(self._resolve())


class TestInputToItemCarriesAccountId(_Base):
    """`registry.create` 必须把 `account_id` 从输入复制到条目。

    漏了这一行时**没有任何报错** —— 批次照跑、结果照显示、zip 照导出，
    只是事件永远不发。第一版就漏了，靠「跑一次真实检测再查事件」才发现。
    """

    def test_item_keeps_account_id(self):
        from web.backend.services.account_check_service import AccountCheckRegistry

        a = self.account("reg-1")
        rows = self.session.query(Account).filter(Account.id == a.id).all()
        inputs = svc.build_inputs_from_accounts(rows, [a.id])

        runner = AccountCheckRegistry().create(
            accounts=inputs,
            created_by=self.user.id,
            concurrency=1,
            max_login_retries=0,
            max_total_time=60,
        )
        assert [i.account_id for i in runner.batch.items] == [a.id]

    def test_text_mode_items_have_no_account_id(self):
        from web.backend.services.account_check_service import AccountCheckRegistry

        inputs = svc.parse_account_check_lines("u----p----e----ep")
        runner = AccountCheckRegistry().create(
            accounts=inputs,
            created_by=self.user.id,
            concurrency=1,
            max_login_retries=0,
            max_total_time=60,
        )
        assert [i.account_id for i in runner.batch.items] == [None]