urbnywrt
21437c78ad
Implements MihomoClient wrapping Mihomo's REST API with provider refresh polling, config reload, and startup readiness check. Also fixes a Python 3.14 + respx 0.21.1 compatibility issue where the httpcore mocker passes bytes method to httpx.Request causing Method pattern matching to fail; resolved by switching respx DEFAULT_MOCKER to the httpx-level mocker. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
99 lines
3.1 KiB
Python
99 lines
3.1 KiB
Python
import pytest
|
|
import respx
|
|
import httpx
|
|
from mihomo import MihomoClient
|
|
|
|
|
|
@pytest.fixture
|
|
def client():
|
|
return MihomoClient("http://mihomo:9090", "secret123")
|
|
|
|
|
|
async def test_trigger_provider_refresh(client):
|
|
with respx.mock:
|
|
respx.put("http://mihomo:9090/providers/proxies/myprovider").mock(
|
|
return_value=httpx.Response(204)
|
|
)
|
|
await client.trigger_provider_refresh("myprovider")
|
|
|
|
|
|
async def test_get_provider(client):
|
|
payload = {
|
|
"name": "myprovider",
|
|
"updatedAt": "2024-01-01T00:00:00Z",
|
|
"proxies": [{"name": "node1", "type": "ss"}],
|
|
}
|
|
with respx.mock:
|
|
respx.get("http://mihomo:9090/providers/proxies/myprovider").mock(
|
|
return_value=httpx.Response(200, json=payload)
|
|
)
|
|
data = await client.get_provider("myprovider")
|
|
assert data["updatedAt"] == "2024-01-01T00:00:00Z"
|
|
|
|
|
|
async def test_refresh_and_collect_detects_update(client):
|
|
call_count = 0
|
|
|
|
def provider_handler(request):
|
|
nonlocal call_count
|
|
call_count += 1
|
|
if call_count == 1:
|
|
return httpx.Response(200, json={"updatedAt": "t1", "proxies": []})
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"updatedAt": "t2",
|
|
"proxies": [
|
|
{"name": "n1", "type": "ss", "server": "1.2.3.4", "port": 443}
|
|
],
|
|
},
|
|
)
|
|
|
|
with respx.mock:
|
|
respx.put("http://mihomo:9090/providers/proxies/myprovider").mock(
|
|
return_value=httpx.Response(204)
|
|
)
|
|
respx.get("http://mihomo:9090/providers/proxies/myprovider").mock(
|
|
side_effect=provider_handler
|
|
)
|
|
proxies = await client.refresh_and_collect("myprovider", timeout=5)
|
|
|
|
assert len(proxies) == 1
|
|
assert proxies[0]["name"] == "n1"
|
|
|
|
|
|
async def test_refresh_and_collect_timeout(client):
|
|
with respx.mock:
|
|
respx.put("http://mihomo:9090/providers/proxies/slow").mock(
|
|
return_value=httpx.Response(204)
|
|
)
|
|
respx.get("http://mihomo:9090/providers/proxies/slow").mock(
|
|
return_value=httpx.Response(200, json={"updatedAt": "never-changes", "proxies": []})
|
|
)
|
|
with pytest.raises(TimeoutError):
|
|
await client.refresh_and_collect("slow", timeout=2)
|
|
|
|
|
|
async def test_reload_config(client):
|
|
with respx.mock:
|
|
respx.put("http://mihomo:9090/configs").mock(return_value=httpx.Response(204))
|
|
await client.reload_config()
|
|
|
|
|
|
async def test_wait_ready_succeeds(client):
|
|
with respx.mock:
|
|
respx.get("http://mihomo:9090/version").mock(
|
|
return_value=httpx.Response(200, json={"version": "1.0"})
|
|
)
|
|
await client.wait_ready(retries=3)
|
|
|
|
|
|
async def test_wait_ready_raises_after_retries():
|
|
slow_client = MihomoClient("http://missing:9090", "")
|
|
with respx.mock:
|
|
respx.get("http://missing:9090/version").mock(
|
|
return_value=httpx.Response(503)
|
|
)
|
|
with pytest.raises(RuntimeError, match="did not become available"):
|
|
await slow_client.wait_ready(retries=2)
|