66 lines
2.3 KiB
Python
66 lines
2.3 KiB
Python
from __future__ import annotations
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from broker_sync.metrics import push_pushgateway
|
|
|
|
|
|
async def test_push_pushgateway_posts_text_format() -> None:
|
|
captured: dict[str, str] = {}
|
|
|
|
def transport_handler(request: httpx.Request) -> httpx.Response:
|
|
captured["url"] = str(request.url)
|
|
captured["method"] = request.method
|
|
captured["body"] = request.content.decode("utf-8")
|
|
return httpx.Response(200)
|
|
|
|
transport = httpx.MockTransport(transport_handler)
|
|
await push_pushgateway(
|
|
job="broker-sync-ibkr",
|
|
metrics=[
|
|
("ibkr_position_drift_shares", {"symbol": "VUAG.L"}, 0.0),
|
|
("ibkr_sync_last_success_timestamp_seconds", {}, 1779830000.0),
|
|
],
|
|
pushgateway_url="http://pg.example/metrics",
|
|
transport=transport,
|
|
)
|
|
assert captured["method"] == "POST"
|
|
assert captured["url"] == "http://pg.example/metrics/job/broker-sync-ibkr"
|
|
body = captured["body"]
|
|
assert 'ibkr_position_drift_shares{symbol="VUAG.L"} 0.0' in body
|
|
assert "ibkr_sync_last_success_timestamp_seconds 1779830000.0" in body
|
|
|
|
|
|
async def test_push_pushgateway_raises_on_non_2xx() -> None:
|
|
transport = httpx.MockTransport(lambda r: httpx.Response(500, text="boom"))
|
|
with pytest.raises(RuntimeError, match="pushgateway.*500"):
|
|
await push_pushgateway(
|
|
job="x",
|
|
metrics=[("m", {}, 1.0)],
|
|
pushgateway_url="http://pg/metrics",
|
|
transport=transport,
|
|
)
|
|
|
|
|
|
async def test_push_pushgateway_uses_env_var(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
captured: dict[str, str] = {}
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
captured["url"] = str(request.url)
|
|
return httpx.Response(200)
|
|
|
|
transport = httpx.MockTransport(handler)
|
|
monkeypatch.setenv("PUSHGATEWAY_URL", "http://from-env/metrics")
|
|
await push_pushgateway(
|
|
job="j",
|
|
metrics=[("m", {}, 1.0)],
|
|
transport=transport,
|
|
)
|
|
assert captured["url"] == "http://from-env/metrics/job/j"
|
|
|
|
|
|
async def test_push_pushgateway_raises_when_url_missing(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.delenv("PUSHGATEWAY_URL", raising=False)
|
|
with pytest.raises(RuntimeError, match="PUSHGATEWAY_URL not set"):
|
|
await push_pushgateway(job="j", metrics=[("m", {}, 1.0)])
|