84 lines
3.2 KiB
Python
84 lines
3.2 KiB
Python
"""REQ-119 provider source and refresh lifecycle regressions."""
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from trex_core.runtime.broker_provider import (
|
|
BrokerCredentialUnavailable,
|
|
ConfigBrokerProvider,
|
|
)
|
|
from trex_core.runtime.config import DaemonConfig, KISConfig, PortfolioHubConfig
|
|
|
|
|
|
def _cfg(*, fetched=True, broker="kis"):
|
|
return DaemonConfig(
|
|
dsn="postgresql://test/test", broker=broker, execution_interval_sec=1,
|
|
kis=KISConfig("https://openapivts.example", "env-key", "env-secret", "env-account", 1, False),
|
|
portfolio_hub=PortfolioHubConfig("http://hub"), credential_fetch_enabled=fetched,
|
|
)
|
|
|
|
|
|
class _Broker:
|
|
def __init__(self): self.credentials = []
|
|
def reinitialize(self, credential): self.credentials.append(credential)
|
|
|
|
|
|
class _Read:
|
|
def __init__(self, ctx=None, error=None): self.ctx, self.error, self.calls = ctx, error, 0
|
|
def fetch_broker_credential(self):
|
|
self.calls += 1
|
|
if self.error:
|
|
raise self.error
|
|
return self.ctx
|
|
|
|
|
|
def _ctx(version="v1"):
|
|
return SimpleNamespace(trading_enabled=True, broker=SimpleNamespace(
|
|
account_ref="synthetic-account", app_key="synthetic-key", app_secret="synthetic-secret",
|
|
credential_version=version,
|
|
))
|
|
|
|
|
|
def test_fetched_provider_constructs_once_reuses_and_refreshes_without_env_fallback(monkeypatch):
|
|
broker = _Broker()
|
|
monkeypatch.setattr("trex_core.runtime.broker_provider._make_env_broker", lambda cfg: broker)
|
|
read = _Read(_ctx())
|
|
provider = ConfigBrokerProvider(_cfg(), read_port=read)
|
|
|
|
got, source = provider.get_broker_for_read()
|
|
assert got is broker
|
|
assert source.mode == "fetched"
|
|
assert source.identity_label == "credential-version:v1"
|
|
assert broker.credentials[0].app_key == "synthetic-key"
|
|
assert read.calls == 1
|
|
assert provider.refresh_day_start_credential() == source
|
|
assert read.calls == 1 # same day cache, no broker/token churn
|
|
|
|
|
|
def test_fetched_provider_failure_is_sanitized_and_never_constructs_env_broker(monkeypatch):
|
|
monkeypatch.setattr("trex_core.runtime.broker_provider._make_env_broker", lambda cfg: pytest.fail("env fallback"))
|
|
provider = ConfigBrokerProvider(_cfg(), read_port=_Read(ctx=None, error=RuntimeError("secret payload")))
|
|
with pytest.raises(BrokerCredentialUnavailable) as raised:
|
|
provider.get_broker_for_read()
|
|
assert raised.value.code == "portfolio_hub_unavailable"
|
|
assert "secret" not in str(raised.value)
|
|
|
|
|
|
def test_enabled_paper_never_fetches_kis_credentials(monkeypatch):
|
|
read = _Read(_ctx())
|
|
provider = ConfigBrokerProvider(_cfg(broker="paper"), read_port=read)
|
|
_, source = provider.get_broker_for_read()
|
|
assert source.mode == "env"
|
|
assert read.calls == 0
|
|
|
|
|
|
def test_break_glass_is_distinct_and_skips_fetch(monkeypatch):
|
|
broker = _Broker()
|
|
monkeypatch.setenv("TREX_KIS_BROKER_BREAK_GLASS", "true")
|
|
monkeypatch.setattr("trex_core.runtime.broker_provider._make_env_broker", lambda cfg: broker)
|
|
read = _Read(_ctx())
|
|
_, source = ConfigBrokerProvider(_cfg(), read_port=read).get_broker_for_read()
|
|
assert source.mode == "break_glass"
|
|
assert source.identity_label == "operator-break-glass"
|
|
assert read.calls == 0
|