]*>(.*?)", 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/components/rsptx/code_runner/jobe.py b/components/rsptx/code_runner/jobe.py
new file mode 100644
index 000000000..f0f306769
--- /dev/null
+++ b/components/rsptx/code_runner/jobe.py
@@ -0,0 +1,130 @@
+"""Run Python programs on JOBE.
+
+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/projects/book_server/pyproject.toml b/projects/book_server/pyproject.toml
index da9c54c78..e7a0d3abe 100644
--- a/projects/book_server/pyproject.toml
+++ b/projects/book_server/pyproject.toml
@@ -109,3 +109,5 @@ bypass-selection = true
"../../components/rsptx/lti1p3" = "rsptx/lti1p3"
"../../components/rsptx/lti1p1" = "rsptx/lti1p1"
"../../components/rsptx/data_types" = "rsptx/data_types"
+"../../components/rsptx/ai_helpers" = "rsptx/ai_helpers"
+"../../components/rsptx/code_runner" = "rsptx/code_runner"
diff --git a/test/bases/rsptx/book_server_api/test_java_jobe_failures.py b/test/bases/rsptx/book_server_api/test_java_jobe_failures.py
new file mode 100644
index 000000000..1dd46bda8
--- /dev/null
+++ b/test/bases/rsptx/book_server_api/test_java_jobe_failures.py
@@ -0,0 +1,45 @@
+"""
+A Java test run that never reaches JOBE's verdict must count as failing.
+"""
+
+from types import SimpleNamespace
+
+from rsptx.book_server_api.routers.personalized_parsons import evaluate_fixed_code
+
+_JAVA = "public class Main {\n public static int one() { return 1; }\n}"
+_TESTS = "public class MainTest extends CodeTestHelper {\n}"
+
+
+class _FakeSession:
+ def __init__(self, put_status=204, run_body=None):
+ self.headers = {}
+ self.put_status = put_status
+ self.run_body = run_body
+
+ def head(self, url, timeout=None):
+ return SimpleNamespace(status_code=404)
+
+ def put(self, url, json=None, timeout=None):
+ return SimpleNamespace(status_code=self.put_status, text="nope")
+
+ def post(self, url, json=None, timeout=None):
+ def bad_json():
+ raise ValueError("not json")
+
+ return SimpleNamespace(status_code=502, text=self.run_body, json=bad_json)
+
+
+def _use_session(monkeypatch, session):
+ monkeypatch.setattr(evaluate_fixed_code.rq, "Session", lambda: session)
+
+
+def test_failed_file_push_is_a_failure(monkeypatch):
+ _use_session(monkeypatch, _FakeSession(put_status=500))
+ assert evaluate_fixed_code.load_and_run_java_tests(_JAVA, _TESTS) is False
+ passed, _ = evaluate_fixed_code.java_unittest_evaluation(_JAVA, "", "", _TESTS)
+ assert passed is False
+
+
+def test_non_json_reply_is_a_failure(monkeypatch):
+ _use_session(monkeypatch, _FakeSession(run_body="Bad Gateway"))
+ assert evaluate_fixed_code.load_and_run_java_tests(_JAVA, _TESTS) is False
diff --git a/test/components/rsptx/ai_helpers/__init__.py b/test/components/rsptx/ai_helpers/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/test/components/rsptx/ai_helpers/test_api_token.py b/test/components/rsptx/ai_helpers/test_api_token.py
new file mode 100644
index 000000000..e2df76cef
--- /dev/null
+++ b/test/components/rsptx/ai_helpers/test_api_token.py
@@ -0,0 +1,69 @@
+"""
+Tests for choosing the course's LLM token and provider for CodeTailor.
+"""
+
+from types import SimpleNamespace
+
+import pytest
+
+from rsptx.ai_helpers import api_token
+from rsptx.ai_helpers.api_token import LLMToken, fetch_llm_token
+
+# Session loop scope keeps later DB tests on the shared loop (see AGENTS.md)
+_asyncio = pytest.mark.asyncio(loop_scope="session")
+
+_COURSE = SimpleNamespace(id=7)
+
+
+def _fake_tokens(monkeypatch, tokens):
+ """Make fetch_api_token return tokens[provider] for _COURSE."""
+
+ async def fetch_api_token(course_id, provider):
+ assert course_id == _COURSE.id
+ token = tokens.get(provider)
+ return SimpleNamespace(token=token) if token is not None else None
+
+ monkeypatch.setattr(api_token, "fetch_api_token", fetch_api_token)
+
+
+@_asyncio
+async def test_umgpt_token_used_when_course_has_no_openai_token(monkeypatch):
+ _fake_tokens(monkeypatch, {"umgpt": "um-key"})
+ assert await fetch_llm_token(_COURSE) == LLMToken("um-key", "umgpt")
+
+
+@_asyncio
+async def test_openai_token_preferred_when_course_has_both(monkeypatch):
+ _fake_tokens(monkeypatch, {"openai": "sk-key", "umgpt": "um-key"})
+ assert await fetch_llm_token(_COURSE) == LLMToken("sk-key", "openai")
+
+
+@_asyncio
+async def test_no_token(monkeypatch):
+ _fake_tokens(monkeypatch, {"openai": ""})
+ assert await fetch_llm_token(_COURSE) is None
+
+
+def test_client_and_model_follow_provider():
+ um = LLMToken("um-key", "umgpt")
+ assert str(um.client().base_url).startswith(
+ "https://api.toolkit.umgpt.umich.edu/v1"
+ )
+ assert um.model == "gpt-4o-mini"
+
+ oa = LLMToken("sk-key")
+ assert str(oa.client().base_url).startswith("https://api.openai.com/v1")
+ assert oa.model == "gpt-5-nano"
+
+
+def test_repr_masks_key():
+ token = LLMToken("sk-abSECRETSECRETSECRETxyz9", "openai")
+ assert token.masked() == "sk-a****xyz9"
+ assert "SECRET" not in repr(token)
+
+
+def test_short_key_shows_less():
+ token = LLMToken("abcd1234", "openai")
+ assert token.masked() == "a****4"
+ assert "abcd" not in repr(token) and "1234" not in repr(token)
+ assert LLMToken("abc").masked() == "****"
diff --git a/test/components/rsptx/code_runner/__init__.py b/test/components/rsptx/code_runner/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/test/components/rsptx/code_runner/test_activecode_questions.py b/test/components/rsptx/code_runner/test_activecode_questions.py
new file mode 100644
index 000000000..c5752fcbf
--- /dev/null
+++ b/test/components/rsptx/code_runner/test_activecode_questions.py
@@ -0,0 +1,97 @@
+"""
+Tests for the helpers that read an activecode question's code, tests and data
+files from its stored source, and for the JOBE test runner. CodeTailor uses
+them to check generated code against a question's own unit tests.
+"""
+
+from types import SimpleNamespace
+
+from rsptx.code_runner import jobe
+from rsptx.code_runner.activecode_questions import (
+ data_file_candidates,
+ extract_suffix_code_from_htmlsrc,
+ question_test_code,
+ starter_code_from_question,
+)
+
+_HTMLSRC = """\
+
+
+
"""
+
+
+def _question(htmlsrc=_HTMLSRC, question_json=None):
+ return SimpleNamespace(htmlsrc=htmlsrc, question_json=question_json)
+
+
+def test_starter_code_drops_hidden_parts_and_markers():
+ assert starter_code_from_question(_question()) == (
+ "# Return the average of nums\n"
+ "def average(nums):\n"
+ " return sum(nums) / len(nums)\n"
+ "print(average([1, 2, 3]))\n"
+ )
+
+
+def test_starter_code_prefers_question_json():
+ question = _question(question_json={"starter_code": "def f():\n pass\n"})
+ assert starter_code_from_question(question) == "def f():\n pass\n"
+
+
+def test_suffix_code_is_the_text_after_the_invisible_marker():
+ assert extract_suffix_code_from_htmlsrc(_HTMLSRC).startswith(
+ "from unittest.gui import TestCaseGui"
+ )
+ assert extract_suffix_code_from_htmlsrc("") == ""
+
+
+def test_question_test_code_prefers_question_json():
+ question = _question(question_json={"suffix_code": "import unittest"})
+ assert question_test_code(question) == "import unittest"
+ assert question_test_code(_question()) == extract_suffix_code_from_htmlsrc(_HTMLSRC)
+
+
+def test_data_file_candidates_lists_declared_files_first():
+ assert data_file_candidates(_HTMLSRC, 'open("mine.txt")') == [
+ "scores.csv",
+ "words.txt",
+ "mine.txt",
+ ]
+
+
+def test_load_and_run_tests_reads_the_last_line(monkeypatch):
+ runs = []
+
+ def run(program, time_limit, data_files):
+ runs.append(program)
+ return "debug print\nPASS\n", ""
+
+ monkeypatch.setattr(jobe, "run_python_program", run)
+ assert jobe.load_and_run_tests("import unittest", "x = 1").wasSuccessful()
+ assert runs[0].startswith('__name__ = "__runestone__"\nx = 1\n')
+
+ monkeypatch.setattr(jobe, "run_python_program", lambda *a: ("PASS\nFAIL\n", ""))
+ assert not jobe.load_and_run_tests("import unittest", "x = 1").wasSuccessful()
+
+
+def test_load_and_run_tests_fails_when_jobe_does(monkeypatch):
+ def run(*args):
+ raise ConnectionError("JOBE is down")
+
+ monkeypatch.setattr(jobe, "run_python_program", run)
+ assert not jobe.load_and_run_tests("import unittest", "x = 1").wasSuccessful()
diff --git a/test/bases/rsptx/book_server_api/test_clean_python_testcase.py b/test/components/rsptx/code_runner/test_clean_python_testcase.py
similarity index 94%
rename from test/bases/rsptx/book_server_api/test_clean_python_testcase.py
rename to test/components/rsptx/code_runner/test_clean_python_testcase.py
index 103fc25e9..f18cb06ff 100644
--- a/test/bases/rsptx/book_server_api/test_clean_python_testcase.py
+++ b/test/components/rsptx/code_runner/test_clean_python_testcase.py
@@ -10,7 +10,7 @@
import pytest
-from rsptx.book_server_api.routers.coach import clean_python_testcase
+from rsptx.code_runner.activecode_questions import clean_python_testcase
_TESTS = """\
import unittest