#!/usr/bin/env python3
from __future__ import annotations
import argparse
import json
import os
import sys
import time
from datetime import UTC, datetime
from pathlib import Path
from typing import Any
from google.oauth2 import service_account
from googleapiclient.discovery import build
ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
from server import ( # noqa: E402
classroom_add_student_to_course,
classroom_create_assignment,
classroom_create_topic,
classroom_get_course,
classroom_list_course_students,
classroom_list_student_submissions,
classroom_remove_student_from_course,
classroom_return_submission,
classroom_set_draft_grade,
classroom_turn_in_submission,
)
def _arg_or_env(value: str | None, env_name: str) -> str | None:
if value:
return value
return os.getenv(env_name)
def _require_arg_or_env(value: str | None, env_name: str) -> str:
resolved = _arg_or_env(value, env_name)
if not resolved:
raise RuntimeError(
f"Missing required value. Provide --{env_name.lower()} or set {env_name}."
)
return resolved
def _contains_email(student_item: Dict[str, Any], email: str) -> bool:
profile = student_item.get("profile") or {}
found = (profile.get("emailAddress") or "").lower()
return found == email.lower()
def _build_teacher_cleanup_service(service_account_file: str, teacher_email: str):
creds = service_account.Credentials.from_service_account_file(
service_account_file,
scopes=[
"https://www.googleapis.com/auth/classroom.topics",
"https://www.googleapis.com/auth/classroom.coursework.students",
"https://www.googleapis.com/auth/classroom.rosters",
],
subject=teacher_email,
)
return build("classroom", "v1", credentials=creds, cache_discovery=False)
def _append_test(
report: Dict[str, Any],
name: str,
ok: bool,
summary: Dict[str, Any] | None = None,
error: str | None = None,
) -> None:
item: Dict[str, Any] = {"name": name, "ok": ok}
if summary is not None:
item["summary"] = summary
if error is not None:
item["error"] = error
report["tests"].append(item)
print(("PASS" if ok else "FAIL"), name, ("" if ok else error))
def _append_cleanup(
report: Dict[str, Any],
name: str,
ok: bool,
detail: str,
) -> None:
report["cleanup"].append({"name": name, "ok": ok, "detail": detail})
print(("OK" if ok else "WARN"), f"cleanup::{name}", detail)
def main() -> int:
parser = argparse.ArgumentParser(
description="Run end-to-end Google Classroom MCP smoke test with cleanup."
)
parser.add_argument("--teacher-email")
parser.add_argument("--student-email")
parser.add_argument("--course-id")
parser.add_argument("--draft-grade", type=float, default=97.0)
parser.add_argument(
"--keep-artifacts",
action="store_true",
help="Do not delete created topic/assignment or remove added student.",
)
args = parser.parse_args()
teacher_email = _require_arg_or_env(args.teacher_email, "SMOKE_TEST_TEACHER_EMAIL")
student_email = _require_arg_or_env(args.student_email, "SMOKE_TEST_STUDENT_EMAIL")
course_id = _require_arg_or_env(args.course_id, "SMOKE_TEST_COURSE_ID")
service_account_file = os.getenv("GOOGLE_SERVICE_ACCOUNT_FILE")
if not service_account_file:
raise RuntimeError("Missing GOOGLE_SERVICE_ACCOUNT_FILE in environment.")
report: Dict[str, Any] = {
"timestamp_utc": datetime.now(UTC).isoformat().replace("+00:00", "Z"),
"phase": "smoke_test",
"inputs": {
"teacher_email": teacher_email,
"student_email": student_email,
"course_id": course_id,
"draft_grade": args.draft_grade,
"keep_artifacts": args.keep_artifacts,
},
"tests": [],
"cleanup": [],
}
created_topic_id: str | None = None
created_assignment_id: str | None = None
submission_id: str | None = None
added_student_this_run = False
success = True
stamp = datetime.now(UTC).strftime("%Y%m%d-%H%M%S")
teacher_cleanup_service = _build_teacher_cleanup_service(
service_account_file, teacher_email
)
try:
course = classroom_get_course(course_id, acting_user_email=teacher_email)
_append_test(
report,
"get_course_as_teacher",
True,
{
"id": course.get("id"),
"name": course.get("name"),
"state": course.get("courseState"),
},
)
students_before = classroom_list_course_students(
course_id, acting_user_email=teacher_email
)
already_member = any(
_contains_email(s, student_email)
for s in students_before.get("students", [])
)
if not already_member:
try:
add_resp = classroom_add_student_to_course(
course_id=course_id,
student_email=student_email,
acting_user_email=teacher_email,
)
added_student_this_run = True
_append_test(
report,
"ensure_student_membership",
True,
{
"added": True,
"userId": add_resp.get("userId"),
},
)
except Exception as exc:
msg = str(exc)
if "HTTP 409" in msg or "already exists" in msg.lower():
_append_test(
report,
"ensure_student_membership",
True,
{"added": False, "reason": "already_member_conflict"},
)
else:
raise
else:
_append_test(
report,
"ensure_student_membership",
True,
{"added": False, "reason": "already_member"},
)
topic = classroom_create_topic(
course_id=course_id,
name=f"ZZ SMOKE TEST TOPIC {stamp}",
acting_user_email=teacher_email,
)
created_topic_id = topic.get("topicId")
_append_test(
report,
"create_topic",
True,
{"topic_id": created_topic_id},
)
assignment = classroom_create_assignment(
course_id=course_id,
title=f"ZZ SMOKE TEST ASSIGNMENT {stamp}",
instructions="Smoke test: turn in + grade + return",
state="PUBLISHED",
max_points=100.0,
topic_id=created_topic_id,
acting_user_email=teacher_email,
)
created_assignment_id = assignment.get("id")
_append_test(
report,
"create_assignment_published",
True,
{"assignment_id": created_assignment_id, "state": assignment.get("state")},
)
submission = None
for _ in range(12):
subs = classroom_list_student_submissions(
course_id=course_id,
course_work_id=created_assignment_id,
user_id=student_email,
acting_user_email=teacher_email,
)
items = subs.get("studentSubmissions", [])
if items:
submission = items[0]
break
time.sleep(2)
if not submission:
raise RuntimeError("No submission found for target student.")
submission_id = submission.get("id")
_append_test(
report,
"find_submission_for_student",
True,
{"submission_id": submission_id, "state": submission.get("state")},
)
try:
classroom_turn_in_submission(
course_id=course_id,
course_work_id=created_assignment_id,
submission_id=submission_id,
acting_user_email=student_email,
)
except Exception:
pass
turned_state = None
for _ in range(8):
subs = classroom_list_student_submissions(
course_id=course_id,
course_work_id=created_assignment_id,
user_id=student_email,
acting_user_email=teacher_email,
)
items = subs.get("studentSubmissions", [])
if items:
turned_state = items[0].get("state")
if turned_state == "TURNED_IN":
break
time.sleep(2)
if turned_state != "TURNED_IN":
raise RuntimeError(f"Expected TURNED_IN, got {turned_state!r}.")
_append_test(
report,
"turn_in_submission_as_student",
True,
{"state_after": turned_state},
)
graded = classroom_set_draft_grade(
course_id=course_id,
course_work_id=created_assignment_id,
submission_id=submission_id,
draft_grade=args.draft_grade,
acting_user_email=teacher_email,
)
_append_test(
report,
"set_draft_grade",
True,
{
"draftGrade": graded.get("draftGrade"),
"state": graded.get("state"),
},
)
classroom_return_submission(
course_id=course_id,
course_work_id=created_assignment_id,
submission_id=submission_id,
acting_user_email=teacher_email,
)
final_state = None
for _ in range(8):
subs = classroom_list_student_submissions(
course_id=course_id,
course_work_id=created_assignment_id,
user_id=student_email,
acting_user_email=teacher_email,
)
items = subs.get("studentSubmissions", [])
if items:
final_state = items[0].get("state")
if final_state == "RETURNED":
break
time.sleep(2)
if final_state != "RETURNED":
raise RuntimeError(f"Expected RETURNED, got {final_state!r}.")
_append_test(
report,
"return_submission",
True,
{"final_state": final_state},
)
except Exception as exc:
success = False
_append_test(
report,
"smoke_test_flow",
False,
error=f"{type(exc).__name__}: {exc}",
)
finally:
if args.keep_artifacts:
_append_cleanup(report, "keep_artifacts", True, "enabled")
else:
try:
if created_assignment_id:
teacher_cleanup_service.courses().courseWork().delete(
courseId=course_id,
id=created_assignment_id,
).execute()
_append_cleanup(
report,
"delete_assignment",
True,
created_assignment_id,
)
else:
_append_cleanup(report, "delete_assignment", True, "skipped")
except Exception as exc:
_append_cleanup(
report,
"delete_assignment",
False,
f"{type(exc).__name__}: {exc}",
)
success = False
try:
if created_topic_id:
teacher_cleanup_service.courses().topics().delete(
courseId=course_id,
id=created_topic_id,
).execute()
_append_cleanup(report, "delete_topic", True, created_topic_id)
else:
_append_cleanup(report, "delete_topic", True, "skipped")
except Exception as exc:
_append_cleanup(
report,
"delete_topic",
False,
f"{type(exc).__name__}: {exc}",
)
success = False
try:
if added_student_this_run:
classroom_remove_student_from_course(
course_id=course_id,
student_email=student_email,
acting_user_email=teacher_email,
)
_append_cleanup(
report,
"remove_student_from_course",
True,
student_email,
)
else:
_append_cleanup(
report, "remove_student_from_course", True, "skipped"
)
except Exception as exc:
_append_cleanup(
report,
"remove_student_from_course",
False,
f"{type(exc).__name__}: {exc}",
)
success = False
print("\n=== SMOKE TEST REPORT ===")
print(json.dumps(report, ensure_ascii=False, indent=2))
return 0 if success else 1
if __name__ == "__main__":
raise SystemExit(main())
