Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
83 changes: 71 additions & 12 deletions cmscontrib/SolutionChecker.py
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@
- "WrongAnswer" (at least one testcase produced a wrong answer)
- "TimeLimitExceeded" (at least one testcase exceeded CPU time limit)
- "WallTimeLimitExceeded" (at least one testcase exceeded wall time limit)
- "MemoryLimitExceeded" (at least one testcase exceeded memory limit)
- "RuntimeError" (at least one testcase failed due to runtime error)

Such a file can be generated with `task-maker-rust export-solution-checks`.
Expand Down Expand Up @@ -256,6 +257,40 @@ def has_slow_testcases(
)
return False

@staticmethod
def get_max_execution_time(
details: list[dict[str, Any]] | None,
) -> float | None:
if not isinstance(details, list):
return None
max_time = None
for item in details:
if not isinstance(item, dict):
continue
testcases = item.get("testcases")
if isinstance(testcases, list):
for tc in testcases:
if not isinstance(tc, dict):
continue
t = tc.get("time")
if t is not None:
try:
val = float(t)
if max_time is None or val > max_time:
max_time = val
except (ValueError, TypeError):
pass
else:
t = item.get("time")
if t is not None:
try:
val = float(t)
if max_time is None or val > max_time:
max_time = val
except (ValueError, TypeError):
pass
return max_time

@staticmethod
def get_testcase_status(tc: dict[str, Any]) -> str:
outcome = tc.get("outcome")
Expand All @@ -265,24 +300,23 @@ def get_testcase_status(tc: dict[str, Any]) -> str:
if isinstance(text_list, list) and text_list
else str(text_list)
)
if "wall clock" in text_str.lower():
combined = f"{outcome} {text_str}".lower()
if "wall clock" in combined:
return "WallTimeLimitExceeded"
if tc.get("time_limit_was_exceeded", False) or "timed out" in text_str.lower():
if tc.get("time_limit_was_exceeded", False) or "timed out" in combined:
return "TimeLimitExceeded"
if (
"signal" in text_str.lower()
or "return code" in text_str.lower()
or "memory limit" in text_str.lower()
):
if "memory limit" in combined:
return "MemoryLimitExceeded"
if "signal" in combined or "return code" in combined:
return "RuntimeError"
if outcome == "Correct" or "output is correct" in text_str.lower():
if outcome == "Correct" or "output is correct" in combined:
return "Accepted"
if outcome == "Partially correct":
return "PartialScore"
if (
outcome == "Not correct"
or "output isn't correct" in text_str.lower()
or "wrong answer" in text_str.lower()
or "output isn't correct" in combined
or "wrong answer" in combined
):
return "WrongAnswer"
return "Unknown"
Expand Down Expand Up @@ -355,6 +389,7 @@ def check_single_subtask(
"WrongAnswer",
"TimeLimitExceeded",
"WallTimeLimitExceeded",
"MemoryLimitExceeded",
"RuntimeError",
]:
if check not in statuses:
Expand Down Expand Up @@ -407,6 +442,7 @@ def is_subtask_slow(st: dict[str, Any], time_limit: float) -> bool:
"WrongAnswer": "WA",
"TimeLimitExceeded": "TLE",
"WallTimeLimitExceeded": "WTL",
"MemoryLimitExceeded": "MLE",
"RuntimeError": "RTE",
"PartialScore": "PS",
"Zero": "0",
Expand Down Expand Up @@ -476,6 +512,28 @@ def format_report_table(

row_cells.append((cell_text, cell_color))

# Longest execution time
max_time = r.get("max_time")
if max_time is None:
max_time = self.get_max_execution_time(details)

if compilation_failed:
time_text = "-"
time_color = RED
elif max_time is not None:
time_text = f"{max_time:.3f}s"
if time_limit > 0 and max_time > time_limit:
time_color = RED
elif time_limit > 0 and max_time > time_limit * 0.5:
time_color = YELLOW
else:
time_color = GREEN
else:
time_text = "-"
time_color = ""

row_cells.append((time_text, time_color))

if compilation_failed:
score_text = "CE"
expected_str = ""
Expand Down Expand Up @@ -505,8 +563,8 @@ def format_report_table(

rows.append((row_cells, score_text, expected_str, total_color))

num_subtask_cols = num_subtasks + 1 # sol_name + subtasks
col_widths = [0] * num_subtask_cols
num_cols = max(len(row_cells) for row_cells, _, _, _ in rows)
col_widths = [0] * num_cols
score_col_width = 0

for row_cells, score_text, _, _ in rows:
Expand Down Expand Up @@ -657,6 +715,7 @@ def main():
"score": score,
"compilation_failed": compilation_failed,
"details": details,
"max_time": checker.get_max_execution_time(details),
"failed": failed,
"slow": slow,
}
Expand Down
68 changes: 68 additions & 0 deletions cmstestsuite/unit_tests/cmscontrib/SolutionCheckerTest.py
Original file line number Diff line number Diff line change
Expand Up @@ -245,6 +245,40 @@ def test_check_subtasks_runtime_error(self):
[],
)

def test_check_subtasks_memory_limit_exceeded(self):
details = [
{
"idx": 0,
"testcases": [
{
"outcome": "Not correct",
"text": ["Memory limit exceeded"],
}
],
}
]
self.assertEqual(
self.checker.check_subtasks(details, ["MemoryLimitExceeded"]), []
)

details_no_mle = [
{
"idx": 0,
"testcases": [
{
"outcome": "Not correct",
"text": ["Output isn't correct"],
}
],
}
]
errors = self.checker.check_subtasks(details_no_mle, ["MemoryLimitExceeded"])
self.assertEqual(len(errors), 1)
self.assertIn(
"expected MemoryLimitExceeded, got statuses ['WrongAnswer']",
errors[0],
)

def test_check_subtasks_multiple_statuses(self):
# Subtask with multiple different testcase failures
details = [
Expand Down Expand Up @@ -362,6 +396,33 @@ def test_has_slow_testcases(self):
details_flat = [{"idx": 0, "time": 0.8}]
self.assertTrue(self.checker.has_slow_testcases(details_flat, time_limit))

def test_get_max_execution_time(self):
# Subtask structure
details_subtasks = [
{
"idx": 0,
"testcases": [{"time": 0.1}, {"time": 0.35}],
},
{
"idx": 1,
"testcases": [{"time": 0.2}, {"time": 0.7}],
},
]
self.assertEqual(SolutionChecker.get_max_execution_time(details_subtasks), 0.7)

# Flat testcase structure
details_flat = [{"time": 0.2}, {"time": 0.85}, {"time": 0.5}]
self.assertEqual(SolutionChecker.get_max_execution_time(details_flat), 0.85)

# None / empty / no times
self.assertIsNone(SolutionChecker.get_max_execution_time(None))
self.assertIsNone(SolutionChecker.get_max_execution_time([]))
self.assertIsNone(
SolutionChecker.get_max_execution_time(
[{"idx": 0, "testcases": [{"outcome": "Correct"}]}]
)
)

def test_login_validations(self):
# Password without username
c1 = SolutionChecker(base_url="http://localhost:8888", password="pwd")
Expand Down Expand Up @@ -651,6 +712,10 @@ def test_format_report_table(self):
self.assertIn("AC", table_colored)
self.assertIn("WA", table_colored)
self.assertIn("CE", table_colored)
# Verify longest execution times
self.assertIn("0.100s", table_colored)
self.assertIn("0.700s", table_colored)
self.assertIn("0.001s", table_colored)
# Verify ANSI colors
self.assertIn("\x1b[32;1m", table_colored) # Green
self.assertIn("\x1b[33;1m", table_colored) # Yellow
Expand All @@ -664,6 +729,9 @@ def test_format_report_table(self):
self.assertIn("CE", table_plain)
self.assertIn("71 (expected 110-120)", table_plain)
self.assertIn("100", table_plain)
self.assertIn("0.100s", table_plain)
self.assertIn("0.700s", table_plain)
self.assertIn("0.001s", table_plain)

# Empty results -> empty string
self.assertEqual(self.checker.format_report_table([], time_limit), "")
Expand Down
Loading