tests/test_tasks.py - master - artyuhovs/AIFeedBackTrainingBot

tests/test_tasks.py
300 строк11 KB
Codex
Выпустить версию 0.8.0
19 июл 2026, 18:59
19 июл 2026, 18:59b788979
from __future__ import annotations

import asyncio
from datetime import timedelta, timezone
from types import SimpleNamespace

from app.bot.handlers import (
    course_actions_keyboard,
    learning_overview_text,
    send_reviews,
    send_tasks,
    task_view_callback,
)
from app.bot.handlers import (
    progress as progress_handler,
)
from tests.time_support import frozen_clock, iso_at

CLOCK = frozen_clock(zone=timezone(timedelta(hours=3)))
SCHEDULED_AT = iso_at(CLOCK, days=-1, minutes=-30)
DEADLINE_AT = iso_at(CLOCK, minutes=-30)
BEFORE_DEADLINE = iso_at(CLOCK, minutes=-30, milliseconds=-1)
REVIEW_DEADLINE_AT = iso_at(CLOCK, hours=1, minutes=30)
REVIEW_QUALITY_DEADLINE_AT = iso_at(CLOCK, hours=3, minutes=30)
DEFAULT_AT = iso_at(CLOCK, hours=2)


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

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


class FakeLearnerWorkflow:
    def __init__(self, owner: FakeRepository) -> None:
        self.owner = owner

    def assignments(self, user_id: int, stream_id: int) -> list[dict[str, object]]:
        return self.owner.participant_assignments(user_id, stream_id)

    def assignment(
        self,
        user_id: int,
        _stream_id: int,
        assignment_id: int,
    ) -> dict[str, object] | None:
        return self.owner.participant_assignment(user_id, assignment_id)

    def review_tasks(self, user_id: int, stream_id: int) -> list[dict[str, object]]:
        return self.owner.peer_review_tasks(user_id, stream_id=stream_id)

    def received_reviews(self, assignment_ids: list[int]) -> list[dict[str, object]]:
        return self.owner.peer_reviews_for_assignments(assignment_ids)

    def written_review_feedback(
        self,
        user_id: int,
        publication_ids: list[int],
    ) -> list[dict[str, object]]:
        return self.owner.peer_review_feedback_for_reviewer_publications(
            user_id,
            publication_ids,
        )


class FakeAssignments:
    def __init__(self, owner: FakeRepository) -> None:
        self.owner = owner

    def latest_assignment_answer(
        self,
        user_id: int,
        assignment_id: int,
    ) -> dict[str, object]:
        return self.owner.latest_assignment_answer(user_id, assignment_id)


class FakeReviewResults:
    def __init__(self, owner: FakeRepository) -> None:
        self.owner = owner

    def peer_reviews_for_assignment(self, assignment_id: int) -> list[dict[str, object]]:
        return self.owner.peer_reviews_for_assignment(assignment_id)


class FakeRepository:
    def __init__(self, at: str = DEFAULT_AT) -> None:
        self.at = at
        self.database = self
        self.access_commands = self
        self.identities = self
        self.assignments = FakeAssignments(self)
        self.review_results = FakeReviewResults(self)
        self.learner_workflow = FakeLearnerWorkflow(self)
        self.learner_results = self

    def now_iso(self) -> str:
        return self.at

    def current_stream_for_user_role(self, _user_id: int, _role: object) -> SimpleNamespace:
        return SimpleNamespace(
            id=3,
            title="Тестовый поток",
            status="active",
            course_version_id=1,
        )

    def current_stream_for_user(self, _user_id: int) -> SimpleNamespace:
        return SimpleNamespace(
            id=3,
            title="Тестовый поток",
            status="active",
            course_version_id=1,
        )

    def get_stream(self, _stream_id: int) -> SimpleNamespace:
        return SimpleNamespace(id=3, title="Тестовый поток")

    def score_rows(self, _stream_id: int) -> list[dict[str, object]]:
        return []

    def course_completion(self, _stream_id: int, *, at: str) -> dict[str, object]:
        return {
            "course_version_id": 1,
            "crowd_publication_count": 0,
            "closed_crowd_publication_count": 0,
        }

    def participant_assignments(self, _user_id: int, _stream_id: int | None = None) -> list[dict[str, object]]:
        return [
            {
                "id": 10,
                "stream_id": 3,
                "stream_publication_id": 5,
                "topic_title": "0. Онбординг",
                "content_text": "Первое задание",
                "status": "submitted",
                "collection_closed_status": "sent",
                "scheduled_at": SCHEDULED_AT,
                "deadline_at": DEADLINE_AT,
                "review_deadline_at": REVIEW_DEADLINE_AT,
                "review_closed_at": (None if self.at <= DEADLINE_AT else REVIEW_DEADLINE_AT),
                "review_quality_deadline_at": REVIEW_QUALITY_DEADLINE_AT,
                "review_quality_closed_at": None,
                "response_format": "text",
            }
        ]

    def participant_assignment(
        self,
        user_id: int,
        assignment_id: int,
    ) -> dict[str, object] | None:
        return next(
            (row for row in self.participant_assignments(user_id) if int(row["id"]) == assignment_id),
            None,
        )

    def peer_review_tasks(
        self,
        _user_id: int,
        _status: str | None = None,
        stream_id: int | None = None,
    ) -> list[dict[str, object]]:
        assert stream_id in {None, 3}
        return [
            {
                "id": 27,
                "stream_publication_id": 5,
                "stream_id": 3,
                "review_number": 1,
                "topic_title": "Оценка ОС",
                "status": "completed",
                "deadline_at": REVIEW_DEADLINE_AT,
            }
        ]

    def latest_assignment_answer(self, _user_id: int, _assignment_id: int) -> dict[str, object]:
        return {"answer_text": "Мой прошлый ответ", "answer_file_path": None}

    def peer_reviews_for_assignment(self, _assignment_id: int) -> list[dict[str, object]]:
        return [
            {
                "id": 31,
                "score": 5.0,
                "quality_score": None,
                "quality_comment": None,
                "comment": "Развивающий комментарий <&>",
            }
        ]

    def peer_reviews_for_assignments(self, assignment_ids: list[int]) -> list[dict[str, object]]:
        return [
            {**review, "target_assignment_id": assignment_id}
            for assignment_id in assignment_ids
            for review in self.peer_reviews_for_assignment(assignment_id)
        ]

    def peer_review_feedback_for_reviewer(self, _user_id: int, _stream_publication_id: int) -> list[dict[str, object]]:
        return []

    def peer_review_feedback_for_reviewer_publications(
        self,
        _user_id: int,
        _stream_publication_ids: list[int],
    ) -> list[dict[str, object]]:
        return []


def test_tasks_shows_received_review_rating_instead_of_past_stages() -> None:
    message = FakeMessage()

    asyncio.run(send_tasks(message, FakeRepository(), 42))  # type: ignore[arg-type]

    assert len(message.answers) == 4
    answer = message.answers[0]
    assert "Оценить полученные рецензии" in str(answer["text"])
    assert "Оценено: 0 из 1" in str(answer["text"])
    assert "Первое задание" not in str(answer["text"])
    assert "Разобрать и оценить ОС" not in str(answer["text"])
    assert all("5 из 10" not in str(item["text"]) for item in message.answers)
    assert "Полученные рецензии · 0. Онбординг" in str(message.answers[1]["text"])
    quote = message.answers[2]
    assert quote["kwargs"]["parse_mode"] == "HTML"
    assert "<blockquote expandable>Развивающий комментарий &lt;&amp;&gt;</blockquote>" in str(quote["text"])
    keyboard = message.answers[3]["kwargs"]["reply_markup"]
    assert keyboard.inline_keyboard[0][0].text == "Рецензия 1"
    assert keyboard.inline_keyboard[0][0].callback_data == "ratereviewitem:31"


def test_course_and_reviews_use_the_same_current_stage_as_tasks() -> None:
    repository = FakeRepository()
    stream = SimpleNamespace(
        id=3,
        title="Тестовый поток",
        starts_at=iso_at(CLOCK, days=-1),
        ends_at=iso_at(CLOCK, days=1),
    )

    overview = learning_overview_text(repository, 42, stream)  # type: ignore[arg-type]
    keyboard = course_actions_keyboard(repository, 42, stream)  # type: ignore[arg-type]
    message = FakeMessage()
    asyncio.run(send_reviews(message, repository, 42))  # type: ignore[arg-type]

    assert "оценить полезность полученных рецензий: осталось 1" in overview
    assert [button.callback_data for row in keyboard.inline_keyboard for button in row] == ["nav:tasks"]
    assert "Этап проверки чужих работ завершён" in str(message.answers[0]["text"])
    assert "ОС 1" not in str(message.answers[0]["text"])

    progress_message = FakeMessage()
    asyncio.run(progress_handler(progress_message, repository))  # type: ignore[arg-type]
    progress_text = str(progress_message.answers[0]["text"])
    assert "оценка полученных рецензий" in progress_text
    assert "балл" not in progress_text.lower()


class FakeCallback:
    def __init__(self) -> None:
        self.data = "taskview:10"
        self.from_user = SimpleNamespace(id=42)
        self.message = FakeMessage()
        self.callback_answers: list[dict[str, object]] = []

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


def test_past_assignment_opens_saved_answer_without_edit_button() -> None:
    callback = FakeCallback()

    asyncio.run(task_view_callback(callback, FakeRepository()))  # type: ignore[arg-type]

    assert "Первое задание" in str(callback.message.answers[0]["text"])
    detail = callback.message.answers[1]
    assert "Мой прошлый ответ" in str(detail["text"])
    assert detail["kwargs"]["reply_markup"] is None


def test_assignment_editability_uses_repository_clock_at_deadline_boundary() -> None:
    before = FakeCallback()
    at_deadline = FakeCallback()

    asyncio.run(
        task_view_callback(
            before,
            FakeRepository(BEFORE_DEADLINE),
        )
    )  # type: ignore[arg-type]
    asyncio.run(
        task_view_callback(
            at_deadline,
            FakeRepository(DEADLINE_AT),
        )
    )  # type: ignore[arg-type]

    assert before.message.answers[1]["kwargs"]["reply_markup"] is not None
    assert at_deadline.message.answers[1]["kwargs"]["reply_markup"] is None