120 lines
3.9 KiB
Python
120 lines
3.9 KiB
Python
"""Smoke против реального шлюза Gravitee (тестовый контур).
|
||
|
||
Запуск только явно и только из сети банка:
|
||
DRRK_SMOKE_GATEWAY=1 python -m pytest tests/smoke/test_gateway_smoke.py -v
|
||
Ключи и адрес шлюза берутся из обычных настроек (OPENBAO__SETTINGS__* / .env).
|
||
Без DRRK_SMOKE_GATEWAY=1 или без ключей тесты пропускаются.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
from dataclasses import replace
|
||
from datetime import date, timedelta
|
||
|
||
import pytest
|
||
|
||
from app.core.config import settings
|
||
from app.integrations.cbr import CbrClient
|
||
from app.integrations.gateway import GatewayAuthError, GatewayNotFoundError
|
||
from app.integrations.moex import MoexClient
|
||
|
||
REO_02 = "RU000A106169"
|
||
|
||
pytestmark = pytest.mark.skipif(
|
||
os.getenv("DRRK_SMOKE_GATEWAY") != "1",
|
||
reason="smoke с реальным шлюзом: задайте DRRK_SMOKE_GATEWAY=1",
|
||
)
|
||
|
||
|
||
def _require(*keys: str) -> None:
|
||
missing = [k for k in keys if not getattr(settings, k).get_secret_value()]
|
||
if missing:
|
||
pytest.skip(f"не заданы {', '.join(missing)}")
|
||
|
||
|
||
def _recent_weekdays(count: int = 7) -> list[date]:
|
||
days: list[date] = []
|
||
day = date.today()
|
||
while len(days) < count:
|
||
day -= timedelta(days=1)
|
||
if day.weekday() < 5:
|
||
days.append(day)
|
||
return days
|
||
|
||
|
||
def test_moex_corporate_bond_chain() -> None:
|
||
_require("MOEX_API_KEY")
|
||
client = MoexClient.from_settings()
|
||
|
||
hit = client.find_by_isin(REO_02)
|
||
assert hit is not None and hit.secid == REO_02
|
||
|
||
info = client.get_security(hit.secid)
|
||
assert info is not None and info.isin == REO_02
|
||
|
||
# Бумага неликвидная: торгов за конкретный день может не быть — важно,
|
||
# что запрос проходит и разбирается.
|
||
for day in _recent_weekdays(3):
|
||
client.get_history(hit.secid, day)
|
||
|
||
|
||
def test_moex_ofz_chain_isin_to_secid_to_history() -> None:
|
||
"""У ОФЗ SECID != ISIN: проверяем, что ISIN -> SECID -> history работает."""
|
||
_require("MOEX_API_KEY")
|
||
client = MoexClient.from_settings()
|
||
|
||
ofz = next(
|
||
(
|
||
h
|
||
for h in client.search("SU26")
|
||
if h.secid.startswith("SU26") and h.is_traded and h.isin.startswith("RU")
|
||
),
|
||
None,
|
||
)
|
||
assert ofz is not None, "в поиске q=SU26 нет торгуемой ОФЗ"
|
||
assert ofz.secid != ofz.isin
|
||
|
||
hit = client.find_by_isin(ofz.isin)
|
||
assert hit is not None and hit.secid == ofz.secid
|
||
|
||
info = client.get_security(hit.secid)
|
||
assert info is not None and info.isin == ofz.isin
|
||
|
||
# ОФЗ торгуются каждый рабочий день — за последнюю неделю строка найдётся.
|
||
quote = next(
|
||
(q for d in _recent_weekdays() if (q := client.get_history(hit.secid, d))),
|
||
None,
|
||
)
|
||
assert quote is not None
|
||
assert quote.board == "TQOB"
|
||
assert quote.clean_price is not None
|
||
|
||
|
||
def test_cbr_cny_rates() -> None:
|
||
_require("CBR_API_KEY")
|
||
today = date.today()
|
||
|
||
rates = CbrClient.from_settings().get_rates(today - timedelta(days=14), today)
|
||
|
||
assert rates
|
||
assert all(r.nominal == 1 and r.rate_per_unit > 0 for r in rates)
|
||
|
||
|
||
def test_wrong_gateway_path_is_not_found() -> None:
|
||
_require("MOEX_API_KEY")
|
||
client = MoexClient.from_settings()
|
||
client.gateway = replace(client.gateway, path_prefix="/external/moex")
|
||
|
||
with pytest.raises(GatewayNotFoundError):
|
||
client.get_security(REO_02)
|
||
|
||
|
||
def test_cbr_key_is_rejected_by_moex() -> None:
|
||
_require("MOEX_API_KEY", "CBR_API_KEY")
|
||
client = MoexClient.from_settings()
|
||
client.gateway = replace(client.gateway, api_key=settings.CBR_API_KEY, retries=0)
|
||
|
||
with pytest.raises(GatewayAuthError):
|
||
client.get_security(REO_02)
|