797 lines
26 KiB
Python
797 lines
26 KiB
Python
import asyncio
|
|
import threading
|
|
from datetime import UTC, datetime, timedelta
|
|
from decimal import Decimal
|
|
|
|
import pytest
|
|
|
|
from backend import today_environment as environment
|
|
|
|
SGE_HTML = """
|
|
<html><body>
|
|
<h1>上海黄金交易所2026年09月16日延时行情</h1>
|
|
<table>
|
|
<tr><th>合约</th><th>最新价</th><th>最高价</th><th>最低价</th><th>今开盘</th></tr>
|
|
<tr><td>Au99.95</td><td>927.2</td><td>949.8</td><td>927.2</td><td>949.8</td></tr>
|
|
<tr><td><span>Au99.99</span></td><td>935.99</td><td>936.4</td><td>924.5</td><td>925.0</td></tr>
|
|
</table>
|
|
</body></html>
|
|
"""
|
|
|
|
|
|
def test_parse_sge_delayed_page_selects_exact_au9999_contract():
|
|
quote = environment.parse_sge_au9999(SGE_HTML)
|
|
|
|
assert quote == (Decimal("935.99"), "2026-09-16")
|
|
|
|
|
|
def test_parse_sge_uses_named_columns_when_the_table_is_reordered():
|
|
html = SGE_HTML.replace(
|
|
'<tr><th>合约</th><th>最新价</th><th>最高价</th><th>最低价</th><th>今开盘</th></tr>',
|
|
'<tr><th>最高价</th><th>合约</th><th>今开盘</th><th>最新价</th></tr>',
|
|
).replace(
|
|
'<tr><td>Au99.95</td><td>927.2</td><td>949.8</td><td>927.2</td><td>949.8</td></tr>',
|
|
'<tr><td>949.8</td><td>Au99.95</td><td>927.2</td><td>927.2</td></tr>',
|
|
).replace(
|
|
'<tr><td><span>Au99.99</span></td><td>935.99</td><td>936.4</td><td>924.5</td><td>925.0</td></tr>',
|
|
'<tr><td>936.4</td><td><span>Au99.99</span></td><td>925.0</td><td>935.99</td></tr>',
|
|
)
|
|
|
|
assert environment.parse_sge_au9999(html) == (Decimal("935.99"), "2026-09-16")
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"html",
|
|
[
|
|
SGE_HTML.replace("最新价", "收盘价"),
|
|
SGE_HTML.replace("Au99.99", "Au99.999"),
|
|
SGE_HTML.replace("935.99", "NaN"),
|
|
SGE_HTML.replace("935.99", "Infinity"),
|
|
SGE_HTML.replace("935.99", "0"),
|
|
SGE_HTML.replace("2026年09月16日", "2026年02月30日"),
|
|
],
|
|
)
|
|
def test_parse_sge_rejects_changed_structure_or_invalid_quote(html):
|
|
with pytest.raises(ValueError):
|
|
environment.parse_sge_au9999(html)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_gold_falls_back_to_latest_trading_day_when_delayed_quote_is_zero():
|
|
delayed_html = SGE_HTML.replace("2026年09月16日", "2026年09月20日").replace(
|
|
"935.99", "0.0"
|
|
)
|
|
history_html = """
|
|
<table>
|
|
<tr><th>序号</th><th>日期</th><th>合约</th><th>开盘价</th><th>收盘价</th></tr>
|
|
<tr><td>1</td><td>2026-09-17</td><td>Au99.99</td><td>937.50</td><td>934.81</td></tr>
|
|
<tr><td>2</td><td>2026-09-18</td><td>Au99.99</td><td>937.00</td><td>947.09</td></tr>
|
|
<tr><td>3</td><td>2026-09-16</td><td>Au99.99</td><td>-</td><td>-</td></tr>
|
|
</table>
|
|
"""
|
|
seen = []
|
|
|
|
async def request(url, **_kwargs):
|
|
seen.append(url)
|
|
return delayed_html if url == environment.GOLD_URL else history_html
|
|
|
|
result = await environment.fetch_gold(request)
|
|
|
|
assert seen == [
|
|
environment.GOLD_URL,
|
|
environment.GOLD_HISTORY_URL.format(start_date="2026-08-20", end_date="2026-09-20"),
|
|
]
|
|
assert result["latest_price"] == Decimal("947.09")
|
|
assert result["market_date"] == "2026-09-18"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gold_source_has_time_for_two_sequential_upstream_requests(monkeypatch):
|
|
environment.clear_cache()
|
|
monkeypatch.setattr(environment, "SOURCE_TIMEOUT_SECONDS", 0.01)
|
|
monkeypatch.setattr(environment, "GOLD_SOURCE_TIMEOUT_SECONDS", 0.05, raising=False)
|
|
monkeypatch.setattr(environment, "TOTAL_TIMEOUT_SECONDS", 0.06)
|
|
|
|
async def weather():
|
|
return {"temperature_c": 25, "source": "Open-Meteo"}
|
|
|
|
async def gold():
|
|
await asyncio.sleep(0.03)
|
|
return {
|
|
"contract": "Au99.99",
|
|
"latest_price": Decimal("947.09"),
|
|
"market_date": "2026-09-18",
|
|
"delayed": True,
|
|
"source": "上海黄金交易所",
|
|
}
|
|
|
|
result = await environment.get_environment(weather_fetcher=weather, gold_fetcher=gold)
|
|
|
|
assert result["gold"]["latest_price"] == "947.09"
|
|
assert result["errors"] == {}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_weather_uses_fixed_haishu_location_and_current_conditions():
|
|
seen = {}
|
|
|
|
async def request(url, **kwargs):
|
|
seen["url"] = url
|
|
seen["params"] = kwargs["params"]
|
|
return {
|
|
"timezone": "Asia/Shanghai",
|
|
"current": {
|
|
"time": "2026-09-16T14:15",
|
|
"temperature_2m": 28.4,
|
|
"apparent_temperature": 30.1,
|
|
"weather_code": 2,
|
|
}
|
|
}
|
|
|
|
result = await environment.fetch_weather(request)
|
|
|
|
assert seen == {
|
|
"url": environment.WEATHER_URL,
|
|
"params": {
|
|
"latitude": 29.88,
|
|
"longitude": 121.55,
|
|
"current": "temperature_2m,apparent_temperature,weather_code",
|
|
"timezone": "Asia/Shanghai",
|
|
},
|
|
}
|
|
assert result == {
|
|
"temperature_c": 28.4,
|
|
"apparent_temperature_c": 30.1,
|
|
"weather_code": 2,
|
|
"observed_at": "2026-09-16T14:15:00+08:00",
|
|
"source": "Open-Meteo",
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"current",
|
|
[
|
|
{},
|
|
{"time": "2026-09-16T14:15", "temperature_2m": True, "apparent_temperature": 30.1, "weather_code": 2},
|
|
{"time": "2026-09-16T14:15", "temperature_2m": float("nan"), "apparent_temperature": 30.1, "weather_code": 2},
|
|
{"time": "2026-09-16T14:15", "temperature_2m": 28.4, "apparent_temperature": float("inf"), "weather_code": 2},
|
|
{"time": "2026-09-16T14:15", "temperature_2m": 101, "apparent_temperature": 30.1, "weather_code": 2},
|
|
{"time": "2026-09-16T14:15", "temperature_2m": 28.4, "apparent_temperature": -101, "weather_code": 2},
|
|
{"time": "2026-09-16T14:15", "temperature_2m": 28.4, "apparent_temperature": 30.1, "weather_code": 100},
|
|
{"time": "2026-02-30T14:15", "temperature_2m": 28.4, "apparent_temperature": 30.1, "weather_code": 2},
|
|
{"time": "2026-09-16T14:15+00:00", "temperature_2m": 28.4, "apparent_temperature": 30.1, "weather_code": 2},
|
|
],
|
|
)
|
|
async def test_fetch_weather_rejects_invalid_contract(current):
|
|
async def request(*_args, **_kwargs):
|
|
return {"timezone": "Asia/Shanghai", "current": current}
|
|
|
|
with pytest.raises((KeyError, TypeError, ValueError)):
|
|
await environment.fetch_weather(request)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_weather_rejects_wrong_response_timezone():
|
|
async def request(*_args, **_kwargs):
|
|
return {
|
|
"timezone": "UTC",
|
|
"current": {
|
|
"time": "2026-09-16T14:15",
|
|
"temperature_2m": 28.4,
|
|
"apparent_temperature": 30.1,
|
|
"weather_code": 2,
|
|
},
|
|
}
|
|
|
|
with pytest.raises(ValueError):
|
|
await environment.fetch_weather(request)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_environment_fetches_sources_concurrently_and_formats_gold_as_string():
|
|
now = datetime(2026, 9, 16, 8, 30, tzinfo=UTC)
|
|
environment.clear_cache()
|
|
both_started = asyncio.Event()
|
|
started = set()
|
|
|
|
async def weather():
|
|
started.add("weather")
|
|
if len(started) == 2:
|
|
both_started.set()
|
|
await asyncio.wait_for(both_started.wait(), timeout=0.1)
|
|
return {"temperature_c": 28.4, "source": "Open-Meteo"}
|
|
|
|
async def gold():
|
|
started.add("gold")
|
|
if len(started) == 2:
|
|
both_started.set()
|
|
await asyncio.wait_for(both_started.wait(), timeout=0.1)
|
|
return {
|
|
"contract": "Au99.99",
|
|
"latest_price": Decimal("935.990"),
|
|
"market_date": "2026-09-16",
|
|
"delayed": True,
|
|
"source": "上海黄金交易所",
|
|
}
|
|
|
|
result = await environment.get_environment(now=now, weather_fetcher=weather, gold_fetcher=gold)
|
|
|
|
assert result["date"] == {
|
|
"solar_date": "2026-09-16",
|
|
"weekday": "星期三",
|
|
"lunar": "农历八月初六",
|
|
"timezone": "Asia/Shanghai",
|
|
}
|
|
assert result["weather"]["temperature_c"] == 28.4
|
|
assert result["weather"]["stale"] is False
|
|
assert result["gold"]["latest_price"] == "935.990"
|
|
assert result["gold"]["delayed"] is True
|
|
assert result["gold"]["stale"] is False
|
|
assert result["errors"] == {}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_each_source_has_independent_ttl_and_stale_if_error():
|
|
base = datetime(2026, 9, 16, 8, tzinfo=UTC)
|
|
environment.clear_cache()
|
|
calls = {"weather": 0, "gold": 0}
|
|
|
|
async def weather():
|
|
calls["weather"] += 1
|
|
if calls["weather"] > 1:
|
|
raise RuntimeError("weather down")
|
|
return {"temperature_c": 25, "source": "Open-Meteo"}
|
|
|
|
async def gold():
|
|
calls["gold"] += 1
|
|
if calls["gold"] > 1:
|
|
raise RuntimeError("gold down")
|
|
return {
|
|
"contract": "Au99.99",
|
|
"latest_price": Decimal("900.10"),
|
|
"market_date": "2026-09-16",
|
|
"delayed": True,
|
|
"source": "上海黄金交易所",
|
|
}
|
|
|
|
await environment.get_environment(now=base, weather_fetcher=weather, gold_fetcher=gold)
|
|
after_ten_minutes = await environment.get_environment(
|
|
now=base + timedelta(minutes=10), weather_fetcher=weather, gold_fetcher=gold
|
|
)
|
|
assert calls == {"weather": 1, "gold": 2}
|
|
assert after_ten_minutes["weather"]["stale"] is False
|
|
assert after_ten_minutes["gold"]["stale"] is True
|
|
assert after_ten_minutes["errors"] == {"gold": "upstream_unavailable"}
|
|
|
|
after_twenty_minutes = await environment.get_environment(
|
|
now=base + timedelta(minutes=20), weather_fetcher=weather, gold_fetcher=gold
|
|
)
|
|
assert calls == {"weather": 2, "gold": 3}
|
|
assert after_twenty_minutes["weather"]["stale"] is True
|
|
assert after_twenty_minutes["gold"]["stale"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_failure_without_stale_cache_returns_partial_success():
|
|
environment.clear_cache()
|
|
|
|
async def weather():
|
|
raise RuntimeError("weather down")
|
|
|
|
async def gold():
|
|
return {
|
|
"contract": "Au99.99",
|
|
"latest_price": Decimal("901.2"),
|
|
"market_date": "2026-09-16",
|
|
"delayed": True,
|
|
"source": "上海黄金交易所",
|
|
}
|
|
|
|
result = await environment.get_environment(weather_fetcher=weather, gold_fetcher=gold)
|
|
|
|
assert result["weather"] is None
|
|
assert result["gold"]["latest_price"] == "901.2"
|
|
assert result["errors"] == {"weather": "upstream_unavailable"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_expired_stale_values_are_not_returned():
|
|
base = datetime(2026, 9, 1, tzinfo=UTC)
|
|
environment.clear_cache()
|
|
|
|
async def weather_ok():
|
|
return {"temperature_c": 25, "source": "Open-Meteo"}
|
|
|
|
async def gold_ok():
|
|
return {
|
|
"contract": "Au99.99",
|
|
"latest_price": Decimal("900.10"),
|
|
"market_date": "2026-09-01",
|
|
"delayed": True,
|
|
"source": "上海黄金交易所",
|
|
}
|
|
|
|
await environment.get_environment(
|
|
now=base, weather_fetcher=weather_ok, gold_fetcher=gold_ok
|
|
)
|
|
|
|
async def failed():
|
|
raise RuntimeError("down")
|
|
|
|
result = await environment.get_environment(
|
|
now=base + timedelta(days=8), weather_fetcher=failed, gold_fetcher=failed
|
|
)
|
|
|
|
assert result["weather"] is None
|
|
assert result["gold"] is None
|
|
assert result["errors"] == {
|
|
"weather": "upstream_unavailable",
|
|
"gold": "upstream_unavailable",
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_each_source_single_flights_concurrent_cache_misses():
|
|
environment.clear_cache()
|
|
release = asyncio.Event()
|
|
calls = {"weather": 0, "gold": 0}
|
|
|
|
async def weather():
|
|
calls["weather"] += 1
|
|
await release.wait()
|
|
return {"temperature_c": 25, "source": "Open-Meteo"}
|
|
|
|
async def gold():
|
|
calls["gold"] += 1
|
|
await release.wait()
|
|
return {
|
|
"contract": "Au99.99",
|
|
"latest_price": Decimal("900.10"),
|
|
"market_date": "2026-09-16",
|
|
"delayed": True,
|
|
"source": "上海黄金交易所",
|
|
}
|
|
|
|
tasks = [asyncio.create_task(environment.get_environment(weather_fetcher=weather, gold_fetcher=gold)) for _ in range(5)]
|
|
for _ in range(10):
|
|
if calls == {"weather": 1, "gold": 1}:
|
|
break
|
|
await asyncio.sleep(0)
|
|
assert calls == {"weather": 1, "gold": 1}
|
|
release.set()
|
|
results = await asyncio.gather(*tasks)
|
|
|
|
assert len(results) == 5
|
|
assert calls == {"weather": 1, "gold": 1}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_each_source_single_flights_concurrent_upstream_failures():
|
|
environment.clear_cache()
|
|
release = asyncio.Event()
|
|
calls = {"weather": 0, "gold": 0}
|
|
|
|
async def failed(name):
|
|
calls[name] += 1
|
|
await release.wait()
|
|
raise RuntimeError(f"{name} down")
|
|
|
|
tasks = [asyncio.create_task(environment.get_environment(
|
|
weather_fetcher=lambda: failed("weather"),
|
|
gold_fetcher=lambda: failed("gold"),
|
|
)) for _ in range(5)]
|
|
for _ in range(10):
|
|
if calls == {"weather": 1, "gold": 1}:
|
|
break
|
|
await asyncio.sleep(0)
|
|
assert calls == {"weather": 1, "gold": 1}
|
|
release.set()
|
|
results = await asyncio.gather(*tasks)
|
|
|
|
assert calls == {"weather": 1, "gold": 1}
|
|
assert all(result["weather"] is None and result["gold"] is None for result in results)
|
|
assert all(
|
|
result["errors"]
|
|
== {"weather": "upstream_unavailable", "gold": "upstream_unavailable"}
|
|
for result in results
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_slow_old_fetch_cannot_overwrite_a_newer_cache_value():
|
|
environment.clear_cache()
|
|
old_started = asyncio.Event()
|
|
release_old = asyncio.Event()
|
|
|
|
async def old_weather():
|
|
old_started.set()
|
|
await release_old.wait()
|
|
return {"temperature_c": 10, "source": "Open-Meteo"}
|
|
|
|
old = asyncio.create_task(environment._cached_source(
|
|
"weather", old_weather, datetime(2026, 9, 16, 8, tzinfo=UTC),
|
|
environment.WEATHER_TTL, environment.WEATHER_STALE_TTL,
|
|
))
|
|
await old_started.wait()
|
|
|
|
# Simulate a cache generation reset while the old upstream request is still running.
|
|
# The replacement request must be able to finish first, and the old completion must
|
|
# neither clear its in-flight slot nor overwrite its newer value.
|
|
environment.clear_cache()
|
|
release_new = asyncio.Event()
|
|
|
|
async def new_weather():
|
|
await release_new.wait()
|
|
return {"temperature_c": 20, "source": "Open-Meteo"}
|
|
|
|
newer_task = asyncio.create_task(environment._cached_source(
|
|
"weather", new_weather, datetime(2026, 9, 16, 8, 20, tzinfo=UTC),
|
|
environment.WEATHER_TTL, environment.WEATHER_STALE_TTL,
|
|
))
|
|
await asyncio.sleep(0)
|
|
release_new.set()
|
|
newer, _ = await newer_task
|
|
release_old.set()
|
|
await old
|
|
|
|
async def should_not_fetch():
|
|
raise AssertionError("new cache value should still be fresh")
|
|
|
|
cached, _ = await environment._cached_source(
|
|
"weather", should_not_fetch, datetime(2026, 9, 16, 8, 21, tzinfo=UTC),
|
|
environment.WEATHER_TTL, environment.WEATHER_STALE_TTL,
|
|
)
|
|
|
|
assert newer["temperature_c"] == 20
|
|
assert cached["temperature_c"] == 20
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_source_timeout_preserves_other_source_success(monkeypatch):
|
|
environment.clear_cache()
|
|
monkeypatch.setattr(environment, "SOURCE_TIMEOUT_SECONDS", 0.01)
|
|
|
|
async def weather():
|
|
await asyncio.sleep(1)
|
|
return {"temperature_c": 25, "source": "Open-Meteo"}
|
|
|
|
async def gold():
|
|
return {"contract": "Au99.99", "latest_price": Decimal("901.2"), "market_date": "2026-09-16", "delayed": True, "source": "上海黄金交易所"}
|
|
|
|
result = await environment.get_environment(weather_fetcher=weather, gold_fetcher=gold)
|
|
|
|
assert result["weather"] is None
|
|
assert result["gold"]["latest_price"] == "901.2"
|
|
assert result["errors"] == {"weather": "timeout"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("failure", ["exception", "timeout"])
|
|
async def test_cancelled_waiters_do_not_leak_failed_shared_source(monkeypatch, failure):
|
|
environment.clear_cache()
|
|
monkeypatch.setattr(environment, "SOURCE_TIMEOUT_SECONDS", 0.01)
|
|
started = asyncio.Event()
|
|
release = asyncio.Event()
|
|
loop_errors = []
|
|
loop = asyncio.get_running_loop()
|
|
previous_handler = loop.get_exception_handler()
|
|
loop.set_exception_handler(lambda _loop, context: loop_errors.append(context))
|
|
calls = 0
|
|
|
|
async def source():
|
|
nonlocal calls
|
|
calls += 1
|
|
started.set()
|
|
if failure == "timeout":
|
|
await asyncio.sleep(1)
|
|
else:
|
|
await release.wait()
|
|
raise RuntimeError("secret https://provider.invalid/feed")
|
|
return {"temperature_c": 25, "source": "Open-Meteo"}
|
|
|
|
try:
|
|
waiters = [
|
|
asyncio.create_task(
|
|
environment._cached_source(
|
|
"weather",
|
|
source,
|
|
datetime(2026, 9, 16, 8, tzinfo=UTC),
|
|
environment.WEATHER_TTL,
|
|
environment.WEATHER_STALE_TTL,
|
|
)
|
|
)
|
|
for _ in range(2)
|
|
]
|
|
await started.wait()
|
|
for waiter in waiters:
|
|
waiter.cancel()
|
|
await asyncio.gather(*waiters, return_exceptions=True)
|
|
release.set()
|
|
await asyncio.sleep(0.03)
|
|
|
|
assert all(not runtime.inflight for runtime in environment._runtimes.values())
|
|
assert not [
|
|
context
|
|
for context in loop_errors
|
|
if context.get("message") == "Task exception was never retrieved"
|
|
]
|
|
|
|
async def replacement():
|
|
nonlocal calls
|
|
calls += 1
|
|
return {"temperature_c": 26, "source": "Open-Meteo"}
|
|
|
|
value, error = await environment._cached_source(
|
|
"weather",
|
|
replacement,
|
|
datetime(2026, 9, 16, 8, 20, tzinfo=UTC),
|
|
environment.WEATHER_TTL,
|
|
environment.WEATHER_STALE_TTL,
|
|
)
|
|
assert value["temperature_c"] == 26
|
|
assert error is None
|
|
assert calls == 2
|
|
finally:
|
|
loop.set_exception_handler(previous_handler)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
("exc", "expected_code"),
|
|
[
|
|
(TimeoutError("secret timeout https://provider.invalid"), "timeout"),
|
|
(ValueError("secret malformed payload https://provider.invalid"), "invalid_upstream_response"),
|
|
(RuntimeError("secret outage https://provider.invalid"), "upstream_unavailable"),
|
|
],
|
|
)
|
|
async def test_source_errors_use_stable_codes_without_exception_details(exc, expected_code):
|
|
environment.clear_cache()
|
|
|
|
async def failed():
|
|
raise exc
|
|
|
|
value, error = await environment._cached_source(
|
|
"weather",
|
|
failed,
|
|
datetime(2026, 9, 16, 8, tzinfo=UTC),
|
|
environment.WEATHER_TTL,
|
|
environment.WEATHER_STALE_TTL,
|
|
)
|
|
|
|
assert value is None
|
|
assert error == expected_code
|
|
assert "secret" not in error
|
|
assert "provider.invalid" not in error
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stale_source_error_uses_stable_code_without_exception_details():
|
|
base = datetime(2026, 9, 16, 8, tzinfo=UTC)
|
|
environment.clear_cache()
|
|
|
|
async def initial():
|
|
return {"temperature_c": 25, "source": "Open-Meteo"}
|
|
|
|
await environment._cached_source(
|
|
"weather", initial, base, environment.WEATHER_TTL, environment.WEATHER_STALE_TTL
|
|
)
|
|
|
|
async def failed():
|
|
raise RuntimeError("secret https://provider.invalid/feed")
|
|
|
|
value, error = await environment._cached_source(
|
|
"weather",
|
|
failed,
|
|
base + timedelta(minutes=20),
|
|
environment.WEATHER_TTL,
|
|
environment.WEATHER_STALE_TTL,
|
|
)
|
|
|
|
assert value["temperature_c"] == 25
|
|
assert value["stale"] is True
|
|
assert error == "upstream_unavailable"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_creator_cancellation_does_not_close_shared_default_fetch_client(monkeypatch):
|
|
environment.clear_cache()
|
|
started = asyncio.Event()
|
|
release = asyncio.Event()
|
|
calls = {"weather": 0, "gold": 0}
|
|
|
|
class Response:
|
|
def raise_for_status(self):
|
|
return None
|
|
|
|
def json(self):
|
|
return {
|
|
"timezone": "Asia/Shanghai",
|
|
"current": {
|
|
"time": "2026-09-16T14:15",
|
|
"temperature_2m": 25,
|
|
"apparent_temperature": 26,
|
|
"weather_code": 1,
|
|
},
|
|
}
|
|
|
|
@property
|
|
def text(self):
|
|
return SGE_HTML
|
|
|
|
class Client:
|
|
closed = False
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, *_args):
|
|
self.closed = True
|
|
|
|
async def get(self, url, **_kwargs):
|
|
source = "weather" if url == environment.WEATHER_URL else "gold"
|
|
calls[source] += 1
|
|
if calls == {"weather": 1, "gold": 1}:
|
|
started.set()
|
|
await release.wait()
|
|
if self.closed:
|
|
raise RuntimeError("client closed while shared refresh was running")
|
|
return Response()
|
|
|
|
monkeypatch.setattr(environment.httpx, "AsyncClient", lambda **_kwargs: Client())
|
|
creator = asyncio.create_task(environment.get_environment())
|
|
await started.wait()
|
|
survivor = asyncio.create_task(environment.get_environment())
|
|
await asyncio.sleep(0)
|
|
creator.cancel()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await creator
|
|
release.set()
|
|
|
|
result = await survivor
|
|
|
|
assert result["weather"]["temperature_c"] == 25
|
|
assert result["gold"]["latest_price"] == "935.99"
|
|
assert result["errors"] == {}
|
|
assert calls == {"weather": 1, "gold": 1}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_clear_cache_generation_blocks_old_refill_and_old_done_cleanup():
|
|
environment.clear_cache()
|
|
base = datetime(2026, 9, 16, 8, tzinfo=UTC)
|
|
old_started = asyncio.Event()
|
|
release_old = asyncio.Event()
|
|
release_new = asyncio.Event()
|
|
|
|
async def old_fetch():
|
|
old_started.set()
|
|
await release_old.wait()
|
|
return {"temperature_c": 10}
|
|
|
|
old_waiter = asyncio.create_task(
|
|
environment._cached_source(
|
|
"weather", old_fetch, base, environment.WEATHER_TTL, environment.WEATHER_STALE_TTL
|
|
)
|
|
)
|
|
await old_started.wait()
|
|
environment.clear_cache()
|
|
|
|
async def new_fetch():
|
|
await release_new.wait()
|
|
return {"temperature_c": 20}
|
|
|
|
new_waiter = asyncio.create_task(
|
|
environment._cached_source(
|
|
"weather",
|
|
new_fetch,
|
|
base + timedelta(minutes=20),
|
|
environment.WEATHER_TTL,
|
|
environment.WEATHER_STALE_TTL,
|
|
)
|
|
)
|
|
await asyncio.sleep(0)
|
|
release_old.set()
|
|
assert (await old_waiter)[0]["temperature_c"] == 10
|
|
|
|
# The old task's callback must not remove the new generation's in-flight task.
|
|
third_waiter = asyncio.create_task(
|
|
environment._cached_source(
|
|
"weather",
|
|
lambda: pytest.fail("old done callback cleared the new task"),
|
|
base + timedelta(minutes=20),
|
|
environment.WEATHER_TTL,
|
|
environment.WEATHER_STALE_TTL,
|
|
)
|
|
)
|
|
release_new.set()
|
|
new_value, third_value = await asyncio.gather(new_waiter, third_waiter)
|
|
assert new_value[0]["temperature_c"] == 20
|
|
assert third_value[0]["temperature_c"] == 20
|
|
|
|
async def no_refetch():
|
|
raise AssertionError("old generation refilled or new generation was lost")
|
|
|
|
cached, _ = await environment._cached_source(
|
|
"weather",
|
|
no_refetch,
|
|
base + timedelta(minutes=21),
|
|
environment.WEATHER_TTL,
|
|
environment.WEATHER_STALE_TTL,
|
|
)
|
|
assert cached["temperature_c"] == 20
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_runtime_singleflight_state_is_isolated_per_event_loop():
|
|
environment.clear_cache()
|
|
barrier = threading.Barrier(2)
|
|
calls = []
|
|
results = []
|
|
failures = []
|
|
|
|
def worker(label):
|
|
async def run():
|
|
async def fetch():
|
|
calls.append(label)
|
|
await asyncio.to_thread(barrier.wait)
|
|
return {"temperature_c": label}
|
|
|
|
value, error = await environment._cached_source(
|
|
"weather",
|
|
fetch,
|
|
datetime(2026, 9, 16, 8, tzinfo=UTC),
|
|
environment.WEATHER_TTL,
|
|
environment.WEATHER_STALE_TTL,
|
|
)
|
|
results.append((value["temperature_c"], error))
|
|
|
|
try:
|
|
asyncio.run(run())
|
|
except RuntimeError as exc:
|
|
failures.append(exc)
|
|
|
|
threads = [threading.Thread(target=worker, args=(label,)) for label in (11, 22)]
|
|
for thread in threads:
|
|
thread.start()
|
|
for thread in threads:
|
|
thread.join(timeout=2)
|
|
|
|
assert not any(thread.is_alive() for thread in threads)
|
|
assert failures == []
|
|
assert sorted(calls) == [11, 22]
|
|
assert sorted(results) == [(11, None), (22, None)]
|
|
|
|
|
|
def test_today_environment_endpoint_requires_authentication(client):
|
|
response = client.get("/api/v1/today/environment")
|
|
|
|
assert response.status_code == 401
|
|
|
|
|
|
def test_today_environment_endpoint_returns_aggregated_payload(client, monkeypatch):
|
|
client.post(
|
|
"/api/v1/setup/initialize",
|
|
json={"username": "owner", "password": "correct horse battery staple"},
|
|
)
|
|
payload = {
|
|
"date": {
|
|
"solar_date": "2026-09-16",
|
|
"weekday": "星期三",
|
|
"lunar": "农历八月初六",
|
|
"timezone": "Asia/Shanghai",
|
|
},
|
|
"weather": None,
|
|
"gold": None,
|
|
"errors": {"weather": "unavailable", "gold": "unavailable"},
|
|
}
|
|
|
|
async def aggregate():
|
|
return payload
|
|
|
|
monkeypatch.setattr("backend.main.get_today_environment", aggregate)
|
|
response = client.get("/api/v1/today/environment")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == payload
|