jiangchang-platform-kit/tests/test_rpa_wait_and_env.py

# -*- coding: utf-8 -*-
"""poll_until / 薄错误码 / browser launch timeout / machine_power 探测。"""
from __future__ import annotations

import asyncio
from pathlib import Path
from unittest.mock import MagicMock

import pytest

from jiangchang_skill_core.machine_power import probe_machine_power
from jiangchang_skill_core.rpa import errors
from jiangchang_skill_core.rpa.browser import launch_persistent_browser
from jiangchang_skill_core.rpa.wait import poll_until, poll_until_sync


def test_poll_until_succeeds_when_predicate_becomes_true(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    sleeps: list[float] = []

    async def _fake_sleep(seconds: float, *, chunk=None):
        del chunk
        sleeps.append(seconds)

    monkeypatch.setattr(
        "jiangchang_skill_core.rpa.wait.interruptible_sleep",
        _fake_sleep,
    )
    state = {"n": 0}

    def ready() -> bool:
        state["n"] += 1
        return state["n"] >= 3

    result = asyncio.run(poll_until(ready, timeout_sec=2.0, interval_sec=0.1))
    assert result.ok is True
    assert result.timed_out is False
    assert state["n"] == 3
    assert sleeps


def test_poll_until_times_out(monkeypatch: pytest.MonkeyPatch) -> None:
    async def _fake_sleep(seconds: float, *, chunk=None):
        del chunk, seconds

    monkeypatch.setattr(
        "jiangchang_skill_core.rpa.wait.interruptible_sleep",
        _fake_sleep,
    )
    result = asyncio.run(poll_until(lambda: False, timeout_sec=0.01, interval_sec=0.01))
    assert result.ok is False
    assert result.timed_out is True


def test_poll_until_sync_ok(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(
        "jiangchang_skill_core.rpa.wait.interruptible_sleep_sync",
        lambda seconds, *, chunk=None: None,
    )
    assert poll_until_sync(lambda: True, timeout_sec=1.0).ok is True


def test_thin_error_codes_exist() -> None:
    assert errors.ERR_UI_NOT_READY == "ERROR:UI_NOT_READY"
    assert errors.ERR_BROWSER_START_TIMEOUT == "ERROR:BROWSER_START_TIMEOUT"
    assert errors.ERR_ENV_DEGRADED == "ERROR:ENV_DEGRADED"
    assert errors.ERR_MISSING_BROWSER == "ERROR:MISSING_BROWSER"


def test_launch_persistent_browser_timeout_maps_code(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setattr(
        "jiangchang_skill_core.rpa.browser.find_chrome_executable",
        lambda: r"C:\fake\chrome.exe",
    )
    monkeypatch.setattr(
        "jiangchang_skill_core.rpa.browser.persistent_context_launch_parts",
        lambda extra_args=None: ([], None),
    )
    monkeypatch.setattr(
        "jiangchang_skill_core.rpa.browser.stealth_enabled",
        lambda: False,
    )

    async def _hang(**kwargs):
        del kwargs
        await asyncio.sleep(10)
        raise AssertionError("should have been cancelled by wait_for")

    playwright = MagicMock()
    playwright.chromium.launch_persistent_context = _hang

    async def _run():
        with pytest.raises(RuntimeError) as ctx:
            await launch_persistent_browser(
                playwright,
                profile_dir=r"C:\fake\profile",
                timeout_ms=50,
            )
        assert str(ctx.value).startswith(errors.ERR_BROWSER_START_TIMEOUT)

    asyncio.run(_run())


def test_launch_persistent_browser_missing_browser(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setattr(
        "jiangchang_skill_core.rpa.browser.find_chrome_executable",
        lambda: None,
    )

    async def _run():
        with pytest.raises(RuntimeError) as ctx:
            await launch_persistent_browser(MagicMock(), profile_dir=r"C:\fake\profile")
        assert str(ctx.value).startswith(errors.ERR_MISSING_BROWSER)

    asyncio.run(_run())


def test_probe_machine_power_shape_and_no_raise() -> None:
    payload = probe_machine_power()
    assert "probed" in payload
    assert "ac_power" in payload
    assert "nvidia" in payload
    assert isinstance(payload["probed"], bool)


def test_collect_runtime_diagnostics_includes_machine_power_without_fatal(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from jiangchang_skill_core import media_assets as ma
    from jiangchang_skill_core.runtime_diagnostics import (
        collect_runtime_diagnostics,
        runtime_diagnostics_dict,
    )

    env = {"JIANGCHANG_DATA_ROOT": str(tmp_path / "data"), "OPENCLAW_RECORD_VIDEO": "0"}
    monkeypatch.setattr(
        "jiangchang_skill_core.runtime_diagnostics.probe_media_assets",
        lambda *a, **k: ma.MediaAssetsStatus(
            root=tmp_path,
            exists=True,
            ready=True,
            source="local",
            warnings=[],
            ffmpeg_path=tmp_path / "ffmpeg.exe",
            music_dir=tmp_path / "music",
            music_ready=True,
        ),
    )
    monkeypatch.setattr(
        "jiangchang_skill_core.runtime_diagnostics.probe_background_music",
        lambda *a, **k: {
            "music_root": str(tmp_path / "music"),
            "mp3_count": 1,
            "usable_count": 1,
            "sample_path": str(tmp_path / "music" / "a.mp3"),
            "issue": None,
        },
    )
    monkeypatch.setattr(
        "jiangchang_skill_core.runtime_diagnostics._platform_kit_version",
        lambda: "1.2.11",
    )
    monkeypatch.setattr(
        "jiangchang_skill_core.runtime_diagnostics.probe_machine_power",
        lambda: {
            "probed": True,
            "ac_power": {
                "ac_line_status": 1,
                "on_ac_power": True,
                "battery_life_percent": 100,
            },
            "nvidia": {
                "gpu_name": "Fake",
                "enforced_power_limit_w": 140.0,
                "power_limit_w": 140.0,
                "clocks_throttle_reasons_active": "None",
            },
        },
    )

    diag = collect_runtime_diagnostics(skill_slug="demo", env=env)
    payload = runtime_diagnostics_dict(diag)
    assert payload["machine_power"]["probed"] is True
    assert diag.has_fatal_issues is False