"""The HTTP client and the ``quantcore-api`` CLI, against the app running in this process. ``TestClient`` satisfies the transport contract of :class:`QuantcoreClient` (``request(method, url, params=, json=, timeout=)``), so the same code paths that talk to the Spark are exercised here without a server: the URLs, the query parameters, the request bodies and the error handling. """ from __future__ import annotations import pytest from fastapi.testclient import TestClient import quantcore.api.app as app_module from quantcore import apicli from quantcore.client import QuantcoreClient, QuantcoreError @pytest.fixture(scope="module") def api(settings, store): app_module.state = app_module.State(settings) with TestClient(app_module.app) as transport: yield QuantcoreClient("http://testserver", session=transport) @pytest.fixture def cli(api, monkeypatch): """Run the CLI against the in-process app: every command builds its client through ``_client``.""" monkeypatch.setattr(apicli, "_client", lambda args: api) return apicli.main # ------------------------------------------------------------------------------------------------ # client # ------------------------------------------------------------------------------------------------ def test_the_client_covers_the_api(api): health = api.health() assert health["status"] == "ok" and health["store"]["prices"] is True assert len(api.universe(region="Europe")) == 6 assert api.symbol("S0")["symbol"] == "S0" table = api.symbols_table(limit=5) assert len(table["symbols"]) == 5 and "sectors" in table["facets"] assert table["n_universe"] >= table["n_matched"] >= 5 px = api.prices(["S0", "S1"], freq="weekly") assert px["columns"] == ["S0", "S1"] and px["freq"] == "weekly" assert api.returns(["S0"])["kind"] == "log_returns" assert api.market_index(freq="weekly")["n"] > 100 assert sum(api.market_weights()["weights"].values()) == pytest.approx(1.0) rows = api.asset_risk(["S0", "S1"], horizon="week") assert len(rows) == 2 and rows[0]["cvar95_hist"] > rows[0]["var95_hist"] assert api.risk_series("S0", horizon="week")["index"] assert api.covariance(["S0", "S1"])["columns"] == ["S0", "S1"] assert api.correlation(["S0", "S1"])["data"][0][0] == pytest.approx(1.0) assert api.portfolio_risk({"S0": 100, "S1": 50}, cash=1000.0, horizon="week")["value"] > 0 assert len(api.asset_performance(["S0", "S1"])) == 2 assert api.rolling_performance("S0", window=252)["index"] assert api.portfolio_performance({"S0": 100}) assert api.expected_returns(max_assets=6)["market_return_ann"] != 0 assert api.model_covariance(["S0", "S1"])["conditioning"]["n_assets"] == 2 scen = api.scenarios(["S0", "S1"], model="gbm", n_scenarios=1024, holdings={"S0": 10}) assert scen["description"]["n_scenarios"] == 1024 and scen["portfolio"]["invested"] > 0 # order-independent: another module may have created a snapshot on the session-scoped store assert isinstance(api.jobs(), list) and isinstance(api.snapshots(), list) def test_the_client_optimises(api): res = api.optimize(cash=1_000_000, max_assets=6, horizon="week", scenario_model="gaussian_exact", solver="highs", time_limit=120) assert res["status"] in ("Optimal", "FeasibleFound") assert res["solver"] == "highs" and res["positions"] # a rebalancing run carries the starting point, so before/after can be compared holdings = {p["symbol"]: p["shares"] for p in res["positions"][:2]} reb = api.optimize(mode="rebalance", holdings=holdings, cash=10_000, max_assets=6, horizon="week", scenario_model="gaussian_exact", solver="highs", time_limit=120) assert reb["before"] is not None and reb["before"]["invested"] > 0 def test_none_arguments_are_left_to_the_api(api): """Optional arguments that are ``None`` must not be sent, or the API would see "None".""" sent: dict = {} original = api.session.request def spy(method, url, **kw): sent.update({"params": kw.get("params"), "json": kw.get("json")}) return original(method, url, **kw) api.session.request = spy try: api.prices(["S0"], start=None, end=None) assert "start" not in (sent["params"] or {}) and "end" not in (sent["params"] or {}) api.expected_returns(symbols=None, max_assets=6) assert "symbols" not in (sent["json"] or {}) finally: api.session.request = original def test_the_client_reports_api_errors(api): with pytest.raises(QuantcoreError) as exc: api.symbol("NOPE") assert exc.value.status_code == 404 and not exc.value.unreachable with pytest.raises(QuantcoreError) as exc: # 422: the field name must reach the caller api.asset_risk(["S0"], window=1) assert exc.value.status_code == 422 and "window" in str(exc.value) assert "greater than or equal to 10" in str(exc.value) def test_the_client_reports_an_unreachable_server(): with QuantcoreClient("http://127.0.0.1:9", timeout=2.0) as qc, pytest.raises(QuantcoreError) as exc: qc.health() assert exc.value.unreachable and "127.0.0.1:9" in str(exc.value) def test_the_client_retries_dropped_gets_but_never_posts(): """A poll on a kept-alive connection the server just closed must not abort `update --wait`; a POST (optimisation, job start) must never be sent twice.""" with QuantcoreClient("http://127.0.0.1:9") as qc: retry = qc.session.get_adapter("http://127.0.0.1:9/health").max_retries assert retry.read >= 1 and "GET" in retry.allowed_methods and "POST" not in retry.allowed_methods def test_wait_for_job_polls_until_the_job_ends(): states = iter(["queued", "running", "running", "done"]) class Stub: def request(self, method, url, **kw): class Res: status_code = 200 @staticmethod def json(): return {"job_id": "abc", "status": next(states)} return Res() qc = QuantcoreClient("http://stub", session=Stub()) seen: list[str] = [] record = qc.wait_for_job("abc", poll=0.0, on_poll=lambda r: seen.append(r["status"])) assert record["status"] == "done" and seen == ["queued", "running", "running"] with pytest.raises(QuantcoreError): # a job that never finishes hits the deadline QuantcoreClient("http://stub", session=type("S", (), { "request": lambda self, m, u, **k: type("R", (), { "status_code": 200, "json": staticmethod(lambda: {"status": "running"})})()})() ).wait_for_job("abc", poll=0.0, deadline=-1.0) # ------------------------------------------------------------------------------------------------ # CLI # ------------------------------------------------------------------------------------------------ def test_every_documented_command_parses(): parser = apicli.build_parser() for argv in ( ["health"], ["endpoints"], ["universe", "--region", "Europe", "--sector", "Health Care"], ["symbol", "S0"], ["prices", "S0", "S1", "--freq", "weekly", "--tail", "5"], ["returns", "S0", "--currency", "local"], ["market", "weights", "--max-assets", "10"], ["risk", "S0", "S1", "--horizon", "week", "--drift", "zero", "--space", "value"], ["risk-portfolio", "--holdings", '{"S0": 10}'], ["correlation", "S0", "S1"], ["covariance", "S0", "S1", "--method", "rolling"], ["performance", "S0"], ["expected-returns", "--max-assets", "50", "--view", "S0:0.6:0.4"], ["model-covariance", "S0", "S1"], ["scenarios", "S0", "--model", "gbm", "--n-scenarios", "1024"], ["optimize", "--cash", "1000000", "--cvar-limit", "40000", "--risk-drift", "zero"], ["optimize", "--request", "r.json", "--json"], ["jobs", "--job-id", "abc"], ["setup", "--offline", "--wait"], ["update", "--full", "--wait"], ["snapshots", "--create", "--name", "demo"], ["selftest", "--full", "--write"], ): args = parser.parse_args(argv) assert callable(args.func), argv def test_sentiment_views_are_parsed(): assert apicli._views(["S0:0.6"]) == [{"symbol": "S0", "sentiment": 0.6}] assert apicli._views(["S0:0.6:0.4:0.3"]) == [ {"symbol": "S0", "sentiment": 0.6, "confidence": 0.4, "strength": 0.3}] with pytest.raises(SystemExit): apicli._views(["S0"]) @pytest.mark.parametrize("argv", [ ["health"], ["endpoints"], ["universe", "--region", "Europe"], ["symbol", "S0"], ["prices", "S0", "S1", "--tail", "3"], ["returns", "S0", "--tail", "3"], ["market", "index"], ["market", "weights"], ["risk", "S0", "S1", "--drift", "zero"], ["risk-portfolio", "--holdings", '{"S0": 10}'], ["correlation", "S0", "S1"], ["covariance", "S0", "S1"], ["performance", "S0"], ["expected-returns", "--max-assets", "6"], ["model-covariance", "S0", "S1"], ["scenarios", "S0", "S1", "--n-scenarios", "1024"], ["jobs"], ["snapshots"], ]) def test_the_cli_runs_every_read_only_command(cli, capsys, argv): assert cli(argv) == 0 assert capsys.readouterr().out.strip() def test_the_cli_optimises_and_prints_a_summary(cli, capsys): assert cli(["optimize", "--max-assets", "6", "--scenario-model", "gaussian_exact", "--solver", "highs", "--time-limit", "120"]) == 0 out = capsys.readouterr().out assert "status Optimal" in out and "E[P&L]" in out and "CVaR" in out def test_the_selftest_passes_against_a_healthy_service(cli, capsys): """The selftest is the deployment check: every endpoint, one table, exit 0 when all answered.""" assert cli(["selftest", "--n-symbols", "2"]) == 0 out = capsys.readouterr().out assert "FAIL" not in out assert "0 failed" in out and "3 skipped" in out # the writing endpoints stay untouched for endpoint in ("GET /health", "POST /risk/portfolio", "POST /optimize", "GET /snapshots"): assert endpoint in out def test_the_selftest_fails_loudly_when_the_service_is_down(capsys, monkeypatch): monkeypatch.setattr(apicli, "_client", lambda args: QuantcoreClient("http://127.0.0.1:9", timeout=1.0)) assert apicli.main(["selftest"]) == 1 assert "not usable" in capsys.readouterr().out def test_the_cli_exits_non_zero_when_the_api_is_unreachable(capsys): assert apicli.main(["--url", "http://127.0.0.1:9", "--timeout", "1", "health"]) == 1 assert "is the service running?" in capsys.readouterr().err