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 = """
上海黄金交易所2026年09月16日延时行情
| 合约 | 最新价 | 最高价 | 最低价 | 今开盘 |
|---|
| Au99.95 | 927.2 | 949.8 | 927.2 | 949.8 |
| Au99.99 | 935.99 | 936.4 | 924.5 | 925.0 |
"""
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(
'| 合约 | 最新价 | 最高价 | 最低价 | 今开盘 |
|---|
',
'| 最高价 | 合约 | 今开盘 | 最新价 |
|---|
',
).replace(
'| Au99.95 | 927.2 | 949.8 | 927.2 | 949.8 |
',
'| 949.8 | Au99.95 | 927.2 | 927.2 |
',
).replace(
'| Au99.99 | 935.99 | 936.4 | 924.5 | 925.0 |
',
'| 936.4 | Au99.99 | 925.0 | 935.99 |
',
)
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_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