gitadmin/dzentra_bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

build 059.4: validate websocket OHLC values

Browse Source
This commit is contained in:
Sergey
2026-07-16 23:40:46 +03:00
parent c8c0bb6951
commit 925b447d45
4 changed files with 1235 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

387
app/tests/unit/market_data/acquisition/validation/test_websocket_ohlc_values.py Normal file
View File

@@ -0,0 +1,387 @@
# app/tests/unit/market_data/acquisition/validation/test_websocket_ohlc_values.py
from __future__ import annotations
import pytest
from src.market_data.acquisition.adapters.dzengi.models import (
DzengiWebSocketOhlcEvent,
)
from src.market_data.acquisition.exceptions import (
CandleWebSocketValueError,
)
from src.market_data.acquisition.validation.values import (
validate_dzengi_websocket_ohlc_values,
)
def _event(
**overrides: object,
) -> DzengiWebSocketOhlcEvent:
values: dict[str, object] = {
"symbol": "BTC/USD_LEVERAGE",
"interval": "1m",
"candle_type": "classic",
"open_time": 1784224740000,
"open_price": "63992.00",
"high_price": "64032.55",
"low_price": "63984.00",
"close_price": "64032.55",
}
values.update(overrides)
return DzengiWebSocketOhlcEvent(
symbol=values["symbol"], # type: ignore[arg-type]
interval=values["interval"], # type: ignore[arg-type]
candle_type=values["candle_type"], # type: ignore[arg-type]
open_time=values["open_time"], # type: ignore[arg-type]
open_price=values["open_price"], # type: ignore[arg-type]
high_price=values["high_price"], # type: ignore[arg-type]
low_price=values["low_price"], # type: ignore[arg-type]
close_price=values["close_price"], # type: ignore[arg-type]
)
@pytest.mark.parametrize(
"interval",
[
"1m",
"5m",
"15m",
"1h",
],
)
def test_validate_websocket_ohlc_values_accepts_supported_intervals(
interval: str,
) -> None:
validate_dzengi_websocket_ohlc_values(
_event(interval=interval)
)
@pytest.mark.parametrize(
"candle_type",
[
"classic",
"heikin-ashi",
],
)
def test_validate_websocket_ohlc_values_accepts_supported_types(
candle_type: str,
) -> None:
validate_dzengi_websocket_ohlc_values(
_event(candle_type=candle_type)
)
@pytest.mark.parametrize(
("field_name", "value"),
[
("open_price", "63992.00"),
("open_price", 63992),
("open_price", 63992.0),
("high_price", "64032.55"),
("low_price", 63984),
("close_price", 64032.55),
],
)
def test_validate_websocket_ohlc_values_accepts_numeric_representations(
field_name: str,
value: object,
) -> None:
validate_dzengi_websocket_ohlc_values(
_event(**{field_name: value})
)
@pytest.mark.parametrize(
"symbol",
[
"",
" ",
"\t",
"\n",
],
)
def test_validate_websocket_ohlc_values_rejects_empty_symbol(
symbol: str,
) -> None:
with pytest.raises(
CandleWebSocketValueError,
match=r"\$\.payload\.symbol не должен быть пустым",
):
validate_dzengi_websocket_ohlc_values(
_event(symbol=symbol)
)
@pytest.mark.parametrize(
"interval",
[
"",
"1s",
"4h",
"1M",
" 1m ",
],
)
def test_validate_websocket_ohlc_values_rejects_unsupported_interval(
interval: str,
) -> None:
with pytest.raises(
CandleWebSocketValueError,
match=r"\$\.payload\.interval содержит неподдерживаемый интервал",
):
validate_dzengi_websocket_ohlc_values(
_event(interval=interval)
)
@pytest.mark.parametrize(
"candle_type",
[
"",
"bid",
"ask",
"Classic",
"unknown",
" classic ",
],
)
def test_validate_websocket_ohlc_values_rejects_unsupported_type(
candle_type: str,
) -> None:
with pytest.raises(
CandleWebSocketValueError,
match=r"\$\.payload\.type содержит неподдерживаемый тип свечи",
):
validate_dzengi_websocket_ohlc_values(
_event(candle_type=candle_type)
)
@pytest.mark.parametrize(
"open_time",
[
0,
-1,
-1784224740000,
],
)
def test_validate_websocket_ohlc_values_rejects_non_positive_open_time(
open_time: int,
) -> None:
with pytest.raises(
CandleWebSocketValueError,
match=r"\$\.payload\.t должно быть целым числом больше нуля",
):
validate_dzengi_websocket_ohlc_values(
_event(open_time=open_time)
)
@pytest.mark.parametrize(
"field_name",
[
"open_price",
"high_price",
"low_price",
"close_price",
],
)
@pytest.mark.parametrize(
"value",
[
0,
-1,
"-0.01",
],
)
def test_validate_websocket_ohlc_values_rejects_non_positive_prices(
field_name: str,
value: object,
) -> None:
path_by_field = {
"open_price": "o",
"high_price": "h",
"low_price": "l",
"close_price": "c",
}
with pytest.raises(
CandleWebSocketValueError,
match=(
rf"\$\.payload\.{path_by_field[field_name]} "
r"должно быть больше нуля"
),
):
validate_dzengi_websocket_ohlc_values(
_event(**{field_name: value})
)
@pytest.mark.parametrize(
"field_name",
[
"open_price",
"high_price",
"low_price",
"close_price",
],
)
@pytest.mark.parametrize(
"value",
[
"NaN",
"Infinity",
"-Infinity",
float("nan"),
float("inf"),
float("-inf"),
],
)
def test_validate_websocket_ohlc_values_rejects_non_finite_prices(
field_name: str,
value: object,
) -> None:
path_by_field = {
"open_price": "o",
"high_price": "h",
"low_price": "l",
"close_price": "c",
}
with pytest.raises(
CandleWebSocketValueError,
match=(
rf"\$\.payload\.{path_by_field[field_name]} "
r"должно быть конечным числом"
),
):
validate_dzengi_websocket_ohlc_values(
_event(**{field_name: value})
)
@pytest.mark.parametrize(
"field_name",
[
"open_price",
"high_price",
"low_price",
"close_price",
],
)
@pytest.mark.parametrize(
"value",
[
"",
"invalid",
"--1",
],
)
def test_validate_websocket_ohlc_values_rejects_invalid_numeric_strings(
field_name: str,
value: str,
) -> None:
path_by_field = {
"open_price": "o",
"high_price": "h",
"low_price": "l",
"close_price": "c",
}
with pytest.raises(
CandleWebSocketValueError,
match=(
rf"\$\.payload\.{path_by_field[field_name]} "
r"должно быть корректным числом"
),
):
validate_dzengi_websocket_ohlc_values(
_event(**{field_name: value})
)
def test_validate_websocket_ohlc_values_rejects_high_below_low() -> None:
with pytest.raises(
CandleWebSocketValueError,
match=r"\$\.payload\.h не должно быть меньше \$\.payload\.l",
):
validate_dzengi_websocket_ohlc_values(
_event(
high_price="63980",
low_price="63984",
)
)
def test_validate_websocket_ohlc_values_rejects_high_below_open() -> None:
with pytest.raises(
CandleWebSocketValueError,
match=r"\$\.payload\.h не должно быть меньше \$\.payload\.o",
):
validate_dzengi_websocket_ohlc_values(
_event(
open_price="64000",
high_price="63999",
low_price="63980",
close_price="63990",
)
)
def test_validate_websocket_ohlc_values_rejects_high_below_close() -> None:
with pytest.raises(
CandleWebSocketValueError,
match=r"\$\.payload\.h не должно быть меньше \$\.payload\.c",
):
validate_dzengi_websocket_ohlc_values(
_event(
open_price="63990",
high_price="64000",
low_price="63980",
close_price="64001",
)
)
def test_validate_websocket_ohlc_values_rejects_low_above_open() -> None:
with pytest.raises(
CandleWebSocketValueError,
match=r"\$\.payload\.l не должно превышать \$\.payload\.o",
):
validate_dzengi_websocket_ohlc_values(
_event(
open_price="63990",
high_price="64020",
low_price="64000",
close_price="64010",
)
)
def test_validate_websocket_ohlc_values_rejects_low_above_close() -> None:
with pytest.raises(
CandleWebSocketValueError,
match=r"\$\.payload\.l не должно превышать \$\.payload\.c",
):
validate_dzengi_websocket_ohlc_values(
_event(
open_price="64010",
high_price="64020",
low_price="64000",
close_price="63999",
)
)
def test_validate_websocket_ohlc_values_accepts_flat_candle() -> None:
validate_dzengi_websocket_ohlc_values(
_event(
open_price="64000",
high_price="64000",
low_price="64000",
close_price="64000",
)
)