tests/test_crowd_test_data.py - master - artyuhovs/AIFeedBackTrainingBot

tests/test_crowd_test_data.py
387 строк15 KB
Codex
Завершить архитектурное оздоровление и исправить итоги
15 июл 2026, 11:05
15 июл 2026, 11:05c65f9a6
import asyncio
from pathlib import Path
from types import SimpleNamespace

from app.bot.handlers import (
    admin_stream_keyboard,
    progress,
    send_course_overview,
    send_reviews,
    send_tasks,
    stream_callback,
)
from app.crowd_test_data import seed_crowd_test_data
from app.db.repositories import Repositories, text_sha256
from app.utils.time_utils import now_iso


class FakeMessage:
    def __init__(self, user_id: int) -> None:
        self.from_user = SimpleNamespace(id=user_id)
        self.answers: list[dict[str, object]] = []

    async def answer(self, text: str, **kwargs: object) -> None:
        self.answers.append({"text": text, "kwargs": kwargs})


class FakeStreamCallback:
    def __init__(self, user_id: int, stream_id: int) -> None:
        self.data = f"stream:{stream_id}"
        self.from_user = SimpleNamespace(id=user_id)
        self.message = FakeMessage(user_id)
        self.answers: list[dict[str, object]] = []

    async def answer(self, text: str, **kwargs: object) -> None:
        self.answers.append({"text": text, "kwargs": kwargs})


def message_text(message: FakeMessage) -> str:
    return "\n".join(str(answer["text"]) for answer in message.answers)


def test_seed_crowd_data_is_reusable_and_idempotent(tmp_path: Path) -> None:
    repository = Repositories(tmp_path / "bot.sqlite3")
    repository.schema.init()
    first = repository.identities.upsert_user(101, "sergeart", "Sergey", "")
    second = repository.identities.upsert_user(202, "annikart", "Anna", "")

    for _ in range(2):
        result = seed_crowd_test_data(
            repository,
            stream_title="тестпоток2",
            synthetic_count=6,
            works_per_real_user=4,
            reviews_per_real_user=5,
        )

    assert result.synthetic_users == 6
    stream = repository.identities.get_stream(result.stream_id)
    assert stream is not None
    assert stream.status == "draft"
    assert repository.learner_workflow.assignments(first.telegram_user_id, result.stream_id) == []
    assert repository.peer_review_task_query.load(first.telegram_user_id, "assigned") == []
    assert repository.scheduler_queries.delivery.due_collection_closed(now_iso()) == []
    hidden_messages = [FakeMessage(first.telegram_user_id) for _ in range(4)]
    asyncio.run(send_tasks(hidden_messages[0], repository, first.telegram_user_id))
    asyncio.run(send_reviews(hidden_messages[1], repository, first.telegram_user_id))
    asyncio.run(progress(hidden_messages[2], repository))
    asyncio.run(
        send_course_overview(
            hidden_messages[3],
            SimpleNamespace(),
            repository,
            first.telegram_user_id,
        )
    )
    assert all("не стартовал" in str(message.answers[-1]["text"]) for message in hidden_messages)
    keyboard = admin_stream_keyboard(repository, stream)
    assert keyboard is not None
    assert keyboard.inline_keyboard[0][0].text == "Стартовать поток"
    assert keyboard.inline_keyboard[0][0].callback_data == f"startstream:{result.stream_id}"

    started, changed = repository.course_streams.start_stream(result.stream_id)
    assert changed is True
    assert started.status == "active"
    assert repository.access_commands.current_stream_for_user(first.telegram_user_id).id == result.stream_id
    assert repository.scheduler_queries.delivery.due_collection_closed(now_iso())
    assert len(repository.peer_review_task_query.load(first.telegram_user_id, "assigned")) == 4
    assert len(repository.peer_review_task_query.load(second.telegram_user_id, "assigned")) == 4
    for learner in (first, second):
        assignment = repository.learner_workflow.assignments(learner.telegram_user_id, result.stream_id)[0]
        assert len(repository.review_results.peer_reviews_for_assignment(int(assignment["id"]))) == 5
    with repository.database.connect() as db:
        assert db.execute("SELECT COUNT(*) FROM bot_users WHERE username LIKE 'seed_%'").fetchone()[0] == 6


def test_selected_stream_scopes_commands_actions_and_notifications(tmp_path: Path) -> None:
    repository = Repositories(tmp_path / "bot.sqlite3")
    repository.schema.init()
    first = repository.identities.upsert_user(101, "sergeart", "Sergey", "")
    second = repository.identities.upsert_user(202, "annikart", "Anna", "")

    first_result = seed_crowd_test_data(
        repository,
        stream_title="тестпоток1",
        synthetic_count=3,
        works_per_real_user=2,
        reviews_per_real_user=2,
    )
    repository.course_streams.start_stream(first_result.stream_id)
    assert repository.access_commands.current_stream_for_user(first.telegram_user_id).id == first_result.stream_id

    second_result = seed_crowd_test_data(
        repository,
        stream_title="тестпоток2",
        synthetic_count=3,
        works_per_real_user=2,
        reviews_per_real_user=2,
    )
    repository.course_streams.start_stream(second_result.stream_id)

    # Preparing another test stream must not silently change an existing selection.
    assert repository.access_commands.current_stream_for_user(first.telegram_user_id).id == first_result.stream_id
    assert repository.access_commands.current_stream_for_user(second.telegram_user_id).id == first_result.stream_id

    markers = {
        first_result.stream_id: "Работа только потока 1",
        second_result.stream_id: "Работа только потока 2",
    }
    with repository.database.connect() as db:
        for stream_id, answer_marker in markers.items():
            db.execute(
                "UPDATE stream_publications SET scheduled_at='2020-01-01T00:00:00+00:00' WHERE stream_id=?",
                (stream_id,),
            )
            db.execute(
                """
                UPDATE messages
                SET transcript_text=?
                WHERE participant_assignment_id IN (
                    SELECT participant_assignments.id
                    FROM participant_assignments
                    JOIN stream_publications
                      ON stream_publications.id=participant_assignments.stream_publication_id
                    WHERE stream_publications.stream_id=?
                )
                """,
                (answer_marker, stream_id),
            )

    settings = SimpleNamespace(
        telegram_admin_ids=(),
        bot_admin_usernames=(),
        app_env="development",
        access_mode="closed",
    )

    def assert_selected_stream(
        expected_result: object,
        expected_title: str,
        forbidden_title: str,
        expected_marker: str,
        forbidden_marker: str,
    ) -> None:
        expected_stream_id = int(expected_result.stream_id)
        assignments = repository.learner_workflow.assignments(first.telegram_user_id)
        reviews = repository.peer_review_task_query.load(first.telegram_user_id)
        assert assignments
        assert reviews
        assert {int(row["stream_id"]) for row in assignments} == {expected_stream_id}
        assert {int(row["stream_id"]) for row in reviews} == {expected_stream_id}

        command_messages = [FakeMessage(first.telegram_user_id) for _ in range(4)]
        asyncio.run(send_tasks(command_messages[0], repository, first.telegram_user_id))
        asyncio.run(send_reviews(command_messages[1], repository, first.telegram_user_id))
        asyncio.run(progress(command_messages[2], repository))
        asyncio.run(
            send_course_overview(
                command_messages[3],
                SimpleNamespace(),
                repository,
                first.telegram_user_id,
            )
        )
        rendered = "\n".join(message_text(message) for message in command_messages)
        assert expected_title in message_text(command_messages[0])
        assert expected_title in message_text(command_messages[3])
        assert forbidden_title not in rendered
        assert expected_marker in rendered
        assert forbidden_marker not in rendered

        due_for_user = [
            row
            for row in repository.scheduler_queries.delivery.due_collection_closed(now_iso())
            if row.recipient_id == first.telegram_user_id
        ]
        assert due_for_user
        assert {row.stream_id for row in due_for_user} == {expected_stream_id}

        selected_review_ids = {int(row["id"]) for row in reviews}
        due_review_ids = {
            row.review_assignment_id
            for row in repository.scheduler_queries.reviews.due_peer_review_reminders("9999-12-31T23:59:59+00:00")
            if row.reviewer_id == first.telegram_user_id
        }
        assert due_review_ids
        assert due_review_ids <= selected_review_ids

    assert_selected_stream(
        first_result,
        "тестпоток1",
        "тестпоток2",
        "Работа только потока 1",
        "Работа только потока 2",
    )

    switch = FakeStreamCallback(first.telegram_user_id, second_result.stream_id)
    asyncio.run(stream_callback(switch, repository, settings))
    assert switch.answers[-1]["text"] == "Поток выбран."
    assert repository.access_commands.current_stream_for_user(first.telegram_user_id).id == second_result.stream_id

    assert_selected_stream(
        second_result,
        "тестпоток2",
        "тестпоток1",
        "Работа только потока 2",
        "Работа только потока 1",
    )


def test_quality_stage_opens_received_review_rating(tmp_path: Path) -> None:
    repository = Repositories(tmp_path / "bot.sqlite3")
    repository.schema.init()
    repository.identities.upsert_user(101, "sergeart", "Sergey", "")
    repository.identities.upsert_user(202, "annikart", "Anna", "")

    result = seed_crowd_test_data(
        repository,
        stream_title="тестпоток3",
        synthetic_count=3,
        works_per_real_user=2,
        reviews_per_real_user=3,
        stage="quality",
    )
    repository.course_streams.start_stream(result.stream_id)

    assignment = repository.learner_workflow.assignments(101, result.stream_id)[0]
    review = repository.review_results.peer_reviews_for_assignment(int(assignment["id"]))[0]
    assert repository.review_quality.start_peer_review_quality_draft(int(review["id"]), 101, 9.0) is not None


def test_rerun_returns_owned_seeded_stream_to_manual_start(tmp_path: Path) -> None:
    repository = Repositories(tmp_path / "bot.sqlite3")
    repository.schema.init()
    repository.identities.upsert_user(101, "sergeart", "Sergey", "")
    repository.identities.upsert_user(202, "annikart", "Anna", "")
    result = seed_crowd_test_data(
        repository,
        stream_title="тестпоток старой версии",
        synthetic_count=2,
        works_per_real_user=1,
        reviews_per_real_user=1,
    )
    repository.course_streams.start_stream(result.stream_id)
    seed_crowd_test_data(
        repository,
        stream_title="тестпоток старой версии",
        synthetic_count=2,
        works_per_real_user=1,
        reviews_per_real_user=1,
    )

    migrated = repository.identities.get_stream(result.stream_id)
    assert migrated is not None
    assert migrated.status == "draft"
    assert repository.learner_workflow.assignments(101, result.stream_id) == []


def test_seed_shrink_preserves_unowned_rows_and_newer_participant_state(tmp_path: Path) -> None:
    repository = Repositories(tmp_path / "bot.sqlite3")
    repository.schema.init()
    first = repository.identities.upsert_user(101, "sergeart", "Sergey", "")
    repository.identities.upsert_user(202, "annikart", "Anna", "")
    observer = repository.identities.upsert_user(303, "observer", "Observer", "")
    initial = seed_crowd_test_data(
        repository,
        stream_title="тест ownership",
        synthetic_count=5,
        works_per_real_user=3,
        reviews_per_real_user=3,
    )
    timestamp = repository.database.now_iso()
    with repository.database.connect() as db:
        assignment = db.execute(
            """
            SELECT id FROM participant_assignments
            WHERE stream_publication_id=? AND telegram_user_id=?
            """,
            (initial.publication_id, first.telegram_user_id),
        ).fetchone()
        assignment_id = int(assignment["id"])
        real_text = "Реальная работа, созданная после fixture seed"
        db.execute(
            """
            INSERT INTO messages
                (stream_id, participant_assignment_id, submission_kind, telegram_user_id,
                 telegram_chat_id, telegram_message_id, telegram_file_id, media_type,
                 mime_type, file_size_bytes, status, transcript_text, transcript_sha256,
                 created_at, updated_at, transcribed_at)
            VALUES (?, ?, 'original', ?, ?, 9999, 'real:9999', 'text', 'text/plain', ?,
                    'transcribed', ?, ?, ?, ?, ?)
            """,
            (
                initial.stream_id,
                assignment_id,
                first.telegram_user_id,
                first.telegram_user_id,
                len(real_text.encode()),
                real_text,
                text_sha256(real_text),
                timestamp,
                timestamp,
                timestamp,
            ),
        )
        db.execute(
            """
            UPDATE participant_assignments
            SET status='completed', submitted_at='real-submission',
                self_assessment_status='completed', updated_at='real-update'
            WHERE id=?
            """,
            (assignment_id,),
        )
        unowned_review = db.execute(
            """
            INSERT INTO peer_review_assignments
                (target_assignment_id, reviewer_user_id, status, assigned_at, deadline_at)
            VALUES (?, ?, 'assigned', ?, '2099-01-01T00:00:00+00:00')
            """,
            (assignment_id, observer.telegram_user_id, timestamp),
        )
        unowned_review_id = int(unowned_review.lastrowid)

    seed_crowd_test_data(
        repository,
        stream_title="тест ownership",
        synthetic_count=2,
        works_per_real_user=1,
        reviews_per_real_user=1,
    )

    with repository.database.connect() as db:
        state = db.execute(
            """
            SELECT status, submitted_at, self_assessment_status, updated_at
            FROM participant_assignments WHERE id=?
            """,
            (assignment_id,),
        ).fetchone()
        assert tuple(state) == (
            "completed",
            "real-submission",
            "completed",
            "real-update",
        )
        assert (
            db.execute("SELECT transcript_text FROM messages WHERE telegram_file_id='real:9999'").fetchone()[0]
            == real_text
        )
        assert (
            db.execute(
                "SELECT status FROM peer_review_assignments WHERE id=?",
                (unowned_review_id,),
            ).fetchone()[0]
            == "assigned"
        )
        assert db.execute("SELECT COUNT(*) FROM bot_users WHERE identity_kind='synthetic'").fetchone()[0] == 2
        audit = db.execute(
            """
            SELECT result, details_json FROM audit_events
            WHERE event_type='devtool_mutation'
              AND details_json LIKE '%\"tool\": \"crowd_test_data\"%'
            ORDER BY id DESC LIMIT 1
            """
        ).fetchone()
        assert audit is not None
        assert audit["result"] == "completed"