fix(ops): avoid truncating stockplatform runtime readback
Some checks failed
CD Pipeline / workflow-shape (push) Successful in 0s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Successful in 1m3s
CD Pipeline / post-deploy-checks (push) Has been cancelled
CD Pipeline / build-and-deploy (push) Has been cancelled
AWOOOI Harbor 110 Local Repair / workflow-shape (push) Successful in 1s
AWOOOI Harbor 110 Local Repair / harbor-110-local-repair (push) Successful in 26s

This commit is contained in:
Your Name
2026-07-02 19:11:10 +08:00
parent 9302933a95
commit a13038db87
2 changed files with 37 additions and 2 deletions

View File

@@ -7,6 +7,7 @@ from fastapi.testclient import TestClient
from src.api.v1 import agents
from src.api.v1.agents import router
from src.services import stockplatform_public_api_runtime_readback as runtime_readback
from src.services.stockplatform_public_api_runtime_readback import (
load_latest_stockplatform_public_api_runtime_readback,
)
@@ -235,6 +236,37 @@ def test_stockplatform_public_api_runtime_endpoint_returns_readback(monkeypatch)
assert data["active_blocker_count"] == 5
def test_http_probe_reads_full_large_json_body(monkeypatch):
body = json.dumps({"padding": "x" * 9000, "status": "ok"}).encode()
class Response:
status = 200
def __enter__(self):
return self
def __exit__(self, exc_type, exc, traceback):
return False
def read(self, size: int) -> bytes:
assert size >= len(body)
return body
monkeypatch.setattr(
runtime_readback.urllib.request,
"urlopen",
lambda request, timeout: Response(),
)
result = runtime_readback._http_probe(
"https://stock.wooo.work/api/v1/system/ingestion",
10,
)
assert result["http_status"] == 200
assert json.loads(result["body"])["status"] == "ok"
def _probe_public_web_ok_api_502(url: str, timeout_seconds: float) -> dict:
del timeout_seconds
if url.endswith("/healthz") and "/api/" not in url: