Jesse_Chen
45d132588f
Advance the one-way import to git commit a6f47abd with consultation keypath golden, switch official VedAstro comparison to the REST Calculate bridge, and receive the 25 new registry entries behind research_only_blocked. Co-authored-by: Cursor <cursoragent@cursor.com>
112 lines
3.3 KiB
Python
112 lines
3.3 KiB
Python
import argparse
|
|
import http.client
|
|
import urllib.error
|
|
from io import BytesIO
|
|
|
|
import pytest
|
|
|
|
from scripts.vedastro_rest_bridge import (
|
|
SMOKE_DEFAULT,
|
|
_date_to_vedastro_std,
|
|
build_dasa_payload,
|
|
build_horoscope_payload,
|
|
build_parser,
|
|
call,
|
|
match_not_implemented,
|
|
probe_calculate_health,
|
|
reset_rate_limiter,
|
|
)
|
|
|
|
|
|
def _args(**overrides):
|
|
values = {
|
|
"birth": SMOKE_DEFAULT["birth"],
|
|
"name": SMOKE_DEFAULT["name"],
|
|
"lat": SMOKE_DEFAULT["lat"],
|
|
"lon": SMOKE_DEFAULT["lon"],
|
|
"ayanamsa": SMOKE_DEFAULT["ayanamsa"],
|
|
"start": "2026-01-01 00:00 +08:00",
|
|
"end": "2026-03-31 00:00 +08:00",
|
|
}
|
|
values.update(overrides)
|
|
return argparse.Namespace(**values)
|
|
|
|
|
|
def test_default_horoscope_payload_uses_fictional_smoke_without_network():
|
|
payload = build_horoscope_payload(_args())
|
|
|
|
assert payload == {
|
|
"Ayanamsa": "LAHIRI",
|
|
"BirthTime": {
|
|
"StdTime": "12:00 01/01/1990 +08:00",
|
|
"Location": {"Name": "Beijing, China", "Longitude": 116.4074, "Latitude": 39.9042},
|
|
},
|
|
"SortByWeight": False,
|
|
}
|
|
assert "1993" not in payload["BirthTime"]["StdTime"]
|
|
assert "14:49" not in payload["BirthTime"]["StdTime"]
|
|
|
|
|
|
def test_dasa_payload_formats_vedastro_range_times_without_network():
|
|
payload = build_dasa_payload(_args())
|
|
|
|
assert payload["Ayanamsa"] == "LAHIRI"
|
|
assert payload["StartTime"]["StdTime"] == "00:00 01/01/2026 +08:00"
|
|
assert payload["EndTime"]["StdTime"] == "00:00 31/03/2026 +08:00"
|
|
assert payload["Levels"] == 3
|
|
assert payload["PrecisionHours"] == 100
|
|
|
|
|
|
def test_date_parser_rejects_ambiguous_times():
|
|
with pytest.raises(ValueError):
|
|
_date_to_vedastro_std("2026-01-01")
|
|
|
|
|
|
def test_parser_match_is_registered_as_official_blocked():
|
|
parser = build_parser()
|
|
args = parser.parse_args(["match"])
|
|
assert args.cmd == "match"
|
|
blocked = match_not_implemented()
|
|
assert blocked["official_closure_state"] == "official_blocked"
|
|
assert blocked["official_closure_reason"] == "match_subcommand_not_implemented"
|
|
|
|
|
|
class _Handle(BytesIO):
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *args):
|
|
return False
|
|
|
|
|
|
def test_recorded_horoscope_status_pass_does_not_hit_network():
|
|
reset_rate_limiter()
|
|
recorded = {"Status": "Pass", "Payload": {"source": "recorded_smoke"}}
|
|
|
|
def fake_opener(_req, timeout=90):
|
|
return _Handle(b'{"Status": "Pass", "Payload": {"source": "recorded_smoke"}}')
|
|
|
|
result = call("HoroscopePredictions", build_horoscope_payload(_args()), opener=fake_opener)
|
|
assert result == recorded
|
|
|
|
|
|
def test_probe_maps_recorded_pass_and_rate_limit_without_network():
|
|
reset_rate_limiter()
|
|
health = probe_calculate_health(opener=lambda _req, timeout=8: _Handle(b'{"Status": "Pass"}'))
|
|
assert health["status"] == "official_verified"
|
|
assert health["transport"] == "rest"
|
|
|
|
def limited(_req, timeout=8):
|
|
raise urllib.error.HTTPError(
|
|
"https://api.vedastro.org/api/Calculate/HoroscopePredictions",
|
|
429,
|
|
"rate",
|
|
hdrs=http.client.HTTPMessage(),
|
|
fp=BytesIO(b""),
|
|
)
|
|
|
|
reset_rate_limiter()
|
|
blocked = probe_calculate_health(opener=limited)
|
|
assert blocked["status"] == "official_blocked"
|
|
assert blocked["reason"] == "rate_limited"
|