trex/trex-core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: align portfolio-hub read adapter to live API contract

Browse Source 
whoami now hits GET /v1/user_info (the live endpoint; /v1/auth/whoami
does not exist), get_portfolios parses the {portfolios:[...],total} envelope
and filters by managed_by_daemon_id client-side (the list endpoint is
token-scoped, no such query param), and a missing URL scheme now defaults
to http:// so a bare host can't silently degrade the daemon.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
genichin
2026-06-06 10:34:35 +09:00
co-authored by Claude Opus 4.7
parent ca757009f5
commit 43bad2424e
2 changed files with 73 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/adapters/portfolio/test_hub_read.py
+48 -18
View File
@@ -18,46 +18,76 @@ def test_whoami_returns_token_daemon_id():
def handler(req: httpx.Request) -> httpx.Response:
captured["url"] = str(req.url)
captured["auth"] = req.headers.get("authorization")
return httpx.Response(200, json={"daemon_id": "d-7", "scope": ["read"]})
return httpx.Response(200, json={"daemon_id": "d-7", "user_id": "u-1",
"username": "bot", "trading_enabled": True})
adapter = PortfolioHubReadAdapter("https://pf.test", token="tok",
client=_client(handler))
assert adapter.whoami() == "d-7"
assert captured["url"] == "https://pf.test/v1/auth/whoami"
assert captured["url"] == "https://pf.test/v1/user_info"
assert captured["auth"] == "Bearer tok"
def test_get_portfolios_scoped_by_managed_by_daemon_id():
def test_get_portfolios_parses_envelope_and_filters_by_daemon():
captured = {}
def handler(req: httpx.Request) -> httpx.Response:
captured["url"] = str(req.url)
return httpx.Response(200, json={
"items": [
{"portfolio_id": "pf-1", "symbol": "005930",
"cash_balance": 1_000_000, "asset_quantity": 10,
"active_strategy_id": "target-value-v1",
"strategy_params": {"target_value": 1_000_000}},
{"portfolio_id": "pf-2", "symbol": "000660",
"cash_balance": 2_000_000, "asset_quantity": 0,
"active_strategy_id": None, "strategy_params": {}},
],
})
# live shape: {"portfolios": [PortfolioItem...], "total": N};
# PortfolioItem has no active_strategy_id/strategy_params and the
# endpoint is token-scoped (no managed_by_daemon_id query param).
return httpx.Response(200, json={"portfolios": [
{"portfolio_id": "pf-1", "asset_id": "a-1", "symbol": "005930",
"cash_balance": 1_000_000, "asset_quantity": 10,
"managed_by_daemon_id": "d-7", "price": 70_000},
{"portfolio_id": "pf-2", "asset_id": "a-2", "symbol": "000660",
"cash_balance": 2_000_000, "asset_quantity": 0,
"managed_by_daemon_id": "d-7"},
{"portfolio_id": "pf-other", "symbol": "035720",
"cash_balance": 0, "asset_quantity": 0,
"managed_by_daemon_id": "d-OTHER"},
], "total": 3})
adapter = PortfolioHubReadAdapter("https://pf.test", token="tok",
client=_client(handler))
items = adapter.get_portfolios("d-7")
assert "managed_by_daemon_id=d-7" in captured["url"]
assert captured["url"] == "https://pf.test/v1/portfolios"
# client-side filter drops the other daemon's portfolio
assert {it.portfolio_id for it in items} == {"pf-1", "pf-2"}
assert items[0].symbol == "005930"
assert items[0].cash_balance == 1_000_000
assert items[0].active_strategy_id == "target-value-v1"
assert items[0].strategy_params == {"target_value": 1_000_000}
# no price field
# strategy fields absent on PortfolioItem -> graceful defaults
assert items[0].active_strategy_id is None
assert items[0].strategy_params == {}
# no price field on the domain item
for it in items:
assert not hasattr(it, "price")
def test_get_portfolios_includes_items_without_daemon_field():
def handler(req: httpx.Request) -> httpx.Response:
return httpx.Response(200, json={"portfolios": [
{"portfolio_id": "pf-1", "symbol": "005930",
"cash_balance": 0, "asset_quantity": 0},
], "total": 1})
adapter = PortfolioHubReadAdapter("https://pf.test", client=_client(handler))
items = adapter.get_portfolios("d-7")
assert {it.portfolio_id for it in items} == {"pf-1"}
def test_base_url_without_scheme_defaults_to_http():
captured = {}
def handler(req: httpx.Request) -> httpx.Response:
captured["url"] = str(req.url)
return httpx.Response(200, json={"daemon_id": "d-7"})
adapter = PortfolioHubReadAdapter("127.0.0.1:18000", client=_client(handler))
adapter.whoami()
assert captured["url"] == "http://127.0.0.1:18000/v1/user_info"
def test_whoami_5xx_raises_unavailable():
def handler(req):
return httpx.Response(503, text="upstream down")