Initial import: selectel-ml-tui TUI for Selectel AI-router token usage
This commit is contained in:
@@ -0,0 +1,296 @@
|
||||
"""Unit-тесты API-клиентов с моками httpx.MockTransport."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from decimal import Decimal
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
from selectel_ml_tui.api.billing import (
|
||||
BillingApiError,
|
||||
BillingClient,
|
||||
normalize_dt,
|
||||
)
|
||||
from selectel_ml_tui.api.client import SelectelApiError
|
||||
from selectel_ml_tui.api.gateway import GatewayApiError, GatewayClient
|
||||
|
||||
|
||||
class NoopSleep:
|
||||
"""Пауза-заглушка для ретраев (не ждём в тестах)."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.calls: list[float] = []
|
||||
|
||||
async def __call__(self, seconds: float) -> None:
|
||||
self.calls.append(seconds)
|
||||
|
||||
|
||||
def _transport(handler: Any) -> httpx.MockTransport:
|
||||
return httpx.MockTransport(handler)
|
||||
|
||||
|
||||
def _billing_client(
|
||||
handler: Any, max_retries: int = 2, sleep: NoopSleep | None = None
|
||||
) -> BillingClient:
|
||||
return BillingClient(
|
||||
api_base_url="https://api.selectel.ru",
|
||||
static_token="static-token",
|
||||
max_retries=max_retries,
|
||||
transport=_transport(handler),
|
||||
sleep=sleep or NoopSleep(),
|
||||
)
|
||||
|
||||
|
||||
def _ok_consumption() -> dict[str, Any]:
|
||||
return {
|
||||
"status": "success",
|
||||
"data": [
|
||||
{
|
||||
"account_id": "239633",
|
||||
"period": "2026-08-01",
|
||||
"provider_key": "aig",
|
||||
"metric": {
|
||||
"id": "air_tokens_cost",
|
||||
"name": "Стоимость токенов",
|
||||
"unit": "nanorub",
|
||||
"quantity": "85483000000",
|
||||
},
|
||||
"value": 854.83,
|
||||
"project": "7bf2d2a328494e25b13d05a1926ca75a",
|
||||
}
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
async def test_billing_normalize_dt() -> None:
|
||||
assert normalize_dt("2026-08-01") == "2026-08-01T00:00:00.000"
|
||||
assert normalize_dt("2026-08-01T10:30:00") == "2026-08-01T10:30:00"
|
||||
assert normalize_dt("2026-08-01T00:00:00.000") == "2026-08-01T00:00:00.000"
|
||||
|
||||
|
||||
async def test_billing_gets_normalized_dates() -> None:
|
||||
seen: dict[str, str] = {}
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
seen.update(dict(request.url.params))
|
||||
return httpx.Response(200, json=_ok_consumption())
|
||||
|
||||
client = _billing_client(handler)
|
||||
await client.get_consumption("2026-08-01", "2026-08-19")
|
||||
assert seen["start"] == "2026-08-01T00:00:00.000"
|
||||
assert seen["end"] == "2026-08-19T00:00:00.000"
|
||||
await client.aclose()
|
||||
|
||||
|
||||
async def test_billing_get_consumption() -> None:
|
||||
client = _billing_client(
|
||||
lambda request: httpx.Response(200, json=_ok_consumption())
|
||||
)
|
||||
items = await client.get_consumption("2026-08-01", "2026-08-19")
|
||||
assert len(items) == 1
|
||||
assert items[0].value == Decimal("854.83")
|
||||
assert items[0].metric is not None
|
||||
assert items[0].metric.id == "air_tokens_cost"
|
||||
assert items[0].project_id == "7bf2d2a328494e25b13d05a1926ca75a"
|
||||
await client.aclose()
|
||||
|
||||
|
||||
async def test_billing_get_summary_sums_values() -> None:
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
assert "provider_keys=aig" in request.url.query.decode()
|
||||
return httpx.Response(
|
||||
200,
|
||||
json={
|
||||
"status": "success",
|
||||
"data": [
|
||||
{"value": "100.25"},
|
||||
{"value": 42.5},
|
||||
{"value": 0},
|
||||
],
|
||||
},
|
||||
)
|
||||
|
||||
client = _billing_client(handler)
|
||||
total = await client.get_summary("2026-08-01", "2026-08-19")
|
||||
assert total == Decimal("142.75")
|
||||
await client.aclose()
|
||||
|
||||
|
||||
async def test_billing_5xx_retries_then_success() -> None:
|
||||
sleep = NoopSleep()
|
||||
attempts = {"count": 0}
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
attempts["count"] += 1
|
||||
if attempts["count"] == 1:
|
||||
return httpx.Response(500, text="internal error")
|
||||
return httpx.Response(200, json=_ok_consumption())
|
||||
|
||||
client = _billing_client(handler, sleep=sleep)
|
||||
items = await client.get_consumption("2026-08-01", "2026-08-19")
|
||||
assert attempts["count"] == 2
|
||||
assert len(items) == 1
|
||||
assert sleep.calls == [0.5]
|
||||
await client.aclose()
|
||||
|
||||
|
||||
async def test_billing_429_retries_then_raises() -> None:
|
||||
sleep = NoopSleep()
|
||||
attempts = {"count": 0}
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
attempts["count"] += 1
|
||||
return httpx.Response(429, text="rate limit")
|
||||
|
||||
client = _billing_client(handler, max_retries=2, sleep=sleep)
|
||||
with pytest.raises(SelectelApiError) as exc_info:
|
||||
await client.get_consumption("2026-08-01", "2026-08-19")
|
||||
assert exc_info.value.status_code == 429
|
||||
assert attempts["count"] == 3
|
||||
assert sleep.calls == [0.5, 1.0]
|
||||
await client.aclose()
|
||||
|
||||
|
||||
async def test_billing_4xx_no_retry() -> None:
|
||||
sleep = NoopSleep()
|
||||
attempts = {"count": 0}
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
attempts["count"] += 1
|
||||
return httpx.Response(400, text="bad request")
|
||||
|
||||
client = _billing_client(handler, sleep=sleep)
|
||||
with pytest.raises(BillingApiError) as exc_info:
|
||||
await client.get_consumption("2026-08-01", "2026-08-19")
|
||||
assert exc_info.value.status_code == 400
|
||||
assert "HTTP 400" in str(exc_info.value)
|
||||
assert attempts["count"] == 1
|
||||
assert sleep.calls == []
|
||||
await client.aclose()
|
||||
|
||||
|
||||
async def test_billing_network_error_retries() -> None:
|
||||
sleep = NoopSleep()
|
||||
attempts = {"count": 0}
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
attempts["count"] += 1
|
||||
if attempts["count"] == 1:
|
||||
raise httpx.ConnectError("connection refused")
|
||||
return httpx.Response(200, json=_ok_consumption())
|
||||
|
||||
client = _billing_client(handler, sleep=sleep)
|
||||
items = await client.get_consumption("2026-08-01", "2026-08-19")
|
||||
assert attempts["count"] == 2
|
||||
assert len(items) == 1
|
||||
await client.aclose()
|
||||
|
||||
|
||||
async def test_billing_timeout_retries_then_raises() -> None:
|
||||
sleep = NoopSleep()
|
||||
attempts = {"count": 0}
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
attempts["count"] += 1
|
||||
raise httpx.ReadTimeout("timed out")
|
||||
|
||||
client = _billing_client(handler, max_retries=2, sleep=sleep)
|
||||
with pytest.raises(SelectelApiError) as exc_info:
|
||||
await client.get_consumption("2026-08-01", "2026-08-19")
|
||||
assert exc_info.value.status_code is None
|
||||
assert "Ошибка сети" in str(exc_info.value)
|
||||
assert attempts["count"] == 3
|
||||
assert sleep.calls == [0.5, 1.0]
|
||||
await client.aclose()
|
||||
|
||||
|
||||
async def test_billing_non_success_status_raises() -> None:
|
||||
client = _billing_client(
|
||||
lambda request: httpx.Response(200, json={"status": "error", "data": []})
|
||||
)
|
||||
with pytest.raises(BillingApiError):
|
||||
await client.get_consumption("2026-08-01", "2026-08-19")
|
||||
await client.aclose()
|
||||
|
||||
|
||||
class FakeTokenProvider:
|
||||
"""Токен-провайдер для GatewayClient в тестах."""
|
||||
|
||||
def __init__(self, tokens: list[str] | None = None) -> None:
|
||||
self._tokens = tokens or ["token-1"]
|
||||
self._index = 0
|
||||
self.calls = 0
|
||||
|
||||
async def get_token(self) -> str:
|
||||
self.calls += 1
|
||||
token = self._tokens[min(self._index, len(self._tokens) - 1)]
|
||||
self._index += 1
|
||||
return token
|
||||
|
||||
def invalidate(self) -> None:
|
||||
pass
|
||||
|
||||
|
||||
async def test_gateway_list_gateways() -> None:
|
||||
provider = FakeTokenProvider()
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
assert request.headers["X-Auth-Token"] == "token-1"
|
||||
assert request.headers["X-Project-Id"] == "proj-1"
|
||||
assert request.headers["X-Roles"] == "reader"
|
||||
return httpx.Response(200, json={"data": [{"id": "gw-1", "name": "Router"}]})
|
||||
|
||||
client = GatewayClient(
|
||||
api_base_url="https://api.selectel.ru",
|
||||
token_provider=provider,
|
||||
project_id="proj-1",
|
||||
domain_id="239633",
|
||||
transport=_transport(handler),
|
||||
)
|
||||
gateways = await client.list_gateways()
|
||||
assert gateways == [{"id": "gw-1", "name": "Router"}]
|
||||
assert provider.calls == 1
|
||||
await client.aclose()
|
||||
|
||||
|
||||
async def test_gateway_401_refreshes_token() -> None:
|
||||
provider = FakeTokenProvider(tokens=["token-old", "token-new"])
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
if request.headers["X-Auth-Token"] == "token-old":
|
||||
return httpx.Response(401, text="unauthorized")
|
||||
return httpx.Response(200, json={"data": [{"id": "gw-2"}]})
|
||||
|
||||
client = GatewayClient(
|
||||
api_base_url="https://api.selectel.ru",
|
||||
token_provider=provider,
|
||||
project_id="proj-1",
|
||||
domain_id="239633",
|
||||
transport=_transport(handler),
|
||||
)
|
||||
gateways = await client.list_gateways()
|
||||
assert gateways == [{"id": "gw-2"}]
|
||||
assert provider.calls == 2
|
||||
await client.aclose()
|
||||
|
||||
|
||||
async def test_gateway_401_without_refresh_raises() -> None:
|
||||
provider = FakeTokenProvider(tokens=["token-only"])
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
return httpx.Response(401, text="unauthorized")
|
||||
|
||||
client = GatewayClient(
|
||||
api_base_url="https://api.selectel.ru",
|
||||
token_provider=provider,
|
||||
project_id="proj-1",
|
||||
domain_id="239633",
|
||||
max_retries=0,
|
||||
transport=_transport(handler),
|
||||
)
|
||||
with pytest.raises(GatewayApiError) as exc_info:
|
||||
await client.list_gateways()
|
||||
assert exc_info.value.status_code == 401
|
||||
await client.aclose()
|
||||
@@ -0,0 +1,119 @@
|
||||
"""Unit-тесты IAM-авторизации и кэша токена."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import datetime as dt
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
from selectel_ml_tui.auth import (
|
||||
IamAuthError,
|
||||
IamClient,
|
||||
IamToken,
|
||||
IamTokenManager,
|
||||
)
|
||||
from selectel_ml_tui.config import Credentials, Settings
|
||||
|
||||
|
||||
class FakeIamClient:
|
||||
"""Заглушка IamClient для проверки логики кэширования."""
|
||||
|
||||
def __init__(self, token: IamToken) -> None:
|
||||
self._token = token
|
||||
self.calls = 0
|
||||
|
||||
async def get_token(self) -> IamToken:
|
||||
self.calls += 1
|
||||
return self._token
|
||||
|
||||
|
||||
def _far_future() -> dt.datetime:
|
||||
return dt.datetime.now(dt.timezone.utc) + dt.timedelta(hours=23)
|
||||
|
||||
|
||||
def _expired() -> dt.datetime:
|
||||
return dt.datetime.now(dt.timezone.utc) - dt.timedelta(minutes=1)
|
||||
|
||||
|
||||
async def test_iam_get_token_success() -> None:
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
body = request.content.decode()
|
||||
assert '"name":"svc-user"' in body
|
||||
assert '"password":"secret"' in body
|
||||
assert '"id":"proj-1"' in body
|
||||
return httpx.Response(
|
||||
201,
|
||||
headers={"X-Subject-Token": "iam-token-abc"},
|
||||
json={
|
||||
"token": {
|
||||
"expires_at": "2026-08-20T10:00:00.000000Z",
|
||||
}
|
||||
},
|
||||
)
|
||||
|
||||
client = IamClient(
|
||||
Settings(),
|
||||
Credentials(
|
||||
service_user="svc-user",
|
||||
password="secret",
|
||||
domain_id="239633",
|
||||
project_id="proj-1",
|
||||
),
|
||||
transport=httpx.MockTransport(handler),
|
||||
)
|
||||
token = await client.get_token()
|
||||
assert token.value == "iam-token-abc"
|
||||
assert token.expires_at == dt.datetime(2026, 8, 20, 10, 0, tzinfo=dt.timezone.utc)
|
||||
|
||||
|
||||
async def test_iam_get_token_missing_credentials() -> None:
|
||||
client = IamClient(Settings(), Credentials())
|
||||
with pytest.raises(IamAuthError, match="логин/пароль"):
|
||||
await client.get_token()
|
||||
|
||||
|
||||
async def test_iam_get_token_missing_subject_token() -> None:
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
return httpx.Response(
|
||||
201, json={"token": {"expires_at": "2026-08-20T10:00:00Z"}}
|
||||
)
|
||||
|
||||
client = IamClient(
|
||||
Settings(),
|
||||
Credentials(
|
||||
service_user="svc", password="pwd", domain_id="239633", project_id="p"
|
||||
),
|
||||
transport=httpx.MockTransport(handler),
|
||||
)
|
||||
with pytest.raises(IamAuthError, match="X-Subject-Token"):
|
||||
await client.get_token()
|
||||
|
||||
|
||||
async def test_iam_token_manager_caches() -> None:
|
||||
fake = FakeIamClient(IamToken(value="t1", expires_at=_far_future()))
|
||||
manager = IamTokenManager(fake) # type: ignore[arg-type]
|
||||
assert await manager.get_token() == "t1"
|
||||
assert await manager.get_token() == "t1"
|
||||
assert fake.calls == 1
|
||||
assert manager.is_valid()
|
||||
|
||||
|
||||
async def test_iam_token_manager_refreshes_after_expiry() -> None:
|
||||
fake = FakeIamClient(IamToken(value="t-expired", expires_at=_expired()))
|
||||
manager = IamTokenManager(fake) # type: ignore[arg-type]
|
||||
assert await manager.get_token() == "t-expired"
|
||||
assert not manager.is_valid()
|
||||
assert await manager.get_token() == "t-expired"
|
||||
assert fake.calls == 2
|
||||
|
||||
|
||||
async def test_iam_token_manager_invalidate() -> None:
|
||||
fake = FakeIamClient(IamToken(value="t1", expires_at=_far_future()))
|
||||
manager = IamTokenManager(fake) # type: ignore[arg-type]
|
||||
await manager.get_token()
|
||||
assert manager.is_valid()
|
||||
manager.invalidate()
|
||||
assert not manager.is_valid()
|
||||
await manager.get_token()
|
||||
assert fake.calls == 2
|
||||
@@ -0,0 +1,120 @@
|
||||
"""Базовые тесты CLI."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
from click.testing import CliRunner
|
||||
|
||||
from selectel_ml_tui import __version__
|
||||
from selectel_ml_tui.cli import main
|
||||
from selectel_ml_tui.config import load_credentials, load_settings
|
||||
|
||||
|
||||
def test_version() -> None:
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(main, ["--version"])
|
||||
assert result.exit_code == 0
|
||||
assert __version__ in result.output
|
||||
|
||||
|
||||
def test_main_module_help() -> None:
|
||||
result = subprocess.run(
|
||||
[sys.executable, "-m", "selectel_ml_tui", "--help"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
assert result.returncode == 0
|
||||
assert "doctor" in result.stdout
|
||||
assert "run" in result.stdout
|
||||
|
||||
|
||||
def test_config_paths(tmp_path, monkeypatch) -> None:
|
||||
monkeypatch.setattr("selectel_ml_tui.config.config_dir", lambda: tmp_path)
|
||||
monkeypatch.setattr("selectel_ml_tui.cli.config_dir", lambda: tmp_path)
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(main, ["config", "--path"])
|
||||
assert result.exit_code == 0
|
||||
assert str(tmp_path) in result.output
|
||||
|
||||
|
||||
def test_config_show_masks_secrets(tmp_path, monkeypatch) -> None:
|
||||
monkeypatch.setattr("selectel_ml_tui.config.config_dir", lambda: tmp_path)
|
||||
monkeypatch.setattr("selectel_ml_tui.cli.config_dir", lambda: tmp_path)
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(main, ["config", "--show"])
|
||||
assert result.exit_code == 0
|
||||
assert "Static token" in result.output
|
||||
assert "секрет" not in result.output.lower()
|
||||
|
||||
|
||||
def test_config_wizard_saves(tmp_path, monkeypatch) -> None:
|
||||
monkeypatch.setattr("selectel_ml_tui.config.config_dir", lambda: tmp_path)
|
||||
monkeypatch.setattr("selectel_ml_tui.cli.config_dir", lambda: tmp_path)
|
||||
input_data = "\n".join(
|
||||
[
|
||||
"https://api.example.test", # API base URL
|
||||
"30", # интервал обновления
|
||||
"gw-test", # ID роутера
|
||||
"ru-3", # регион
|
||||
"svc-user", # сервисный пользователь
|
||||
"239633", # домен (аккаунт)
|
||||
"7bf2d2a328494e25b13d05a1926ca75a", # project id
|
||||
"secret-pass", # пароль
|
||||
"secret-token", # статический токен
|
||||
]
|
||||
)
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(main, ["config"], input=input_data)
|
||||
assert result.exit_code == 0, result.output
|
||||
assert "Сохранено" in result.output
|
||||
|
||||
settings = load_settings()
|
||||
assert settings.router_id == "gw-test"
|
||||
assert settings.refresh_interval == 30
|
||||
assert settings.api_base_url == "https://api.example.test"
|
||||
assert settings.region == "ru-3"
|
||||
|
||||
creds = load_credentials()
|
||||
assert creds.service_user == "svc-user"
|
||||
assert creds.domain_id == "239633"
|
||||
assert creds.project_id == "7bf2d2a328494e25b13d05a1926ca75a"
|
||||
assert creds.password == "secret-pass"
|
||||
assert creds.static_token == "secret-token"
|
||||
|
||||
|
||||
def test_run_requires_credentials(tmp_path, monkeypatch) -> None:
|
||||
monkeypatch.setattr("selectel_ml_tui.config.config_dir", lambda: tmp_path)
|
||||
monkeypatch.setattr("selectel_ml_tui.cli.config_dir", lambda: tmp_path)
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(main, ["run"])
|
||||
assert result.exit_code == 1
|
||||
assert "не настроены" in result.output
|
||||
|
||||
|
||||
def test_config_wizard_rejects_small_interval(tmp_path, monkeypatch) -> None:
|
||||
monkeypatch.setattr("selectel_ml_tui.config.config_dir", lambda: tmp_path)
|
||||
monkeypatch.setattr("selectel_ml_tui.cli.config_dir", lambda: tmp_path)
|
||||
input_data = "\n".join(
|
||||
[
|
||||
"https://api.example.test", # API base URL
|
||||
"5", # интервал < 10 — будет отклонён
|
||||
"30", # повторный ввод, валидный
|
||||
"gw-test", # ID роутера
|
||||
"", # регион (оставить)
|
||||
"", # сервисный пользователь (оставить)
|
||||
"", # домен (оставить)
|
||||
"", # project id (оставить)
|
||||
"", # пароль (оставить)
|
||||
"", # статический токен (оставить)
|
||||
"", # хвостовая строка (CliRunner отбрасывает последний пустой элемент)
|
||||
]
|
||||
)
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(main, ["config"], input=input_data)
|
||||
assert result.exit_code == 0, result.output
|
||||
assert "не может быть меньше 10" in result.output
|
||||
settings = load_settings()
|
||||
assert settings.refresh_interval == 30
|
||||
@@ -0,0 +1,89 @@
|
||||
"""Тесты модуля конфигурации."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from selectel_ml_tui.config import (
|
||||
Credentials,
|
||||
Settings,
|
||||
load_credentials,
|
||||
load_settings,
|
||||
save_credentials,
|
||||
save_settings,
|
||||
)
|
||||
|
||||
|
||||
def test_settings_defaults() -> None:
|
||||
settings = load_settings()
|
||||
assert settings.api_base_url == "https://api.selectel.ru"
|
||||
assert settings.refresh_interval >= 10
|
||||
|
||||
|
||||
def test_save_settings_roundtrip(tmp_path, monkeypatch) -> None:
|
||||
cfg_file = tmp_path / "config.toml"
|
||||
monkeypatch.setattr("selectel_ml_tui.config._config_file", lambda: cfg_file)
|
||||
settings = Settings(router_id="gw-test", refresh_interval=30)
|
||||
save_settings(settings)
|
||||
loaded = load_settings()
|
||||
assert loaded.router_id == "gw-test"
|
||||
assert loaded.refresh_interval == 30
|
||||
|
||||
|
||||
def test_load_settings_repairs_small_interval(tmp_path, monkeypatch) -> None:
|
||||
cfg_file = tmp_path / "config.toml"
|
||||
cfg_file.write_text("refresh_interval = 5\n", encoding="utf-8")
|
||||
monkeypatch.setattr("selectel_ml_tui.config._config_file", lambda: cfg_file)
|
||||
settings = load_settings()
|
||||
assert settings.refresh_interval == 10
|
||||
|
||||
|
||||
def test_load_settings_repairs_garbage_interval(tmp_path, monkeypatch) -> None:
|
||||
cfg_file = tmp_path / "config.toml"
|
||||
cfg_file.write_text('refresh_interval = "abc"\n', encoding="utf-8")
|
||||
monkeypatch.setattr("selectel_ml_tui.config._config_file", lambda: cfg_file)
|
||||
settings = load_settings()
|
||||
assert settings.refresh_interval == 60
|
||||
|
||||
|
||||
def test_load_settings_ignores_bad_env_interval(tmp_path, monkeypatch) -> None:
|
||||
cfg_file = tmp_path / "config.toml"
|
||||
monkeypatch.setattr("selectel_ml_tui.config._config_file", lambda: cfg_file)
|
||||
monkeypatch.setenv("SELCTEL_REFRESH_INTERVAL", "5")
|
||||
settings = load_settings()
|
||||
assert settings.refresh_interval == 60
|
||||
|
||||
|
||||
def test_credentials_defaults() -> None:
|
||||
creds = Credentials()
|
||||
assert creds.service_user == ""
|
||||
assert creds.static_token == ""
|
||||
|
||||
|
||||
def test_settings_env_api_url(monkeypatch, tmp_path) -> None:
|
||||
cfg_file = tmp_path / "config.toml"
|
||||
monkeypatch.setattr("selectel_ml_tui.config._config_file", lambda: cfg_file)
|
||||
monkeypatch.setenv("SELCTEL_API_URL", "https://api.example.test")
|
||||
settings = load_settings()
|
||||
assert settings.api_base_url == "https://api.example.test"
|
||||
|
||||
|
||||
def test_credentials_env(monkeypatch, tmp_path) -> None:
|
||||
cred_file = tmp_path / "credentials.json"
|
||||
monkeypatch.setattr("selectel_ml_tui.config._credentials_file", lambda: cred_file)
|
||||
monkeypatch.setenv("SELCTEL_TOKEN", "tok123")
|
||||
monkeypatch.setenv("SELCTEL_SERVICE_USER", "svc-user")
|
||||
monkeypatch.setenv("SELCTEL_DOMAIN_ID", "239633")
|
||||
creds = load_credentials()
|
||||
assert creds.static_token == "tok123"
|
||||
assert creds.service_user == "svc-user"
|
||||
assert creds.domain_id == "239633"
|
||||
|
||||
|
||||
def test_credentials_file_precedence(monkeypatch, tmp_path) -> None:
|
||||
cred_file = tmp_path / "credentials.json"
|
||||
monkeypatch.setattr("selectel_ml_tui.config._credentials_file", lambda: cred_file)
|
||||
save_credentials(Credentials(static_token="from-file", service_user="file-user"))
|
||||
monkeypatch.setenv("SELCTEL_TOKEN", "from-env")
|
||||
creds = load_credentials()
|
||||
# env имеет приоритет над файлом
|
||||
assert creds.static_token == "from-env"
|
||||
assert creds.service_user == "file-user"
|
||||
@@ -0,0 +1,215 @@
|
||||
"""Unit-тесты pydantic-моделей данных."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timezone
|
||||
from decimal import Decimal
|
||||
|
||||
import pytest
|
||||
from pydantic import ValidationError
|
||||
|
||||
from selectel_ml_tui.models import (
|
||||
Consumption,
|
||||
ConsumptionItem,
|
||||
HistoryPoint,
|
||||
Metrics,
|
||||
ModelMetrics,
|
||||
ProjectRef,
|
||||
format_percent,
|
||||
format_rub,
|
||||
format_tokens,
|
||||
)
|
||||
|
||||
|
||||
class TestConsumption:
|
||||
def test_parse_with_z_timestamp(self) -> None:
|
||||
item = Consumption.model_validate(
|
||||
{
|
||||
"period": "2026-08-01T00:00:00.000Z",
|
||||
"amount_rub": "854.83",
|
||||
"input_tokens": 100,
|
||||
"output_tokens": 50,
|
||||
"cache_read_tokens": 10,
|
||||
"cache_write_tokens": 5,
|
||||
}
|
||||
)
|
||||
assert item.period == datetime(2026, 8, 1, tzinfo=timezone.utc)
|
||||
assert item.amount_rub == Decimal("854.83")
|
||||
assert item.input_tokens == 100
|
||||
|
||||
def test_parse_naive_timestamp(self) -> None:
|
||||
item = Consumption.model_validate({"period": "2026-08-01T00:00:00"})
|
||||
assert item.period == datetime(2026, 8, 1)
|
||||
assert item.amount_rub == Decimal("0")
|
||||
assert item.input_tokens == 0
|
||||
|
||||
def test_negative_amount_rejected(self) -> None:
|
||||
with pytest.raises(ValidationError):
|
||||
Consumption.model_validate(
|
||||
{"period": "2026-08-01T00:00:00", "amount_rub": "-1"}
|
||||
)
|
||||
|
||||
def test_negative_tokens_rejected(self) -> None:
|
||||
with pytest.raises(ValidationError):
|
||||
Consumption.model_validate(
|
||||
{
|
||||
"period": "2026-08-01T00:00:00",
|
||||
"input_tokens": -5,
|
||||
}
|
||||
)
|
||||
|
||||
def test_invalid_date_rejected(self) -> None:
|
||||
with pytest.raises(ValidationError):
|
||||
Consumption.model_validate({"period": "not-a-date"})
|
||||
|
||||
|
||||
class TestMetrics:
|
||||
def test_parse_metrics(self) -> None:
|
||||
metrics = Metrics.model_validate(
|
||||
{
|
||||
"availability_pct": 99.912,
|
||||
"avg_ttft_ms": 0.45,
|
||||
"input_tokens": 1000,
|
||||
"output_tokens": 900,
|
||||
"cache_read_tokens": 800,
|
||||
"cache_write_tokens": 700,
|
||||
}
|
||||
)
|
||||
assert metrics.availability_pct == 99.912
|
||||
assert metrics.avg_ttft_ms == 0.45
|
||||
assert metrics.input_tokens == 1000
|
||||
|
||||
def test_missing_fields_defaults(self) -> None:
|
||||
metrics = Metrics.model_validate({})
|
||||
assert metrics.availability_pct == 100.0
|
||||
assert metrics.input_tokens == 0
|
||||
assert metrics.avg_ttft_ms == 0.0
|
||||
|
||||
def test_availability_bounds_rejected(self) -> None:
|
||||
with pytest.raises(ValidationError):
|
||||
Metrics.model_validate({"availability_pct": 101})
|
||||
with pytest.raises(ValidationError):
|
||||
Metrics.model_validate({"availability_pct": -1})
|
||||
|
||||
|
||||
class TestModelMetrics:
|
||||
def test_parse_model_metrics(self) -> None:
|
||||
model = ModelMetrics.model_validate(
|
||||
{
|
||||
"model_id": "gpt-4o",
|
||||
"model_name": "GPT-4o",
|
||||
"availability_pct": 99.5,
|
||||
"avg_ttft_ms": 0.3,
|
||||
}
|
||||
)
|
||||
assert model.model_id == "gpt-4o"
|
||||
assert model.model_name == "GPT-4o"
|
||||
assert model.availability_pct == 99.5
|
||||
|
||||
|
||||
class TestHistoryPoint:
|
||||
def test_parse(self) -> None:
|
||||
point = HistoryPoint.model_validate(
|
||||
{
|
||||
"timestamp": "2026-08-01T12:00:00Z",
|
||||
"amount_rub": "12.5",
|
||||
"tokens": {
|
||||
"input_tokens": 100,
|
||||
"output_tokens": 200,
|
||||
"cache_read_tokens": 30,
|
||||
"cache_write_tokens": 40,
|
||||
},
|
||||
}
|
||||
)
|
||||
assert point.timestamp == datetime(2026, 8, 1, 12, 0, tzinfo=timezone.utc)
|
||||
assert point.amount_rub == Decimal("12.5")
|
||||
assert point.tokens.output_tokens == 200
|
||||
|
||||
def test_missing_tokens_defaults(self) -> None:
|
||||
point = HistoryPoint.model_validate({"timestamp": "2026-08-01T00:00:00"})
|
||||
assert point.tokens.input_tokens == 0
|
||||
assert point.amount_rub == Decimal("0")
|
||||
|
||||
|
||||
class TestConsumptionItem:
|
||||
def _base(self, **overrides: object) -> dict[str, object]:
|
||||
data: dict[str, object] = {
|
||||
"account_id": "239633",
|
||||
"period": "2026-08-01T00:00:00",
|
||||
"provider_key": "aig",
|
||||
"value": 854.83,
|
||||
}
|
||||
data.update(overrides)
|
||||
return data
|
||||
|
||||
def test_project_as_string(self) -> None:
|
||||
item = ConsumptionItem.model_validate(
|
||||
self._base(project="7bf2d2a328494e25b13d05a1926ca75a")
|
||||
)
|
||||
assert item.project_id == "7bf2d2a328494e25b13d05a1926ca75a"
|
||||
|
||||
def test_project_as_object(self) -> None:
|
||||
item = ConsumptionItem.model_validate(
|
||||
self._base(
|
||||
project={"id": "proj-1", "name": "239633"},
|
||||
)
|
||||
)
|
||||
assert isinstance(item.project, ProjectRef)
|
||||
assert item.project_id == "proj-1"
|
||||
assert item.project.name == "239633"
|
||||
|
||||
def test_project_null(self) -> None:
|
||||
item = ConsumptionItem.model_validate(self._base(project=None))
|
||||
assert item.project_id is None
|
||||
|
||||
def test_object_as_object(self) -> None:
|
||||
item = ConsumptionItem.model_validate(
|
||||
self._base(
|
||||
object={
|
||||
"id": "obj-1",
|
||||
"name": "key-1",
|
||||
"type": "ai_api_key",
|
||||
"type_name": "Api ключ",
|
||||
}
|
||||
)
|
||||
)
|
||||
assert item.object_id == "obj-1"
|
||||
assert item.object.type == "ai_api_key" # type: ignore[union-attr]
|
||||
|
||||
def test_object_as_string(self) -> None:
|
||||
item = ConsumptionItem.model_validate(self._base(object="obj-1"))
|
||||
assert item.object_id == "obj-1"
|
||||
|
||||
def test_metric_info(self) -> None:
|
||||
item = ConsumptionItem.model_validate(
|
||||
self._base(
|
||||
metric={
|
||||
"id": "air_tokens_cost",
|
||||
"name": "Стоимость",
|
||||
"unit": "nanorub",
|
||||
"quantity": "854857248637",
|
||||
}
|
||||
)
|
||||
)
|
||||
assert item.metric is not None
|
||||
assert item.metric.id == "air_tokens_cost"
|
||||
assert item.metric.quantity == Decimal("854857248637")
|
||||
|
||||
def test_value_decimal(self) -> None:
|
||||
item = ConsumptionItem.model_validate(self._base(value=42.5))
|
||||
assert item.value == Decimal("42.5")
|
||||
|
||||
|
||||
class TestFormatting:
|
||||
def test_format_rub(self) -> None:
|
||||
assert format_rub(Decimal("854.83")) == "854,83 ₽"
|
||||
assert format_rub(0) == "0,00 ₽"
|
||||
assert format_rub("1234.5") == "1 234,50 ₽"
|
||||
|
||||
def test_format_tokens(self) -> None:
|
||||
assert format_tokens(0) == "0"
|
||||
assert format_tokens(1234567) == "1 234 567"
|
||||
|
||||
def test_format_percent(self) -> None:
|
||||
assert format_percent(99.912) == "99,9%"
|
||||
assert format_percent(100.0) == "100,0%"
|
||||
@@ -0,0 +1,43 @@
|
||||
"""Тесты платформенно-зависимой диагностики (platform.py)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from selectel_ml_tui import platform as pl
|
||||
|
||||
|
||||
def test_platform_info_fields(monkeypatch) -> None:
|
||||
monkeypatch.setenv("TERM", "xterm-256color")
|
||||
monkeypatch.setenv("LANG", "ru_RU.UTF-8")
|
||||
info = pl.get_platform_info()
|
||||
assert info.system in ("Linux", "Darwin", "Windows")
|
||||
assert info.term == "xterm-256color"
|
||||
assert info.lang == "ru_RU.UTF-8"
|
||||
assert info.utf8 is True
|
||||
assert info.python_version
|
||||
|
||||
|
||||
def test_check_environment_ok(monkeypatch) -> None:
|
||||
monkeypatch.setenv("TERM", "tmux-256color")
|
||||
monkeypatch.setenv("LANG", "en_US.UTF-8")
|
||||
results = {c.name: c for c in pl.check_environment()}
|
||||
assert all(r.ok for r in results.values())
|
||||
assert results["python"].ok
|
||||
|
||||
|
||||
def test_check_environment_warns_without_term_and_lang(monkeypatch) -> None:
|
||||
monkeypatch.delenv("TERM", raising=False)
|
||||
monkeypatch.delenv("LANG", raising=False)
|
||||
results = {c.name: c for c in pl.check_environment()}
|
||||
assert not results["TERM"].ok
|
||||
assert not results["LANG"].ok
|
||||
|
||||
|
||||
def test_check_environment_warns_wrong_term(monkeypatch) -> None:
|
||||
monkeypatch.setenv("TERM", "dumb")
|
||||
monkeypatch.setenv("LANG", "C")
|
||||
results = {c.name: c for c in pl.check_environment()}
|
||||
assert not results["TERM"].ok
|
||||
|
||||
|
||||
def test_min_python_version() -> None:
|
||||
assert pl.MIN_PYTHON == (3, 10)
|
||||
@@ -0,0 +1,324 @@
|
||||
"""Unit-тесты сервисного слоя."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import datetime as dt
|
||||
from decimal import Decimal
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
|
||||
from selectel_ml_tui.config import Credentials, Settings
|
||||
from selectel_ml_tui.models import Consumption, Metrics
|
||||
from selectel_ml_tui.services import (
|
||||
DashboardData,
|
||||
DataService,
|
||||
parse_keys,
|
||||
parse_metrics,
|
||||
parse_models,
|
||||
parse_overview,
|
||||
parse_tokens,
|
||||
)
|
||||
from selectel_ml_tui.storage import Storage
|
||||
|
||||
|
||||
def _summary_response(value: str = "85483") -> dict[str, Any]:
|
||||
return {
|
||||
"status": "success",
|
||||
"data": [{"value": value}],
|
||||
}
|
||||
|
||||
|
||||
async def test_parse_tokens() -> None:
|
||||
tokens = parse_tokens(
|
||||
{"input_tokens": 10, "output_tokens": 20, "cache_read": 30, "cache_write": 40}
|
||||
)
|
||||
assert tokens.input_tokens == 10
|
||||
assert tokens.output_tokens == 20
|
||||
assert tokens.cache_read_tokens == 30
|
||||
assert tokens.cache_write_tokens == 40
|
||||
|
||||
|
||||
async def test_parse_tokens_missing_fields_defaults() -> None:
|
||||
tokens = parse_tokens({})
|
||||
assert tokens.input_tokens == 0
|
||||
assert tokens.output_tokens == 0
|
||||
assert tokens.cache_read_tokens == 0
|
||||
assert tokens.cache_write_tokens == 0
|
||||
|
||||
|
||||
async def test_parse_metrics() -> None:
|
||||
metrics = parse_metrics(
|
||||
{
|
||||
"global": {
|
||||
"requests": 1035,
|
||||
"success_rate_pct": 100,
|
||||
"avg_ttft_seconds": 3.67,
|
||||
"avg_latency_seconds": 11.85,
|
||||
"tokens": {
|
||||
"input": 228542299,
|
||||
"output": 747268,
|
||||
"cached_input": 222355840,
|
||||
"cache_write": 0,
|
||||
},
|
||||
}
|
||||
}
|
||||
)
|
||||
assert metrics is not None
|
||||
assert metrics.availability_pct == 100
|
||||
assert metrics.avg_ttft_ms == 3670
|
||||
assert metrics.avg_latency_ms == 11850
|
||||
assert metrics.requests == 1035
|
||||
assert metrics.input_tokens == 228542299
|
||||
assert metrics.output_tokens == 747268
|
||||
assert metrics.cache_read_tokens == 222355840
|
||||
assert metrics.cache_write_tokens == 0
|
||||
|
||||
|
||||
async def test_parse_metrics_empty() -> None:
|
||||
assert parse_metrics({}) is None
|
||||
|
||||
|
||||
async def test_parse_models() -> None:
|
||||
models = parse_models(
|
||||
{
|
||||
"by_model": [
|
||||
{
|
||||
"model": "deepseek/deepseek-v4-flash",
|
||||
"requests": 69,
|
||||
"success_rate_pct": 100,
|
||||
"avg_ttft_seconds": 3.35,
|
||||
},
|
||||
{
|
||||
"model": "deepseek/deepseek-v4-pro",
|
||||
"requests": 966,
|
||||
"success_rate_pct": 100,
|
||||
"avg_ttft_seconds": 3.69,
|
||||
},
|
||||
]
|
||||
}
|
||||
)
|
||||
assert len(models) == 2
|
||||
assert models[0].model_name == "deepseek/deepseek-v4-flash"
|
||||
assert models[0].requests == 69
|
||||
assert models[0].availability_pct == 100
|
||||
assert models[0].avg_ttft_ms == 3350
|
||||
assert models[1].model_id == "deepseek/deepseek-v4-pro"
|
||||
|
||||
|
||||
async def test_parse_overview() -> None:
|
||||
import datetime as _dt
|
||||
|
||||
consumption = parse_overview(
|
||||
{
|
||||
"totals": {
|
||||
"requests": 1034,
|
||||
"cost_rub_nano": "1075311673934",
|
||||
"tokens": {
|
||||
"input": 228314832,
|
||||
"output": 746702,
|
||||
"cached_input": 222128640,
|
||||
"cache_write": 0,
|
||||
},
|
||||
"tokens_total": {"total": 451607518},
|
||||
}
|
||||
},
|
||||
_dt.datetime(2026, 8, 19, tzinfo=_dt.timezone.utc),
|
||||
)
|
||||
assert consumption.amount_rub == Decimal("1075.311673934")
|
||||
assert consumption.requests == 1034
|
||||
assert consumption.tokens_total == 451607518
|
||||
assert consumption.input_tokens == 228314832
|
||||
assert consumption.cache_read_tokens == 222128640
|
||||
|
||||
|
||||
async def test_parse_keys() -> None:
|
||||
keys = parse_keys(
|
||||
[
|
||||
{
|
||||
"name": "Mei",
|
||||
"prefix": "sk-sl-v1-9dd61ec",
|
||||
"is_active": True,
|
||||
"budget": {
|
||||
"spend_limit_rub_nano": "1300000000000",
|
||||
"current_usage": {"spend_rub_nano": "1082949415164"},
|
||||
},
|
||||
},
|
||||
{
|
||||
"name": "Ollama Mac Air",
|
||||
"prefix": "sk-sl-v1-65fdb2e",
|
||||
"is_active": False,
|
||||
"budget": {"spend_limit_rub_nano": "100000000000"},
|
||||
},
|
||||
]
|
||||
)
|
||||
assert len(keys) == 2
|
||||
assert keys[0].name == "Mei"
|
||||
assert keys[0].is_active is True
|
||||
assert keys[0].spend_limit_rub == Decimal("1300")
|
||||
assert keys[0].spend_rub == Decimal("1082.949415164")
|
||||
assert keys[0].remaining_rub == Decimal("217.050584836")
|
||||
assert keys[1].is_active is False
|
||||
assert keys[1].spend_limit_rub == Decimal("100")
|
||||
|
||||
|
||||
async def test_refresh_billing_only(tmp_path) -> None:
|
||||
"""Только billing (без IAM): стоимость за месяц есть, gateway отмечен."""
|
||||
settings = Settings()
|
||||
creds = Credentials(static_token="static-token")
|
||||
storage = Storage(tmp_path / "cache.db")
|
||||
seen_end: dict[str, str] = {}
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
if "summary_stats" in request.url.path:
|
||||
seen_end["summary"] = request.url.params.get("end", "")
|
||||
return httpx.Response(200, json=_summary_response())
|
||||
if "consumption" in request.url.path:
|
||||
seen_end["consumption"] = request.url.params.get("end", "")
|
||||
return httpx.Response(
|
||||
200,
|
||||
json={
|
||||
"status": "success",
|
||||
"data": [{"period": "2026-08-19T00:00:00", "value": "7133"}],
|
||||
},
|
||||
)
|
||||
return httpx.Response(200, json=_summary_response())
|
||||
|
||||
service = DataService(
|
||||
settings, creds, storage, transport=httpx.MockTransport(handler)
|
||||
)
|
||||
data = await service.refresh()
|
||||
assert data.month_consumption is not None
|
||||
assert data.month_consumption.amount_rub == Decimal("854.83")
|
||||
# Пункт A: end должен содержать время (не усечённую дату) и быть naive
|
||||
# (без смещения часового пояса — иначе API отвечает 422).
|
||||
assert "T" in seen_end["summary"]
|
||||
assert "T" in seen_end["consumption"]
|
||||
assert "+" not in seen_end["summary"]
|
||||
# Пункт B: дневная стоимость сегодня выделена отдельно.
|
||||
assert data.today_amount_rub == Decimal("71.33")
|
||||
assert not data.offline
|
||||
assert data.error is not None
|
||||
assert "сервисный пользователь" in data.error
|
||||
assert storage.get_latest_consumption() is not None
|
||||
history = storage.get_history(days=30)
|
||||
assert len(history) == 1
|
||||
# История хранит дневную стоимость (не месячную сумму).
|
||||
assert history[0].amount_rub == Decimal("71.33")
|
||||
|
||||
|
||||
async def test_refresh_billing_failure_uses_cache(tmp_path) -> None:
|
||||
"""Сбой billing + кэш → офлайн-данные из Storage."""
|
||||
settings = Settings()
|
||||
creds = Credentials(static_token="static-token")
|
||||
storage = Storage(tmp_path / "cache.db")
|
||||
storage.save_snapshot(
|
||||
consumption=Consumption(
|
||||
period=dt.datetime(2026, 8, 1, tzinfo=dt.timezone.utc),
|
||||
amount_rub=Decimal("10.00"),
|
||||
)
|
||||
)
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
return httpx.Response(500, text="boom")
|
||||
|
||||
service = DataService(
|
||||
settings, creds, storage, transport=httpx.MockTransport(handler)
|
||||
)
|
||||
data = await service.refresh()
|
||||
assert data.offline is True
|
||||
assert data.source == "cache"
|
||||
assert data.month_consumption is not None
|
||||
assert data.month_consumption.amount_rub == Decimal("10.00")
|
||||
assert data.error is not None
|
||||
|
||||
|
||||
async def test_refresh_with_gateway(tmp_path) -> None:
|
||||
"""Billing + gateway (IAM): заполнены метрики, модели, ключи, бюджет."""
|
||||
settings = Settings()
|
||||
creds = Credentials(
|
||||
static_token="static-token",
|
||||
service_user="svc",
|
||||
password="pwd",
|
||||
project_id="proj-1",
|
||||
domain_id="239633",
|
||||
)
|
||||
storage = Storage(tmp_path / "cache.db")
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
path = request.url.path
|
||||
if path.endswith("/auth/tokens"):
|
||||
return httpx.Response(
|
||||
201,
|
||||
headers={"X-Subject-Token": "iam-token"},
|
||||
json={"token": {"expires_at": "2030-01-01T00:00:00Z"}},
|
||||
)
|
||||
if path.endswith("/gateways"):
|
||||
return httpx.Response(200, json={"data": [{"id": "gw-1"}]})
|
||||
if path.endswith("/overview"):
|
||||
return httpx.Response(
|
||||
200,
|
||||
json={
|
||||
"totals": {
|
||||
"requests": 1034,
|
||||
"cost_rub_nano": "1075311673934",
|
||||
"tokens": {"input": 100, "output": 200},
|
||||
"tokens_total": {"total": 300},
|
||||
},
|
||||
"budget": {
|
||||
"spend_limit_rub_nano": "2000000000000",
|
||||
"spend_rub_nano": "1077874156239",
|
||||
},
|
||||
},
|
||||
)
|
||||
if path.endswith("/metrics"):
|
||||
return httpx.Response(
|
||||
200,
|
||||
json={
|
||||
"global": {"success_rate_pct": 99.5, "avg_ttft_seconds": 0.3},
|
||||
"by_model": [
|
||||
{"model": "deepseek/deepseek-v4-flash", "requests": 69}
|
||||
],
|
||||
},
|
||||
)
|
||||
if path.endswith("/keys"):
|
||||
return httpx.Response(
|
||||
200,
|
||||
json={
|
||||
"data": [
|
||||
{
|
||||
"name": "Mei",
|
||||
"prefix": "sk-sl-v1-9dd61ec",
|
||||
"is_active": True,
|
||||
"budget": {
|
||||
"spend_limit_rub_nano": "1300000000000",
|
||||
"current_usage": {"spend_rub_nano": "1082949415164"},
|
||||
},
|
||||
}
|
||||
]
|
||||
},
|
||||
)
|
||||
return httpx.Response(200, json=_summary_response())
|
||||
|
||||
service = DataService(
|
||||
settings, creds, storage, transport=httpx.MockTransport(handler)
|
||||
)
|
||||
data = await service.refresh()
|
||||
assert data.month_consumption is not None
|
||||
assert data.month_consumption.amount_rub == Decimal("1075.311673934")
|
||||
assert data.metrics is not None
|
||||
assert data.metrics.availability_pct == 99.5
|
||||
assert data.metrics.avg_ttft_ms == 300
|
||||
assert len(data.models) == 1
|
||||
assert data.models[0].model_name == "deepseek/deepseek-v4-flash"
|
||||
assert len(data.keys) == 1
|
||||
assert data.keys[0].name == "Mei"
|
||||
assert data.budget is not None
|
||||
assert data.budget.spend_limit_rub == Decimal("2000")
|
||||
|
||||
|
||||
async def test_dashboard_data_has_data() -> None:
|
||||
empty = DashboardData()
|
||||
assert not empty.has_data
|
||||
full = DashboardData(metrics=Metrics())
|
||||
assert full.has_data
|
||||
@@ -0,0 +1,238 @@
|
||||
"""Unit-тесты SQLite-кэша и истории."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import datetime as dt
|
||||
import json
|
||||
import sqlite3
|
||||
from decimal import Decimal
|
||||
|
||||
import pytest
|
||||
|
||||
from selectel_ml_tui.models import (
|
||||
Consumption,
|
||||
HistoryPoint,
|
||||
Metrics,
|
||||
ModelMetrics,
|
||||
TokenUsage,
|
||||
)
|
||||
from selectel_ml_tui.storage import Storage, default_db_path
|
||||
|
||||
|
||||
def _consumption(period: str, amount: str) -> Consumption:
|
||||
return Consumption.model_validate(
|
||||
{
|
||||
"period": period,
|
||||
"amount_rub": amount,
|
||||
"input_tokens": 100,
|
||||
"output_tokens": 200,
|
||||
"cache_read_tokens": 30,
|
||||
"cache_write_tokens": 40,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def _metrics() -> Metrics:
|
||||
return Metrics.model_validate(
|
||||
{
|
||||
"availability_pct": 99.9,
|
||||
"avg_ttft_ms": 0.4,
|
||||
"input_tokens": 1000,
|
||||
"output_tokens": 900,
|
||||
"cache_read_tokens": 800,
|
||||
"cache_write_tokens": 700,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def _models() -> list[ModelMetrics]:
|
||||
return [
|
||||
ModelMetrics.model_validate(
|
||||
{
|
||||
"model_id": "gpt-4o",
|
||||
"model_name": "GPT-4o",
|
||||
"availability_pct": 99.9,
|
||||
"avg_ttft_ms": 0.3,
|
||||
}
|
||||
),
|
||||
ModelMetrics.model_validate({"model_id": "llama-3", "model_name": "Llama 3"}),
|
||||
]
|
||||
|
||||
|
||||
def test_default_db_path(monkeypatch, tmp_path) -> None:
|
||||
monkeypatch.setattr("selectel_ml_tui.storage.data_dir", lambda: tmp_path)
|
||||
assert default_db_path() == tmp_path / "cache.db"
|
||||
assert default_db_path().parent == tmp_path
|
||||
|
||||
|
||||
def test_save_and_get_latest_consumption(tmp_path) -> None:
|
||||
storage = Storage(tmp_path / "cache.db")
|
||||
item = _consumption("2026-08-01T00:00:00", "854.83")
|
||||
storage.save_consumption(item)
|
||||
latest = storage.get_latest_consumption()
|
||||
assert latest is not None
|
||||
assert latest.amount_rub == Decimal("854.83")
|
||||
assert latest.period == dt.datetime(2026, 8, 1)
|
||||
assert latest.input_tokens == 100
|
||||
storage.close()
|
||||
|
||||
|
||||
def test_get_latest_empty(tmp_path) -> None:
|
||||
storage = Storage(tmp_path / "cache.db")
|
||||
assert storage.get_latest_consumption() is None
|
||||
assert storage.get_latest_metrics() is None
|
||||
assert storage.get_latest_model_metrics() == []
|
||||
storage.close()
|
||||
|
||||
|
||||
def test_save_and_get_latest_metrics(tmp_path) -> None:
|
||||
storage = Storage(tmp_path / "cache.db")
|
||||
storage.save_metrics(_metrics())
|
||||
latest = storage.get_latest_metrics()
|
||||
assert latest is not None
|
||||
assert latest.availability_pct == 99.9
|
||||
assert latest.input_tokens == 1000
|
||||
storage.close()
|
||||
|
||||
|
||||
def test_save_and_get_latest_model_metrics(tmp_path) -> None:
|
||||
storage = Storage(tmp_path / "cache.db")
|
||||
storage.save_model_metrics(_models())
|
||||
latest = storage.get_latest_model_metrics()
|
||||
assert len(latest) == 2
|
||||
assert {m.model_id for m in latest} == {"gpt-4o", "llama-3"}
|
||||
gpt = next(m for m in latest if m.model_id == "gpt-4o")
|
||||
assert gpt.availability_pct == 99.9
|
||||
storage.close()
|
||||
|
||||
|
||||
def test_history_upsert_same_day(tmp_path) -> None:
|
||||
storage = Storage(tmp_path / "cache.db")
|
||||
point = HistoryPoint(
|
||||
timestamp=dt.datetime(2026, 8, 1, 12, 0),
|
||||
amount_rub=Decimal("10.00"),
|
||||
tokens=TokenUsage(input_tokens=5),
|
||||
)
|
||||
storage.save_history(point)
|
||||
point2 = HistoryPoint(
|
||||
timestamp=dt.datetime(2026, 8, 1, 12, 0),
|
||||
amount_rub=Decimal("15.00"),
|
||||
tokens=TokenUsage(input_tokens=8),
|
||||
)
|
||||
storage.save_history(point2)
|
||||
history = storage.get_history(days=30)
|
||||
assert len(history) == 1 # upsert, без дублей
|
||||
assert history[0].amount_rub == Decimal("15.00")
|
||||
storage.close()
|
||||
|
||||
|
||||
def test_history_days_filter(tmp_path) -> None:
|
||||
storage = Storage(tmp_path / "cache.db")
|
||||
now = dt.datetime.now(dt.timezone.utc)
|
||||
storage.save_history(
|
||||
HistoryPoint(timestamp=now, amount_rub=Decimal("1"), tokens=TokenUsage())
|
||||
)
|
||||
storage.save_history(
|
||||
HistoryPoint(
|
||||
timestamp=now - dt.timedelta(days=30),
|
||||
amount_rub=Decimal("2"),
|
||||
tokens=TokenUsage(),
|
||||
)
|
||||
)
|
||||
assert len(storage.get_history(days=7)) == 1
|
||||
assert len(storage.get_history(days=60)) == 2
|
||||
storage.close()
|
||||
|
||||
|
||||
def test_save_snapshot_combined(tmp_path) -> None:
|
||||
storage = Storage(tmp_path / "cache.db")
|
||||
storage.save_snapshot(
|
||||
consumption=_consumption("2026-08-01T00:00:00", "854.83"),
|
||||
metrics=_metrics(),
|
||||
models=_models(),
|
||||
history_point=HistoryPoint(
|
||||
timestamp=dt.datetime(2026, 8, 1, 12, 0),
|
||||
amount_rub=Decimal("854.83"),
|
||||
tokens=TokenUsage(input_tokens=100),
|
||||
),
|
||||
)
|
||||
assert storage.get_latest_consumption() is not None
|
||||
assert storage.get_latest_metrics() is not None
|
||||
assert len(storage.get_latest_model_metrics()) == 2
|
||||
assert len(storage.get_history(days=30)) == 1
|
||||
storage.close()
|
||||
|
||||
|
||||
def test_cleanup_removes_old(tmp_path) -> None:
|
||||
storage = Storage(tmp_path / "cache.db")
|
||||
old = dt.datetime(2020, 1, 1, 0, 0)
|
||||
storage.save_history(
|
||||
HistoryPoint(timestamp=old, amount_rub=Decimal("1"), tokens=TokenUsage())
|
||||
)
|
||||
storage.save_consumption(_consumption("2020-01-01T00:00:00", "1"))
|
||||
deleted = storage.cleanup(max_age_days=30)
|
||||
# history удаляется по timestamp; consumption_snapshots — по created_at (сейчас)
|
||||
assert deleted == 1
|
||||
assert storage.get_history(days=3650) == []
|
||||
assert storage.get_latest_consumption() is not None
|
||||
storage.close()
|
||||
|
||||
|
||||
def test_export_json(tmp_path) -> None:
|
||||
storage = Storage(tmp_path / "cache.db")
|
||||
storage.save_history(
|
||||
HistoryPoint(
|
||||
timestamp=dt.datetime(2026, 8, 1, 12, 0),
|
||||
amount_rub=Decimal("854.83"),
|
||||
tokens=TokenUsage(input_tokens=100, output_tokens=200),
|
||||
)
|
||||
)
|
||||
out = tmp_path / "history.json"
|
||||
storage.export_json(out)
|
||||
data = json.loads(out.read_text(encoding="utf-8"))
|
||||
assert len(data) == 1
|
||||
assert data[0]["amount_rub"] == "854.83"
|
||||
assert data[0]["input_tokens"] == 100
|
||||
storage.close()
|
||||
|
||||
|
||||
def test_export_csv(tmp_path) -> None:
|
||||
storage = Storage(tmp_path / "cache.db")
|
||||
storage.save_history(
|
||||
HistoryPoint(
|
||||
timestamp=dt.datetime(2026, 8, 1, 12, 0),
|
||||
amount_rub=Decimal("854.83"),
|
||||
tokens=TokenUsage(input_tokens=100, output_tokens=200),
|
||||
)
|
||||
)
|
||||
out = tmp_path / "history.csv"
|
||||
storage.export_csv(out)
|
||||
lines = out.read_text(encoding="utf-8").strip().splitlines()
|
||||
assert lines[0].startswith("timestamp,amount_rub")
|
||||
assert "854.83" in lines[1]
|
||||
assert "100" in lines[1]
|
||||
storage.close()
|
||||
|
||||
|
||||
def test_close_idempotent(tmp_path) -> None:
|
||||
storage = Storage(tmp_path / "cache.db")
|
||||
storage.close()
|
||||
with pytest.raises(sqlite3.ProgrammingError):
|
||||
storage.get_history()
|
||||
|
||||
|
||||
def test_clear_removes_all_rows(tmp_path) -> None:
|
||||
storage = Storage(tmp_path / "cache.db")
|
||||
storage.save_history(
|
||||
HistoryPoint(
|
||||
timestamp=dt.datetime(2026, 8, 18),
|
||||
amount_rub=Decimal("1"),
|
||||
tokens=TokenUsage(input_tokens=1),
|
||||
)
|
||||
)
|
||||
storage.save_consumption(Consumption(period="2026-08-01T00:00:00"))
|
||||
assert storage.get_history() != []
|
||||
deleted = storage.clear()
|
||||
assert deleted >= 1
|
||||
assert storage.get_history() == []
|
||||
assert storage.get_latest_consumption() is None
|
||||
@@ -0,0 +1,296 @@
|
||||
"""Headless-тесты TUI (Textual pilot)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import datetime as dt
|
||||
from decimal import Decimal
|
||||
|
||||
from textual.widgets import DataTable, Input, Static
|
||||
|
||||
from selectel_ml_tui.config import Credentials, Settings
|
||||
from selectel_ml_tui.models import (
|
||||
ApiKeyUsage,
|
||||
BudgetInfo,
|
||||
Consumption,
|
||||
HistoryPoint,
|
||||
Metrics,
|
||||
ModelMetrics,
|
||||
TokenUsage,
|
||||
)
|
||||
from selectel_ml_tui.services import DashboardData
|
||||
from selectel_ml_tui.storage import Storage
|
||||
from selectel_ml_tui.ui.app import SelectelMLTUI
|
||||
|
||||
|
||||
class FakeService:
|
||||
"""Сервис-заглушка: возвращает готовые данные."""
|
||||
|
||||
def __init__(self, data: DashboardData) -> None:
|
||||
self.data = data
|
||||
self.refresh_count = 0
|
||||
|
||||
async def refresh(self) -> DashboardData:
|
||||
self.refresh_count += 1
|
||||
return self.data
|
||||
|
||||
|
||||
def _dashboard_data() -> DashboardData:
|
||||
return DashboardData(
|
||||
month_consumption=Consumption.model_validate(
|
||||
{
|
||||
"period": "2026-08-01T00:00:00",
|
||||
"amount_rub": "854.83",
|
||||
"requests": 1034,
|
||||
"tokens_total": 451607518,
|
||||
"input_tokens": 100,
|
||||
"output_tokens": 200,
|
||||
"cache_read_tokens": 30,
|
||||
"cache_write_tokens": 40,
|
||||
}
|
||||
),
|
||||
metrics=Metrics.model_validate(
|
||||
{
|
||||
"availability_pct": 99.9,
|
||||
"avg_ttft_ms": 0.4,
|
||||
"avg_latency_ms": 1.2,
|
||||
"requests": 1035,
|
||||
"input_tokens": 5,
|
||||
}
|
||||
),
|
||||
models=[
|
||||
ModelMetrics.model_validate(
|
||||
{
|
||||
"model_id": "gpt-4o",
|
||||
"model_name": "GPT-4o",
|
||||
"requests": 69,
|
||||
"availability_pct": 100.0,
|
||||
"avg_ttft_ms": 3.35,
|
||||
}
|
||||
)
|
||||
],
|
||||
keys=[
|
||||
ApiKeyUsage(
|
||||
name="Mei",
|
||||
prefix="sk-sl-v1-9dd61ec",
|
||||
is_active=True,
|
||||
spend_limit_rub=Decimal("1300"),
|
||||
spend_rub=Decimal("1082.95"),
|
||||
remaining_rub=Decimal("217.05"),
|
||||
)
|
||||
],
|
||||
budget=BudgetInfo(
|
||||
spend_limit_rub=Decimal("2000"),
|
||||
spend_rub=Decimal("1077.87"),
|
||||
remaining_rub=Decimal("922.13"),
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def _make_app(tmp_path, data: DashboardData | None = None) -> SelectelMLTUI:
|
||||
settings = Settings(refresh_interval=3600)
|
||||
creds = Credentials(static_token="tok")
|
||||
storage = Storage(tmp_path / "cache.db")
|
||||
service = FakeService(data or _dashboard_data())
|
||||
return SelectelMLTUI(
|
||||
settings=settings,
|
||||
credentials=creds,
|
||||
storage=storage,
|
||||
service=service,
|
||||
)
|
||||
|
||||
|
||||
async def test_dashboard_renders_data(tmp_path) -> None:
|
||||
app = _make_app(tmp_path)
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.pause()
|
||||
month = app.screen.query_one("#month-panel", Static)
|
||||
rendered = str(month.render())
|
||||
assert "854,83" in rendered
|
||||
assert "Потребление за месяц" in rendered
|
||||
metrics = app.screen.query_one("#metrics-panel", Static)
|
||||
assert "99,9%" in str(metrics.render())
|
||||
table = app.screen.query_one("#models-table", DataTable)
|
||||
assert table.row_count == 1
|
||||
assert "GPT-4o" in table.get_row_at(0)
|
||||
|
||||
|
||||
async def test_refresh_hotkey_reruns_service(tmp_path) -> None:
|
||||
app = _make_app(tmp_path)
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.pause()
|
||||
service = app._service
|
||||
assert service.refresh_count == 1
|
||||
await pilot.press("r")
|
||||
await pilot.pause()
|
||||
assert service.refresh_count == 2
|
||||
|
||||
|
||||
async def test_navigate_to_history_and_settings(tmp_path) -> None:
|
||||
app = _make_app(tmp_path)
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.pause()
|
||||
# history
|
||||
await pilot.press("h")
|
||||
await pilot.pause()
|
||||
assert app.screen.__class__.__name__ == "HistoryScreen"
|
||||
# назад
|
||||
await pilot.press("escape")
|
||||
await pilot.pause()
|
||||
assert app.screen.__class__.__name__ == "DashboardScreen"
|
||||
# settings
|
||||
await pilot.press("s")
|
||||
await pilot.pause()
|
||||
assert app.screen.__class__.__name__ == "SettingsScreen"
|
||||
info = app.screen.query_one("#settings-info", Static).render()
|
||||
assert "API base URL" in str(info)
|
||||
assert "tok" in str(info)
|
||||
await pilot.press("escape")
|
||||
await pilot.pause()
|
||||
assert app.screen.__class__.__name__ == "DashboardScreen"
|
||||
|
||||
|
||||
async def test_history_screen_shows_points(tmp_path) -> None:
|
||||
storage = Storage(tmp_path / "cache.db")
|
||||
storage.save_history(
|
||||
HistoryPoint(
|
||||
timestamp=dt.datetime(2026, 8, 18, 0, 0),
|
||||
amount_rub=Decimal("854.83"),
|
||||
tokens=TokenUsage(input_tokens=100),
|
||||
)
|
||||
)
|
||||
app = _make_app(tmp_path)
|
||||
app._storage = storage
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.press("h")
|
||||
await pilot.pause()
|
||||
table = app.screen.query_one("#history-table", DataTable)
|
||||
assert table.row_count == 1
|
||||
cells = [str(cell) for cell in table.get_row_at(0)]
|
||||
assert any("854,83" in cell for cell in cells)
|
||||
bars = app.screen.query_one("#history-bars")
|
||||
assert bars.render() != "Нет данных для графика"
|
||||
|
||||
|
||||
async def test_history_period_switch(tmp_path) -> None:
|
||||
storage = Storage(tmp_path / "cache.db")
|
||||
for day in range(20):
|
||||
storage.save_history(
|
||||
HistoryPoint(
|
||||
timestamp=dt.datetime(2026, 8, day + 1, 0, 0),
|
||||
amount_rub=Decimal(str(day + 1)),
|
||||
tokens=TokenUsage(input_tokens=day + 1),
|
||||
)
|
||||
)
|
||||
app = _make_app(tmp_path)
|
||||
app._storage = storage
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.press("h")
|
||||
await pilot.pause()
|
||||
title = app.screen.query_one("#history-title", Static)
|
||||
assert "30 дней" in str(title.render())
|
||||
table = app.screen.query_one("#history-table", DataTable)
|
||||
rows_30 = table.row_count
|
||||
await pilot.click("#period-7")
|
||||
await pilot.pause()
|
||||
title = app.screen.query_one("#history-title", Static)
|
||||
assert "7 дней" in str(title.render())
|
||||
table = app.screen.query_one("#history-table", DataTable)
|
||||
assert table.row_count < rows_30
|
||||
await pilot.click("#period-90")
|
||||
await pilot.pause()
|
||||
assert "90 дней" in str(app.screen.query_one("#history-title", Static).render())
|
||||
assert app.screen.query_one("#history-table", DataTable).row_count >= rows_30
|
||||
|
||||
|
||||
async def test_model_detail_on_enter(tmp_path) -> None:
|
||||
app = _make_app(tmp_path)
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.pause()
|
||||
await pilot.press("enter")
|
||||
await pilot.pause()
|
||||
assert app.screen.__class__.__name__ == "ModelDetailScreen"
|
||||
detail = app.screen.query_one("#model-detail", Static)
|
||||
rendered = str(detail.render())
|
||||
assert "GPT-4o" in rendered
|
||||
assert "Запросы" in rendered
|
||||
await pilot.press("escape")
|
||||
await pilot.pause()
|
||||
assert app.screen.__class__.__name__ == "DashboardScreen"
|
||||
|
||||
|
||||
async def test_settings_save_interval(tmp_path, monkeypatch) -> None:
|
||||
saved = {}
|
||||
|
||||
def fake_save(settings) -> None:
|
||||
saved["interval"] = settings.refresh_interval
|
||||
|
||||
monkeypatch.setattr("selectel_ml_tui.ui.app.save_settings", fake_save)
|
||||
app = _make_app(tmp_path)
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.press("s")
|
||||
await pilot.pause()
|
||||
inp = app.screen.query_one("#interval-input", Input)
|
||||
inp.value = "120"
|
||||
await pilot.click("#save-settings")
|
||||
await pilot.pause()
|
||||
assert app._settings.refresh_interval == 120
|
||||
assert saved["interval"] == 120
|
||||
|
||||
|
||||
async def test_settings_clear_cache(tmp_path) -> None:
|
||||
storage = Storage(tmp_path / "cache.db")
|
||||
storage.save_history(
|
||||
HistoryPoint(
|
||||
timestamp=dt.datetime(2026, 8, 18, 0, 0),
|
||||
amount_rub=Decimal("854.83"),
|
||||
tokens=TokenUsage(input_tokens=100),
|
||||
)
|
||||
)
|
||||
app = _make_app(tmp_path)
|
||||
app._storage = storage
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.press("s")
|
||||
await pilot.pause()
|
||||
await pilot.click("#clear-cache")
|
||||
await pilot.pause()
|
||||
assert storage.get_history(days=30) == []
|
||||
|
||||
|
||||
async def test_offline_status_in_subtitle(tmp_path) -> None:
|
||||
data = _dashboard_data()
|
||||
data.offline = True
|
||||
app = _make_app(tmp_path, data)
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.pause()
|
||||
assert "offline" in app.sub_title
|
||||
|
||||
|
||||
async def test_window_resize(tmp_path) -> None:
|
||||
app = _make_app(tmp_path)
|
||||
async with app.run_test(size=(120, 40)) as pilot:
|
||||
await pilot.pause()
|
||||
assert app.screen.__class__.__name__ == "DashboardScreen"
|
||||
await pilot.resize_terminal(80, 24)
|
||||
await pilot.pause()
|
||||
assert app.screen.__class__.__name__ == "DashboardScreen"
|
||||
month = app.screen.query_one("#month-panel", Static)
|
||||
assert "854,83" in str(month.render())
|
||||
|
||||
|
||||
async def test_keys_screen(tmp_path) -> None:
|
||||
app = _make_app(tmp_path)
|
||||
async with app.run_test() as pilot:
|
||||
await pilot.pause()
|
||||
await pilot.press("k")
|
||||
await pilot.pause()
|
||||
assert app.screen.__class__.__name__ == "KeysScreen"
|
||||
table = app.screen.query_one("#keys-table", DataTable)
|
||||
assert table.row_count == 1
|
||||
cells = [str(cell) for cell in table.get_row_at(0)]
|
||||
assert any("Mei" in cell for cell in cells)
|
||||
assert any("1 300" in cell for cell in cells)
|
||||
budget = app.screen.query_one("#keys-budget", Static)
|
||||
assert "Бюджет роутера" in str(budget.render())
|
||||
await pilot.press("escape")
|
||||
await pilot.pause()
|
||||
assert app.screen.__class__.__name__ == "DashboardScreen"
|
||||
Reference in New Issue
Block a user