chuanqi/cq2_sdk/tests/unit/test_http_client.py

import pytest
from cq2_sdk.transport.http import HttpClient


@pytest.mark.asyncio
async def test_sign_injected_into_params(monkeypatch):
    captured = {}

    class FakeResp:
        status_code = 200
        headers = {"content-type": "application/json"}
        text = '{"ok": true}'

        def raise_for_status(self):
            pass

        def json(self):
            return {"ok": True}

    class FakeAsyncClient:
        def __init__(self, **kw):
            pass

        async def __aenter__(self):
            return self

        async def __aexit__(self, *a):
            return False

        async def request(self, method, url, params=None, data=None, headers=None):
            captured["url"] = url
            captured["params"] = params
            captured["method"] = method
            return FakeResp()

    import httpx
    monkeypatch.setattr(httpx, "AsyncClient", FakeAsyncClient)

    cli = HttpClient("https://example.com", signing_secret="SECRET", sign_version=1)
    result = await cli.request("GET", "/api/foo", params={"a": "1"})
    assert "sign" in captured["params"]
    assert captured["params"]["a"] == "1"
    assert result["ok"] is True
    assert result["_status"] == 200


@pytest.mark.asyncio
async def test_no_sign_when_version_unknown(monkeypatch):
    captured = {}

    class FakeResp:
        status_code = 200
        headers = {"content-type": "application/json"}

        def raise_for_status(self):
            pass

        def json(self):
            return {}

    class FakeAsyncClient:
        def __init__(self, **kw):
            pass

        async def __aenter__(self):
            return self

        async def __aexit__(self, *a):
            return False

        async def request(self, method, url, params=None, **kw):
            captured["params"] = params
            return FakeResp()

    import httpx
    monkeypatch.setattr(httpx, "AsyncClient", FakeAsyncClient)

    cli = HttpClient("https://example.com", signing_secret="SECRET", sign_version=99)
    await cli.request("GET", "/api/foo", params={"a": "1"})
    assert "sign" not in captured["params"]


@pytest.mark.asyncio
async def test_request_passes_method_and_path(monkeypatch):
    captured = {}

    class FakeResp:
        status_code = 200
        headers = {"content-type": "application/json"}

        def raise_for_status(self):
            pass

        def json(self):
            return {}

    class FakeAsyncClient:
        def __init__(self, **kw):
            pass

        async def __aenter__(self):
            return self

        async def __aexit__(self, *a):
            return False

        async def request(self, method, url, params=None, **kw):
            captured["method"] = method
            captured["url"] = url
            return FakeResp()

    import httpx
    monkeypatch.setattr(httpx, "AsyncClient", FakeAsyncClient)

    cli = HttpClient("https://api.996hezi.com", signing_secret="X")
    await cli.request("POST", "/userApi/v1/Login/mobile", data={"phone": "13800138000"})
    assert captured["method"] == "POST"
    assert captured["url"] == "https://api.996hezi.com/userApi/v1/Login/mobile"