script-build-and-framework-tmp/agent/tests/test_browser_governance.py

from __future__ import annotations

from pathlib import Path
from typing import Any, Dict, Optional

import pytest

from agent.tools.builtin.browser import downloads, providers


class _FakeResponse:
    def __init__(self, status_code: int, chunks: tuple[bytes, ...] = ()) -> None:
        self.status_code = status_code
        self._chunks = chunks

    async def __aenter__(self) -> "_FakeResponse":
        return self

    async def __aexit__(self, *_args: object) -> None:
        return None

    async def aiter_bytes(self):
        for chunk in self._chunks:
            yield chunk


class _FakeClient:
    response = _FakeResponse(200)

    def __init__(self, **_kwargs: object) -> None:
        pass

    async def __aenter__(self) -> "_FakeClient":
        return self

    async def __aexit__(self, *_args: object) -> None:
        return None

    def stream(self, _method: str, _url: str) -> _FakeResponse:
        return self.response


class _ContainerProvider:
    async def create(self, url: str, account_name: str) -> Dict[str, Any]:
        return {"success": True, "url": url, "account_name": account_name}


class _CookieProvider:
    def get_cookie_row(self, cookie_type: str) -> Optional[Dict[str, Any]]:
        return {"type": cookie_type, "cookies": "session=test"}


@pytest.fixture(autouse=True)
def _reset_providers() -> None:
    yield
    providers.reset_browser_providers()


@pytest.mark.asyncio
async def test_download_accepts_partial_content_and_sanitizes_name(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    _FakeClient.response = _FakeResponse(206, (b"abc", b"def"))
    monkeypatch.setattr(downloads.httpx, "AsyncClient", _FakeClient)

    result = await downloads.download_direct_url(
        "https://files.example.invalid/path/report.txt",
        "../../escaped.txt",
        download_dir=tmp_path,
    )

    assert result.error is None
    assert result.metadata["status_code"] == 206
    assert result.metadata["bytes"] == 6
    assert Path(result.metadata["path"]) == tmp_path / "escaped.txt"
    assert (tmp_path / "escaped.txt").read_bytes() == b"abcdef"
    assert not (tmp_path / ".escaped.txt.part").exists()


@pytest.mark.asyncio
async def test_download_failure_always_returns_tool_result(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    _FakeClient.response = _FakeResponse(404)
    monkeypatch.setattr(downloads.httpx, "AsyncClient", _FakeClient)
    stale_partial = tmp_path / ".missing.bin.part"
    stale_partial.write_bytes(b"stale")

    result = await downloads.download_direct_url(
        "https://files.example.invalid/missing.bin",
        "",
        download_dir=tmp_path,
    )

    assert result.error == "HTTP 404"
    assert result.metadata["status_code"] == 404
    assert not list(tmp_path.iterdir())


@pytest.mark.asyncio
async def test_download_infers_and_decodes_url_filename(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    _FakeClient.response = _FakeResponse(200, (b"content",))
    monkeypatch.setattr(downloads.httpx, "AsyncClient", _FakeClient)

    result = await downloads.download_direct_url(
        "https://files.example.invalid/%E6%8A%A5%E5%91%8A.txt",
        "",
        download_dir=tmp_path,
    )

    assert result.error is None
    assert Path(result.metadata["path"]).name == "报告.txt"
    assert (tmp_path / "报告.txt").read_bytes() == b"content"


@pytest.mark.asyncio
async def test_container_provider_normalizes_unexpected_failures(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    class _BrokenClient:
        def __init__(self, **_kwargs: object) -> None:
            raise TypeError("malformed service response")

    monkeypatch.setattr(providers.httpx, "AsyncClient", _BrokenClient)
    provider = providers.HttpContainerProvider(
        base_url="https://browser.example.invalid",
        startup_delay=0,
        retry_delay=0,
    )

    result = await provider.create("https://example.invalid", "account")

    assert result["success"] is False
    assert result["error"] == "malformed service response"


def test_browser_providers_are_host_replaceable() -> None:
    container = _ContainerProvider()
    cookie = _CookieProvider()

    providers.configure_browser_providers(
        container_provider=container,
        cookie_provider=cookie,
    )

    assert providers.get_browser_container_provider() is container
    assert providers.get_browser_cookie_provider() is cookie


def test_base_class_no_longer_knows_host_endpoint_or_cookie_table() -> None:
    source = (
        Path(__file__).parents[1]
        / "agent"
        / "tools"
        / "builtin"
        / "browser"
        / "baseClass.py"
    ).read_text(encoding="utf-8")

    assert "47.84.182.56" not in source
    assert "agent_channel_cookies" not in source