From 2f0b624fe749fa039bd8e0335649839c6a5afec3 Mon Sep 17 00:00:00 2001 From: Aadarsh Padiyath Date: Mon, 26 Oct 2026 13:38:29 -0400 Subject: [PATCH 1/6] Move shared CodeTailor helpers into common book server modules No behavior change. Explain in Plain English questions (#1627) need the same question, token and JOBE plumbing CodeTailor has, so move it out of coach.py and evaluate_fixed_code.py: - activecode_questions.py: a question's starter code, tests and data files, plus find_question (base course first, then a global match) and question_test_code (question_json.suffix_code or the textarea tests) - openai_token.py: the course's OpenAI token and the dev-course key - jobe.py: the JOBE session, data file upload, and load_and_run_tests, now split around run_python_program, which returns a run's output Adds tests for the moved helpers, which had none. Co-Authored-By: Claude Opus 5.5 --- .../book_server_api/activecode_questions.py | 212 ++++++++++++++++ bases/rsptx/book_server_api/jobe.py | 130 ++++++++++ bases/rsptx/book_server_api/openai_token.py | 29 +++ bases/rsptx/book_server_api/routers/coach.py | 239 ++---------------- .../evaluate_fixed_code.py | 126 +-------- .../test_activecode_questions.py | 97 +++++++ .../test_clean_python_testcase.py | 2 +- 7 files changed, 500 insertions(+), 335 deletions(-) create mode 100644 bases/rsptx/book_server_api/activecode_questions.py create mode 100644 bases/rsptx/book_server_api/jobe.py create mode 100644 bases/rsptx/book_server_api/openai_token.py create mode 100644 test/bases/rsptx/book_server_api/test_activecode_questions.py diff --git a/bases/rsptx/book_server_api/activecode_questions.py b/bases/rsptx/book_server_api/activecode_questions.py new file mode 100644 index 000000000..c17c5960b --- /dev/null +++ b/bases/rsptx/book_server_api/activecode_questions.py @@ -0,0 +1,212 @@ +"""Read an activecode question's code, tests and data files from the database. + +CodeTailor uses these to check generated code against a question's own unit +tests on JOBE, with the files the question's program opens shipped alongside. +They split a question's source the way activecode.js does in the browser. +""" + +import html +import re + +from rsptx.db.crud import fetch_source_code +from rsptx.db.crud.question import fetch_question +from rsptx.logging import rslogger + + +async def find_question(name, basecourse): + """ + The question named ``name``, or None. Question names are unique only + within a base course, so look in ``basecourse`` first. A cloned course + keeps the original book's base_course on its question rows, and + selectquestion can pull an exercise from another book, so fall back to a + global name match -- the same resolution get_question_source and /htmlsrc + use. + """ + question = await fetch_question(name, basecourse=basecourse) if basecourse else None + if question: + return question + question = await fetch_question(name) + if question and basecourse: + rslogger.warning( + f"'{name}' not in base course '{basecourse}' " + f"(likely a cloned course); resolved via global name match" + ) + return question + + +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):" + ) + # A test file's own runner call (myTests().main() or unittest.main(...)) + # exits before JOBE's PASS line prints, which reads as a failed test. + result = re.sub( + r"^\s*(?:myTests\(\)|unittest)\.main\(.*\)\s*$", + "", + result, + flags=re.MULTILINE, + ) + return result + + +# Activecode source markers, matched the way activecode.js matches them: the +# first occurrence anywhere, newline or not. The invisible (test) suffix is +# four or more "=" -- activecode.js accepts "=====" and longer from older +# books -- and the invisible prefix four or more "^". "===!" and "^^^!" mark +# visible suffixes and prefixes, which stay in the editor. +_INVISIBLE_SUFFIX_RE = re.compile(r"={4,}") +_INVISIBLE_PREFIX_RE = re.compile(r"\^{4,}") +_VISIBLE_SUFFIX = "===!" +_VISIBLE_PREFIX = "^^^!" + + +def _drop_marker(text, marker): + """Remove the first ``marker`` and the newline right after it, if any.""" + start = text.find(marker) + if start == -1: + return text + end = start + len(marker) + if text[end : end + 1] == "\n": + end += 1 + return text[:start] + text[end:] + + +def extract_suffix_code_from_htmlsrc(htmlsrc): + """ + Book-authored activecode questions never get question_json.suffix_code + populated by the build pipeline (only instructor-authored custom + exercises do) -- their test code instead lives in the raw ", htmlsrc, flags=re.DOTALL) + text = html.unescape(match.group(1) if match else htmlsrc) + marker = _INVISIBLE_SUFFIX_RE.search(text) + if marker: + return text[marker.end() :].strip() + if _VISIBLE_SUFFIX in text: + return text.partition(_VISIBLE_SUFFIX)[2].strip() + return "" + + +def question_test_code(question): + """ + The question's unit tests: a custom exercise's question_json.suffix_code, + or for a book-authored question (whose question_json has none) the tests + parsed out of the raw textarea source. + """ + return (question.question_json or {}).get( + "suffix_code" + ) or extract_suffix_code_from_htmlsrc(question.htmlsrc) + + +def starter_code_from_question(question): + """ + The code a student sees in the editor before changing anything: a custom + exercise's question_json.starter_code, or for a book-authored question the + activecode ", question.htmlsrc or "", flags=re.DOTALL + ) + if not match: + return "" + text = html.unescape(match.group(1)) + prefix = _INVISIBLE_PREFIX_RE.search(text) + if prefix: + text = _drop_marker(text[prefix.start() :], prefix.group(0)) + text = _drop_marker(text, _VISIBLE_PREFIX) + suffix = _INVISIBLE_SUFFIX_RE.search(text) + if suffix: + text = text[: suffix.start()] + return _drop_marker(text, _VISIBLE_SUFFIX) + + +# A quoted name with a file extension, e.g. open("scores.csv") or +# new File("words.txt") -- a possible data file to look up in source_code. +# Single spaces inside the name are allowed, since uploaded datafiles keep +# theirs (open("my data.csv")). +_FILENAME_LITERAL_RE = re.compile( + r"""["']([\w\-./]+(?: [\w\-./]+)*\.[A-Za-z0-9]{1,8})["']""" +) +# Bounds the source_code lookups that guessed filename literals can trigger. +# Names the question declares in data-datafile are always kept. +_MAX_DATA_FILE_CANDIDATES = 20 + + +def data_file_candidates(htmlsrc, *code_sources): + """ + Filenames the question's program may open. Livecode (JOBE) questions list + them in data-datafile, but Python questions don't declare them anywhere -- + Skulpt resolves open() at runtime -- so also take every quoted + filename-like literal from the question's code, its tests and the + student's code. Names that turn out not to be data files simply miss in + the source_code lookup. + """ + declared = [] + htmlsrc = htmlsrc or "" + match = re.search(r'data-datafile\s*=\s*"([^"]*)"', htmlsrc) + if match: + declared.extend(name.strip() for name in match.group(1).split(",")) + declared = list(dict.fromkeys(name for name in declared if name)) + textarea = re.search(r"]*>(.*?)", htmlsrc, flags=re.DOTALL) + question_code = html.unescape(textarea.group(1)) if textarea else "" + inferred = [] + for text in (question_code, *code_sources): + inferred.extend(_FILENAME_LITERAL_RE.findall(text or "")) + inferred = [name for name in dict.fromkeys(inferred) if name not in declared] + return declared + inferred[:_MAX_DATA_FILE_CANDIDATES] + + +async def fetch_data_files(candidates, course_scopes): + """ + Resolve candidate filenames against source_code -- the same store + Skulpt's file reader and livecode.js fall back to -- so the files can be + shipped to JOBE with every test run. + + course_scopes is a list of (base_course, course_name) pairs tried in + order: the requesting course first, then the book the question came from + (which differs for cloned courses and cross-book selectquestions). + Returns dicts in the shape jobe.push_data_files expects. + """ + data_files = [] + for name in candidates: + for base_course, course_name in course_scopes: + try: + row = await fetch_source_code( + base_course=base_course, course_name=course_name, filename=name + ) + except Exception: + # fetch_source_code raises when nothing matches + row = None + if row and row.main_code is not None: + filename = row.filename or name + # data-datafile names the acid while the code names the file, + # so the same file can be resolved twice -- ship it once + if filename not in {data_file["filename"] for data_file in data_files}: + data_files.append( + { + "filename": filename, + "content": row.main_code, + "is_binary": bool(row.is_binary), + } + ) + break + return data_files diff --git a/bases/rsptx/book_server_api/jobe.py b/bases/rsptx/book_server_api/jobe.py new file mode 100644 index 000000000..6a8d4b3a0 --- /dev/null +++ b/bases/rsptx/book_server_api/jobe.py @@ -0,0 +1,130 @@ +"""Run Python programs on JOBE from the book server. + +Used by CodeTailor to check the code it generates against a question's unit +tests. Data files the program opens are pushed to JOBE first, the way +livecode.js ships them with a run. +""" + +import base64 +import hashlib + +import requests as rq + +from rsptx.configuration import settings + + +def runestone_file_id(filename: str, content: str) -> str: + # Exactly: "runestone" + MD5(fileName + fileContent) + md5 = hashlib.md5((filename + content).encode("utf-8")).hexdigest() + return "runestone" + md5 + + +def b64_text_utf8(s: str) -> str: + return base64.b64encode(s.encode("utf-8")).decode("ascii") + + +def jobe_session(): + s = rq.Session() + s.headers["Content-type"] = "application/json; charset=utf-8" + s.headers["Accept"] = "application/json" + if getattr(settings, "jobe_key", None): + s.headers["X-API-KEY"] = settings.jobe_key + return s + + +def push_data_files(sess, data_files): + """ + Upload the question's data files to JOBE so the program under test can + open() them, mirroring how livecode.js ships datafiles with a run. + Inputs: + sess: requests session with the JOBE headers set + data_files (list): dicts with "filename", "content" and "is_binary"; + binary contents are already base64 (as stored in source_code) + Output: list: [file_id, filename] pairs for the run_spec file_list + """ + file_list = [] + for data_file in data_files or []: + filename = data_file["filename"] + content = data_file["content"] + if data_file.get("is_binary"): + b64 = content + else: + b64 = b64_text_utf8(content) + file_id = runestone_file_id(filename, content) + file_url = settings.jobe_server + "/jobe/index.php/restapi/files/" + file_id + if sess.head(file_url, timeout=10).status_code != 204: + put = sess.put(file_url, json={"file_contents": b64}, timeout=10) + if put.status_code != 204: + raise RuntimeError( + f"Failed to push data file {filename} to JOBE: {put.status_code}" + ) + file_list.append([file_id, filename]) + return file_list + + +def run_python_program(program, time_limit=6, data_files=None): + """ + Run a Python program on JOBE with the question's data files beside it. + Returns the run's (stdout, stderr). Raises on a JOBE or network failure. + """ + sess = jobe_session() + runspec = { + "language_id": "python3", + "sourcecode": program, + "sourcefilename": "solution.py", + "parameters": {"timelimitsecs": time_limit}, + } + if data_files: + runspec["file_list"] = push_data_files(sess, data_files) + resp = sess.post( + settings.jobe_server + "/jobe/index.php/restapi/runs/", + json={"run_spec": runspec}, + timeout=time_limit + 10, + ) + result = resp.json() + return result.get("stdout") or "", result.get("stderr") or "" + + +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, data_files=None): + """ + Run Python test cases against the provided code via JOBE. + + Inputs: + 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 + data_files (list): the question's data files to place beside the program (see push_data_files) + Output: _JobeTestResult with wasSuccessful() method + """ + # 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' + ) + + try: + stdout, _ = run_python_program(combined, time_limit, data_files) + except Exception: + return _JobeTestResult(False) + + out = stdout.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 "" + return _JobeTestResult(last_line == "PASS") diff --git a/bases/rsptx/book_server_api/openai_token.py b/bases/rsptx/book_server_api/openai_token.py new file mode 100644 index 000000000..f5aab58d0 --- /dev/null +++ b/bases/rsptx/book_server_api/openai_token.py @@ -0,0 +1,29 @@ +"""The OpenAI API token a course's instructor added for its LLM features. + +CodeTailor calls OpenAI with the course's own token, never a server-wide key. +""" + +from rsptx.db.crud.crud import fetch_api_token + +# for dev/test -- replace with your own key for local testing +DEV_API_KEY = "" +# The test course for development, which uses DEV_API_KEY +DEV_COURSE = "personalized_parsons" + + +def uses_dev_api_key(course_name): + return course_name is None or course_name == DEV_COURSE + + +async def fetch_openai_token(course): + """ + The course's least recently used OpenAI token (decrypted), or None if + its instructor hasn't added one. + """ + token_record = await fetch_api_token( + course_id=course.id, + provider="openai", # from add_token.html