diff --git a/bases/rsptx/book_server_api/routers/coach.py b/bases/rsptx/book_server_api/routers/coach.py index ce1b543f2..31ccfef7a 100644 --- a/bases/rsptx/book_server_api/routers/coach.py +++ b/bases/rsptx/book_server_api/routers/coach.py @@ -14,7 +14,7 @@ # Third-party imports # ------------------- -from fastapi import APIRouter, Request +from fastapi import APIRouter, Depends, Request from pyflakes import checker as pyflakes_checker # Local application imports @@ -32,8 +32,10 @@ from .assessment import get_question_source, SelectQRequest # Import function for fetching api - comment out for DEV purposes +from rsptx.auth.session import auth_manager from rsptx.db.crud.crud import fetch_api_token from rsptx.db.crud.course import fetch_course +from rsptx.db.crud.question import fetch_question # .. _APIRouter config: # @@ -85,6 +87,24 @@ async def python_check(request: Request): # for dev/test -- replace with your own key for local testing +def clean_python_testcase(raw_test_code: str) -> str: + """ + Transform Runestone browser-style test code to standard unittest format. + Mirrors the cleanTestcase() transformation in activecode.js so that + suffix_code from the DB can be run by JOBE (which has no unittest.gui). + """ + result = re.sub( + r"from unittest\.gui import TestCaseGui\s*\n", + "import unittest\n", + raw_test_code, + ) + result = result.replace( + "class myTests(TestCaseGui):", "class myTests(unittest.TestCase):" + ) + result = re.sub(r"^\s*myTests\(\)\.main\(\)\s*$", "", result, flags=re.MULTILINE) + return result + + def extract_parsons_code(html_block): """ Given the full HTML/pre block for a Parsons problem extracted from DB, @@ -145,7 +165,9 @@ async def get_question_html(request: Request, div_id: str): # @router.post("/ns/coach/parsons_scaffolding") @router.post("/parsons_scaffolding") -async def parsons_scaffolding(request: Request, course: Optional[str]): +async def parsons_scaffolding( + request: Request, course: Optional[str], user=Depends(auth_manager) +): # Get `course` directly from the query string rslogger.warning(f"URL seen: {request.url}") rslogger.warning(f"Query parameters: {request.query_params}") @@ -211,24 +233,57 @@ async def parsons_scaffolding(request: Request, course: Optional[str]): _ = req_bytes.decode("utf-8") data = await request.json() - language = data.get("language") # Capture the question language from the front end - student_code = data.get( - "student_code" - ) # Capture the student code from the front end - problem_id = data.get("problem_id") # Capture the problem name from the front end - personalization_level = data.get( - "personalization_level" - ) # Capture the personalization level set by the instructor from the front end - parsonsexample = data.get( - "parsonsexample" - ) # Capture whether the scaffolding puzzle is a pre-defined example or LLM-example - problem_description = data.get( - "problem_description" - ) # Capture the problem description from the front end - internal_test_case = data.get( - "internal_test_case" - ) # Capture the internal test case from the front end + language = data.get("language") + student_code = data.get("student_code") + problem_id = data.get("problem_id") + personalization_level = data.get("personalization_level") + parsonsexample = data.get("parsonsexample") + problem_description = data.get("problem_description") parsons_personalized = data.get("parsons_personalized", True) + + if not problem_id: + return JSONResponse( + content={"error": "CodeTailor: problem_id is required"}, + status_code=status.HTTP_400_BAD_REQUEST, + ) + + # Fetch the test code from the database using the problem_id. + try: + basecourse = getattr(course, "base_course", None) + question = await fetch_question(problem_id, basecourse=basecourse) + if question and question.question_json: + internal_test_case = question.question_json.get("suffix_code", "") or "" + else: + rslogger.error( + f"CodeTailor: no question found for problem_id '{problem_id}'" + ) + return JSONResponse( + content={"error": f"CodeTailor: question '{problem_id}' not found"}, + status_code=status.HTTP_400_BAD_REQUEST, + ) + except Exception as e: + rslogger.error(f"CodeTailor: could not fetch test code for '{problem_id}': {e}") + return JSONResponse( + content={ + "error": f"CodeTailor: could not fetch test code for '{problem_id}'" + }, + status_code=status.HTTP_400_BAD_REQUEST, + ) + + if not internal_test_case: + rslogger.error( + f"CodeTailor: question '{problem_id}' has no suffix_code in question_json — cannot validate generated code" + ) + return JSONResponse( + content={ + "error": f"CodeTailor: question '{problem_id}' has no test code in the database" + }, + status_code=status.HTTP_400_BAD_REQUEST, + ) + + if language and language.lower() == "python": + internal_test_case = clean_python_testcase(internal_test_case) + print("start_to: get_parsons_help", api_token, language, personalization_level) adaptive_attr = 'data-adaptive="true"' diff --git a/bases/rsptx/book_server_api/routers/personalized_parsons/evaluate_fixed_code.py b/bases/rsptx/book_server_api/routers/personalized_parsons/evaluate_fixed_code.py index 8a335ed22..4d93f0300 100644 --- a/bases/rsptx/book_server_api/routers/personalized_parsons/evaluate_fixed_code.py +++ b/bases/rsptx/book_server_api/routers/personalized_parsons/evaluate_fixed_code.py @@ -1,9 +1,5 @@ -import unittest import re import difflib -from types import ModuleType -import threading -import signal import requests as rq import hashlib import base64 @@ -11,22 +7,6 @@ from ..rsproxy import settings -class NullOutput: - def write(self, _): - pass - - def flush(self): - pass - - -class TimeoutError(Exception): - pass - - -def handler(signum, frame): - raise TimeoutError("Test execution exceeded time limit") - - def _runestone_file_id(filename: str, content: str) -> str: # Exactly: "runestone" + MD5(fileName + fileContent) md5 = hashlib.md5((filename + content).encode("utf-8")).hexdigest() @@ -229,45 +209,63 @@ def extract_class_name(code): return False +class _JobeTestResult: + """Minimal stand-in for unittest.TestResult returned by load_and_run_tests.""" + + def __init__(self, passed: bool): + self._passed = passed + + def wasSuccessful(self) -> bool: + return self._passed + + def load_and_run_tests(unittest_case, code_to_test, time_limit=6): """ - Load and run Python test cases against the provided code. + Run Python test cases against the provided code via JOBE. + Inputs: - unittest_case (str): The Python test cases. The test code is automatically reformatted based on the unittest_code provided by instructors in the RST file. - code_to_test (str): The Python code to be tested. - time_limit (int): The time limit for running the tests in seconds. - Output: unittest.TestResult: The result of the test run. + unittest_case (str): unittest source (class myTests(unittest.TestCase): ...) + code_to_test (str): the Python solution code to validate + time_limit (int): JOBE wall-clock time limit in seconds + Output: _JobeTestResult with wasSuccessful() method """ - # Set the alarm signal for timeout - if threading.current_thread() is threading.main_thread(): - signal.signal(signal.SIGALRM, handler) - signal.alarm(time_limit) + # Suppress __main__ guards in student code — JOBE runs as top-level script + combined = ( + '__name__ = "__runestone__"\n' + + code_to_test + + "\n\n" + + unittest_case + + "\n\nimport unittest as _ut\n" + + "_result = _ut.main(verbosity=0, exit=False)\n" + + 'print("PASS" if _result.result.wasSuccessful() else "FAIL")\n' + ) + + sess = _jobe_session() + runs_url = settings.jobe_server + "/jobe/index.php/restapi/runs/" try: - # Create a dummy module to hold the test cases - test_module = ModuleType("test_module") - test_module.unittest = unittest - - # Execute the test cases string within the dummy module's namespace - exec(unittest_case, test_module.__dict__) - # Execute the code to test within the desired scope - exec(code_to_test, test_module.__dict__) - # Retrieve the loaded test cases - test_suite = unittest.TestLoader().loadTestsFromModule(test_module) - print("test_suite", test_suite) - # Run the test suite - test_results = unittest.TextTestRunner( - verbosity=0, failfast=True, stream=NullOutput() - ).run(test_suite) - print("test_results", test_results) - - except TimeoutError: - print("test_results", test_results) - return False - finally: - signal.alarm(0) + runspec = { + "language_id": "python3", + "sourcecode": combined, + "sourcefilename": "solution.py", + "parameters": {"timelimitsecs": time_limit}, + } + resp = sess.post(runs_url, json={"run_spec": runspec}, timeout=time_limit + 10) + try: + result = resp.json() + except Exception: + return _JobeTestResult(False) - return test_results + out = (result.get("stdout") or "").strip() + # Check the last line so student debug prints don't cause false failures. + # Trust stdout over outcome code: JOBE/Python3 may return outcome 12 + # even when tests pass. + last_line = out.splitlines()[-1].strip() if out else "" + passed = last_line == "PASS" + return _JobeTestResult(passed) + + except Exception: + return _JobeTestResult(False) def fix_indentation(text):