tests/test_db_models.py - master - artyuhovs/AIFeedBackTrainingBot

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

from datetime import timedelta, timezone
from pathlib import Path

import pytest

from app.db.base import SCHEMA_SQL
from app.db.models import (
    AnalysisStatus,
    CourseVersionStatus,
    IdentityKind,
    MembershipStatus,
    ProcessingStatus,
    StreamRole,
    StreamStatus,
)
from app.db.repositories import Repositories, text_sha256
from app.models.media import IncomingMediaMeta
from app.services.identity_claim import IdentityClaimService, TelegramProfile
from tests.time_support import date_at, frozen_clock, iso_at


def media() -> IncomingMediaMeta:
    return IncomingMediaMeta(
        telegram_user_id=42,
        telegram_chat_id=100,
        telegram_message_id=7,
        telegram_file_id="file-id",
        telegram_file_unique_id="unique-id",
        media_type="voice",
        original_file_name="voice.oga",
        mime_type="audio/ogg",
        file_size_bytes=123,
    )


def test_message_record_is_created(tmp_path: Path) -> None:
    repo = Repositories(tmp_path / "bot.sqlite3")
    repo.schema.init()
    record = repo.messages.create_message(media())
    assert record.id > 0
    assert record.status == ProcessingStatus.RECEIVED.value
    assert record.telegram_user_id == 42


def test_paused_stream_maps_to_typed_status(tmp_path: Path) -> None:
    repo = Repositories(tmp_path / "bot.sqlite3")
    repo.schema.init()
    stream = repo.course_catalog.default_stream()
    with repo.database.connect() as db:
        db.execute("UPDATE streams SET status='paused' WHERE id=?", (stream.id,))

    paused = repo.identities.get_stream(stream.id)

    assert paused is not None
    assert paused.status == StreamStatus.PAUSED


def test_access_queries_preserve_typed_database_boundary(tmp_path: Path) -> None:
    repo = Repositories(tmp_path / "bot.sqlite3")
    repo.schema.init()
    stream = repo.course_catalog.default_stream()
    user = repo.identities.upsert_user(42, "learner", "Learner", None)
    repo.access_commands.approve_membership(stream.id, user.telegram_user_id, None)
    repo.access_commands.set_current_stream(user.telegram_user_id, stream.id)

    current = repo.user_access.current_stream_for_role(user.telegram_user_id, StreamRole.LEARNER)
    membership = repo.user_access.membership(stream.id, user.telegram_user_id)
    summary = next(
        item for item in repo.user_access.users_with_memberships() if item["telegram_user_id"] == user.telegram_user_id
    )
    counted = next(item for item in repo.identities.streams_with_counts() if item["id"] == stream.id)

    assert current is not None
    assert current.is_active is True
    assert current.status == StreamStatus.DRAFT
    assert membership is not None
    assert membership.status == MembershipStatus.APPROVED
    assert membership.roles == (StreamRole.LEARNER,)
    assert summary["identity_kind"] == IdentityKind.TELEGRAM
    assert summary["is_admin"] is False
    assert counted["is_active"] is True
    assert counted["status"] == StreamStatus.DRAFT


def test_existing_database_is_migrated_before_hash_indexes(tmp_path: Path) -> None:
    repo = Repositories(tmp_path / "bot.sqlite3")
    old_schema = (
        SCHEMA_SQL.replace("    transcript_sha256 TEXT,\n    transcript_file_path TEXT,\n", "")
        .replace("    input_text_sha256 TEXT,\n", "")
        .replace("    role TEXT NOT NULL DEFAULT 'learner' CHECK(role IN ('learner', 'curator')),\n", "")
    )
    with repo.database.connect() as db:
        db.executescript(old_schema)

    with pytest.raises(RuntimeError, match="unversioned"):
        repo.schema.init()
    repo.schema.prepare_legacy_upgrade()

    with repo.database.connect() as db:
        message_columns = {row["name"] for row in db.execute("PRAGMA table_info(messages)")}
        analysis_columns = {row["name"] for row in db.execute("PRAGMA table_info(analysis_runs)")}
        membership_columns = {row["name"] for row in db.execute("PRAGMA table_info(stream_memberships)")}
        message_indexes = {row["name"] for row in db.execute("PRAGMA index_list(messages)")}
        analysis_indexes = {row["name"] for row in db.execute("PRAGMA index_list(analysis_runs)")}

    assert {"transcript_sha256", "transcript_file_path"}.issubset(message_columns)
    assert "input_text_sha256" in analysis_columns
    assert "role" in membership_columns
    assert "idx_messages_transcript_sha256" in message_indexes
    assert "idx_analysis_runs_input_text_sha256" in analysis_indexes


@pytest.mark.parametrize(
    "response_format",
    ["text|voice", "audio_or_video_note", "group", "both"],
)
def test_unsupported_composite_response_formats_are_rejected(
    tmp_path: Path,
    response_format: str,
) -> None:
    repo = Repositories(tmp_path / "bot.sqlite3")
    repo.schema.init()
    course = repo.course_catalog.create_course("Обратная связь")
    version = repo.course_catalog.create_course_version(course.id)
    topic = repo.course_catalog.add_course_topic(version.id, "Практика", position=0)

    with pytest.raises(ValueError, match="Unsupported assignment response format"):
        repo.course_catalog.add_course_publication(
            topic.id,
            position=0,
            publication_type="Задание",
            content_text="Дайте обратную связь по кейсу.",
            publish_offset_minutes=0,
            deadline_offset_minutes=60,
            response_format=response_format,
        )


def test_status_and_error_are_saved(tmp_path: Path) -> None:
    repo = Repositories(tmp_path / "bot.sqlite3")
    repo.schema.init()
    record = repo.messages.create_message(media())
    repo.processing.mark_failed(record.id, "download_failed", "network")
    updated = repo.messages.get_message(record.id)
    assert updated is not None
    assert updated.status == ProcessingStatus.FAILED.value
    assert updated.error_code == "download_failed"
    assert updated.error_message == "network"


def test_transcript_is_saved(tmp_path: Path) -> None:
    repo = Repositories(tmp_path / "bot.sqlite3")
    repo.schema.init()
    record = repo.messages.create_message(media())
    repo.processing.mark_transcribed(record.id, "Привет, это тест.", tmp_path / "raw.json")
    updated = repo.messages.get_message(record.id)
    assert updated is not None
    assert updated.status == ProcessingStatus.TRANSCRIBED.value
    assert updated.transcript_text == "Привет, это тест."
    assert updated.transcript_sha256 == text_sha256("Привет, это тест.")
    assert updated.transcript_file_path is None
    assert updated.provider_raw_response_path == str(tmp_path / "raw.json")


def test_analysis_run_is_saved(tmp_path: Path) -> None:
    repo = Repositories(tmp_path / "bot.sqlite3")
    repo.schema.init()
    record = repo.messages.create_message(media())
    run = repo.analysis.create(
        message_id=record.id,
        provider="gigachat",
        model="GigaChat",
        prompt_version="feedback_general_quality_v4",
        input_text="Текст обратной связи.",
    )

    repo.analysis.complete(
        run.id,
        '{"course_score_10": 8}',
        tmp_path / "gigachat.json",
        prompt_tokens=120,
        completion_tokens=30,
        total_tokens=150,
    )

    updated = repo.analysis.by_id(run.id)
    message = repo.messages.get_message(record.id)
    assert updated is not None
    assert updated.status == AnalysisStatus.ANALYZED.value
    assert updated.input_text is None
    assert updated.input_text_sha256 == text_sha256("Текст обратной связи.")
    assert updated.output_text == '{"course_score_10": 8}'
    assert updated.raw_response_path == str(tmp_path / "gigachat.json")
    assert (updated.prompt_tokens, updated.completion_tokens, updated.total_tokens) == (120, 30, 150)
    assert message is not None
    assert message.status == ProcessingStatus.RECEIVED.value


def test_completed_framework_analysis_can_be_reused(tmp_path: Path) -> None:
    repo = Repositories(tmp_path / "bot.sqlite3")
    repo.schema.init()
    record = repo.messages.create_message(media())
    run = repo.analysis.create(
        message_id=record.id,
        provider="gigachat",
        model="GigaChat",
        framework="COIN",
        prompt_version="feedback_framework_quality_v6",
        input_text="Текст обратной связи.",
    )
    repo.analysis.complete(run.id, '{"framework_name": "COIN"}', tmp_path / "coin.json")

    cached = repo.analysis.latest_completed(record.id, "COIN", "feedback_framework_quality_v6")

    assert cached is not None
    assert cached.id == run.id
    assert cached.output_text == '{"framework_name": "COIN"}'


def test_completed_quality_analysis_can_be_reused(tmp_path: Path) -> None:
    repo = Repositories(tmp_path / "bot.sqlite3")
    repo.schema.init()
    record = repo.messages.create_message(media())
    run = repo.analysis.create(
        message_id=record.id,
        provider="gigachat",
        model="GigaChat",
        prompt_version="feedback_general_quality_v4",
        input_text="Текст обратной связи.",
    )
    repo.analysis.complete(run.id, '{"course_score_10": 8}', tmp_path / "quality.json")

    cached = repo.analysis.latest_completed(record.id, None, "feedback_general_quality_v4")

    assert cached is not None
    assert cached.id == run.id
    assert cached.output_text == '{"course_score_10": 8}'


def test_completed_analysis_can_be_reused_for_same_transcript(tmp_path: Path) -> None:
    repo = Repositories(tmp_path / "bot.sqlite3")
    repo.schema.init()
    first = repo.messages.create_message(media())
    repo.processing.mark_transcribed(first.id, "Текст обратной связи.", None)
    run = repo.analysis.create(
        message_id=first.id,
        provider="gigachat",
        model="GigaChat",
        prompt_version="feedback_general_quality_v4",
        input_text="Текст обратной связи.",
    )
    repo.analysis.complete(run.id, '{"course_score_10": 8}', tmp_path / "quality.json")
    repeated = repo.messages.create_message(
        media().model_copy(
            update={
                "telegram_message_id": 8,
                "telegram_file_id": "file-id-2",
                "telegram_file_unique_id": "unique-id-2",
            }
        )
    )
    repo.processing.mark_transcribed(repeated.id, "Текст обратной связи.", None)
    other_user = repo.messages.create_message(
        media().model_copy(
            update={
                "telegram_user_id": 43,
                "telegram_message_id": 9,
                "telegram_file_id": "file-id-3",
                "telegram_file_unique_id": "unique-id-3",
            }
        )
    )
    repo.processing.mark_transcribed(other_user.id, "Текст обратной связи.", None)

    cached = repo.analysis.latest_completed_for_transcript(repeated.id, None, "feedback_general_quality_v4")
    hidden_from_other_user = repo.analysis.latest_completed_for_transcript(
        other_user.id, None, "feedback_general_quality_v4"
    )

    assert cached is not None
    assert cached.id == run.id
    assert hidden_from_other_user is None


def test_completed_analysis_cache_can_be_cleared(tmp_path: Path) -> None:
    repo = Repositories(tmp_path / "bot.sqlite3")
    repo.schema.init()
    record = repo.messages.create_message(media())
    run = repo.analysis.create(
        message_id=record.id,
        provider="gigachat",
        model="GigaChat",
        prompt_version="feedback_general_quality_v4",
        input_text="Текст обратной связи.",
    )
    repo.analysis.complete(run.id, '{"course_score_10": 8}', tmp_path / "quality.json")

    deleted_count = repo.analysis.clear_completed(record.id)
    cached = repo.analysis.latest_completed(record.id, None, "feedback_general_quality_v4")

    assert deleted_count == 1
    assert cached is None


def test_completed_analysis_memory_can_be_cleared_for_same_transcript(tmp_path: Path) -> None:
    repo = Repositories(tmp_path / "bot.sqlite3")
    repo.schema.init()
    first = repo.messages.create_message(media())
    repo.processing.mark_transcribed(first.id, "Одна и та же ОС.", None)
    first_run = repo.analysis.create(
        message_id=first.id,
        provider="gigachat",
        model="GigaChat",
        prompt_version="feedback_general_quality_v4",
        input_text="Одна и та же ОС.",
    )
    repo.analysis.complete(first_run.id, '{"course_score_10": 8}', None)

    repeated = repo.messages.create_message(
        media().model_copy(
            update={
                "telegram_message_id": 8,
                "telegram_file_id": "file-id-2",
                "telegram_file_unique_id": "unique-id-2",
            }
        )
    )
    repo.processing.mark_transcribed(repeated.id, "Одна и та же ОС.", None)
    repeated_run = repo.analysis.create(
        message_id=repeated.id,
        provider="gigachat",
        model="GigaChat",
        framework="COIN",
        prompt_version="feedback_framework_quality_v6",
        input_text="Одна и та же ОС.",
    )
    repo.analysis.complete(repeated_run.id, '{"framework_name": "COIN"}', None)

    different_text = repo.messages.create_message(
        media().model_copy(
            update={
                "telegram_message_id": 9,
                "telegram_file_id": "file-id-3",
                "telegram_file_unique_id": "unique-id-3",
            }
        )
    )
    repo.processing.mark_transcribed(different_text.id, "Другая ОС.", None)
    different_run = repo.analysis.create(
        message_id=different_text.id,
        provider="gigachat",
        model="GigaChat",
        prompt_version="feedback_general_quality_v4",
        input_text="Другая ОС.",
    )
    repo.analysis.complete(different_run.id, '{"course_score_10": 5}', None)

    other_user = repo.messages.create_message(
        media().model_copy(
            update={
                "telegram_user_id": 43,
                "telegram_message_id": 10,
                "telegram_file_id": "file-id-4",
                "telegram_file_unique_id": "unique-id-4",
            }
        )
    )
    repo.processing.mark_transcribed(other_user.id, "Одна и та же ОС.", None)
    other_user_run = repo.analysis.create(
        message_id=other_user.id,
        provider="gigachat",
        model="GigaChat",
        prompt_version="feedback_general_quality_v4",
        input_text="Одна и та же ОС.",
    )
    repo.analysis.complete(other_user_run.id, '{"course_score_10": 7}', None)

    deleted_count = repo.analysis.clear_completed(repeated.id, same_transcript=True)

    assert deleted_count == 2
    assert repo.analysis.by_id(first_run.id) is None
    assert repo.analysis.by_id(repeated_run.id) is None
    assert repo.analysis.by_id(different_run.id) is not None
    assert repo.analysis.by_id(other_user_run.id) is not None


def test_text_message_record_is_created(tmp_path: Path) -> None:
    repo = Repositories(tmp_path / "bot.sqlite3")
    repo.schema.init()
    record = repo.messages.create_text_message(
        telegram_user_id=42,
        telegram_chat_id=100,
        telegram_message_id=8,
        text="Можно дам обратную связь?",
    )

    assert record.id > 0
    assert record.media_type == "text"
    assert record.status == ProcessingStatus.TRANSCRIBED.value
    assert record.transcript_text == "Можно дам обратную связь?"
    assert record.transcript_sha256 == text_sha256("Можно дам обратную связь?")
    assert repo.messages.messages_by_transcript_hash(record.transcript_sha256) == [record]
    assert record.telegram_file_id == "text:100:8"


def test_free_training_is_available_without_stream_and_consumed_by_next_message(
    tmp_path: Path,
) -> None:
    repo = Repositories(tmp_path / "bot.sqlite3")
    repo.schema.init()
    user = repo.identities.upsert_user(42, "external", "External", "")

    assert repo.input_intents.start_free_training(user.telegram_user_id) is True
    assert repo.input_intents.free_training_is_pending(user.telegram_user_id) is True
    record = repo.messages.create_text_message(user.telegram_user_id, 100, 9, "Свободная ОС")

    assert record.participant_assignment_id is None
    assert repo.input_intents.free_training_is_pending(user.telegram_user_id) is False


def test_default_stream_is_created(tmp_path: Path) -> None:
    repo = Repositories(tmp_path / "bot.sqlite3")
    repo.schema.init()
    stream = repo.course_catalog.default_stream()
    assert stream.title == "Основной поток"
    assert stream.slug == "default"


def test_course_version_contains_ordered_topics_and_publications(tmp_path: Path) -> None:
    repo = Repositories(tmp_path / "bot.sqlite3")
    repo.schema.init()
    course = repo.course_catalog.create_course("Обратная связь", "Асинхронный тренинг")
    version = repo.course_catalog.create_course_version(course.id)
    boff = repo.course_catalog.add_course_topic(version.id, "1. BOFF", position=1)
    onboarding = repo.course_catalog.add_course_topic(version.id, "0. Онбординг", position=0)

    material = repo.course_catalog.add_course_publication(
        onboarding.id,
        position=0,
        publication_type="Изучить",
        content_text="Признаки хорошей обратной связи",
        publish_offset_minutes=9 * 60,
    )
    assignment = repo.course_catalog.add_course_publication(
        onboarding.id,
        position=1,
        publication_type="Задание",
        content_text="Запишите обратную связь по кейсу",
        publish_offset_minutes=21 * 60 + 5,
        deadline_offset_minutes=24 * 60 + 21 * 60,
        response_format="audio",
    )

    assert course.title == "Обратная связь"
    assert version.version_number == 1
    assert version.status == CourseVersionStatus.DRAFT.value
    assert repo.course_catalog.course_topics(version.id) == [onboarding, boff]
    assert repo.course_catalog.course_publications(onboarding.id) == [material, assignment]
    assert material.is_assignment is False
    assert assignment.is_assignment is True
    assert assignment.publish_offset_minutes == 1265
    assert assignment.deadline_offset_minutes == 2700


def test_new_course_versions_get_sequential_numbers(tmp_path: Path) -> None:
    repo = Repositories(tmp_path / "bot.sqlite3")
    repo.schema.init()
    course = repo.course_catalog.create_course("Обратная связь")

    first = repo.course_catalog.create_course_version(course.id)
    second = repo.course_catalog.create_course_version(course.id)

    assert first.version_number == 1
    assert second.version_number == 2


def test_assignment_requires_future_deadline_and_response_format(tmp_path: Path) -> None:
    repo = Repositories(tmp_path / "bot.sqlite3")
    repo.schema.init()
    course = repo.course_catalog.create_course("Обратная связь")
    version = repo.course_catalog.create_course_version(course.id)
    topic = repo.course_catalog.add_course_topic(version.id, "Онбординг", position=0)

    try:
        repo.course_catalog.add_course_publication(
            topic.id,
            position=0,
            publication_type="Задание",
            content_text="Запишите ответ",
            publish_offset_minutes=600,
            deadline_offset_minutes=600,
            response_format="audio",
        )
    except ValueError as error:
        assert str(error) == "Deadline must be after publication"
    else:
        raise AssertionError("Deadline equal to publication time must be rejected")

    try:
        repo.course_catalog.add_course_publication(
            topic.id,
            position=0,
            publication_type="Задание",
            content_text="Запишите ответ",
            publish_offset_minutes=600,
            deadline_offset_minutes=1200,
        )
    except ValueError as error:
        assert str(error) == "Assignment response format is required"
    else:
        raise AssertionError("Assignment without response format must be rejected")

    try:
        repo.course_catalog.add_course_publication(
            topic.id,
            position=0,
            publication_type="Задание",
            content_text="Запишите ответ",
            publish_offset_minutes=600,
            deadline_offset_minutes=1200,
            response_format="text",
            crowd_training_enabled=True,
            required_reviews=1,
            assigned_reviews=21,
        )
    except ValueError as error:
        assert str(error) == "Assigned reviews must not exceed 20"
    else:
        raise AssertionError("An unreachable peer-review keyboard must be rejected")


def test_published_course_version_cannot_be_edited(tmp_path: Path) -> None:
    repo = Repositories(tmp_path / "bot.sqlite3")
    repo.schema.init()
    course = repo.course_catalog.create_course("Обратная связь")
    version = repo.course_catalog.create_course_version(course.id)
    topic = repo.course_catalog.add_course_topic(version.id, "Онбординг", position=0)
    repo.course_catalog.add_course_publication(
        topic.id,
        position=0,
        publication_type="Изучить",
        content_text="Вводный материал",
        publish_offset_minutes=540,
    )

    published = repo.course_catalog.publish_course_version(version.id)

    assert published.status == CourseVersionStatus.PUBLISHED.value
    assert published.published_at is not None
    try:
        repo.course_catalog.add_course_topic(version.id, "BOFF", position=1)
    except ValueError as error:
        assert str(error) == "Published course version cannot be edited"
    else:
        raise AssertionError("Published course version must be immutable")


def test_published_course_version_expands_into_stream_calendar_and_assignments(
    tmp_path: Path,
) -> None:
    clock = frozen_clock(zone=timezone(timedelta(hours=3)))
    repo = Repositories(tmp_path / "bot.sqlite3", clock=clock)
    repo.schema.init()
    course = repo.course_catalog.create_course("Обратная связь")
    version = repo.course_catalog.create_course_version(course.id)
    topic = repo.course_catalog.add_course_topic(version.id, "Онбординг", position=0)
    repo.course_catalog.add_course_publication(
        topic.id,
        position=0,
        publication_type="Изучить",
        content_text="Вводный материал",
        publish_offset_minutes=540,
    )
    repo.course_catalog.add_course_publication(
        topic.id,
        position=1,
        publication_type="Задание",
        content_text="Запишите ОС",
        publish_offset_minutes=1265,
        deadline_offset_minutes=2700,
        response_format="video_note",
    )
    published = repo.course_catalog.publish_course_version(version.id)
    stream = repo.course_catalog.default_stream()
    repo.identities.upsert_user(42, "learner", "", "")
    learner = repo.access_commands.add_user_by_username("@learner", stream.id)

    start_date = date_at(clock)
    configured = repo.course_streams.configure_stream_course(
        stream.id,
        published.id,
        start_date,
        "Europe/Moscow",
    )
    publications = repo.course_streams.stream_publications(stream.id)
    assignments = repo.learner_workflow.assignments(learner.telegram_user_id, stream.id)

    assert configured.course_version_id == published.id
    assert configured.status == "active"
    assert configured.starts_at == iso_at(clock, hours=-15)
    assert configured.ends_at == iso_at(clock, hours=30)
    assert [publication.scheduled_at for publication in publications] == [
        iso_at(clock, hours=-6),
        iso_at(clock, hours=6, minutes=5),
    ]
    assert publications[1].deadline_at == iso_at(clock, hours=30)
    assert len(assignments) == 1
    assert assignments[0]["status"] == "not_started"
    reminders = repo.scheduler_queries.delivery.due_assignment_reminders("9999-12-31T23:59:59+00:00")
    assert [row.scheduled_at for row in reminders] == [
        iso_at(clock, hours=18),
        iso_at(clock, hours=25, minutes=10),
    ]

    repeated = repo.course_streams.configure_stream_course(
        stream.id,
        published.id,
        start_date,
        "Europe/Moscow",
    )
    assert repeated == configured
    assert len(repo.course_streams.stream_publications(stream.id)) == 2
    assert len(repo.learner_workflow.assignments(learner.telegram_user_id, stream.id)) == 1


def test_learner_approved_after_course_start_gets_assignments(tmp_path: Path) -> None:
    clock = frozen_clock()
    repo = Repositories(tmp_path / "bot.sqlite3", clock=clock)
    repo.schema.init()
    course = repo.course_catalog.create_course("Обратная связь")
    version = repo.course_catalog.create_course_version(course.id)
    topic = repo.course_catalog.add_course_topic(version.id, "Онбординг", position=0)
    repo.course_catalog.add_course_publication(
        topic.id,
        position=0,
        publication_type="Задание",
        content_text="Запишите ОС",
        publish_offset_minutes=0,
        deadline_offset_minutes=1440,
        response_format="video_note",
    )
    published = repo.course_catalog.publish_course_version(version.id)
    stream = repo.course_catalog.default_stream()
    repo.course_streams.configure_stream_course(stream.id, published.id, date_at(clock, days=-1))
    user = repo.identities.upsert_user(42, "learner", "Learner", "")

    repo.access_commands.approve_membership(stream.id, user.telegram_user_id, decided_by_user_id=None)

    assert len(repo.learner_workflow.assignments(user.telegram_user_id, stream.id)) == 1


def test_selected_assignment_is_linked_to_next_message_only(tmp_path: Path) -> None:
    clock = frozen_clock()
    repo = Repositories(tmp_path / "bot.sqlite3", clock=clock)
    repo.schema.init()
    course = repo.course_catalog.create_course("Обратная связь")
    version = repo.course_catalog.create_course_version(course.id)
    topic = repo.course_catalog.add_course_topic(version.id, "Онбординг", position=0)
    repo.course_catalog.add_course_publication(
        topic.id,
        position=0,
        publication_type="Задание",
        content_text="Запишите ОС",
        publish_offset_minutes=0,
        deadline_offset_minutes=1440,
        response_format="text",
    )
    published = repo.course_catalog.publish_course_version(version.id)
    stream = repo.course_catalog.default_stream()
    user = repo.identities.upsert_user(42, "learner", "Learner", "")
    repo.access_commands.approve_membership(stream.id, user.telegram_user_id, decided_by_user_id=None)
    repo.course_streams.configure_stream_course(stream.id, published.id, date_at(clock))
    assignment = repo.learner_workflow.assignments(user.telegram_user_id, stream.id)[0]

    assert repo.submissions.set_current_assignment(user.telegram_user_id, assignment["id"]) is True
    assert repo.input_intents.current_submission_response_format(user.telegram_user_id) == "text"
    submitted = repo.messages.create_text_message(user.telegram_user_id, 100, 1, "Моя ОС")
    free_practice = repo.messages.create_text_message(user.telegram_user_id, 100, 2, "Свободная практика")
    updated_assignment = repo.learner_workflow.assignments(user.telegram_user_id, stream.id)[0]

    assert submitted.participant_assignment_id == assignment["id"]
    assert free_practice.participant_assignment_id is None
    assert updated_assignment["status"] == "submitted"
    assert updated_assignment["submitted_at"] is not None


def test_submitted_assignment_can_only_be_reselected_before_its_deadline(tmp_path: Path) -> None:
    clock = frozen_clock()
    repo = Repositories(tmp_path / "bot.sqlite3", clock=clock)
    repo.schema.init()
    course = repo.course_catalog.create_course("Обратная связь")
    version = repo.course_catalog.create_course_version(course.id)
    topic = repo.course_catalog.add_course_topic(version.id, "Онбординг", position=0)
    repo.course_catalog.add_course_publication(
        topic.id,
        position=0,
        publication_type="Задание",
        content_text="Запишите ОС",
        publish_offset_minutes=0,
        deadline_offset_minutes=1,
        response_format="text",
        crowd_training_enabled=True,
        required_reviews=1,
        assigned_reviews=1,
    )
    published = repo.course_catalog.publish_course_version(version.id)
    stream = repo.course_catalog.default_stream()
    user = repo.identities.upsert_user(42, "learner", "Learner", "")
    repo.access_commands.approve_membership(stream.id, user.telegram_user_id, decided_by_user_id=None)
    repo.course_streams.configure_stream_course(stream.id, published.id, date_at(clock, days=-1))
    assignment = repo.learner_workflow.assignments(user.telegram_user_id, stream.id)[0]
    with repo.database.connect() as db:
        db.execute("UPDATE participant_assignments SET status='submitted' WHERE id=?", (assignment["id"],))
        db.execute(
            "UPDATE stream_publications SET deadline_at=? WHERE id=?",
            (iso_at(clock, days=1), assignment["stream_publication_id"]),
        )

    assert repo.submissions.set_current_assignment(user.telegram_user_id, assignment["id"]) is True

    with repo.database.connect() as db:
        db.execute(
            "UPDATE stream_publications SET deadline_at=? WHERE id=?",
            (iso_at(clock, days=-1), assignment["stream_publication_id"]),
        )

    assert repo.input_intents.input_mode(user.telegram_user_id) is None
    assert repo.submissions.set_current_assignment(user.telegram_user_id, assignment["id"]) is False


def test_assignment_cannot_be_selected_before_publication(tmp_path: Path) -> None:
    clock = frozen_clock()
    repo = Repositories(tmp_path / "bot.sqlite3", clock=clock)
    repo.schema.init()
    course = repo.course_catalog.create_course("Обратная связь")
    version = repo.course_catalog.create_course_version(course.id)
    topic = repo.course_catalog.add_course_topic(version.id, "Онбординг", position=0)
    repo.course_catalog.add_course_publication(
        topic.id,
        position=0,
        publication_type="Задание",
        content_text="Будущее задание",
        publish_offset_minutes=60,
        deadline_offset_minutes=120,
        response_format="text",
    )
    published = repo.course_catalog.publish_course_version(version.id)
    stream = repo.course_catalog.default_stream()
    user = repo.identities.upsert_user(42, "learner", "Learner", "")
    repo.access_commands.approve_membership(stream.id, user.telegram_user_id, decided_by_user_id=None)
    repo.course_streams.configure_stream_course(stream.id, published.id, date_at(clock, days=1))
    assignment = repo.learner_workflow.assignments(user.telegram_user_id, stream.id)[0]

    assert repo.submissions.set_current_assignment(user.telegram_user_id, assignment["id"]) is False


def test_assignment_write_callbacks_stay_inside_selected_stream(tmp_path: Path) -> None:
    clock = frozen_clock()
    repo = Repositories(tmp_path / "bot.sqlite3", clock=clock)
    repo.schema.init()
    course = repo.course_catalog.create_course("Обратная связь")
    version = repo.course_catalog.create_course_version(course.id)
    topic = repo.course_catalog.add_course_topic(version.id, "Онбординг", position=0)
    repo.course_catalog.add_course_publication(
        topic.id,
        position=0,
        publication_type="Задание",
        content_text="Запишите ОС",
        publish_offset_minutes=0,
        deadline_offset_minutes=60,
        response_format="text",
    )
    published = repo.course_catalog.publish_course_version(version.id)
    first_stream = repo.course_catalog.default_stream()
    second_stream = repo.identities.create_stream("Второй поток")
    user = repo.identities.upsert_user(42, "learner", "Learner", "")
    repo.access_commands.add_user_by_username("@learner", first_stream.id)
    repo.access_commands.add_user_by_username("@learner", second_stream.id)
    start_date = date_at(clock, days=-1)
    repo.course_streams.configure_stream_course(first_stream.id, published.id, start_date)
    repo.course_streams.configure_stream_course(second_stream.id, published.id, start_date)
    with repo.database.connect() as db:
        db.execute(
            "UPDATE stream_publications SET scheduled_at=?, deadline_at=?",
            (iso_at(clock, days=-1), iso_at(clock, days=1)),
        )
    first_assignment = repo.learner_workflow.assignments(user.telegram_user_id, first_stream.id)[0]
    second_assignment = repo.learner_workflow.assignments(user.telegram_user_id, second_stream.id)[0]

    assert repo.access_commands.set_current_stream(user.telegram_user_id, second_stream.id)
    assert not repo.submissions.set_current_assignment(user.telegram_user_id, int(first_assignment["id"]))
    assert repo.access_commands.current_stream_for_user(user.telegram_user_id).id == second_stream.id
    assert repo.input_intents.input_mode(user.telegram_user_id) is None
    assert repo.submissions.set_current_assignment(user.telegram_user_id, int(second_assignment["id"]))
    assert repo.access_commands.current_stream_for_user(user.telegram_user_id).id == second_stream.id
    assert repo.input_intents.input_mode(user.telegram_user_id) == "course"

    assert repo.access_commands.set_current_stream(user.telegram_user_id, first_stream.id)
    assert repo.submissions.set_current_assignment(user.telegram_user_id, int(first_assignment["id"]))
    assert repo.access_commands.current_stream_for_user(user.telegram_user_id).id == first_stream.id
    assert repo.input_intents.input_mode(user.telegram_user_id) == "course"


def test_stream_request_can_be_approved(tmp_path: Path) -> None:
    repo = Repositories(tmp_path / "bot.sqlite3")
    repo.schema.init()
    user = repo.identities.upsert_user(42, "SergeArt", "Serge", "Art")
    stream = repo.course_catalog.default_stream()

    membership, should_notify = repo.access_commands.request_stream_access(stream.id, user.telegram_user_id)
    assert should_notify is True
    assert membership.status == MembershipStatus.PENDING.value

    approved = repo.access_commands.approve_membership(stream.id, user.telegram_user_id, decided_by_user_id=None)
    assert approved is not None
    assert approved.status == MembershipStatus.APPROVED.value
    assert repo.access_commands.current_stream_for_user(user.telegram_user_id) == stream


def test_added_username_access_moves_to_real_user_id(tmp_path: Path) -> None:
    repo = Repositories(tmp_path / "bot.sqlite3")
    repo.schema.init()
    stream = repo.course_catalog.default_stream()

    placeholder = repo.access_commands.add_user_by_username("@SergeArt", stream.id)
    assert placeholder.telegram_user_id < 0

    sync = IdentityClaimService(repo).sync_profile(TelegramProfile(42, "sergeart", "Serge", "Art"))
    user = sync.user
    assert sync.claim_result == "claimed"
    assert user.telegram_user_id == 42
    assert user.identity_kind == "telegram"
    assert repo.access_commands.membership(stream.id, 42).status == MembershipStatus.APPROVED.value  # type: ignore[union-attr]
    assert repo.access_commands.membership(stream.id, placeholder.telegram_user_id) is None
    assert repo.identities.claim_placeholder(placeholder.telegram_user_id, user.telegram_user_id) == "claimed"
    with repo.database.connect() as db:
        audit = db.execute(
            """
            SELECT result FROM audit_events
            WHERE event_type='identity_claim' AND source_user_id=? AND target_user_id=?
            """,
            (placeholder.telegram_user_id, user.telegram_user_id),
        ).fetchone()
    assert audit is not None
    assert audit["result"] == "claimed"


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

    first = repo.identities.upsert_user(41, "shared_name", "First", "User")
    assert repo.user_access.set_admin(first.telegram_user_id, True)

    second = IdentityClaimService(repo).sync_profile(TelegramProfile(42, "SHARED_NAME", "Second", "User")).user

    stored_first = repo.user_access.user(first.telegram_user_id)
    stored_second = repo.user_access.user(second.telegram_user_id)
    assert stored_first is not None
    assert stored_second is not None
    assert stored_first.is_admin == 1
    assert stored_second.is_admin == 0


def test_placeholder_claim_never_grants_username_based_admin_rights(tmp_path: Path) -> None:
    repo = Repositories(tmp_path / "bot.sqlite3")
    repo.schema.init()
    placeholder = repo.access_commands.add_user_by_username("@invited_admin", repo.course_catalog.default_stream().id)
    assert repo.user_access.set_admin(placeholder.telegram_user_id, True)

    real_user = IdentityClaimService(repo).sync_profile(TelegramProfile(42, "invited_admin", "Real", "User")).user

    assert repo.user_access.user(placeholder.telegram_user_id) is None
    assert repo.user_access.user(real_user.telegram_user_id).is_admin == 0  # type: ignore[union-attr]


def test_invalid_positive_identity_claim_is_refused_without_admin_transfer(tmp_path: Path) -> None:
    repo = Repositories(tmp_path / "bot.sqlite3")
    repo.schema.init()
    source = repo.identities.upsert_user(41, "source", "Source", "")
    target = repo.identities.upsert_user(42, "target", "Target", "")
    assert repo.user_access.set_admin(source.telegram_user_id, True)

    assert repo.identities.claim_placeholder(source.telegram_user_id, target.telegram_user_id) == "refused"

    assert repo.user_access.user(source.telegram_user_id).is_admin == 1  # type: ignore[union-attr]
    assert repo.user_access.user(target.telegram_user_id).is_admin == 0  # type: ignore[union-attr]


def test_username_change_does_not_transfer_history_to_new_owner(tmp_path: Path) -> None:
    repo = Repositories(tmp_path / "bot.sqlite3")
    repo.schema.init()
    original_owner = repo.identities.upsert_user(41, "old_name", "Original", "Owner")
    assert repo.user_access.set_admin(original_owner.telegram_user_id, True)

    identity = IdentityClaimService(repo)
    renamed_owner = identity.sync_profile(TelegramProfile(41, "new_name", "Original", "Owner")).user
    new_owner = identity.sync_profile(TelegramProfile(42, "old_name", "New", "Owner")).user

    assert renamed_owner.telegram_user_id == original_owner.telegram_user_id
    assert renamed_owner.username == "new_name"
    assert repo.user_access.user(original_owner.telegram_user_id).is_admin == 1  # type: ignore[union-attr]
    assert repo.user_access.user(new_owner.telegram_user_id).is_admin == 0  # type: ignore[union-attr]


def test_seed_user_is_never_claimed_as_a_username_placeholder(tmp_path: Path) -> None:
    repo = Repositories(tmp_path / "bot.sqlite3")
    repo.schema.init()
    stream = repo.course_catalog.default_stream()
    seed_user = repo.access_commands.add_user_by_username("@seed_learner", stream.id)
    assert seed_user.telegram_user_id < 0
    assert seed_user.identity_kind == "placeholder"

    real_user = IdentityClaimService(repo).sync_profile(TelegramProfile(42, "seed_learner", "Real", "User")).user

    assert repo.user_access.user(seed_user.telegram_user_id) is not None
    assert repo.user_access.user(real_user.telegram_user_id) is not None
    assert repo.access_commands.membership(stream.id, seed_user.telegram_user_id) is not None
    assert repo.access_commands.membership(stream.id, real_user.telegram_user_id) is None


def test_placeholder_claim_conflict_keeps_both_identities(tmp_path: Path) -> None:
    clock = frozen_clock()
    repo = Repositories(tmp_path / "bot.sqlite3", clock=clock)
    repo.schema.init()
    course = repo.course_catalog.create_course("Identity conflict")
    version = repo.course_catalog.create_course_version(course.id)
    topic = repo.course_catalog.add_course_topic(version.id, "Topic", position=0)
    repo.course_catalog.add_course_publication(
        topic.id,
        position=0,
        publication_type="Задание",
        content_text="Task",
        publish_offset_minutes=0,
        deadline_offset_minutes=1440,
        response_format="text",
    )
    published = repo.course_catalog.publish_course_version(version.id)
    stream = repo.course_catalog.default_stream()
    repo.course_streams.configure_stream_course(stream.id, published.id, date_at(clock))

    placeholder = repo.access_commands.add_user_by_username("@claim_conflict", stream.id)
    target = repo.identities.upsert_user(42, "other_name", "Real", "User")
    repo.access_commands.approve_membership(stream.id, target.telegram_user_id, decided_by_user_id=None)

    updated_target = IdentityClaimService(repo).sync_profile(TelegramProfile(42, "claim_conflict", "Real", "User")).user

    assert repo.user_access.user(placeholder.telegram_user_id) is not None
    assert repo.user_access.user(updated_target.telegram_user_id) is not None
    assert len(repo.learner_workflow.assignments(placeholder.telegram_user_id, stream.id)) == 1
    assert len(repo.learner_workflow.assignments(updated_target.telegram_user_id, stream.id)) == 1
    with repo.database.connect() as db:
        audit = db.execute(
            """
            SELECT result, details_json FROM audit_events
            WHERE event_type='identity_claim' AND source_user_id=? AND target_user_id=?
            ORDER BY id DESC LIMIT 1
            """,
            (placeholder.telegram_user_id, updated_target.telegram_user_id),
        ).fetchone()
    assert audit is not None
    assert audit["result"] == "manual_resolution_required"
    assert "participant_assignment" in audit["details_json"]


def test_placeholder_claim_merges_matching_membership_roles_without_duplicates(
    tmp_path: Path,
) -> None:
    repo = Repositories(tmp_path / "bot.sqlite3")
    repo.schema.init()
    stream = repo.course_catalog.default_stream()
    placeholder = repo.access_commands.add_user_by_username("@role_claim", stream.id)
    target = repo.identities.upsert_user(42, "before_claim", "Real", "User")
    repo.access_commands.approve_membership(stream.id, target.telegram_user_id, decided_by_user_id=None)
    repo.access_commands.set_stream_role(stream.id, target.telegram_user_id, "curator")

    IdentityClaimService(repo).sync_profile(TelegramProfile(target.telegram_user_id, "role_claim", "Real", "User"))

    assert repo.access_commands.membership(stream.id, placeholder.telegram_user_id) is None
    merged = repo.access_commands.membership(stream.id, target.telegram_user_id)
    assert merged is not None
    assert merged.status == MembershipStatus.APPROVED.value
    with repo.database.connect() as db:
        roles = db.execute(
            """
            SELECT role FROM stream_membership_roles
            WHERE membership_id=? ORDER BY role
            """,
            (merged.id,),
        ).fetchall()
    assert [row["role"] for row in roles] == ["curator", "learner"]


def test_remove_user_from_stream_preserves_user_and_other_memberships(tmp_path: Path) -> None:
    repo = Repositories(tmp_path / "bot.sqlite3")
    repo.schema.init()
    first_stream = repo.course_catalog.default_stream()
    second_stream = repo.identities.create_stream("Второй поток")
    user = repo.access_commands.add_user_by_username("@SergeArt", first_stream.id)
    repo.access_commands.add_user_by_username("@SergeArt", second_stream.id)

    assert repo.access_commands.remove_user_from_stream_by_username("sergeart", first_stream.id) is True
    assert repo.user_access.user(user.telegram_user_id) is not None
    assert repo.access_commands.membership(first_stream.id, user.telegram_user_id) is None
    assert repo.access_commands.membership(second_stream.id, user.telegram_user_id) is not None
    assert repo.access_commands.current_stream_for_user(user.telegram_user_id) == second_stream
    assert repo.access_commands.remove_user_from_stream_by_username("sergeart", first_stream.id) is False