tests/test_course_bot.py

from __future__ import annotations

import threading
import unittest
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import patch

from selenium.common.exceptions import TimeoutException, WebDriverException

from core.course_bot import CourseBot, RecoverableAutomationError
from core.db import Database
from tests.helpers import workspace_tempdir


class FakeDriver:
    def __init__(self) -> None:
        self.quit_called = False

    def quit(self) -> None:
        self.quit_called = True


class FakeActionDriver:
    def __init__(self, script_results: list[object] | None = None) -> None:
        self.script_results = list(script_results or [])
        self.script_calls: list[str] = []
        self.stop_called = False

    def execute_script(self, script: str, *_args):
        self.script_calls.append(script)
        if script.strip() == "window.stop();":
            self.stop_called = True
            return None
        if self.script_results:
            result = self.script_results.pop(0)
            if isinstance(result, Exception):
                raise result
            return result
        return "scheduled-js-click"


class FakeLoginOutcomeDriver:
    def __init__(self, url_steps: list[object]) -> None:
        self.url_steps = list(url_steps)

    @property
    def current_url(self) -> str:
        if self.url_steps:
            value = self.url_steps.pop(0)
            if isinstance(value, Exception):
                raise value
            return str(value)
        return ""

    def execute_script(self, _script: str):
        return ""


class FakeButton:
    def __init__(self, *, click_exception: Exception | None = None) -> None:
        self.click_count = 0
        self.click_exception = click_exception

    def click(self) -> None:
        self.click_count += 1
        if self.click_exception is not None:
            raise self.click_exception


class CourseBotTests(unittest.TestCase):
    def _make_store(self, temp_dir: Path) -> tuple[Database, dict]:
        store = Database(temp_dir / "test.db", default_parallel_limit=2)
        store.init_db()
        user_id = store.create_user("2023000000001", "encrypted", "Test User")
        store.add_course(user_id, "free", "1001001", "01")
        user = store.get_user(user_id)
        self.assertIsNotNone(user)
        return store, user

    def _make_config(self) -> SimpleNamespace:
        return SimpleNamespace(
            login_retry_limit=2,
            poll_interval_seconds=0,
            task_backoff_seconds=0,
            browser_page_timeout=1,
            selenium_error_limit=2,
            selenium_restart_limit=3,
            submit_captcha_retry_limit=3,
            chrome_binary="/usr/bin/chromium",
            chromedriver_path="/usr/bin/chromedriver",
        )

    def _make_bot(self, store: Database, user: dict, config: SimpleNamespace, logs: list[tuple[str, str]]) -> CourseBot:
        with patch("core.course_bot.onnx_inference.CaptchaONNXInference") as solver_cls:
            solver_cls.return_value = SimpleNamespace(classification=lambda _image: "1234")
            return CourseBot(
                config=config,
                store=store,
                task_id=1,
                user=user,
                password="pw",
                logger=lambda level, message: logs.append((level, message)),
            )

    def test_restart_browser_after_repeated_session_errors(self) -> None:
        with workspace_tempdir("course-bot-") as temp_dir:
            store, user = self._make_store(temp_dir)
            config = self._make_config()
            logs: list[tuple[str, str]] = []
            drivers: list[FakeDriver] = []

            def fake_configure_browser(**_kwargs):
                driver = FakeDriver()
                drivers.append(driver)
                return driver

            bot = self._make_bot(store, user, config, logs)
            stop_event = threading.Event()

            with patch("core.course_bot.webdriver_utils.configure_browser", side_effect=fake_configure_browser), patch.object(
                CourseBot,
                "_login",
                return_value=None,
            ), patch.object(
                CourseBot,
                "_goto_select_course",
                side_effect=RecoverableAutomationError("page unavailable"),
            ):
                result = bot.run(stop_event)

            self.assertEqual(result.status, "failed")
            self.assertIn("\u4efb\u52a1\u7ec8\u6b62", result.error)
            self.assertEqual(len(drivers), config.selenium_restart_limit)
            self.assertTrue(all(driver.quit_called for driver in drivers))
            self.assertTrue(any("\u91cd\u5efa\u6d4f\u89c8\u5668" in message or "\u91cd\u5efa Selenium \u4f1a\u8bdd" in message for _level, message in logs))

    def test_submit_captcha_flow_uses_ocr_branch(self) -> None:
        with workspace_tempdir("submit-captcha-") as temp_dir:
            store, user = self._make_store(temp_dir)
            config = self._make_config()
            logs: list[tuple[str, str]] = []
            bot = self._make_bot(store, user, config, logs)
            fake_button = FakeButton()
            fake_driver = FakeActionDriver(["scheduled-js-click"])
            fake_results = [{"result": True, "subject": "1001001_01", "detail": ""}]

            with patch.object(bot, "_find", return_value=fake_button), patch.object(
                bot,
                "_wait_for_submit_state",
                side_effect=["captcha", "result"],
            ), patch.object(
                bot,
                "_solve_visible_submit_captcha",
                return_value=True,
            ) as solve_mock, patch.object(
                bot,
                "_read_result_page",
                return_value=fake_results,
            ):
                results = bot._submit_with_optional_captcha(fake_driver, object(), "1001001_01")

            self.assertEqual(results, fake_results)
            self.assertEqual(fake_button.click_count, 0)
            solve_mock.assert_called_once()
            self.assertTrue(any("\u68c0\u6d4b\u5230\u9a8c\u8bc1\u7801" in message for _level, message in logs))
            self.assertTrue(any("\u5df2\u89e6\u53d1\u63d0\u4ea4" in message for _level, message in logs))

    def test_trigger_non_blocking_action_prefers_js_submit(self) -> None:
        with workspace_tempdir("non-blocking-js-") as temp_dir:
            store, user = self._make_store(temp_dir)
            config = self._make_config()
            logs: list[tuple[str, str]] = []
            bot = self._make_bot(store, user, config, logs)
            driver = FakeActionDriver(["scheduled-requestSubmit"])
            button = FakeButton()

            method = bot._trigger_non_blocking_action(
                driver,
                button,
                label="login-form",
                allow_form_submit=True,
            )

            self.assertEqual(method, "scheduled-requestSubmit")
            self.assertEqual(button.click_count, 0)
            self.assertFalse(driver.stop_called)

    def test_trigger_non_blocking_action_recovers_native_timeout(self) -> None:
        with workspace_tempdir("non-blocking-timeout-") as temp_dir:
            store, user = self._make_store(temp_dir)
            config = self._make_config()
            logs: list[tuple[str, str]] = []
            bot = self._make_bot(store, user, config, logs)
            driver = FakeActionDriver([WebDriverException("js failed")])
            button = FakeButton(click_exception=TimeoutException("renderer timeout"))

            method = bot._trigger_non_blocking_action(
                driver,
                button,
                label="login-form",
                allow_form_submit=True,
            )

            self.assertEqual(method, "native-click-timeout")
            self.assertEqual(button.click_count, 1)
            self.assertTrue(driver.stop_called)
            self.assertTrue(any("window.stop()" in message for _level, message in logs))

    def test_wait_for_login_outcome_ignores_transient_driver_errors(self) -> None:
        with workspace_tempdir("login-outcome-") as temp_dir:
            store, user = self._make_store(temp_dir)
            config = self._make_config()
            logs: list[tuple[str, str]] = []
            bot = self._make_bot(store, user, config, logs)
            driver = FakeLoginOutcomeDriver([
                WebDriverException("navigation in progress"),
                "https://zhjw.scu.edu.cn/index",
            ])

            with patch("core.course_bot.time.sleep", return_value=None):
                state, error_message = bot._wait_for_login_outcome(driver, timeout_seconds=1)

            self.assertEqual(state, "success")
            self.assertEqual(error_message, "")


if __name__ == "__main__":
    unittest.main()