tests/test_usage_stats.py - master - artyuhovs/AIFeedBackTrainingBot

tests/test_usage_stats.py
104 строки4 KB
Codex
Добавить статистику использования по пользователям
21 июл 2026, 01:34
21 июл 2026, 01:3402504ee
from __future__ import annotations

import asyncio
from pathlib import Path

from app.bot.presentation import usage_stats_text
from app.db.repositories import Repositories
from app.models.media import IncomingMediaMeta, MediaType
from app.services.usage_stats import UsageStatsService


def media(
    user_id: int,
    message_id: int,
    media_type: MediaType,
    *,
    file_size_bytes: int,
) -> IncomingMediaMeta:
    return IncomingMediaMeta(
        telegram_user_id=user_id,
        telegram_chat_id=user_id,
        telegram_message_id=message_id,
        telegram_file_id=f"file-{message_id}",
        telegram_file_unique_id=f"unique-{message_id}",
        media_type=media_type,
        original_file_name=f"input-{message_id}.mp4",
        mime_type="video/mp4",
        file_size_bytes=file_size_bytes,
    )


def test_usage_stats_are_grouped_by_telegram_user(tmp_path: Path) -> None:
    repository = Repositories(tmp_path / "bot.sqlite3")
    repository.schema.init()
    repository.identities.upsert_user(42, "active", "Active", "User")
    repository.identities.upsert_user(77, None, "Second", "User")

    assert repository.input_intents.start_free_training(42)
    text = repository.messages.create_text_message(42, 42, 1, "Текст свободной тренировки")
    video = repository.messages.create_message(media(42, 2, "video_note", file_size_bytes=2048))
    repository.processing.mark_audio_prepared(video.id, tmp_path / "audio.wav", 65.4, 1024)
    repository.processing.mark_transcribed(video.id, "Расшифровка", None)

    failed = repository.messages.create_message(media(77, 3, "video", file_size_bytes=4096))
    repository.processing.mark_failed(failed.id, "download_failed", "network")

    current = repository.analysis.create(
        text.id,
        "gigachat",
        "GigaChat",
        "feedback_general_quality_v4",
        "Текст свободной тренировки",
    )
    repository.analysis.complete(
        current.id,
        "{}",
        None,
        prompt_tokens=120,
        completion_tokens=30,
        total_tokens=150,
    )
    historical = repository.analysis.create(
        video.id,
        "gigachat",
        "GigaChat",
        "feedback_general_quality_v4",
        "Расшифровка",
    )
    repository.analysis.complete(historical.id, "{}", None)

    report = asyncio.run(UsageStatsService(repository).load())

    assert [user.telegram_user_id for user in report.users] == [42, 77]
    active = report.users[0].counts
    assert active.messages == 2
    assert active.free_training_messages == 1
    assert active.video_notes == 1
    assert active.texts == 1
    assert active.duration_seconds == 65.4
    assert active.analysis_runs == 2
    assert active.completed_analyses == 2
    assert active.tokenized_analyses == 1
    assert (active.prompt_tokens, active.completion_tokens, active.total_tokens) == (120, 30, 150)
    assert report.users[1].counts.failed_messages == 1
    assert report.users[1].counts.videos == 1
    assert report.totals.messages == 3
    assert report.totals.input_bytes == len("Текст свободной тренировки".encode()) + 2048 + 4096

    text_report = usage_stats_text(report)
    assert "Статистика использования — за всё время" in text_report
    assert "@active (id 42)" in text_report
    assert "Second User (id 77)" in text_report
    assert "токены 150 (вход 120, выход 30)" in text_report
    assert "usage сохранён для 1 из 2 завершённых" in text_report


def test_usage_stats_report_is_clear_when_no_messages_exist(tmp_path: Path) -> None:
    repository = Repositories(tmp_path / "bot.sqlite3")
    repository.schema.init()

    report = asyncio.run(UsageStatsService(repository).load())

    assert report.users == ()
    assert usage_stats_text(report) == "Статистика использования: обработанных сообщений пока нет."