Skip to content
GitLab
Commit f2a9f00f authored by Kaifmohd's avatar Kaifmohd
Browse files

LLM-ASYST

parents
Hide whitespace changes
Inline Side-by-side
asyst-backend/app/services/normalizer.py 0 → 100644
View file @ f2a9f00f
from __future__ import annotations
import re
from dataclasses import dataclass
from typing import Iterable
import pandas as pd
@dataclass(frozen=True)
class QuestionGroup:
number: int
question_col: str
answer_col: str
reference_col: str
max_points_col: str | None
default_max_points: float | None = None
observed_score_col: str | None = None
_QUESTION_RE = re.compile(r"^(?:question|frage)\s+(\d+)$", re.IGNORECASE)
_ANSWER_RE = re.compile(r"^(?:answer|antwort)\s+(\d+)$", re.IGNORECASE)
_REFERENCE_RE = re.compile(
r"^(?:reference\s*answer|correct\s*answer|richtige\s*antwort|musterloesung|musterlösung)\s+(\d+)$",
re.IGNORECASE,
)
_MAX_POINTS_RE = re.compile(
r"^(?:max[\s_]*points|maximal[\s_]*points|maximale[\s_]*punktzahl|punkte|max[\s_]*punkte)[\s_]*(\d+)$",
re.IGNORECASE,
)
_OBSERVED_SCORE_RE = re.compile(
r"^(?:observed\s*grade|observedgrade|human\s*grade|human\s*score|professor\s*score|score|punktzahl|bewertung)[\s_]*(\d+)$",
re.IGNORECASE,
)
_TOTAL_GRADE_RE = re.compile(r"^(?:bewertung|grade|overall\s*grade)\s*/\s*([0-9]+(?:[,.][0-9]+)?)$", re.IGNORECASE)
def detect_question_groups(columns: Iterable[str]) -> list[QuestionGroup]:
question_cols: dict[int, str] = {}
answer_cols: dict[int, str] = {}
reference_cols: dict[int, str] = {}
max_points_cols: dict[int, str] = {}
observed_score_cols: dict[int, str] = {}
total_max_points: float | None = None
for col in columns:
normalized = col.strip()
if m := _QUESTION_RE.match(normalized):
question_cols[int(m.group(1))] = col
elif m := _ANSWER_RE.match(normalized):
answer_cols[int(m.group(1))] = col
elif m := _REFERENCE_RE.match(normalized):
reference_cols[int(m.group(1))] = col
elif m := _MAX_POINTS_RE.match(normalized):
max_points_cols[int(m.group(1))] = col
elif m := _OBSERVED_SCORE_RE.match(normalized):
observed_score_cols[int(m.group(1))] = col
elif m := _TOTAL_GRADE_RE.match(normalized):
total_max_points = float(m.group(1).replace(",", "."))
all_numbers = sorted(
set(question_cols)
| set(answer_cols)
| set(reference_cols)
| set(max_points_cols)
| set(observed_score_cols)
)
if not all_numbers:
raise ValueError(
"No question groups found. Expected columns like 'Question 1', 'Answer 1', "
"'Reference Answer 1', or German equivalents like 'Frage 1', 'Antwort 1', "
"'Richtige Antwort 1'."
)
fallback_max_points = (
round(total_max_points / len(all_numbers), 2)
if total_max_points is not None and len(all_numbers) > 0
else None
)
groups: list[QuestionGroup] = []
for number in all_numbers:
missing: list[str] = []
if number not in question_cols:
missing.append(f"Question {number}")
if number not in answer_cols:
missing.append(f"Answer {number}")
if number not in reference_cols:
missing.append(f"Reference Answer {number}")
if missing:
raise ValueError(f"Incomplete question group for #{number}. Missing: {', '.join(missing)}")
groups.append(
QuestionGroup(
number=number,
question_col=question_cols[number],
answer_col=answer_cols[number],
reference_col=reference_cols[number],
max_points_col=max_points_cols.get(number),
default_max_points=fallback_max_points,
observed_score_col=observed_score_cols.get(number),
)
)
return groups
def to_text(value: object) -> str:
if value is None or pd.isna(value):
return ""
return str(value).strip()
def first_non_empty_text(series: pd.Series) -> str:
for value in series:
text = to_text(value)
if text:
return text
return ""
def first_non_empty_float(series: pd.Series) -> float | None:
for value in series:
text = to_text(value)
if not text:
continue
try:
return float(text)
except ValueError as exc:
raise ValueError(f"Invalid max points value: {text}") from exc
return None
asyst-backend/app/services/parser.py 0 → 100644
View file @ f2a9f00f
from __future__ import annotations
from pathlib import Path
import pandas as pd
from app.core.config import settings
from app.db import SessionLocal
from app.models.grading import Answer, GradingRun, Question, Student
from app.services.normalizer import (
QuestionGroup,
detect_question_groups,
first_non_empty_float,
first_non_empty_text,
to_text,
)
class ParsingError(Exception):
pass
_STUDENT_COLUMN_ALIASES = {
"first_name": ("FirstName", "First Name", "Vorname"),
"last_name": ("LastName", "Last Name", "Nachname"),
"email": ("Email-Address", "Email", "E-Mail-Adresse", "E-Mail", "E-Mail Address"),
"status": ("Status",),
}
def process_uploaded_file(
*,
upload_id: str,
original_filename: str,
relative_path: str,
file_type: str,
run_type: str = "grading",
) -> None:
normalized_run_type = _normalize_run_type(run_type)
db = SessionLocal()
run = GradingRun(
run_id=upload_id,
filename=original_filename,
file_path=relative_path,
file_type=file_type,
run_type=normalized_run_type,
status="uploaded",
)
db.add(run)
db.flush()
try:
file_path = _resolve_file_path(relative_path)
df = _read_dataframe(file_path=file_path, file_type=file_type)
student_columns = _resolve_student_columns(df)
groups = detect_question_groups(df.columns)
_validate_evaluation_columns(groups, normalized_run_type)
question_map = _create_questions(db=db, run_id=upload_id, df=df, groups=groups)
_create_students_and_answers(
db=db,
run_id=upload_id,
df=df,
student_columns=student_columns,
groups=groups,
question_map=question_map,
run_type=normalized_run_type,
)
run.status = "parsed"
db.commit()
except ParsingError:
run.status = "failed"
db.commit()
raise
except Exception as exc:
run.status = "failed"
db.commit()
raise ParsingError(f"Unexpected parsing failure: {exc}") from exc
finally:
db.close()
def _normalize_run_type(run_type: str) -> str:
normalized = (run_type or "grading").strip().lower()
if normalized not in {"grading", "evaluation"}:
raise ParsingError("run_type must be either 'grading' or 'evaluation'")
return normalized
def _resolve_file_path(relative_path: str) -> Path:
candidate = Path(relative_path)
if candidate.exists():
return candidate
fallback = settings.upload_dir / Path(relative_path).name
if fallback.exists():
return fallback
raise ParsingError(f"Stored file not found at path: {relative_path}")
def _read_dataframe(*, file_path: Path, file_type: str) -> pd.DataFrame:
try:
if file_type.lower() == "csv":
return pd.read_csv(file_path)
if file_type.lower() == "xlsx":
return pd.read_excel(file_path, engine="openpyxl")
except Exception as exc:
raise ParsingError(f"Failed to read {file_type} file: {exc}") from exc
raise ParsingError(f"Unsupported file type for parsing: {file_type}")
def _resolve_student_columns(df: pd.DataFrame) -> dict[str, str | None]:
resolved = {
key: _find_column_alias(df.columns, aliases)
for key, aliases in _STUDENT_COLUMN_ALIASES.items()
}
missing = [
label
for key, label in (
("first_name", "FirstName/Vorname"),
("last_name", "LastName/Nachname"),
("email", "Email-Address/E-Mail-Adresse"),
)
if resolved[key] is None
]
if missing:
raise ParsingError(f"Missing required student columns: {', '.join(missing)}")
return resolved
def _find_column_alias(columns: pd.Index, aliases: tuple[str, ...]) -> str | None:
normalized_columns = {_normalize_column_name(column): column for column in columns}
for alias in aliases:
if found := normalized_columns.get(_normalize_column_name(alias)):
return found
return None
def _normalize_column_name(value: object) -> str:
return "".join(ch for ch in str(value).strip().casefold() if ch.isalnum())
def _validate_evaluation_columns(groups: list[QuestionGroup], run_type: str) -> None:
if run_type != "evaluation":
return
missing = [f"Question {group.number}" for group in groups if group.observed_score_col is None]
if missing:
raise ParsingError(
"Evaluation uploads require a professor score column for every question. "
f"Missing observed score for: {', '.join(missing)}"
)
def _create_questions(
*,
db,
run_id: str,
df: pd.DataFrame,
groups: list[QuestionGroup],
) -> dict[int, Question]:
question_map: dict[int, Question] = {}
for group in groups:
question_text = first_non_empty_text(df[group.question_col])
if not question_text:
raise ParsingError(f"Question text is empty for Question {group.number}")
question = Question(
run_id=run_id,
question_number=group.number,
question_text=question_text,
reference_answer=first_non_empty_text(df[group.reference_col]) or None,
max_points=(
first_non_empty_float(df[group.max_points_col])
if group.max_points_col is not None
else group.default_max_points
),
)
db.add(question)
db.flush()
question_map[group.number] = question
return question_map
def _create_students_and_answers(
*,
db,
run_id: str,
df: pd.DataFrame,
student_columns: dict[str, str | None],
groups: list[QuestionGroup],
question_map: dict[int, Question],
run_type: str,
) -> None:
for row_index, row in df.iterrows():
student = Student(
run_id=run_id,
student_code=f"STUDENT_{row_index + 1:03d}",
first_name=to_text(row.get(student_columns["first_name"])),
last_name=to_text(row.get(student_columns["last_name"])),
email=to_text(row.get(student_columns["email"])),
status=to_text(row.get(student_columns["status"])) if student_columns["status"] else None,
row_index=int(row_index),
)
db.add(student)
db.flush()
for group in groups:
question = question_map[group.number]
observed_score = _observed_score_for_row(row, group=group, max_points=question.max_points, run_type=run_type)
db.add(
Answer(
run_id=run_id,
student_id=student.id,
question_id=question.id,
answer_text=to_text(row.get(group.answer_col)),
observed_score=observed_score,
processing_status="pending",
)
)
def _observed_score_for_row(
row: pd.Series,
*,
group: QuestionGroup,
max_points: float | None,
run_type: str,
) -> float | None:
if group.observed_score_col is None:
return None
raw_value = to_text(row.get(group.observed_score_col))
if not raw_value:
if run_type == "evaluation":
raise ParsingError(f"Missing professor score for Question {group.number}")
return None
try:
observed_score = float(raw_value.replace(",", "."))
except ValueError as exc:
raise ParsingError(f"Invalid professor score for Question {group.number}: {raw_value}") from exc
if observed_score < 0:
raise ParsingError(f"Professor score for Question {group.number} must be greater than or equal to 0")
if max_points is not None and observed_score > max_points:
raise ParsingError(
f"Professor score for Question {group.number} must be less than or equal to max points ({max_points})"
)
return observed_score
asyst-backend/app/services/result_service.py 0 → 100644
View file @ f2a9f00f
from __future__ import annotations
from sqlalchemy.orm import Session
from app.models.grading import Answer, GradingRun, Question, Student
class ResultError(ValueError):
pass
def get_student_result_rows(db: Session, run_id: str) -> list[dict[str, object]]:
run_exists = db.query(GradingRun.id).filter(GradingRun.run_id == run_id).first() is not None
if not run_exists:
raise ResultError("Run Id not found")
students = (
db.query(Student)
.filter(Student.run_id == run_id)
.order_by(Student.last_name.asc(), Student.first_name.asc(), Student.id.asc())
.all()
)
rows: list[dict[str, object]] = []
for student in students:
answer_rows = (
db.query(Answer, Question)
.join(Question, Question.id == Answer.question_id)
.filter(Answer.run_id == run_id, Answer.student_id == student.id)
.all()
)
total_final = 0.0
total_max = 0.0
review_complete = True
for answer, question in answer_rows:
if answer.final_score is not None:
total_final += float(answer.final_score)
if question.max_points is not None:
total_max += float(question.max_points)
if answer.review_status == "pending" or answer.processing_status != "graded":
review_complete = False
percentage = 0.0 if total_max <= 0 else round((total_final / total_max) * 100.0, 2)
rows.append(
{
"student_id": student.id,
"name": f"{student.first_name} {student.last_name}".strip(),
"email": student.email,
"total_final_score": round(total_final, 2),
"total_max_points": round(total_max, 2),
"percentage": percentage,
"review_complete": review_complete,
}
)
return rows
asyst-backend/app/services/review_service.py 0 → 100644
View file @ f2a9f00f
from __future__ import annotations
from datetime import datetime, timezone
from math import floor
from sqlalchemy import func
from sqlalchemy.orm import Session
from app.models.grading import Answer, GradingRun, Question, Student
from app.schemas.review import AnswerReviewRequest
REVIEW_PENDING = "pending"
REVIEW_ACCEPTED = "accepted"
REVIEW_OVERRIDDEN = "overridden"
PROCESSING_GRADED = "graded"
PROCESSING_FAILED = "failed"
class ReviewError(ValueError):
pass
def utc_now() -> datetime:
return datetime.now(timezone.utc)
def apply_answer_review(db: Session, answer_id: int, payload: AnswerReviewRequest) -> Answer:
answer = db.query(Answer).filter(Answer.id == answer_id).one_or_none()
if answer is None:
raise ReviewError("Answer not found")
question = db.query(Question).filter(Question.id == answer.question_id).one()
if answer.processing_status != PROCESSING_GRADED:
raise ReviewError("Answer cannot be reviewed until grading status is 'graded'")
if payload.action == "accept_llm":
_accept_llm(answer=answer, reviewed_by=payload.reviewed_by, review_note=payload.review_note)
elif payload.action == "override_score":
_override_score(
answer=answer,
max_points=question.max_points,
human_score=payload.human_score,
reviewed_by=payload.reviewed_by,
review_note=payload.review_note,
)
else: # pragma: no cover
raise ReviewError(f"Unsupported review action: {payload.action}")
db.commit()
db.refresh(answer)
return answer
def _accept_llm(*, answer: Answer, reviewed_by: str | None, review_note: str | None) -> None:
if answer.llm_score is None:
raise ReviewError("Cannot accept LLM score because llm_score is empty")
answer.accepted_llm = True
answer.human_score = None
answer.final_score = answer.llm_score
answer.review_status = REVIEW_ACCEPTED
answer.review_required = False
answer.reviewed_by = reviewed_by
answer.review_note = review_note
answer.reviewed_at = utc_now()
def _override_score(
*,
answer: Answer,
max_points: float | None,
human_score: float | None,
reviewed_by: str | None,
review_note: str | None,
) -> None:
if human_score is None:
raise ReviewError("human_score is required for override_score")
if human_score < 0:
raise ReviewError("human_score must be greater than or equal to 0")
if max_points is not None and human_score > max_points:
raise ReviewError(f"human_score must be less than or equal to max_points ({max_points})")
answer.accepted_llm = False
answer.human_score = human_score
answer.final_score = human_score
answer.review_status = REVIEW_OVERRIDDEN
answer.review_required = False
answer.reviewed_by = reviewed_by
answer.review_note = review_note
answer.reviewed_at = utc_now()
def get_run_review_summary(db: Session, run_id: str) -> dict[str, int | float | bool | str]:
run = db.query(GradingRun).filter(GradingRun.run_id == run_id).one_or_none()
if run is None:
raise ReviewError("Run not found")
total_students = _count(db, Student.id, Student.run_id == run_id)
total_questions = _count(db, Question.id, Question.run_id == run_id)
total_answers = _count(db, Answer.id, Answer.run_id == run_id)
total_graded = _count(
db,
Answer.id,
Answer.run_id == run_id,
Answer.processing_status == PROCESSING_GRADED,
)
total_failed = _count(
db,
Answer.id,
Answer.run_id == run_id,
Answer.processing_status == PROCESSING_FAILED,
)
total_review_pending = _count(
db,
Answer.id,
Answer.run_id == run_id,
Answer.review_status == REVIEW_PENDING,
)
total_accepted = _count(
db,
Answer.id,
Answer.run_id == run_id,
Answer.review_status == REVIEW_ACCEPTED,
)
total_overridden = _count(
db,
Answer.id,
Answer.run_id == run_id,
Answer.review_status == REVIEW_OVERRIDDEN,
)
reviewed_total = total_accepted + total_overridden
completion = 0.0 if total_answers == 0 else round((reviewed_total / total_answers) * 100.0, 2)
export_ready = total_answers > 0 and total_review_pending == 0 and total_failed == 0 and total_graded == total_answers
return {
"run_id": run.run_id,
"filename": run.filename,
"grading_status": run.status,
"total_students": total_students,
"total_questions": total_questions,
"total_answers": total_answers,
"total_graded": total_graded,
"total_failed": total_failed,
"total_review_pending": total_review_pending,
"total_accepted": total_accepted,
"total_overridden": total_overridden,
"review_completion_percent": completion,
"export_ready": export_ready,
}
def _count(db: Session, col, *filters) -> int:
return int(db.query(func.count(col)).filter(*filters).scalar() or 0)
def get_run_students(
db: Session,
run_id: str,
*,
review_status: str | None,
search: str | None,
page: int,
page_size: int,
) -> dict[str, object]:
_ensure_run_exists(db, run_id)
query = db.query(Student).filter(Student.run_id == run_id)
if search:
term = f"%{search.strip()}%"
query = query.filter(
(Student.first_name.ilike(term))
| (Student.last_name.ilike(term))
| (Student.email.ilike(term))
| (Student.student_code.ilike(term))
)
students = query.order_by(Student.last_name.asc(), Student.first_name.asc(), Student.id.asc()).all()
if review_status:
filtered_students: list[Student] = []
for student in students:
has_status = (
db.query(Answer.id)
.filter(
Answer.run_id == run_id,
Answer.student_id == student.id,
Answer.review_status == review_status,
)
.first()
is not None
)
if has_status:
filtered_students.append(student)
students = filtered_students
total = len(students)
start = (page - 1) * page_size
end = start + page_size
paged_students = students[start:end]
items: list[dict[str, object]] = []
for student in paged_students:
answer_rows = (
db.query(Answer, Question)
.join(Question, Question.id == Answer.question_id)
.filter(Answer.run_id == run_id, Answer.student_id == student.id)
.all()
)
answered_count = len(answer_rows)
reviewed_count = sum(1 for answer, _ in answer_rows if answer.review_status != REVIEW_PENDING)
pending_count = answered_count - reviewed_count
total_final = round(sum(float(answer.final_score or 0.0) for answer, _ in answer_rows), 2)
total_max = round(sum(float(question.max_points or 0.0) for _, question in answer_rows), 2)
review_complete = pending_count == 0 and all(answer.processing_status == PROCESSING_GRADED for answer, _ in answer_rows)
items.append(
{
"student_id": student.id,
"student_code": student.student_code,
"first_name": student.first_name,
"last_name": student.last_name,
"email": student.email,
"answered_count": answered_count,
"reviewed_count": reviewed_count,
"pending_review_count": pending_count,
"total_final_score": total_final,
"total_max_points": total_max,
"review_complete": review_complete,
}
)
return {
"run_id": run_id,
"items": items,
"pagination": {
"page": page,
"page_size": page_size,
"total": total,
},
}
def get_student_answers(db: Session, run_id: str, student_id: int) -> dict[str, object]:
_ensure_run_exists(db, run_id)
student = db.query(Student).filter(Student.id == student_id, Student.run_id == run_id).one_or_none()
if student is None:
raise ReviewError("Student not found")
rows = (
db.query(Answer, Question)
.join(Question, Question.id == Answer.question_id)
.filter(Answer.run_id == run_id, Answer.student_id == student_id)
.order_by(Question.question_number.asc())
.all()
)
answers: list[dict[str, object]] = []
for answer, question in rows:
answers.append(
{
"answer_id": answer.id,
"question_id": question.id,
"question_number": question.question_number,
"question_text": question.question_text,
"answer_text": answer.answer_text,
"reference_answer": question.reference_answer,
"max_points": question.max_points,
"llm_score": answer.llm_score,
"human_score": answer.human_score,
"final_score": answer.final_score,
"review_status": answer.review_status,
"accepted_llm": answer.accepted_llm,
"reviewed_at": answer.reviewed_at,
"review_required": answer.review_required,
"processing_status": answer.processing_status,
"llm_feedback": answer.llm_feedback,
}
)
return {
"run_id": run_id,
"student_id": student.id,
"student_code": student.student_code,
"first_name": student.first_name,
"last_name": student.last_name,
"email": student.email,
"answers": answers,
}
def get_run_questions(db: Session, run_id: str) -> dict[str, object]:
_ensure_run_exists(db, run_id)
questions = (
db.query(Question)
.filter(Question.run_id == run_id)
.order_by(Question.question_number.asc())
.all()
)
items: list[dict[str, object]] = []
for question in questions:
answer_rows = db.query(Answer).filter(Answer.run_id == run_id, Answer.question_id == question.id).all()
attempted_count = len(answer_rows)
reviewed_count = sum(1 for answer in answer_rows if answer.review_status != REVIEW_PENDING)
pending_count = attempted_count - reviewed_count
llm_scores = [float(answer.llm_score) for answer in answer_rows if answer.llm_score is not None]
final_scores = [float(answer.final_score) for answer in answer_rows if answer.final_score is not None]
score_counts: dict[float, int] = {}
for score in llm_scores:
score_counts[score] = score_counts.get(score, 0) + 1
items.append(
{
"question_id": question.id,
"question_number": question.question_number,
"question_text": question.question_text,
"max_points": question.max_points,
"attempted_count": attempted_count,
"reviewed_count": reviewed_count,
"pending_review_count": pending_count,
"avg_llm_score": round(sum(llm_scores) / len(llm_scores), 2) if llm_scores else None,
"avg_final_score": round(sum(final_scores) / len(final_scores), 2) if final_scores else None,
"predicted_score_counts": [
{"score": score, "count": score_counts[score]}
for score in sorted(score_counts)
],
}
)
return {
"run_id": run_id,
"items": items,
}
def get_question_answers(
db: Session,
run_id: str,
question_id: int,
*,
review_status: str | None,
page: int,
page_size: int,
sort: str | None,
predicted_scores: list[float] | None = None,
score_scale: str = "full",
) -> dict[str, object]:
_ensure_run_exists(db, run_id)
question = db.query(Question).filter(Question.id == question_id, Question.run_id == run_id).one_or_none()
if question is None:
raise ReviewError("Question not found")
rows = (
db.query(Answer, Student)
.join(Student, Student.id == Answer.student_id)
.filter(Answer.run_id == run_id, Answer.question_id == question_id)
.all()
)
if review_status:
rows = [(answer, student) for answer, student in rows if answer.review_status == review_status]
if predicted_scores:
def selected_score(answer: Answer) -> bool:
if answer.llm_score is None:
return False
score = float(answer.llm_score)
if score_scale == "whole":
score = float(floor(score + 0.5))
return any(abs(score - selected) < 0.001 for selected in predicted_scores)
rows = [(answer, student) for answer, student in rows if selected_score(answer)]
sort_key = (sort or "student_name").strip().lower()
if sort_key == "llm_score":
rows.sort(key=lambda item: (item[0].llm_score is None, item[0].llm_score if item[0].llm_score is not None else 0.0))
elif sort_key == "final_score":
rows.sort(key=lambda item: (item[0].final_score is None, item[0].final_score if item[0].final_score is not None else 0.0))
elif sort_key == "review_status":
rows.sort(key=lambda item: item[0].review_status)
else:
rows.sort(key=lambda item: (item[1].last_name.lower(), item[1].first_name.lower(), item[1].id))
total = len(rows)
start = (page - 1) * page_size
end = start + page_size
page_rows = rows[start:end]
items: list[dict[str, object]] = []
for answer, student in page_rows:
items.append(
{
"answer_id": answer.id,
"student_id": student.id,
"student_name": f"{student.first_name} {student.last_name}".strip(),
"student_code": student.student_code,
"email": student.email,
"answer_text": answer.answer_text,
"llm_score": answer.llm_score,
"observed_score": answer.observed_score,
"human_score": answer.human_score,
"final_score": answer.final_score,
"review_status": answer.review_status,
"review_required": answer.review_required,
"accepted_llm": answer.accepted_llm,
"reviewed_at": answer.reviewed_at,
"processing_status": answer.processing_status,
"llm_feedback": answer.llm_feedback,
}
)
return {
"run_id": run_id,
"question_id": question.id,
"question_number": question.question_number,
"question_text": question.question_text,
"max_points": question.max_points,
"items": items,
"pagination": {
"page": page,
"page_size": page_size,
"total": total,
},
}
def _ensure_run_exists(db: Session, run_id: str) -> None:
run_exists = db.query(GradingRun.id).filter(GradingRun.run_id == run_id).first() is not None
if not run_exists:
raise ReviewError("Run not found")
asyst-backend/app/services/run_delete_service.py 0 → 100644
View file @ f2a9f00f
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from sqlalchemy.orm import Session
from app.core.config import settings
from app.models.grading import Answer, GradingRun, Question, Student
from app.services.grading import is_grade_run_running
class RunDeleteError(ValueError):
pass
class RunDeleteNotFound(RunDeleteError):
pass
class RunDeleteConflict(RunDeleteError):
pass
@dataclass(frozen=True)
class RunDeleteSummary:
run_id: str
deleted: bool
answers_deleted: int
students_deleted: int
questions_deleted: int
grading_run_deleted: int
file_deleted: bool
file_path: str | None
def delete_run(db: Session, run_id: str) -> RunDeleteSummary:
if is_grade_run_running(run_id):
raise RunDeleteConflict("Run is currently grading and cannot be deleted")
run = db.query(GradingRun).filter(GradingRun.run_id == run_id).one_or_none()
if run is None:
raise RunDeleteNotFound("Run not found")
file_to_delete = _resolve_deletable_upload_path(run.file_path)
reported_file_path = str(file_to_delete) if file_to_delete else run.file_path
answers_deleted = db.query(Answer).filter(Answer.run_id == run_id).delete(synchronize_session=False)
students_deleted = db.query(Student).filter(Student.run_id == run_id).delete(synchronize_session=False)
questions_deleted = db.query(Question).filter(Question.run_id == run_id).delete(synchronize_session=False)
grading_run_deleted = db.query(GradingRun).filter(GradingRun.run_id == run_id).delete(synchronize_session=False)
db.commit()
file_deleted = False
if file_to_delete and file_to_delete.exists():
file_to_delete.unlink()
file_deleted = True
return RunDeleteSummary(
run_id=run_id,
deleted=True,
answers_deleted=int(answers_deleted),
students_deleted=int(students_deleted),
questions_deleted=int(questions_deleted),
grading_run_deleted=int(grading_run_deleted),
file_deleted=file_deleted,
file_path=reported_file_path,
)
def _resolve_deletable_upload_path(stored_path: str | None) -> Path | None:
if not stored_path:
return None
upload_root = settings.upload_dir.resolve()
raw_path = Path(stored_path)
if raw_path.is_absolute():
candidate = raw_path.resolve()
if not _is_relative_to(candidate, upload_root):
raise RunDeleteError("Stored upload file path is outside the upload directory")
return candidate
cwd_candidate = raw_path.resolve()
if cwd_candidate.exists():
if not _is_relative_to(cwd_candidate, upload_root):
raise RunDeleteError("Stored upload file path is outside the upload directory")
return cwd_candidate
fallback = (upload_root / raw_path.name).resolve()
if not _is_relative_to(fallback, upload_root):
raise RunDeleteError("Stored upload file path is outside the upload directory")
return fallback if fallback.exists() else None
def _is_relative_to(path: Path, parent: Path) -> bool:
try:
path.relative_to(parent)
return True
except ValueError:
return False
asyst-backend/app/utils/__init__.py 0 → 100644
View file @ f2a9f00f
"""Utility helpers package."""
asyst-backend/app/utils/file_utils.py 0 → 100644
View file @ f2a9f00f
from __future__ import annotations
from pathlib import Path
def extract_extension(filename: str) -> str:
return Path(filename).suffix.lower().lstrip(".")
def sanitize_filename(filename: str) -> str:
# Drop any path segments and keep only safe characters.
base_name = Path(filename).name
safe = "".join(ch for ch in base_name if ch.isalnum() or ch in {"_", "-", "."})
return safe or "upload"
asyst-backend/diagram/model.puml 0 → 100644
View file @ f2a9f00f
@startuml
hide methods
skinparam classAttributeIconSize 0
entity "grading_runs" as grading_runs {
* id : INTEGER <<PK>>
--
* run_id : VARCHAR(64) <<UNIQUE>>
* filename : VARCHAR(255)
* file_path : VARCHAR(512)
* file_type : VARCHAR(16)
* run_type : VARCHAR(32)
* status : VARCHAR(32)
* created_at : DATETIME
* updated_at : DATETIME
}
entity "students" as students {
* id : INTEGER <<PK>>
--
* run_id : VARCHAR(64) <<FK>>
* student_code : VARCHAR(64)
* first_name : VARCHAR(128)
* last_name : VARCHAR(128)
* email : VARCHAR(255)
status : VARCHAR(64)
* row_index : INTEGER
--
<<INDEX>> ix_students_run_name (run_id, last_name, first_name)
<<INDEX>> ix_students_run_email (run_id, email)
}
entity "questions" as questions {
* id : INTEGER <<PK>>
--
* run_id : VARCHAR(64) <<FK>>
* question_number : INTEGER
* question_text : TEXT
reference_answer : TEXT
max_points : FLOAT
--
<<UNIQUE>> uq_questions_run_qnum (run_id, question_number)
<<INDEX>> ix_questions_run_number (run_id, question_number)
}
entity "answers" as answers {
* id : INTEGER <<PK>>
--
* run_id : VARCHAR(64) <<FK>>
* student_id : INTEGER <<FK>>
* question_id : INTEGER <<FK>>
* answer_text : TEXT
* processing_status : VARCHAR(32)
* review_status : VARCHAR(32)
llm_score : FLOAT
llm_feedback : TEXT
llm_raw_response : TEXT
llm_parse_warning : TEXT
observed_score : FLOAT
human_score : FLOAT
final_score : FLOAT
* accepted_llm : BOOLEAN
reviewed_by : VARCHAR(255)
review_note : TEXT
* attempt_count : INTEGER
last_error : TEXT
graded_at : DATETIME
reviewed_at : DATETIME
* review_required : BOOLEAN
* created_at : DATETIME
* updated_at : DATETIME
--
<<INDEX>> ix_answers_run_review_status (run_id, review_status)
<<INDEX>> ix_answers_run_processing_status (run_id, processing_status)
<<INDEX>> ix_answers_run_student (run_id, student_id)
<<INDEX>> ix_answers_run_question (run_id, question_id)
}
grading_runs ||--o{ students : run_id
grading_runs ||--o{ questions : run_id
grading_runs ||--o{ answers : run_id
students ||--o{ answers : student_id
questions ||--o{ answers : question_id
note right of grading_runs
run_type separates the two workflows:
- grading: AI scores are reviewed and exported.
- evaluation: AI scores are compared with professor scores.
end note
note right of answers
llm_score stores the AI prediction.
observed_score stores the professor grade used in evaluation uploads.
final_score stores the accepted or overridden score for grading exports.
end note
@enduml
asyst-backend/pytest.ini 0 → 100644
View file @ f2a9f00f
[pytest]
pythonpath = .
testpaths = tests
asyst-backend/requirements.txt 0 → 100644
View file @ f2a9f00f
fastapi==0.104.1
pydantic==2.13.4
uvicorn[standard]==0.24.0
python-multipart==0.0.6
python-dotenv==1.2.2
sqlalchemy==2.0.36
pandas==2.2.3
openpyxl==3.1.5
openai==1.79.0
pytest==8.3.5
asyst-backend/tests/test_evaluation_parser.py 0 → 100644
View file @ f2a9f00f
from __future__ import annotations
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
from app.models.base import Base
from app.models.grading import Answer, GradingRun, Question, Student
from app.services import parser
from app.services.parser import ParsingError, process_uploaded_file
@pytest.fixture()
def db_session(tmp_path, monkeypatch) -> Session:
engine = create_engine("sqlite:///:memory:", future=True)
Base.metadata.create_all(bind=engine)
session_factory = sessionmaker(bind=engine, autoflush=False, autocommit=False, expire_on_commit=False)
db = session_factory()
monkeypatch.setattr(parser, "SessionLocal", session_factory)
monkeypatch.setattr(parser.settings, "upload_dir", tmp_path)
try:
yield db
finally:
db.close()
def _write_sample_csv(tmp_path, rows: list[str]) -> str:
path = tmp_path / "sample.csv"
path.write_text("\n".join(rows), encoding="utf-8")
return str(path)
def test_grading_upload_does_not_require_observed_score(db_session: Session, tmp_path) -> None:
csv_path = _write_sample_csv(
tmp_path,
[
"FirstName,LastName,Email-Address,Question 1,Answer 1,Reference Answer 1,max_points_1",
"Ada,Lovelace,ada@example.test,What is RAM?,Temporary memory,Working memory,2",
],
)
process_uploaded_file(
upload_id="grading-run",
original_filename="sample.csv",
relative_path=csv_path,
file_type="csv",
run_type="grading",
)
run = db_session.query(GradingRun).filter(GradingRun.run_id == "grading-run").one()
answer = db_session.query(Answer).filter(Answer.run_id == "grading-run").one()
assert run.run_type == "grading"
assert answer.observed_score is None
def test_evaluation_upload_accepts_professor_score_alias(db_session: Session, tmp_path) -> None:
csv_path = _write_sample_csv(
tmp_path,
[
"FirstName,LastName,Email-Address,Question 1,Answer 1,Reference Answer 1,max_points_1,Professor Score 1",
"Ada,Lovelace,ada@example.test,What is RAM?,Temporary memory,Working memory,2,1.5",
],
)
process_uploaded_file(
upload_id="eval-run",
original_filename="sample.csv",
relative_path=csv_path,
file_type="csv",
run_type="evaluation",
)
run = db_session.query(GradingRun).filter(GradingRun.run_id == "eval-run").one()
answer = db_session.query(Answer).filter(Answer.run_id == "eval-run").one()
assert run.run_type == "evaluation"
assert answer.observed_score == 1.5
def test_grading_upload_accepts_moodle_german_columns_and_infers_max_points(
db_session: Session, tmp_path
) -> None:
csv_path = _write_sample_csv(
tmp_path,
[
'"Nachname","Vorname","E-Mail-Adresse","Status","Begonnen","Beendet","Dauer","Bewertung/30,00","Frage 1","Antwort 1","Richtige Antwort 1","Frage 2","Antwort 2","Richtige Antwort 2","Frage 3","Antwort 3","Richtige Antwort 3"',
'"Lovelace","Ada","ada@example.test","Beendet","2020-07-01 10:21:00","2020-07-01 11:06:00","45 Minuten","0","What is RAM?","Temporary memory","Working memory","What is CPU?","Processor","Central processor","What is ROM?","Permanent memory","Read-only memory"',
],
)
process_uploaded_file(
upload_id="german-moodle-run",
original_filename="moodle.csv",
relative_path=csv_path,
file_type="csv",
run_type="grading",
)
student = db_session.query(Student).filter(Student.run_id == "german-moodle-run").one()
questions = (
db_session.query(Question)
.filter(Question.run_id == "german-moodle-run")
.order_by(Question.question_number.asc())
.all()
)
answers = (
db_session.query(Answer)
.filter(Answer.run_id == "german-moodle-run")
.order_by(Answer.id.asc())
.all()
)
assert student.first_name == "Ada"
assert student.last_name == "Lovelace"
assert student.email == "ada@example.test"
assert student.status == "Beendet"
assert [question.max_points for question in questions] == [10.0, 10.0, 10.0]
assert questions[0].question_text == "What is RAM?"
assert questions[0].reference_answer == "Working memory"
assert answers[0].answer_text == "Temporary memory"
def test_evaluation_upload_rejects_missing_professor_score(db_session: Session, tmp_path) -> None:
csv_path = _write_sample_csv(
tmp_path,
[
"FirstName,LastName,Email-Address,Question 1,Answer 1,Reference Answer 1,max_points_1",
"Ada,Lovelace,ada@example.test,What is RAM?,Temporary memory,Working memory,2",
],
)
with pytest.raises(ParsingError, match="professor score column"):
process_uploaded_file(
upload_id="missing-score",
original_filename="sample.csv",
relative_path=csv_path,
file_type="csv",
run_type="evaluation",
)
def test_evaluation_upload_rejects_score_above_max_points(db_session: Session, tmp_path) -> None:
csv_path = _write_sample_csv(
tmp_path,
[
"FirstName,LastName,Email-Address,Question 1,Answer 1,Reference Answer 1,max_points_1,Human Grade 1",
"Ada,Lovelace,ada@example.test,What is RAM?,Temporary memory,Working memory,2,3",
],
)
with pytest.raises(ParsingError, match="less than or equal to max points"):
process_uploaded_file(
upload_id="bad-score",
original_filename="sample.csv",
relative_path=csv_path,
file_type="csv",
run_type="evaluation",
)
asyst-backend/tests/test_export_service.py 0 → 100644
View file @ f2a9f00f
from __future__ import annotations
import csv
import io
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
from app.models.base import Base
from app.models.grading import Answer, GradingRun, Question, Student
from app.services.export_service import build_moodle_export_csv
def _db_session() -> Session:
engine = create_engine("sqlite:///:memory:", future=True)
Base.metadata.create_all(bind=engine)
session_factory = sessionmaker(bind=engine, autoflush=False, autocommit=False, expire_on_commit=False)
return session_factory()
def test_moodle_export_uses_german_source_shape_with_per_question_scores(tmp_path) -> None:
source_path = tmp_path / "moodle.csv"
source_path.write_text(
"\n".join(
[
'"Nachname","Vorname","E-Mail-Adresse","Status","Begonnen","Beendet","Dauer","Bewertung/20,00","Frage 1","Antwort 1","Richtige Antwort 1","Frage 2","Antwort 2","Richtige Antwort 2"',
'"Lovelace","Ada","ada@example.test","Beendet","2020-07-01 10:21:00","2020-07-01 11:06:00","45 Minuten","0","What is RAM?","Old answer 1","Working memory","What is CPU?","Old answer 2","Central processor"',
]
),
encoding="utf-8",
)
db = _db_session()
try:
run = GradingRun(
run_id="run-1",
filename="moodle.csv",
file_path=str(source_path),
file_type="csv",
run_type="grading",
status="parsed",
)
db.add(run)
db.flush()
student = Student(
run_id="run-1",
student_code="STUDENT_001",
first_name="Ada",
last_name="Lovelace",
email="ada@example.test",
status="Beendet",
row_index=0,
)
db.add(student)
db.flush()
q1 = Question(
run_id="run-1",
question_number=1,
question_text="What is RAM?",
reference_answer="Working memory",
max_points=10.0,
)
q2 = Question(
run_id="run-1",
question_number=2,
question_text="What is CPU?",
reference_answer="Central processor",
max_points=10.0,
)
db.add_all([q1, q2])
db.flush()
db.add_all(
[
Answer(
run_id="run-1",
student_id=student.id,
question_id=q1.id,
answer_text="RAM stores active data.",
processing_status="graded",
review_status="accepted",
llm_score=8.0,
final_score=8.0,
accepted_llm=True,
),
Answer(
run_id="run-1",
student_id=student.id,
question_id=q2.id,
answer_text="The CPU executes instructions.",
processing_status="graded",
review_status="overridden",
llm_score=7.0,
human_score=9.0,
final_score=9.0,
),
]
)
db.commit()
exported = build_moodle_export_csv(db, "run-1")
rows = list(csv.reader(io.StringIO(exported)))
assert rows[0] == [
"Nachname",
"Vorname",
"E-Mail-Adresse",
"Status",
"Begonnen",
"Beendet",
"Dauer",
"Frage 1",
"Antwort 1",
"Richtige Antwort 1",
"Punktzahl 1",
"Frage 2",
"Antwort 2",
"Richtige Antwort 2",
"Punktzahl 2",
"Gesamtbewertung",
]
assert rows[1] == [
"Lovelace",
"Ada",
"ada@example.test",
"Beendet",
"2020-07-01 10:21:00",
"2020-07-01 11:06:00",
"45 Minuten",
"What is RAM?",
"RAM stores active data.",
"Working memory",
"8.0",
"What is CPU?",
"The CPU executes instructions.",
"Central processor",
"9.0",
"17.0",
]
finally:
db.close()
asyst-backend/tests/test_grading_api.py 0 → 100644
View file @ f2a9f00f
from __future__ import annotations
from app.api.routes import grading as grading_route
from app.schemas.grading import GradingStartRequest
def test_start_grading_accepts_runtime_config_without_returning_api_key(monkeypatch) -> None:
captured = {}
def fake_start_grade_run(run_id: str, **kwargs):
kwargs["run_id"] = run_id
captured.update(kwargs)
return {
"run_id": kwargs["run_id"],
"status": "grading_started",
"strategy": kwargs["strategy"],
}
monkeypatch.setattr(grading_route, "start_grade_run", fake_start_grade_run)
response = grading_route.start_grading_for_run(
"run-123",
payload=GradingStartRequest(
strategy="p1_reference_zero_shot",
llm_profile="openai_custom",
model="paid-model",
api_key="sk-secret",
),
)
assert response == {
"run_id": "run-123",
"status": "grading_started",
"strategy": "p1_reference_zero_shot",
}
assert "sk-secret" not in str(response)
runtime_config = captured["runtime_config"]
assert runtime_config.api_key == "sk-secret"
assert runtime_config.provider == "openai"
assert runtime_config.model == "paid-model"
assert runtime_config.base_url == "https://api.openai.com/v1"
def test_start_grading_maps_local_gemma_profile(monkeypatch) -> None:
captured = {}
def fake_start_grade_run(run_id: str, **kwargs):
kwargs["run_id"] = run_id
captured.update(kwargs)
return {
"run_id": kwargs["run_id"],
"status": "grading_started",
"strategy": kwargs["strategy"],
}
monkeypatch.setattr(grading_route, "start_grade_run", fake_start_grade_run)
monkeypatch.setenv("LOCAL_LLM_BASE_URL", "http://local-llm:8080/v1")
monkeypatch.setenv("LOCAL_LLM_MODEL", "gemma-4-E4B-it-GGUF")
grading_route.start_grading_for_run(
"run-local",
payload=GradingStartRequest(
strategy="p1_reference_zero_shot",
llm_profile="local_gemma",
),
)
runtime_config = captured["runtime_config"]
assert runtime_config.provider == "local_llm"
assert runtime_config.base_url == "http://local-llm:8080/v1"
assert runtime_config.model == "gemma-4-E4B-it-GGUF"
assert runtime_config.api_key == "local-not-required"
def test_start_grading_maps_chat_ai_profile(monkeypatch) -> None:
captured = {}
def fake_start_grade_run(run_id: str, **kwargs):
kwargs["run_id"] = run_id
captured.update(kwargs)
return {
"run_id": kwargs["run_id"],
"status": "grading_started",
"strategy": kwargs["strategy"],
}
monkeypatch.setattr(grading_route, "start_grade_run", fake_start_grade_run)
monkeypatch.setenv("CHAT_AI_BASE_URL", "https://chat-ai.academiccloud.de/v1")
monkeypatch.setenv("CHAT_AI_MODEL", "gemma-3-27b-it")
monkeypatch.setenv("CHAT_AI_API_KEY", "chat-ai-secret")
response = grading_route.start_grading_for_run(
"run-chat-ai",
payload=GradingStartRequest(
strategy="p1_reference_zero_shot",
llm_profile="chat_ai",
model="qwen3-30b-a3b-instruct-2507",
),
)
assert "chat-ai-secret" not in str(response)
runtime_config = captured["runtime_config"]
assert runtime_config.provider == "chat_ai"
assert runtime_config.base_url == "https://chat-ai.academiccloud.de/v1"
assert runtime_config.model == "qwen3-30b-a3b-instruct-2507"
assert runtime_config.api_key == "chat-ai-secret"
def test_start_grading_accepts_prompt_options(monkeypatch) -> None:
captured = {}
def fake_start_grade_run(run_id: str, **kwargs):
kwargs["run_id"] = run_id
captured.update(kwargs)
return {
"run_id": kwargs["run_id"],
"status": "grading_started",
"strategy": kwargs["strategy"],
}
monkeypatch.setattr(grading_route, "start_grade_run", fake_start_grade_run)
grading_route.start_grading_for_run(
"run-p3",
payload=GradingStartRequest(
strategy="p3_true_one_shot",
custom_grading_rules="Deduct 0.5 points for missing detail.",
include_feedback=True,
examples=[
{
"question": "What is RAM?",
"answer": "RAM stores active program data.",
"score": "2",
}
],
),
)
assert captured["custom_grading_rules"] == "Deduct 0.5 points for missing detail."
assert captured["include_feedback"] is True
assert len(captured["examples"]) == 1
assert captured["examples"][0].question == "What is RAM?"
assert captured["examples"][0].score == "2"
asyst-backend/tests/test_label_evaluation_service.py 0 → 100644
View file @ f2a9f00f
from __future__ import annotations
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
from app.models.base import Base
from app.models.grading import Answer, GradingRun, Question, Student
from app.schemas.label_evaluation import LabelEvaluationResponse
from app.services.label_evaluation_service import (
build_label_scale,
get_label_evaluation,
round_half_up_grade,
)
def _db_session() -> Session:
engine = create_engine("sqlite:///:memory:", future=True)
Base.metadata.create_all(bind=engine)
factory = sessionmaker(bind=engine, autoflush=False, autocommit=False, expire_on_commit=False)
return factory()
def _seed_run(db: Session) -> Student:
db.add(
GradingRun(
run_id="eval",
filename="evaluation.csv",
file_path="evaluation.csv",
file_type="csv",
run_type="evaluation",
status="parsed",
)
)
student = Student(
run_id="eval",
student_code="S1",
first_name="Ada",
last_name="Lovelace",
email="ada@example.test",
row_index=0,
)
db.add(student)
db.flush()
return student
def _question(db: Session, number: int, max_points: float) -> Question:
question = Question(
run_id="eval",
question_number=number,
question_text=f"Question {number}?",
reference_answer="Reference",
max_points=max_points,
)
db.add(question)
db.flush()
return question
def _answer(
db: Session,
student: Student,
question: Question,
professor: float | None,
llm: float | None,
status: str = "graded",
) -> None:
db.add(
Answer(
run_id="eval",
student_id=student.id,
question_id=question.id,
answer_text="Student answer",
observed_score=professor,
llm_score=llm,
processing_status=status,
)
)
def test_label_metrics_report_precision_recall_support_and_all_observed_labels() -> None:
rows = [(0.0, 0.0), (0.0, 1.0), (1.0, 1.0), (2.0, 1.0)]
scale = build_label_scale(rows, "full")
assert scale["labels"] == [
{
"label": 0.0,
"true_positive": 1,
"predicted_count": 1,
"support": 2,
"precision": 1.0,
"recall": 0.5,
"low_support": True,
},
{
"label": 1.0,
"true_positive": 1,
"predicted_count": 3,
"support": 1,
"precision": 0.3333,
"recall": 1.0,
"low_support": True,
},
{
"label": 2.0,
"true_positive": 0,
"predicted_count": 0,
"support": 1,
"precision": None,
"recall": 0.0,
"low_support": True,
},
]
def test_support_of_five_is_reported_without_a_reliability_verdict() -> None:
scale = build_label_scale([(1.0, 1.0)] * 5, "full")
assert scale["labels"][0]["support"] == 5
assert scale["labels"][0]["low_support"] is False
assert "status" not in scale["labels"][0]
def test_round_half_up_uses_expected_whole_number_mapping() -> None:
assert [round_half_up_grade(value) for value in (0.0, 0.5, 1.5, 2.5)] == [0.0, 1.0, 2.0, 3.0]
def test_large_integer_scale_defaults_to_whole_numbers_without_changing_small_scale_default() -> None:
db = _db_session()
try:
student = _seed_run(db)
small = _question(db, 1, 2.0)
large = _question(db, 2, 5.0)
_answer(db, student, small, 0.5, 1.0)
_answer(db, student, small, 1.0, 1.0)
_answer(db, student, large, 3.5, 4.0)
_answer(db, student, large, 4.0, 4.0)
db.commit()
report = get_label_evaluation(db, "eval")
first, second = report["questions"]
assert first["default_scale"] == "full"
assert first["whole_scale"] is not None
assert second["default_scale"] == "whole"
assert second["whole_scale"]["labels"] == [
{
"label": 4.0,
"true_positive": 2,
"predicted_count": 2,
"support": 2,
"precision": 1.0,
"recall": 1.0,
"low_support": True,
}
]
LabelEvaluationResponse(**report)
finally:
db.close()
def test_non_integer_max_hides_whole_scale_and_invalid_rows_return_warnings() -> None:
db = _db_session()
try:
student = _seed_run(db)
half_max = _question(db, 1, 2.5)
invalid_max = _question(db, 2, 2.3)
_answer(db, student, half_max, 1.0, 1.0)
_answer(db, student, half_max, 1.5, None, "failed")
_answer(db, student, invalid_max, 1.0, 1.0)
db.commit()
report = get_label_evaluation(db, "eval")
assert report["questions"][0]["whole_scale"] is None
assert report["questions"][0]["failed_llm_count"] == 1
assert report["totals"]["skipped_question_count"] == 1
assert report["totals"]["coverage_percent"] == 33.33
assert {warning["code"] for warning in report["warnings"]} == {
"FAILED_LLM_ROWS",
"INVALID_MAX_POINTS",
}
finally:
db.close()
asyst-backend/tests/test_llm_factory.py 0 → 100644
View file @ f2a9f00f
from __future__ import annotations
from app.services.llm.base import LLMRuntimeConfig
from app.services.llm import factory
class _FakeProvider:
def __init__(self, **kwargs: object) -> None:
self.kwargs = kwargs
def test_factory_prefers_runtime_config_over_environment(monkeypatch) -> None:
monkeypatch.setattr(factory, "OpenAICompatibleProvider", _FakeProvider)
monkeypatch.setattr(factory, "load_dotenv", lambda *args, **kwargs: None)
monkeypatch.setenv("LLM_MODEL", "env-model")
monkeypatch.setenv("LLM_API_KEY", "env-key")
provider = factory.build_llm_provider(
"zero_shot",
runtime_config=LLMRuntimeConfig(
provider="openai",
model="runtime-model",
base_url="https://api.openai.com/v1",
api_key="runtime-key",
timeout_seconds=10,
max_tokens=16,
temperature=0.2,
),
)
assert isinstance(provider, _FakeProvider)
assert provider.kwargs["api_key"] == "runtime-key"
assert provider.kwargs["model"] == "runtime-model"
assert provider.kwargs["base_url"] == "https://api.openai.com/v1"
assert provider.kwargs["timeout_seconds"] == 10
assert provider.kwargs["max_tokens"] == 16
assert provider.kwargs["temperature"] == 0.2
def test_factory_allows_local_provider_without_real_api_key(monkeypatch) -> None:
monkeypatch.setattr(factory, "OpenAICompatibleProvider", _FakeProvider)
monkeypatch.setattr(factory, "load_dotenv", lambda *args, **kwargs: None)
monkeypatch.setattr(factory.settings, "llm_api_key", None)
monkeypatch.delenv("LLM_API_KEY", raising=False)
monkeypatch.delenv("API_KEY", raising=False)
provider = factory.build_llm_provider(
"p1_reference_zero_shot",
runtime_config=LLMRuntimeConfig(
provider="local_llm",
model="gemma-4-E4B-it-GGUF",
base_url="http://local-llm:8080/v1",
),
)
assert isinstance(provider, _FakeProvider)
assert provider.kwargs["api_key"] == "local-not-required"
assert provider.kwargs["model"] == "gemma-4-E4B-it-GGUF"
def test_factory_does_not_use_local_dummy_key_for_chat_ai(monkeypatch) -> None:
monkeypatch.setattr(factory, "OpenAICompatibleProvider", _FakeProvider)
monkeypatch.setattr(factory, "load_dotenv", lambda *args, **kwargs: None)
monkeypatch.setenv("LLM_API_KEY", "local-not-required")
monkeypatch.delenv("CHAT_AI_API_KEY", raising=False)
monkeypatch.delenv("API_KEY", raising=False)
try:
factory.build_llm_provider(
"p1_reference_zero_shot",
runtime_config=LLMRuntimeConfig(
provider="chat_ai",
model="gemma-3-27b-it",
base_url="https://chat-ai.academiccloud.de/v1",
),
)
except ValueError as exc:
assert "Missing LLM_API_KEY/API_KEY" in str(exc)
else:
raise AssertionError("Expected missing Chat AI API key error")
def test_factory_uses_chat_ai_specific_key(monkeypatch) -> None:
monkeypatch.setattr(factory, "OpenAICompatibleProvider", _FakeProvider)
monkeypatch.setattr(factory, "load_dotenv", lambda *args, **kwargs: None)
monkeypatch.setenv("LLM_API_KEY", "local-not-required")
monkeypatch.setenv("CHAT_AI_API_KEY", "chat-ai-key")
provider = factory.build_llm_provider(
"p1_reference_zero_shot",
runtime_config=LLMRuntimeConfig(
provider="chat_ai",
model="gemma-3-27b-it",
base_url="https://chat-ai.academiccloud.de/v1",
),
)
assert isinstance(provider, _FakeProvider)
assert provider.kwargs["api_key"] == "chat-ai-key"
asyst-backend/tests/test_llm_profiles.py 0 → 100644
View file @ f2a9f00f
from __future__ import annotations
from app.services.llm.profiles import build_runtime_config_for_profile
def test_local_gemma_profile_uses_local_llm_defaults(monkeypatch) -> None:
monkeypatch.delenv("LOCAL_LLM_BASE_URL", raising=False)
monkeypatch.delenv("LOCAL_LLM_MODEL", raising=False)
config = build_runtime_config_for_profile(llm_profile="local_gemma")
assert config is not None
assert config.provider == "local_llm"
assert config.base_url == "http://127.0.0.1:8080/v1"
assert config.model == "gemma-4-E4B-it-GGUF"
assert config.api_key == "local-not-required"
def test_chat_ai_profile_uses_chat_ai_environment(monkeypatch) -> None:
monkeypatch.setenv("CHAT_AI_BASE_URL", "https://chat-ai.academiccloud.de/v1")
monkeypatch.setenv("CHAT_AI_MODEL", "gemma-chat-ai")
monkeypatch.setenv("CHAT_AI_API_KEY", "chat-key")
config = build_runtime_config_for_profile(llm_profile="chat_ai")
assert config is not None
assert config.provider == "chat_ai"
assert config.base_url == "https://chat-ai.academiccloud.de/v1"
assert config.model == "gemma-chat-ai"
assert config.api_key == "chat-key"
def test_chat_ai_profile_uses_current_default_model(monkeypatch) -> None:
monkeypatch.delenv("CHAT_AI_MODEL", raising=False)
monkeypatch.setenv("CHAT_AI_API_KEY", "chat-key")
config = build_runtime_config_for_profile(llm_profile="chat_ai")
assert config is not None
assert config.model == "gemma-4-31b-it"
def test_chat_ai_profile_prefers_runtime_model(monkeypatch) -> None:
monkeypatch.setenv("CHAT_AI_MODEL", "gemma-4-31b-it")
monkeypatch.setenv("CHAT_AI_API_KEY", "chat-key")
config = build_runtime_config_for_profile(
llm_profile="chat_ai",
model="mistral-large-3-675b-instruct-2512",
)
assert config is not None
assert config.model == "mistral-large-3-675b-instruct-2512"
def test_openai_custom_profile_uses_runtime_model_and_key() -> None:
config = build_runtime_config_for_profile(
llm_profile="openai_custom",
model="gpt-test",
api_key="sk-runtime",
)
assert config is not None
assert config.provider == "openai"
assert config.base_url == "https://api.openai.com/v1"
assert config.model == "gpt-test"
assert config.api_key == "sk-runtime"
def test_no_profile_preserves_direct_runtime_config() -> None:
config = build_runtime_config_for_profile(
llm_profile=None,
provider="openai_compatible",
model="custom-model",
base_url="http://custom:8080/v1",
api_key="custom-key",
)
assert config is not None
assert config.provider == "openai_compatible"
assert config.model == "custom-model"
assert config.base_url == "http://custom:8080/v1"
assert config.api_key == "custom-key"
asyst-backend/tests/test_llm_prompts.py 0 → 100644
View file @ f2a9f00f
from __future__ import annotations
import pytest
from app.services.grading import validate_prompt_run_options
from app.services.llm.base import GradeRequest, GradingExample
from app.services.llm.score_parser import parse_grade_response, parse_score
from app.services.llm.strategies import get_prompt_strategy, normalize_strategy_name
def test_strategy_aliases_resolve_to_thesis_names() -> None:
assert normalize_strategy_name("zero_shot") == "p0_instruction_only"
assert normalize_strategy_name("one_shot") == "p1_reference_zero_shot"
assert normalize_strategy_name("p1_reference_zero_shot") == "p1_reference_zero_shot"
def test_p0_prompt_excludes_reference_answer_and_stays_score_only() -> None:
request = GradeRequest(
question_text="Was ist RAM?",
reference_answer="Temporary memory",
student_answer="RAM speichert temporaere Daten.",
max_points=2.0,
custom_grading_rules="Deduct 0.5 points for missing volatility.",
)
messages = get_prompt_strategy("p0_instruction_only").build_messages(request)
assert messages[0]["role"] == "system"
assert "Studentische Antwort" in messages[1]["content"]
assert "Musterloesung" not in messages[1]["content"]
assert "Bewertungsregeln" not in messages[1]["content"]
assert "Volle Punktzahl" not in messages[1]["content"]
assert "Deduct 0.5 points" not in messages[1]["content"]
assert "Gib nur die Zahl aus" in messages[1]["content"]
assert "feedback" not in messages[1]["content"]
def test_p1_prompt_includes_reference_answer_and_stays_score_only() -> None:
request = GradeRequest(
question_text="Was ist RAM?",
reference_answer="RAM speichert temporaere Daten aktiver Programme.",
student_answer="RAM speichert temporaere Daten.",
max_points=2.0,
custom_grading_rules="Deduct 0.5 points for missing volatility.",
)
messages = get_prompt_strategy("one_shot").build_messages(request)
assert "Musterloesung / Korrekturhinweise" in messages[1]["content"]
assert request.reference_answer in messages[1]["content"]
assert "Bewertungsregeln" not in messages[1]["content"]
assert "Volle Punktzahl" not in messages[1]["content"]
assert "Deduct 0.5 points" not in messages[1]["content"]
assert "[0.0, 0.5, 1.0, 1.5, 2.0]" in messages[1]["content"]
assert "Gib nur die Zahl aus" in messages[1]["content"]
assert "feedback" not in messages[1]["content"]
def test_feedback_strategy_requests_json_feedback() -> None:
request = GradeRequest(
question_text="Was ist RAM?",
reference_answer="RAM speichert temporaere Daten aktiver Programme.",
student_answer="RAM speichert temporaere Daten.",
max_points=2.0,
)
messages = get_prompt_strategy("p1_reference_zero_shot_with_feedback").build_messages(request)
assert '"score": 1.5' in messages[1]["content"]
assert '"feedback"' in messages[1]["content"]
assert "maximal 2 kurze Stichpunkte" in messages[1]["content"]
def test_p2_prompt_includes_default_grading_rules() -> None:
request = GradeRequest(
question_text="Was ist RAM?",
reference_answer="RAM speichert temporaere Daten aktiver Programme.",
student_answer="RAM speichert temporaere Daten.",
max_points=2.0,
)
messages = get_prompt_strategy("p2_reference_discrete").build_messages(request)
assert "Bewertungsregeln" in messages[1]["content"]
assert "Volle Punktzahl nur" in messages[1]["content"]
assert "Teilpunkte vergeben" in messages[1]["content"]
assert "Gib nur die Zahl aus" in messages[1]["content"]
def test_include_feedback_option_works_for_reference_rules_prompt() -> None:
request = GradeRequest(
question_text="Was ist RAM?",
reference_answer="RAM speichert temporaere Daten aktiver Programme.",
student_answer="RAM speichert temporaere Daten.",
max_points=2.0,
include_feedback=True,
)
content = get_prompt_strategy("p2_reference_discrete").build_messages(request)[1]["content"]
assert '"feedback"' in content
assert "maximal 2 kurze Stichpunkte" in content
assert "Gib nur die Zahl aus" not in content
def test_p2_prompt_uses_custom_grading_rules_when_provided() -> None:
request = GradeRequest(
question_text="Was ist RAM?",
reference_answer="RAM speichert temporaere Daten aktiver Programme.",
student_answer="RAM speichert temporaere Daten.",
max_points=2.0,
custom_grading_rules="Deduct 0.5 points if volatility is missing.",
)
content = get_prompt_strategy("p2_reference_discrete").build_messages(request)[1]["content"]
assert "Deduct 0.5 points if volatility is missing." in content
assert "Volle Punktzahl nur" not in content
def test_space_only_custom_rules_fall_back_to_defaults() -> None:
options = validate_prompt_run_options(
strategy="p2_reference_discrete",
custom_grading_rules=" ",
examples=None,
)
request = GradeRequest(
question_text="Was ist RAM?",
reference_answer="RAM speichert temporaere Daten aktiver Programme.",
student_answer="RAM speichert temporaere Daten.",
max_points=2.0,
custom_grading_rules=options.custom_grading_rules,
)
content = get_prompt_strategy("p2_reference_discrete").build_messages(request)[1]["content"]
assert options.custom_grading_rules is None
assert "Volle Punktzahl nur" in content
def test_prompt_options_ignore_custom_rules_for_p0_and_p1() -> None:
p0_options = validate_prompt_run_options(
strategy="p0_instruction_only",
custom_grading_rules="Deduct 0.5 points for missing volatility.",
examples=None,
)
p1_options = validate_prompt_run_options(
strategy="p1_reference_zero_shot",
custom_grading_rules="Deduct 0.5 points for missing volatility.",
examples=None,
)
assert p0_options.custom_grading_rules is None
assert p1_options.custom_grading_rules is None
def test_p3_prompt_includes_scored_examples() -> None:
request = GradeRequest(
question_text="Was ist RAM?",
reference_answer="RAM speichert temporaere Daten aktiver Programme.",
student_answer="RAM ist schneller Speicher.",
max_points=2.0,
examples=(
GradingExample(question="Was ist RAM?", answer="RAM speichert aktive Daten.", score=2.0),
GradingExample(question="Was ist RAM?", answer="RAM ist eine Festplatte.", score=0.0),
),
)
content = get_prompt_strategy("p3_true_one_shot").build_messages(request)[1]["content"]
assert "Bewertungsbeispiel 1" in content
assert "RAM speichert aktive Daten." in content
assert "Punktzahl: 2.0" in content
assert "Bewertungsbeispiel 2" in content
assert "RAM ist eine Festplatte." in content
assert "Zu bewertende Antwort" in content
def test_include_feedback_option_works_for_example_guided_prompt() -> None:
request = GradeRequest(
question_text="Was ist RAM?",
reference_answer="RAM speichert temporaere Daten aktiver Programme.",
student_answer="RAM ist schneller Speicher.",
max_points=2.0,
include_feedback=True,
examples=(
GradingExample(question="Was ist RAM?", answer="RAM speichert aktive Daten.", score=2.0),
),
)
content = get_prompt_strategy("p3_true_one_shot").build_messages(request)[1]["content"]
assert "Bewertungsbeispiel 1" in content
assert '"feedback"' in content
def test_p3_validation_rejects_missing_or_invalid_examples() -> None:
with pytest.raises(ValueError, match="requires at least 1"):
validate_prompt_run_options(
strategy="p3_true_one_shot",
custom_grading_rules=None,
examples=[],
)
with pytest.raises(ValueError, match="at most 3"):
validate_prompt_run_options(
strategy="p3_true_one_shot",
custom_grading_rules=None,
examples=[
{"question": "Q", "answer": "A", "score": "1"},
{"question": "Q", "answer": "A", "score": "1"},
{"question": "Q", "answer": "A", "score": "1"},
{"question": "Q", "answer": "A", "score": "1"},
],
)
with pytest.raises(ValueError, match="student answer"):
validate_prompt_run_options(
strategy="p3_true_one_shot",
custom_grading_rules=None,
examples=[{"question": "Q", "answer": " ", "score": "1"}],
)
with pytest.raises(ValueError, match="must be a number"):
validate_prompt_run_options(
strategy="p3_true_one_shot",
custom_grading_rules=None,
examples=[{"question": "Q", "answer": "A", "score": "full"}],
)
def test_parse_score_accepts_common_model_outputs() -> None:
assert parse_score("1.5", max_points=2.0) == (1.5, None)
assert parse_score("Punktzahl: 1,5", max_points=2.0) == (1.5, None)
assert parse_score('{"score": "1.5"}', max_points=2.0) == (1.5, None)
def test_parse_grade_response_accepts_short_feedback_and_caps_length() -> None:
parsed = parse_grade_response(
'{"score": "1.5", "feedback": "Good core idea, but one detail is missing."}',
max_points=2.0,
)
assert parsed.score == 1.5
assert parsed.feedback == "Good core idea, but one detail is missing."
assert parsed.warning is None
long_feedback = "x" * 400
capped = parse_grade_response(f'{{"score": 1.0, "feedback": "{long_feedback}"}}', max_points=2.0)
assert capped.feedback is not None
assert len(capped.feedback) == 280
def test_parse_grade_response_accepts_fenced_json_feedback() -> None:
parsed = parse_grade_response(
'```json\n{"score": 1.5, "feedback": "Partly correct."}\n```',
max_points=2.0,
)
assert parsed.score == 1.5
assert parsed.feedback == "Partly correct."
assert parsed.warning is None
def test_parse_score_clamps_and_snaps() -> None:
assert parse_score("9", max_points=2.0) == (2.0, None)
assert parse_score("-1", max_points=2.0) == (0.0, None)
score, warning = parse_score("1.2", max_points=2.0)
assert score == 1.0
assert warning == "snapped_from_1.2"
def test_parse_score_fails_cleanly_for_empty_or_text_only_output() -> None:
assert parse_score("", max_points=2.0) == (None, "empty_response")
assert parse_score("keine punktzahl", max_points=2.0) == (None, "no_numeric_score_found")
asyst-backend/tests/test_llm_provider.py 0 → 100644
View file @ f2a9f00f
from __future__ import annotations
from types import SimpleNamespace
from app.services.llm.base import GradeRequest
from app.services.llm.providers import openai_compatible
from app.services.llm.providers.openai_compatible import OpenAICompatibleProvider
from app.services.llm.strategies import get_prompt_strategy
class _FakeRawResponse:
headers = {"retry-after": "1"}
def parse(self) -> object:
message = SimpleNamespace(content="1.5")
choice = SimpleNamespace(message=message)
return SimpleNamespace(choices=[choice])
class _FakeCompletions:
def __init__(self) -> None:
self.calls: list[dict[str, object]] = []
self.with_raw_response = self
def create(self, **kwargs: object) -> _FakeRawResponse:
self.calls.append(kwargs)
return _FakeRawResponse()
class _FakeOpenAI:
last_instance: "_FakeOpenAI | None" = None
def __init__(self, *, api_key: str, base_url: str, timeout: float, max_retries: int) -> None:
self.api_key = api_key
self.base_url = base_url
self.timeout = timeout
self.max_retries = max_retries
self.chat = SimpleNamespace(completions=_FakeCompletions())
_FakeOpenAI.last_instance = self
class _RetryableError(Exception):
status_code = 503
class _FlakyCompletions(_FakeCompletions):
def create(self, **kwargs: object) -> _FakeRawResponse:
self.calls.append(kwargs)
if len(self.calls) == 1:
raise _RetryableError("model is loading")
return _FakeRawResponse()
class _FlakyOpenAI(_FakeOpenAI):
def __init__(self, *, api_key: str, base_url: str, timeout: float, max_retries: int) -> None:
super().__init__(api_key=api_key, base_url=base_url, timeout=timeout, max_retries=max_retries)
self.chat = SimpleNamespace(completions=_FlakyCompletions())
_FakeOpenAI.last_instance = self
def test_openai_compatible_provider_sends_configured_chat_request(monkeypatch) -> None:
monkeypatch.setattr(openai_compatible, "OpenAI", _FakeOpenAI)
provider = OpenAICompatibleProvider(
api_key="sk-test",
base_url="https://api.openai.com/v1",
model="test-model",
timeout_seconds=12.0,
retry_max_attempts=3,
retry_initial_seconds=1,
retry_max_seconds=10,
max_tokens=32,
temperature=0.0,
prompt_strategy=get_prompt_strategy("p1_reference_zero_shot"),
)
result = provider.grade_answer(
GradeRequest(
question_text="Was ist RAM?",
reference_answer="RAM speichert temporaere Daten.",
student_answer="Temporaerer Speicher.",
max_points=2.0,
)
)
fake_client = _FakeOpenAI.last_instance
assert fake_client is not None
assert fake_client.api_key == "sk-test"
assert fake_client.base_url == "https://api.openai.com/v1"
assert fake_client.timeout == 12.0
assert fake_client.max_retries == 0
call = fake_client.chat.completions.calls[0]
assert call["model"] == "test-model"
assert call["temperature"] == 0.0
assert call["max_tokens"] == 32
assert result.score == 1.5
assert result.raw_response == "1.5"
assert result.headers == {"retry-after": "1"}
def test_openai_compatible_provider_retries_transient_llm_errors(monkeypatch) -> None:
sleeps: list[float] = []
monkeypatch.setattr(openai_compatible, "OpenAI", _FlakyOpenAI)
monkeypatch.setattr(openai_compatible.time, "sleep", sleeps.append)
provider = OpenAICompatibleProvider(
api_key="local-not-required",
base_url="http://local-llm:8080/v1",
model="gemma",
timeout_seconds=12.0,
retry_max_attempts=2,
retry_initial_seconds=5,
retry_max_seconds=5,
max_tokens=1024,
temperature=0.0,
prompt_strategy=get_prompt_strategy("p0_instruction_only"),
)
result = provider.grade_answer(
GradeRequest(
question_text="Was ist RAM?",
reference_answer=None,
student_answer="Temporaerer Speicher.",
max_points=2.0,
)
)
fake_client = _FakeOpenAI.last_instance
assert fake_client is not None
assert len(fake_client.chat.completions.calls) == 2
assert sleeps == [5]
assert result.score == 1.5
asyst-backend/tests/test_review_score_filters.py 0 → 100644
View file @ f2a9f00f
from __future__ import annotations
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
from app.models.base import Base
from app.models.grading import Answer, GradingRun, Question, Student
from app.schemas.review import QuestionAnswerReviewResponse, QuestionReviewListResponse
from app.services.review_service import get_question_answers, get_run_questions
def _db_session() -> Session:
engine = create_engine("sqlite:///:memory:", future=True)
Base.metadata.create_all(bind=engine)
factory = sessionmaker(bind=engine, autoflush=False, autocommit=False, expire_on_commit=False)
return factory()
def _seed(db: Session) -> Question:
db.add(
GradingRun(
run_id="run",
filename="grades.csv",
file_path="grades.csv",
file_type="csv",
run_type="evaluation",
status="parsed",
)
)
question = Question(
run_id="run",
question_number=1,
question_text="Question?",
reference_answer="Reference",
max_points=2.0,
)
db.add(question)
db.flush()
for index, score in enumerate((0.5, 1.0, 1.5, 2.0), start=1):
student = Student(
run_id="run",
student_code=f"S{index}",
first_name="Test",
last_name=f"Student{index}",
email=f"student{index}@example.test",
row_index=index,
)
db.add(student)
db.flush()
db.add(
Answer(
run_id="run",
student_id=student.id,
question_id=question.id,
answer_text=f"Answer {index}",
processing_status="graded",
llm_score=score,
observed_score=score,
)
)
db.commit()
return question
def _query(db: Session, question: Question, scores: list[float], scale: str) -> dict[str, object]:
return get_question_answers(
db,
run_id="run",
question_id=question.id,
review_status=None,
page=1,
page_size=50,
sort="llm_score",
predicted_scores=scores,
score_scale=scale,
)
def test_question_answers_filter_exact_ai_scores_and_include_professor_score() -> None:
db = _db_session()
try:
question = _seed(db)
result = _query(db, question, [0.5, 2.0], "full")
assert result["pagination"]["total"] == 2
assert [item["llm_score"] for item in result["items"]] == [0.5, 2.0]
assert [item["observed_score"] for item in result["items"]] == [0.5, 2.0]
QuestionAnswerReviewResponse(**result)
finally:
db.close()
def test_question_answers_filter_uses_half_up_whole_number_categories() -> None:
db = _db_session()
try:
question = _seed(db)
ones = _query(db, question, [1.0], "whole")
twos = _query(db, question, [2.0], "whole")
assert [item["llm_score"] for item in ones["items"]] == [0.5, 1.0]
assert [item["llm_score"] for item in twos["items"]] == [1.5, 2.0]
finally:
db.close()
def test_question_overview_reports_only_ai_categories_that_were_predicted() -> None:
db = _db_session()
try:
_seed(db)
result = get_run_questions(db, "run")
assert result["items"][0]["predicted_score_counts"] == [
{"score": 0.5, "count": 1},
{"score": 1.0, "count": 1},
{"score": 1.5, "count": 1},
{"score": 2.0, "count": 1},
]
QuestionReviewListResponse(**result)
finally:
db.close()
asyst-backend/tests/test_run_delete_api.py 0 → 100644
View file @ f2a9f00f
from __future__ import annotations
import pytest
from fastapi import HTTPException
from app.api.routes import review as review_route
from app.services.run_delete_service import RunDeleteConflict, RunDeleteNotFound, RunDeleteSummary
def test_delete_grading_run_returns_summary(monkeypatch) -> None:
expected = RunDeleteSummary(
run_id="run-123",
deleted=True,
answers_deleted=3,
students_deleted=1,
questions_deleted=3,
grading_run_deleted=1,
file_deleted=True,
file_path="uploads/run-123.csv",
)
monkeypatch.setattr(review_route, "delete_run", lambda db, run_id: expected)
response = review_route.delete_grading_run("run-123")
assert response.run_id == "run-123"
assert response.deleted is True
assert response.answers_deleted == 3
assert response.file_deleted is True
def test_delete_grading_run_returns_404_for_unknown_run(monkeypatch) -> None:
def raise_not_found(db, run_id):
raise RunDeleteNotFound("Run not found")
monkeypatch.setattr(review_route, "delete_run", raise_not_found)
with pytest.raises(HTTPException) as exc_info:
review_route.delete_grading_run("missing")
assert exc_info.value.status_code == 404
def test_delete_grading_run_returns_409_for_running_run(monkeypatch) -> None:
def raise_conflict(db, run_id):
raise RunDeleteConflict("Run is currently grading and cannot be deleted")
monkeypatch.setattr(review_route, "delete_run", raise_conflict)
with pytest.raises(HTTPException) as exc_info:
review_route.delete_grading_run("running")
assert exc_info.value.status_code == 409
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment

Dies ist die Gitlab-Instanz des Transferportals der Hochschule für Technik Stuttgart. Hier geht es zurück zum Portal