mirror of
https://github.com/langbot-app/LangBot.git
synced 2026-10-30 20:35:22 +08:00
268 lines
10 KiB
Python
268 lines
10 KiB
Python
"""QWeather: per-installation config and a non-blocking async HTTP path."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
from langbot_plugin.api.proxies.invocation import bind_invocation, current_config
|
|
from langbot_plugin.entities.io.context import InstallationBinding
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
if str(REPO_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(REPO_ROOT))
|
|
|
|
import pkg.weather_data as weather_data # noqa: E402
|
|
|
|
|
|
def _binding(installation: str, *, workspace: str = "workspace-qweather"):
|
|
return InstallationBinding(
|
|
instance_uuid="instance-1",
|
|
workspace_uuid=workspace,
|
|
installation_uuid=installation,
|
|
runtime_revision=1,
|
|
artifact_digest="b" * 64,
|
|
)
|
|
|
|
|
|
class _StubPlugin:
|
|
"""Per-invocation config, exactly as the SDK's BasePlugin.get_config."""
|
|
|
|
def __init__(self, handler):
|
|
self.plugin_runtime_handler = handler
|
|
|
|
def get_config(self) -> dict:
|
|
return dict(current_config(self.plugin_runtime_handler) or {})
|
|
|
|
def get_plugin_config(self) -> dict:
|
|
return self.get_config()
|
|
|
|
|
|
class _FakeResponse:
|
|
def __init__(self, payload):
|
|
self._payload = payload
|
|
self.status_code = 200
|
|
|
|
def json(self):
|
|
return self._payload
|
|
|
|
|
|
def _payload_for(url: str) -> dict:
|
|
if "city/" in url:
|
|
return {"code": "200", "location": [{"name": "Testville", "id": "101"}]}
|
|
if "warning/now" in url:
|
|
return {"code": "204"}
|
|
if "air/now" in url:
|
|
return {
|
|
"code": "200",
|
|
"now": {"category": "Good", "aqi": "50", "pm2p5": "12", "pm10": "20",
|
|
"o3": "30", "co": "0.5", "no2": "10", "so2": "5"},
|
|
}
|
|
if "weather/24h" in url:
|
|
return {"code": "200", "hourly": [{"fxTime": "2026-01-01T00:00", "temp": "5", "icon": "100", "text": "Clear"}]}
|
|
if "indices" in url:
|
|
return {"code": "200", "daily": [{"name": "Sport", "category": "sport", "text": "Good"}]}
|
|
if "astronomy/sun" in url:
|
|
return {"code": "200", "sunrise": "06:00", "sunset": "18:00"}
|
|
if url.endswith("/weather/now"):
|
|
return {
|
|
"code": "200",
|
|
"now": {"obsTime": "2026-01-01T00:00+08:00", "temp": "15", "icon": "100",
|
|
"text": "Sunny", "windScale": "3", "windDir": "N", "humidity": "45",
|
|
"precip": "0", "vis": "10"},
|
|
}
|
|
return {"code": "200", "daily": [{"fxDate": "2026-01-01", "tempMax": "9", "tempMin": "1",
|
|
"textDay": "Sunny", "textNight": "Clear",
|
|
"iconDay": "100", "iconNight": "150"}]}
|
|
|
|
|
|
class _SlowHttp:
|
|
"""Async stand-in for ``httpx.AsyncClient``'s request (the network boundary)."""
|
|
|
|
def __init__(self, delay: float):
|
|
self.delay = delay
|
|
self.calls: list[tuple[str, dict]] = []
|
|
|
|
async def __call__(self, url, params=None):
|
|
self.calls.append((url, dict(params or {})))
|
|
await asyncio.sleep(self.delay)
|
|
return _FakeResponse(_payload_for(url))
|
|
|
|
|
|
class _ControlHttp:
|
|
"""Network stand-in with controllable failing and hanging branches.
|
|
|
|
``hang_suffixes`` maps a URL suffix to how long that request's cancellation
|
|
teardown takes. The branch never completes on its own; only cancellation can
|
|
end it.
|
|
"""
|
|
|
|
def __init__(self, *, failing_suffix: str | None = None, hang_suffixes: dict[str, float] | None = None):
|
|
self.failing_suffix = failing_suffix
|
|
self.hang_suffixes = dict(hang_suffixes or {})
|
|
self.hang_started = asyncio.Event()
|
|
self.release = asyncio.Event()
|
|
self.hang_tasks: dict[str, asyncio.Task] = {}
|
|
self.tasks: list[asyncio.Task] = []
|
|
|
|
async def wait_for_hangs(self, count: int, timeout: float = 1.0) -> None:
|
|
deadline = asyncio.get_running_loop().time() + timeout
|
|
while len(self.hang_tasks) < count:
|
|
if asyncio.get_running_loop().time() >= deadline:
|
|
raise AssertionError(f"only {len(self.hang_tasks)} of {count} hanging requests started")
|
|
await asyncio.sleep(0.005)
|
|
|
|
async def __call__(self, url, params=None):
|
|
if "city/" not in url:
|
|
self.tasks.append(asyncio.current_task())
|
|
for suffix, cleanup in self.hang_suffixes.items():
|
|
if url.endswith(suffix):
|
|
self.hang_tasks[suffix] = asyncio.current_task()
|
|
self.hang_started.set()
|
|
try:
|
|
await self.release.wait()
|
|
except asyncio.CancelledError:
|
|
# Cancellation still needs a teardown; a caller that returns
|
|
# before this finishes leaves a request task behind.
|
|
await asyncio.sleep(cleanup)
|
|
raise
|
|
if self.failing_suffix is not None and url.endswith(self.failing_suffix):
|
|
# Wait until a sibling request is provably in flight, then fail.
|
|
await self.hang_started.wait()
|
|
raise weather_data.APIError("upstream rejected the request")
|
|
return _FakeResponse(_payload_for(url))
|
|
|
|
|
|
class WeatherSharedRuntimeTests(unittest.IsolatedAsyncioTestCase):
|
|
def setUp(self) -> None:
|
|
self._original_get_data = weather_data._get_data
|
|
|
|
def tearDown(self) -> None:
|
|
weather_data._get_data = self._original_get_data
|
|
|
|
async def _run_command(self, binding, config, slow: _SlowHttp, command, handler):
|
|
weather_data._get_data = slow
|
|
context = SimpleNamespace(crt_params=["Testville"])
|
|
|
|
async def collect():
|
|
return [
|
|
item
|
|
async for item in command.registered_subcommands["*"].subcommand(command, context)
|
|
]
|
|
|
|
with bind_invocation(handler, config=config, binding=binding):
|
|
task = asyncio.ensure_future(collect())
|
|
ticks = 0
|
|
while not task.done():
|
|
await asyncio.sleep(0.02)
|
|
ticks += 1
|
|
results = await task
|
|
return results, ticks
|
|
|
|
async def test_handler_yields_the_loop_and_uses_the_invocation_config(self) -> None:
|
|
from components.commands.weather import Weather as WeatherCommand
|
|
|
|
binding_a = _binding("installation-qweather-a")
|
|
binding_b = _binding("installation-qweather-b")
|
|
slow = _SlowHttp(0.2)
|
|
|
|
# ONE command object and one plugin serve both installations.
|
|
handler = SimpleNamespace()
|
|
command = WeatherCommand()
|
|
command.plugin = _StubPlugin(handler)
|
|
|
|
results_a, ticks_a = await self._run_command(
|
|
binding_a, {"qweather_apikey": "KEY-A", "qweather_apitype": "0"}, slow, command, handler
|
|
)
|
|
results_b, ticks_b = await self._run_command(
|
|
binding_b, {"qweather_apikey": "KEY-B", "qweather_apitype": "0"}, slow, command, handler
|
|
)
|
|
|
|
# The handler awaited the slow network boundary instead of holding the loop:
|
|
# a concurrent 20 ms ticker kept running. Pre-fix the synchronous
|
|
# requests.Session().get fan-out starved it and this stayed near zero.
|
|
self.assertGreaterEqual(ticks_a, 5)
|
|
self.assertGreaterEqual(ticks_b, 5)
|
|
|
|
self.assertIn("温度", results_a[0].text)
|
|
self.assertIsNone(results_a[0].error)
|
|
|
|
keys_a = [params.get("key") for _, params in slow.calls[:8]]
|
|
keys_b = [params.get("key") for _, params in slow.calls[8:]]
|
|
self.assertEqual(keys_a, ["KEY-A"] * 8)
|
|
self.assertEqual(keys_b, ["KEY-B"] * 8)
|
|
self.assertNotIn("KEY-B", keys_a)
|
|
self.assertNotIn("KEY-A", keys_b)
|
|
|
|
async def test_failed_fan_out_leaves_no_request_task_running(self) -> None:
|
|
"""F-01: a failing branch must cancel and await its fan-out siblings.
|
|
|
|
``asyncio.gather`` propagates the first exception without cancelling the
|
|
other tasks, so pre-fix the hanging request outlived the failed call
|
|
while still holding this invocation's API key.
|
|
"""
|
|
from pkg.weather_data import Weather as WeatherData
|
|
|
|
http = _ControlHttp(failing_suffix="/indices/1d", hang_suffixes={"/astronomy/sun": 0.0})
|
|
weather_data._get_data = http
|
|
data = WeatherData(city_name="Testville", api_key="KEY-A", api_type=0)
|
|
|
|
try:
|
|
with self.assertRaises(weather_data.APIError):
|
|
await data.load_data()
|
|
|
|
hang_task = http.hang_tasks.get("/astronomy/sun")
|
|
self.assertIsNotNone(hang_task, "the hanging sibling request never started")
|
|
# The invocation is over (the error already propagated): no request
|
|
# task may still be alive holding this call's API key.
|
|
self.assertTrue(
|
|
all(task.done() for task in http.tasks),
|
|
"a request task outlived the failed load_data() call",
|
|
)
|
|
self.assertTrue(hang_task.cancelled())
|
|
finally:
|
|
for task in http.tasks:
|
|
if not task.done():
|
|
task.cancel()
|
|
await asyncio.gather(task, return_exceptions=True)
|
|
|
|
async def test_cancelled_invocation_leaves_no_request_task_running(self) -> None:
|
|
"""F-01 cancel path: cancellation must await the fan-out teardown.
|
|
|
|
One sibling's teardown finishes immediately while the other's takes
|
|
longer; returning before both complete would leave the slow request
|
|
running after the invocation was cancelled.
|
|
"""
|
|
from pkg.weather_data import Weather as WeatherData
|
|
|
|
http = _ControlHttp(hang_suffixes={"/weather/now": 0.0, "/astronomy/sun": 0.2})
|
|
weather_data._get_data = http
|
|
data = WeatherData(city_name="Testville", api_key="KEY-A", api_type=0)
|
|
|
|
loader = asyncio.ensure_future(data.load_data())
|
|
try:
|
|
await http.wait_for_hangs(2)
|
|
loader.cancel()
|
|
with self.assertRaises(asyncio.CancelledError):
|
|
await loader
|
|
self.assertTrue(
|
|
all(task.done() for task in http.tasks),
|
|
"a request task outlived the cancelled load_data() call",
|
|
)
|
|
self.assertTrue(http.hang_tasks["/astronomy/sun"].cancelled())
|
|
finally:
|
|
if not loader.done():
|
|
loader.cancel()
|
|
await asyncio.gather(loader, return_exceptions=True)
|
|
for task in http.tasks:
|
|
if not task.done():
|
|
task.cancel()
|
|
await asyncio.gather(task, return_exceptions=True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|