diff --git a/cms/grading/scoretypes/Sum.py b/cms/grading/scoretypes/Sum.py index 9df93139fe..1a7a6fbb13 100644 --- a/cms/grading/scoretypes/Sum.py +++ b/cms/grading/scoretypes/Sum.py @@ -18,6 +18,7 @@ # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . +from cms import FEEDBACK_LEVEL_FULL, FEEDBACK_LEVEL_RESTRICTED from . import ScoreTypeAlone @@ -31,6 +32,7 @@ class Sum(ScoreTypeAlone): multiplied by the integer parameter. """ + parameters: int # Mark strings for localization. N_("#") @@ -40,6 +42,8 @@ class Sum(ScoreTypeAlone): N_("Memory used") N_("N/A") TEMPLATE = """\ +{% set show_timing = (details|any("contains", "time") + or details|any("contains", "memory")) %} @@ -52,7 +56,7 @@ class Sum(ScoreTypeAlone): - {% if feedback_level == FEEDBACK_LEVEL_FULL %} + {% if show_timing %} @@ -75,16 +79,16 @@ class Sum(ScoreTypeAlone): - {% if feedback_level == FEEDBACK_LEVEL_FULL %} + {% if show_timing %} + {% if show_timing %} @@ -102,6 +110,38 @@ class Sum(ScoreTypeAlone):
{% trans %}Details{% endtrans %} {% trans %}Execution time{% endtrans %} {{ loop.index }} {{ _(tc["outcome"]) }} {{ tc["text"]|format_status_text }} - {% if tc["time"] is not none %} + {% if "time" in tc and tc["time"] is not none %} {{ tc["time"]|format_duration }} {% else %} {% trans %}N/A{% endtrans %} {% endif %} - {% if tc["memory"] is not none %} + {% if "memory" in tc and tc["memory"] is not none %} {{ tc["memory"]|format_size }} {% else %} {% trans %}N/A{% endtrans %} @@ -93,7 +97,11 @@ class Sum(ScoreTypeAlone): {% endif %} {% else %}
+ {% else %} + + {% endif %} {% trans %}N/A{% endtrans %}
""" + def get_json_details( + self, + score_details: object, + feedback_level: str = FEEDBACK_LEVEL_RESTRICTED, + ) -> object: + """Filter score_details for Sum score type according to the + feedback level. + + """ + if score_details is None: + return None + if not isinstance(score_details, list): + return score_details + + filtered_testcases = [] + for tc in score_details: + if "outcome" in tc and "text" in tc: + filtered_tc = { + "idx": tc["idx"], + "outcome": tc["outcome"], + "text": tc["text"], + } + if feedback_level == FEEDBACK_LEVEL_FULL: + if "time" in tc: + filtered_tc["time"] = tc["time"] + if "memory" in tc: + filtered_tc["memory"] = tc["memory"] + filtered_testcases.append(filtered_tc) + else: + filtered_testcases.append({"idx": tc.get("idx")}) + return filtered_testcases + def max_scores(self): """See ScoreType.max_score.""" public_score = 0.0 @@ -120,8 +160,7 @@ def compute_score(self, submission_result): # XXX Lexicographical order by codename indices = sorted(self.public_testcases.keys()) - evaluations = dict((ev.codename, ev) - for ev in submission_result.evaluations) + evaluations = dict((ev.codename, ev) for ev in submission_result.evaluations) testcases = [] public_testcases = [] score = 0.0 @@ -131,13 +170,15 @@ def compute_score(self, submission_result): this_score = float(evaluations[idx].outcome) * self.parameters tc_outcome = self.get_public_outcome(this_score) score += this_score - testcases.append({ - "idx": idx, - "outcome": tc_outcome, - "text": evaluations[idx].text, - "time": evaluations[idx].execution_time, - "memory": evaluations[idx].execution_memory, - }) + testcases.append( + { + "idx": idx, + "outcome": tc_outcome, + "text": evaluations[idx].text, + "time": evaluations[idx].execution_time, + "memory": evaluations[idx].execution_memory, + } + ) if self.public_testcases[idx]: public_score += this_score public_testcases.append(testcases[-1]) diff --git a/cms/grading/scoretypes/abc.py b/cms/grading/scoretypes/abc.py index 127067f1d4..ad02038316 100644 --- a/cms/grading/scoretypes/abc.py +++ b/cms/grading/scoretypes/abc.py @@ -35,14 +35,17 @@ from typing import TypedDict, NotRequired from abc import ABCMeta, abstractmethod -from cms import FEEDBACK_LEVEL_RESTRICTED +from cms import ( + FEEDBACK_LEVEL_FULL, + FEEDBACK_LEVEL_RESTRICTED, + FEEDBACK_LEVEL_OI_RESTRICTED, +) from cms.db import SubmissionResult from cms.grading.steps import EVALUATION_MESSAGES from cms.locale import Translation, DEFAULT_TRANSLATION from cms.server.jinja2_toolbox import GLOBAL_ENVIRONMENT from jinja2 import Template - logger = logging.getLogger(__name__) @@ -59,8 +62,12 @@ class ScoreType(metaclass=ABCMeta): TEMPLATE = "" - def __init__(self, parameters: object, public_testcases: dict[str, bool], - score_precision: int): + def __init__( + self, + parameters: object, + public_testcases: dict[str, bool], + score_precision: int, + ): """Initializer. parameters: format is specified in the subclasses. @@ -75,12 +82,14 @@ def __init__(self, parameters: object, public_testcases: dict[str, bool], # Preload the maximum possible scores. try: - self.max_score, self.max_public_score, self.ranking_headers = \ + self.max_score, self.max_public_score, self.ranking_headers = ( self.max_scores() + ) except Exception as e: raise ValueError( "Unable to instantiate score type (probably due to invalid " - "values for the score type parameters): %s." % e) + "values for the score type parameters): %s." % e + ) self.template: Template = GLOBAL_ENVIRONMENT.from_string(self.TEMPLATE) @@ -110,7 +119,25 @@ def format_score( """ return "%s / %s" % ( translation.format_decimal(score), - translation.format_decimal(max_score)) + translation.format_decimal(max_score), + ) + + def get_json_details( + self, + score_details: object, + feedback_level: str = FEEDBACK_LEVEL_RESTRICTED, + ) -> object: + """Return a JSON-serializable object representing the score details + of a submission filtered according to the feedback level. + + score_details: the data saved by the score type + itself in the database; can be public or private. + feedback_level: the level of details to show to users. + + return: JSON-serializable filtered score details. + + """ + return score_details def get_html_details( self, @@ -132,20 +159,24 @@ def get_html_details( _ = translation.gettext n_ = translation.ngettext if score_details is None: - logger.error("Found a null score details string. " - "Try invalidating scores.") + logger.error("Found a null score details string. Try invalidating scores.") return _("Score details temporarily unavailable.") else: # FIXME we should provide to the template all the variables # of a typical CWS context as it's entitled to expect them. try: - return self.template.render(details=score_details, - feedback_level=feedback_level, - translation=translation, - gettext=_, ngettext=n_) + filtered_details = self.get_json_details(score_details, feedback_level) + return self.template.render( + details=filtered_details, + feedback_level=feedback_level, + translation=translation, + gettext=_, + ngettext=n_, + ) except Exception: - logger.exception("Found an invalid score details string. " - "Try invalidating scores.") + logger.exception( + "Found an invalid score details string. Try invalidating scores." + ) return _("Score details temporarily unavailable.") @abstractmethod @@ -189,6 +220,7 @@ class ScoreTypeAlone(ScoreType): obtain the score of a single submission and max_scores. """ + pass @@ -202,7 +234,9 @@ class ScoreTypeGroupParametersDict(TypedDict): # the format of parameters is impossible to type-hint correctly, it seems... # this hint is (mostly) correct for the methods this base class implements, # subclasses might need a longer tuple. -ScoreTypeGroupParameters = tuple[float, int | str | list[str]] | ScoreTypeGroupParametersDict +ScoreTypeGroupParameters = ( + tuple[float, int | str | list[str]] | ScoreTypeGroupParametersDict +) class ScoreTypeGroup(ScoreTypeAlone): @@ -223,6 +257,7 @@ class ScoreTypeGroup(ScoreTypeAlone): 'reduce'. """ + parameters: list[ScoreTypeGroupParameters] # Mark strings for localization. @@ -235,6 +270,8 @@ class ScoreTypeGroup(ScoreTypeAlone): N_("N/A") TEMPLATE = """\ {% for st in details %} +{% set show_timing = (st["testcases"]|any("contains", "time") + or st["testcases"]|any("contains", "memory")) %} {% if "score_fraction" in st %} {% if st["score_fraction"] >= 1.0 %}
@@ -274,7 +311,7 @@ class ScoreTypeGroup(ScoreTypeAlone): {% trans %}Details{% endtrans %} - {% if feedback_level == FEEDBACK_LEVEL_FULL %} + {% if show_timing %} {% trans %}Execution time{% endtrans %} @@ -286,13 +323,7 @@ class ScoreTypeGroup(ScoreTypeAlone): {% for tc in st["testcases"] %} - {% set show_tc = "outcome" in tc - and ((feedback_level == FEEDBACK_LEVEL_FULL) - or (feedback_level == FEEDBACK_LEVEL_RESTRICTED - and tc["show_in_restricted_feedback"]) - or (feedback_level == FEEDBACK_LEVEL_OI_RESTRICTED - and tc["show_in_oi_restricted_feedback"])) %} - {% if show_tc %} + {% if "outcome" in tc %} {% if tc["outcome"] == "Correct" %} {% elif tc["outcome"] == "Not correct" %} @@ -305,7 +336,7 @@ class ScoreTypeGroup(ScoreTypeAlone): {{ tc["text"]|format_status_text }} - {% if feedback_level == FEEDBACK_LEVEL_FULL %} + {% if show_timing %} {% if "time_limit_was_exceeded" in tc and tc["time_limit_was_exceeded"] %} > {{ tc["time_limit"]|format_duration }} @@ -325,10 +356,9 @@ class ScoreTypeGroup(ScoreTypeAlone): {% endif %} {% else %} - {% if feedback_level != FEEDBACK_LEVEL_OI_RESTRICTED %} {{ loop.index }} - {% if feedback_level == FEEDBACK_LEVEL_FULL %} + {% if show_timing %} {% else %} @@ -336,7 +366,6 @@ class ScoreTypeGroup(ScoreTypeAlone): {% trans %}N/A{% endtrans %} - {% endif %} {% endif %} {% endfor %} @@ -345,27 +374,86 @@ class ScoreTypeGroup(ScoreTypeAlone):
{% endfor %}""" + def get_json_details( + self, + score_details: object, + feedback_level: str = FEEDBACK_LEVEL_RESTRICTED, + ) -> object: + """Filter score_details for subtask-based score types according to + the feedback level. + + """ + if not isinstance(score_details, list): + return score_details + + filtered_subtasks = [] + for st in score_details: + filtered_st = { + "idx": st["idx"], + "score_fraction": st["score_fraction"], + "score": st["score"], + "max_score": st["max_score"], + } + + filtered_testcases = [] + for tc in st.get("testcases", []): + if feedback_level == FEEDBACK_LEVEL_FULL: + show_tc = True + elif feedback_level == FEEDBACK_LEVEL_RESTRICTED: + show_tc = tc.get("show_in_restricted_feedback", False) + elif feedback_level == FEEDBACK_LEVEL_OI_RESTRICTED: + show_tc = tc.get("show_in_oi_restricted_feedback", False) + else: + raise ValueError(f"Invalid feedback level {feedback_level}") + + if show_tc and "outcome" in tc: + filtered_tc = { + "idx": tc["idx"], + "outcome": tc["outcome"], + "text": tc.get("text"), + } + if feedback_level == FEEDBACK_LEVEL_FULL: + if "time" in tc: + filtered_tc["time"] = tc["time"] + if "time_limit" in tc: + filtered_tc["time_limit"] = tc["time_limit"] + if "time_limit_was_exceeded" in tc: + filtered_tc["time_limit_was_exceeded"] = tc[ + "time_limit_was_exceeded" + ] + if "memory" in tc: + filtered_tc["memory"] = tc["memory"] + filtered_testcases.append(filtered_tc) + else: + if feedback_level != FEEDBACK_LEVEL_OI_RESTRICTED: + filtered_testcases.append({"idx": tc.get("idx")}) + + filtered_st["testcases"] = filtered_testcases + filtered_subtasks.append(filtered_st) + + return filtered_subtasks + def get_max_score(self, group_parameter: ScoreTypeGroupParameters) -> float: if isinstance(group_parameter, tuple) or isinstance(group_parameter, list): score = group_parameter[0] else: score = group_parameter["max_score"] assert ( - round( - score, - self.score_precision - ) == score - ), (f"The max score for a subtask" - "has more precision than the task allows.") + round(score, self.score_precision) == score + ), "The max score for a subtask has more precision than the task allows." return score - def get_testcases(self, group_parameter: ScoreTypeGroupParameters) -> int | str | list[str]: + def get_testcases( + self, group_parameter: ScoreTypeGroupParameters + ) -> int | str | list[str]: if isinstance(group_parameter, tuple) or isinstance(group_parameter, list): return group_parameter[1] else: return group_parameter["testcases"] - def get_always_show_testcases(self, group_parameter: ScoreTypeGroupParameters) -> bool: + def get_always_show_testcases( + self, group_parameter: ScoreTypeGroupParameters + ) -> bool: if isinstance(group_parameter, tuple) or isinstance(group_parameter, list): return False else: @@ -411,18 +499,20 @@ def retrieve_target_testcases(self) -> list[list[str]]: regexp = re.compile(t) target = [tc for tc in indices if regexp.match(tc)] if not target: - raise ValueError( - "No testcase matches against the regexp '%s'" % t) + raise ValueError("No testcase matches against the regexp '%s'" % t) targets.append(target) return targets - elif all(isinstance(t, list) for t in t_params) and all(all(isinstance(t, str) for t in s) for s in t_params): + elif all(isinstance(t, list) for t in t_params) and all( + all(isinstance(t, str) for t in s) for s in t_params + ): return t_params raise ValueError( "In the score type parameters, the second value of each element " - "must have the same type (int, unicode or list of strings)") + "must have the same type (int, unicode or list of strings)" + ) def max_scores(self): """See ScoreType.max_score.""" @@ -472,76 +562,90 @@ def compute_score(self, submission_result): tc_first_lowest_score = None for tc_idx in target: tc_score = float(evaluations[tc_idx].outcome) - tc_outcome = self.get_public_outcome( - tc_score, parameter) + tc_outcome = self.get_public_outcome(tc_score, parameter) time_limit_was_exceeded = False - if evaluations[tc_idx].text == [EVALUATION_MESSAGES.get("timeout").message]: + if evaluations[tc_idx].text == [ + EVALUATION_MESSAGES.get("timeout").message + ]: time_limit_was_exceeded = True - testcases.append({ - "idx": tc_idx, - "outcome": tc_outcome, - "text": evaluations[tc_idx].text, - "time": evaluations[tc_idx].execution_time, - "time_limit": evaluations[tc_idx].dataset.time_limit, - "time_limit_was_exceeded": time_limit_was_exceeded, - "memory": evaluations[tc_idx].execution_memory, - "show_in_restricted_feedback": self.public_testcases[tc_idx], - "show_in_oi_restricted_feedback": self.public_testcases[tc_idx]}) + testcases.append( + { + "idx": tc_idx, + "outcome": tc_outcome, + "text": evaluations[tc_idx].text, + "time": evaluations[tc_idx].execution_time, + "time_limit": evaluations[tc_idx].dataset.time_limit, + "time_limit_was_exceeded": time_limit_was_exceeded, + "memory": evaluations[tc_idx].execution_memory, + "show_in_restricted_feedback": self.public_testcases[tc_idx], + "show_in_oi_restricted_feedback": self.public_testcases[tc_idx], + } + ) if self.public_testcases[tc_idx]: public_testcases.append(testcases[-1]) - if tc_first_lowest_score is None or \ - tc_score < tc_first_lowest_score: + if ( + tc_first_lowest_score is None + or tc_score < tc_first_lowest_score + ): tc_first_lowest_idx = tc_idx tc_first_lowest_score = tc_score else: public_testcases.append({"idx": tc_idx}) st_score_fraction = self.reduce( - [float(evaluations[tc_idx].outcome) for tc_idx in target], - parameter) + [float(evaluations[tc_idx].outcome) for tc_idx in target], parameter + ) st_score = st_score_fraction * self.get_max_score(parameter) rounded_score = round(st_score, score_precision) - if tc_first_lowest_idx is not None and st_score_fraction < 1.0 and not self.get_always_show_testcases(parameter): + if ( + tc_first_lowest_idx is not None + and st_score_fraction < 1.0 + and not self.get_always_show_testcases(parameter) + ): for tc in testcases: if not self.public_testcases[tc["idx"]]: continue - tc["show_in_restricted_feedback"] = ( - tc["idx"] <= tc_first_lowest_idx) + tc["show_in_restricted_feedback"] = tc["idx"] <= tc_first_lowest_idx tc["show_in_oi_restricted_feedback"] = ( - tc["idx"] == tc_first_lowest_idx) + tc["idx"] == tc_first_lowest_idx + ) score += rounded_score - subtasks.append({ - "idx": st_idx, - # We store the fraction so that an "example" testcase - # with a max score of zero is still properly rendered as - # correct or incorrect. - "score_fraction": st_score_fraction, - # But we also want the properly rounded score for display. - "score": rounded_score, - "max_score": self.get_max_score(parameter), - "testcases": testcases}) + subtasks.append( + { + "idx": st_idx, + # We store the fraction so that an "example" testcase + # with a max score of zero is still properly rendered as + # correct or incorrect. + "score_fraction": st_score_fraction, + # But we also want the properly rounded score for display. + "score": rounded_score, + "max_score": self.get_max_score(parameter), + "testcases": testcases, + } + ) if all(self.public_testcases[tc_idx] for tc_idx in target): public_score += rounded_score public_subtasks.append(subtasks[-1]) else: - public_subtasks.append({"idx": st_idx, - "testcases": public_testcases}) + public_subtasks.append({"idx": st_idx, "testcases": public_testcases}) ranking_details.append("%g" % rounded_score) - - # The following line should be unnecessary since subtask scores - # are rounded. However we are using floats not Decimals - # and this can cause errors. So we round again to be sure. + + # The following line should be unnecessary since subtask scores + # are rounded. However we are using floats not Decimals + # and this can cause errors. So we round again to be sure. score = round(score, score_precision) return score, subtasks, public_score, public_subtasks, ranking_details @abstractmethod - def get_public_outcome(self, outcome: float, parameter: ScoreTypeGroupParameters) -> str: + def get_public_outcome( + self, outcome: float, parameter: ScoreTypeGroupParameters + ) -> str: """Return a public outcome from an outcome. The public outcome is shown to the user, and this method @@ -558,7 +662,9 @@ def get_public_outcome(self, outcome: float, parameter: ScoreTypeGroupParameters pass @abstractmethod - def reduce(self, outcomes: list[float], parameter: ScoreTypeGroupParameters) -> float: + def reduce( + self, outcomes: list[float], parameter: ScoreTypeGroupParameters + ) -> float: """Return the score of a subtask given the outcomes. outcomes: the outcomes of the submission in diff --git a/cms/server/contest/handlers/__init__.py b/cms/server/contest/handlers/__init__.py index fef7f6c1a6..94321cbbf0 100644 --- a/cms/server/contest/handlers/__init__.py +++ b/cms/server/contest/handlers/__init__.py @@ -23,89 +23,87 @@ # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . -from .taskusertest import \ - UserTestInterfaceHandler, \ - UserTestHandler, \ - UserTestStatusHandler, \ - UserTestDetailsHandler, \ - UserTestIOHandler, \ - UserTestFileHandler -from .tasksubmission import \ - SubmitHandler, \ - TaskSubmissionsHandler, \ - SubmissionStatusHandler, \ - SubmissionDetailsHandler, \ - SubmissionFileHandler, \ - UseTokenHandler -from .task import \ - TaskDescriptionHandler, \ - TaskStatementViewHandler, \ - TaskAttachmentViewHandler -from .main import \ - LoginHandler, \ - LogoutHandler, \ - RegistrationHandler, \ - StartHandler, \ - NotificationsHandler, \ - DocumentationHandler -from .communication import \ - CommunicationHandler, \ - QuestionHandler -from .api import \ - ApiLoginHandler, \ - ApiSubmissionListHandler, \ - ApiSubmitHandler, \ - ApiTaskListHandler - +from .taskusertest import ( + UserTestInterfaceHandler, + UserTestHandler, + UserTestStatusHandler, + UserTestDetailsHandler, + UserTestIOHandler, + UserTestFileHandler, +) +from .tasksubmission import ( + SubmitHandler, + TaskSubmissionsHandler, + SubmissionStatusHandler, + SubmissionDetailsHandler, + SubmissionFileHandler, + UseTokenHandler, +) +from .task import ( + TaskDescriptionHandler, + TaskStatementViewHandler, + TaskAttachmentViewHandler, +) +from .main import ( + LoginHandler, + LogoutHandler, + RegistrationHandler, + StartHandler, + NotificationsHandler, + DocumentationHandler, +) +from .communication import CommunicationHandler, QuestionHandler +from .api import ( + ApiLoginHandler, + ApiSubmissionDetailsHandler, + ApiSubmissionFullDetailsHandler, + ApiSubmissionListHandler, + ApiSubmitHandler, + ApiTaskListHandler, +) HANDLERS = [ - # Main - (r"/login", LoginHandler), (r"/logout", LogoutHandler), (r"/register", RegistrationHandler), (r"/start", StartHandler), (r"/notifications", NotificationsHandler), (r"/documentation", DocumentationHandler), - # Tasks - (r"/tasks/(.*)/description", TaskDescriptionHandler), (r"/tasks/(.*)/statements/([^/]*)(?:/.*)?", TaskStatementViewHandler), (r"/tasks/(.*)/attachments/(.*)", TaskAttachmentViewHandler), - # Task submissions - (r"/tasks/(.*)/submit", SubmitHandler), (r"/tasks/(.*)/submissions", TaskSubmissionsHandler), (r"/tasks/(.*)/submissions/([1-9][0-9]*)", SubmissionStatusHandler), - (r"/tasks/(.*)/submissions/([1-9][0-9]*)/details", - SubmissionDetailsHandler), - (r"/tasks/(.*)/submissions/([1-9][0-9]*)/files/(.*)", - SubmissionFileHandler), + (r"/tasks/(.*)/submissions/([1-9][0-9]*)/details", SubmissionDetailsHandler), + (r"/tasks/(.*)/submissions/([1-9][0-9]*)/files/(.*)", SubmissionFileHandler), (r"/tasks/(.*)/submissions/([1-9][0-9]*)/token", UseTokenHandler), - # Task usertests - (r"/testing", UserTestInterfaceHandler), (r"/tasks/(.*)/test", UserTestHandler), (r"/tasks/(.*)/tests/([1-9][0-9]*)", UserTestStatusHandler), (r"/tasks/(.*)/tests/([1-9][0-9]*)/details", UserTestDetailsHandler), (r"/tasks/(.*)/tests/([1-9][0-9]*)/(input|output)", UserTestIOHandler), (r"/tasks/(.*)/tests/([1-9][0-9]*)/files/(.*)", UserTestFileHandler), - # Communications - (r"/communication", CommunicationHandler), (r"/question", QuestionHandler), - # API (r"/api/login", ApiLoginHandler), (r"/api/task_list", ApiTaskListHandler), (r"/api/(.*)/submit", ApiSubmitHandler), (r"/api/(.*)/submission_list", ApiSubmissionListHandler), - + ( + r"/api/(.*)/submissions/([1-9][0-9]*)/details", + ApiSubmissionDetailsHandler, + ), + ( + r"/api/(.*)/submissions/([1-9][0-9]*)/full_details", + ApiSubmissionFullDetailsHandler, + ), # The following prefixes are handled by WSGI middlewares: # * /static, defined in cms/io/web_service.py # * /docs, defined in cms/server/contest/server.py diff --git a/cms/server/contest/handlers/api.py b/cms/server/contest/handlers/api.py index 9cba1231ce..92059c86db 100644 --- a/cms/server/contest/handlers/api.py +++ b/cms/server/contest/handlers/api.py @@ -16,18 +16,16 @@ # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . -"""API handlers for CMS. - -""" +"""API handlers for CMS.""" import ipaddress import logging +from cms import FEEDBACK_LEVEL_FULL from cms.db.submission import Submission from cms.server import multi_contest from cms.server.contest.authentication import validate_login -from cms.server.contest.submission import \ - UnacceptableSubmission, accept_submission +from cms.server.contest.submission import UnacceptableSubmission, accept_submission from .contest import ContestHandler, api_login_required from ..phase_management import actual_phase_required @@ -35,9 +33,7 @@ class ApiContestHandler(ContestHandler): - """An extension of ContestHandler marking the request as a part of the API. - - """ + """An extension of ContestHandler marking the request as a part of the API.""" def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) @@ -45,9 +41,8 @@ def __init__(self, *args, **kwargs): class ApiLoginHandler(ApiContestHandler): - """Login handler. + """Login handler.""" - """ @multi_contest def post(self): current_user = self.get_current_user() @@ -59,32 +54,54 @@ def post(self): if current_user is not None: if username != "" and current_user.user.username != username: self.json( - {"error": f"Logged in as {current_user.user.username} but trying to login as {username}"}, 400) + { + "error": f"Logged in as {current_user.user.username} but trying to login as {username}" + }, + 400, + ) else: cookie_name = self.contest.name + "_login" cookie = self.get_secure_cookie(cookie_name) - self.json({"login_data": self.request.headers.get( - "X-CMS-Authorization", cookie if cookie is not None else "Already-Logged-In")}) + self.json( + { + "login_data": self.request.headers.get( + "X-CMS-Authorization", + cookie if cookie is not None else "Already-Logged-In", + ) + } + ) return try: ip_address = ipaddress.ip_address(self.request.remote_ip) except ValueError: - logger.warning("Invalid IP address provided by Tornado: %s", - self.request.remote_ip) + logger.warning( + "Invalid IP address provided by Tornado: %s", self.request.remote_ip + ) return None participation, login_data = validate_login( - self.sql_session, self.contest, self.timestamp, username, password, - ip_address, admin_token=admin_token) + self.sql_session, + self.contest, + self.timestamp, + username, + password, + ip_address, + admin_token=admin_token, + ) if participation is None: self.json({"error": "Login failed"}, 403) elif login_data is not None: cookie_name = self.contest.name + "_login" - self.json({"login_data": self.create_signed_value( - cookie_name, login_data).decode()}) + self.json( + { + "login_data": self.create_signed_value( + cookie_name, login_data + ).decode() + } + ) else: self.json({}) @@ -93,9 +110,8 @@ def check_xsrf_cookie(self): class ApiTaskListHandler(ApiContestHandler): - """Handler to list all tasks and their statements. + """Handler to list all tasks and their statements.""" - """ @api_login_required @actual_phase_required(0, 3) @multi_contest @@ -106,16 +122,19 @@ def get(self): name = task.name statements = [s for s in task.statements] sub_format = task.submission_format - tasks.append({"name": name, - "statements": statements, - "submission_format": sub_format}) + tasks.append( + { + "name": name, + "statements": statements, + "submission_format": sub_format, + } + ) self.json({"tasks": tasks}) class ApiSubmitHandler(ApiContestHandler): - """Handles the received submissions. + """Handles the received submissions.""" - """ @api_login_required @actual_phase_required(0, 3) @multi_contest @@ -133,9 +152,13 @@ def post(self, task_name: str): # of a contestant, allow overriding. if self.impersonated_by_admin: try: - official = self.get_boolean_argument('override_official', official) - override_max_number = self.get_boolean_argument('override_max_number', False) - override_min_interval = self.get_boolean_argument('override_min_interval', False) + official = self.get_boolean_argument("override_official", official) + override_max_number = self.get_boolean_argument( + "override_max_number", False + ) + override_min_interval = self.get_boolean_argument( + "override_min_interval", False + ) except ValueError as err: self.json({"error": str(err)}, 400) return @@ -145,29 +168,32 @@ def post(self, task_name: str): try: submission = accept_submission( - self.sql_session, self.service.file_cacher, self.current_user, - task, self.timestamp, self.request.files, - self.get_argument("language", None), official, + self.sql_session, + self.service.file_cacher, + self.current_user, + task, + self.timestamp, + self.request.files, + self.get_argument("language", None), + official, override_max_number=override_max_number, override_min_interval=override_min_interval, ) self.sql_session.commit() except UnacceptableSubmission as e: - logger.info("API submission rejected: `%s' - `%s'", - e.subject, e.formatted_text) + logger.info( + "API submission rejected: `%s' - `%s'", e.subject, e.formatted_text + ) self.json({"error": e.subject, "details": e.formatted_text}, 422) else: - logger.info( - f'API submission accepted: Submission ID {submission.id}') - self.service.evaluation_service.new_submission( - submission_id=submission.id) - self.json({'id': str(submission.opaque_id)}) + logger.info(f"API submission accepted: Submission ID {submission.id}") + self.service.evaluation_service.new_submission(submission_id=submission.id) + self.json({"id": str(submission.opaque_id)}) class ApiSubmissionListHandler(ApiContestHandler): - """Retrieves the list of submissions on a task. + """Retrieves the list of submissions on a task.""" - """ @api_login_required @actual_phase_required(0, 3) @multi_contest @@ -182,4 +208,79 @@ def get(self, task_name: str): .filter(Submission.task == task) .all() ) - self.json({'list': [{"id": str(s.opaque_id)} for s in submissions]}) + self.json({"list": [{"id": str(s.opaque_id)} for s in submissions]}) + + +class ApiSubmissionDetailsHandler(ApiContestHandler): + """Retrieves the feedback-level-restricted details of a submission + on a task. + + """ + + @api_login_required + @actual_phase_required(0, 1, 2, 3, 4) + @multi_contest + def get(self, task_name: str, opaque_id: str): + task = self.get_task(task_name) + if task is None: + self.json({"error": "Task not found"}, 404) + return + + submission = self.get_submission(task, opaque_id) + if submission is None: + self.json({"error": "Submission not found"}, 404) + return + + sr = submission.get_result(task.active_dataset) + score_type = task.active_dataset.score_type_object + + details = None + if sr is not None and sr.scored(): + is_analysis_mode = self.r_params["actual_phase"] == 3 + if submission.tokened() or is_analysis_mode: + raw_details = sr.score_details + else: + raw_details = sr.public_score_details + + if is_analysis_mode: + feedback_level = FEEDBACK_LEVEL_FULL + else: + feedback_level = task.feedback_level + + details = score_type.get_json_details(raw_details, feedback_level) + + self.json({"details": details}) + + +class ApiSubmissionFullDetailsHandler(ApiContestHandler): + """Retrieves the unfiltered details of a submission on a task + (admin only). + + """ + + @api_login_required + @actual_phase_required(0, 1, 2, 3, 4) + @multi_contest + def get(self, task_name: str, opaque_id: str): + if not self.impersonated_by_admin: + self.json({"error": "Admin impersonation required"}, 403) + return + + task = self.get_task(task_name) + if task is None: + self.json({"error": "Task not found"}, 404) + return + + submission = self.get_submission(task, opaque_id) + if submission is None: + self.json({"error": "Submission not found"}, 404) + return + + sr = submission.get_result(task.active_dataset) + score_type = task.active_dataset.score_type_object + + details = None + if sr is not None and sr.scored(): + details = score_type.get_json_details(sr.score_details, FEEDBACK_LEVEL_FULL) + + self.json({"details": details}) diff --git a/cmscontrib/SolutionChecker.py b/cmscontrib/SolutionChecker.py index e5cc37dc88..1510dcfeb8 100755 --- a/cmscontrib/SolutionChecker.py +++ b/cmscontrib/SolutionChecker.py @@ -23,6 +23,16 @@ - path: path to the solution file. - min_score: minimum expected score. - max_score: maximum expected score. +- checks: optional list of expected subtask outcomes, where each + element can be: + - null (no assertion on this subtask) + - "Accepted" (full score on this subtask) + - "Zero" (0 score on this subtask) + - "PartialScore" (score > 0 and < max_score on this subtask) + - "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) + - "RuntimeError" (at least one testcase failed due to runtime error) Such a file can be generated with `task-maker-rust export-solution-checks`. """ @@ -34,7 +44,7 @@ import re import sys import time -from typing import Optional, Dict, Any +from typing import Optional, Dict, Any, List import requests @@ -52,13 +62,12 @@ class RedAlertFormatter(logging.Formatter): YELLOW_FORMAT = YELLOW + "%(levelname)8s" + RSET + " %(message)s" - FORMATS = { logging.DEBUG: BASE_FORMAT, logging.INFO: BASE_FORMAT, logging.WARNING: YELLOW_FORMAT, logging.ERROR: RED_FORMAT, - logging.CRITICAL: RED_FORMAT + logging.CRITICAL: RED_FORMAT, } def format(self, record): @@ -66,29 +75,90 @@ def format(self, record): formatter = logging.Formatter(log_fmt) return formatter.format(record) + class SolutionChecker: def __init__( - self, base_url: str, username: Optional[str], password: Optional[str] = None + self, + base_url: str, + username: Optional[str] = None, + password: Optional[str] = None, + admin_token: Optional[str] = None, ): self.base_url = base_url.rstrip("/") self.username = username self.password = password + self.admin_token = admin_token self.session = requests.Session() - self.auth_header = {} + self.auth_header: Dict[str, str] = {} + + @staticmethod + def _extract_error_message(err: requests.exceptions.HTTPError) -> str: + if err.response is not None: + try: + json_data = err.response.json() + if "error" in json_data: + return json_data["error"] + except Exception: + pass + if err.response.text: + return err.response.text.strip() + return str(err) def login(self): - if self.password is None or self.username is None: - logger.info("No password provided, assuming IP autologin.") + if self.admin_token is not None: + if self.username is None: + raise ValueError( + "Admin token requires --username to specify user to impersonate." + ) + login_url = f"{self.base_url}/api/login" + data = {"admin_token": self.admin_token, "username": self.username} + if self.password is not None: + data["password"] = self.password + try: + response = self.session.post(login_url, data=data) + response.raise_for_status() + except requests.exceptions.HTTPError as e: + err_msg = self._extract_error_message(e) + raise RuntimeError( + f"Login with admin token failed: {err_msg}" + ) from None + + res_data = response.json() + self.auth_header = {"X-CMS-Authorization": res_data["login_data"]} + logger.info("Successfully logged in with admin token as %s.", self.username) return + if self.username is None and self.password is None: + logger.info("No credentials provided, assuming IP autologin.") + return + + if self.username is None and self.password is not None: + raise ValueError( + "Password provided without username. Please specify " "--username." + ) + + if self.username is not None and self.password is None: + raise ValueError( + f"Username '{self.username}' provided without password. " + f"Please specify --password or use --admin-token." + ) + login_url = f"{self.base_url}/api/login" - response = self.session.post( - login_url, data={"username": self.username, "password": self.password} - ) - response.raise_for_status() + try: + response = self.session.post( + login_url, + data={"username": self.username, "password": self.password}, + ) + response.raise_for_status() + except requests.exceptions.HTTPError as e: + err_msg = self._extract_error_message(e) + raise RuntimeError( + f"Login failed for user '{self.username}': {err_msg}" + ) from None + data = response.json() self.auth_header = {"X-CMS-Authorization": data["login_data"]} - logger.info("Successfully logged in.") + logger.info("Successfully logged in as %s.", self.username) def submit(self, task_name: str, file_path: str) -> str: task_list_url = f"{self.base_url}/api/task_list" @@ -116,7 +186,9 @@ def submit(self, task_name: str, file_path: str) -> str: response.raise_for_status() return response.json().get("id") - def poll_status(self, task_name: str, filename: str, submission_id: str) -> Dict[str, Any]: + def poll_status( + self, task_name: str, filename: str, submission_id: str + ) -> Dict[str, Any]: status_url = f"{self.base_url}/tasks/{task_name}/submissions/{submission_id}" while True: response = self.session.get(status_url, headers=self.auth_header) @@ -135,40 +207,346 @@ def get_time_limit(self, task_name: str) -> float: assert match, "Could not find time limit in task description" return float(match.group(1)) - def has_slow_testcases( - self, task_name: str, submission_id: str, time_limit: float - ) -> bool: - url = f"{self.base_url}/tasks/{task_name}/submissions/{submission_id}/details" + def get_submission_details( + self, task_name: str, submission_id: str + ) -> Optional[List[Dict[str, Any]]]: + endpoint = "full_details" if self.admin_token is not None else "details" + url = ( + f"{self.base_url}/api/{task_name}/submissions/" + f"{submission_id}/{endpoint}" + ) response = self.session.get(url, headers=self.auth_header) response.raise_for_status() + return response.json().get("details") - html = response.text - # Split by subtask. This is quite hacky and relies on subtask - # delimiters having at least two classes (to avoid mixing it - # up with subtask-head/subtask-body). - subtasks = html.split('
\s*\(\s*([\d.]+)\s*/', st) - if score_match and float(score_match.group(1).replace(",", ".")) > 0: - # Find all execution times in this subtask - times = re.findall( - r'\s*(?:>\s*)?([\d.]+)\s*s', st + def has_slow_testcases( + self, details: Optional[List[Dict[str, Any]]], time_limit: float + ) -> bool: + if not details: + return False + + has_times = False + for item in details: + testcases = item.get("testcases") + if testcases is not None: + # Subtask structure + score = item.get("score", 0.0) + score_fraction = item.get("score_fraction", 0.0) + check_slow = (score is not None and score > 0) or ( + score_fraction is not None and score_fraction > 0 ) - if not times: - logger.warning( - "No testcase times found. Ensure feedback levels are configured correctly" - ) - for t in times: - if float(t.replace(",", ".")) > time_limit * 0.5: + for tc in testcases: + t = tc.get("time") + if t is not None: + has_times = True + if check_slow and float(t) > time_limit * 0.5: + return True + else: + # Flat testcase structure (e.g. Sum) + t = item.get("time") + if t is not None: + has_times = True + if float(t) > time_limit * 0.5: return True + + if not has_times and not self.admin_token: + logger.warning( + "No testcase times found. Ensure feedback levels are " + "configured correctly or use --admin-token" + ) + return False + + @staticmethod + def get_testcase_status(tc: Dict[str, Any]) -> str: + outcome = tc.get("outcome") + text_list = tc.get("text", []) + text_str = ( + text_list[0] + if isinstance(text_list, list) and text_list + else str(text_list) + ) + if "wall clock" in text_str.lower(): + return "WallTimeLimitExceeded" + if tc.get("time_limit_was_exceeded", False) or "timed out" in text_str.lower(): + return "TimeLimitExceeded" + if ( + "signal" in text_str.lower() + or "return code" in text_str.lower() + or "memory limit" in text_str.lower() + ): + return "RuntimeError" + if outcome == "Correct" or "output is correct" in text_str.lower(): + 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() + ): + return "WrongAnswer" + return "Unknown" + + def check_single_subtask( + self, st: Optional[Dict[str, Any]], st_idx: int, check: Optional[str] + ) -> Optional[str]: + if check is None: + return None + if st is None: + return f"subtask {st_idx}: not found in submission details" + + score_fraction = st.get("score_fraction") + score = st.get("score") + max_score = st.get("max_score") + testcases = st.get("testcases", []) + statuses = { + self.get_testcase_status(tc) + for tc in testcases + if "outcome" in tc or "text" in tc + } + statuses_list = sorted(statuses) + + if check == "Accepted": + if ( + (statuses and not statuses <= {"Accepted"}) + or (score_fraction is not None and score_fraction < 1.0 - 1e-7) + or ( + score is not None + and max_score is not None + and score < max_score - 1e-7 + ) + ): + return ( + f"subtask {st_idx}: expected Accepted, got statuses " + f"{statuses_list}" + ) + + elif check == "Zero": + if (score_fraction is not None and score_fraction > 1e-7) or ( + score is not None and score > 1e-7 + ): + return ( + f"subtask {st_idx}: expected Zero, got score " + f"{score}/{max_score} and statuses {statuses_list}" + ) + + elif check == "PartialScore": + is_partial = False + if score_fraction is not None and 1e-7 < score_fraction < 1.0 - 1e-7: + is_partial = True + elif ( + score is not None + and max_score is not None + and 1e-7 < score < max_score - 1e-7 + ): + is_partial = True + elif "PartialScore" in statuses or ( + "Accepted" in statuses and len(statuses - {"Accepted"}) > 0 + ): + is_partial = True + + if not is_partial: + return ( + f"subtask {st_idx}: expected PartialScore, got score " + f"{score}/{max_score} and statuses {statuses_list}" + ) + + elif check in [ + "WrongAnswer", + "TimeLimitExceeded", + "WallTimeLimitExceeded", + "RuntimeError", + ]: + if check not in statuses: + return ( + f"subtask {st_idx}: expected {check}, got statuses " + f"{statuses_list}" + ) + + else: + return f"subtask {st_idx}: unknown check type '{check}'" + + return None + + def check_subtasks( + self, details: Optional[List[Dict[str, Any]]], checks: list + ) -> List[str]: + if details is None: + return ["Submission details unavailable for subtask checks."] + + errors = [] + subtasks_by_idx = {st.get("idx", i): st for i, st in enumerate(details)} + + for st_idx, check in enumerate(checks): + st = subtasks_by_idx.get(st_idx) + err = self.check_single_subtask(st, st_idx, check) + if err is not None: + errors.append(err) + + return errors + + @staticmethod + def is_subtask_slow(st: Dict[str, Any], time_limit: float) -> bool: + if not st: + return False + score = st.get("score", 0.0) + score_fraction = st.get("score_fraction", 0.0) + check_slow = (score is not None and score > 0) or ( + score_fraction is not None and score_fraction > 0 + ) + if not check_slow: + return False + for tc in st.get("testcases", []): + t = tc.get("time") + if t is not None and float(t) > time_limit * 0.5: + return True return False + STATUS_SHORT_CODES = { + "Accepted": "AC", + "WrongAnswer": "WA", + "TimeLimitExceeded": "TLE", + "WallTimeLimitExceeded": "WTL", + "RuntimeError": "RTE", + "PartialScore": "PS", + "Zero": "0", + None: "-", + } + + def format_report_table( + self, + results: List[Dict[str, Any]], + time_limit: float, + use_color: bool = True, + ) -> str: + if not results: + return "" + + RED = "\x1b[31;1m" if use_color else "" + GREEN = "\x1b[32;1m" if use_color else "" + YELLOW = "\x1b[33;1m" if use_color else "" + RSET = "\x1b[0m" if use_color else "" + + num_subtasks = 0 + for r in results: + crit_checks = r.get("criteria", {}).get("checks") + if crit_checks: + num_subtasks = max(num_subtasks, len(crit_checks)) + details = r.get("details") + if isinstance(details, list): + num_subtasks = max(num_subtasks, len(details)) + + rows = [] + for r in results: + sol_name = r["name"] + criteria = r.get("criteria", {}) + checks = criteria.get("checks", []) + details = r.get("details") + compilation_failed = r.get("compilation_failed", False) + subtasks_by_idx = {} + if isinstance(details, list): + subtasks_by_idx = { + st.get("idx", i): st for i, st in enumerate(details) + } + + row_cells = [(sol_name, "")] + + for i in range(num_subtasks): + expected = checks[i] if i < len(checks) else None + cell_text = self.STATUS_SHORT_CODES.get( + expected, str(expected) if expected is not None else "-" + ) + cell_color = GREEN + + if compilation_failed: + cell_color = RED + else: + st = subtasks_by_idx.get(i) + if expected is not None: + err = self.check_single_subtask(st, i, expected) + if err is not None: + cell_color = RED + elif self.is_subtask_slow(st, time_limit): + cell_color = YELLOW + else: + cell_color = GREEN + else: + if st and self.is_subtask_slow(st, time_limit): + cell_color = YELLOW + else: + cell_color = GREEN + + row_cells.append((cell_text, cell_color)) + + if compilation_failed: + score_text = "CE" + expected_str = "" + total_color = RED + else: + score = r.get("score", 0.0) + min_score = criteria.get("min_score", 0.0) + max_score = criteria.get("max_score", 100.0) + score_unexpected = ( + score < min_score - 1e-7 or score > max_score + 1e-7 + ) + + score_text = f"{score:g}" + if score_unexpected: + if abs(min_score - max_score) < 1e-7: + expected_note = f"expected {min_score:g}" + else: + expected_note = f"expected {min_score:g}-{max_score:g}" + expected_str = f" ({expected_note})" + else: + expected_str = "" + + if r.get("failed", False): + total_color = RED + elif r.get("slow", False): + total_color = YELLOW + else: + total_color = GREEN + + 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 + score_col_width = 0 + + for row_cells, score_text, _, _ in rows: + for col_idx, (text, _) in enumerate(row_cells): + col_widths[col_idx] = max(col_widths[col_idx], len(text)) + score_col_width = max(score_col_width, len(score_text)) + + lines = [] + for row_cells, score_text, expected_str, total_color in rows: + row_strs = [] + for col_idx, (text, color) in enumerate(row_cells): + w = col_widths[col_idx] + align_fmt = f"{text:<{w}}" if col_idx == 0 else f"{text:>{w}}" + if color: + row_strs.append(f"{color}{align_fmt}{RSET}") + else: + row_strs.append(align_fmt) + + aligned_score = f"{score_text:>{score_col_width}}{expected_str}" + if total_color: + row_strs.append(f"{total_color}{aligned_score}{RSET}") + else: + row_strs.append(aligned_score) + + lines.append(" ".join(row_strs)) + + return "\n".join(lines) + def main(): parser = argparse.ArgumentParser(description="CMS Solution Checker") parser.add_argument( - "--checks-json", "-c", required=True, help="Path to solution_checks.json" + "--checks-json", + "-c", + required=True, + help="Path to solution_checks.json", ) parser.add_argument( "--url", @@ -179,6 +557,9 @@ def main(): parser.add_argument("--task", "-t", required=True, help="Task name") parser.add_argument("--username", "-U", help="CMS username") parser.add_argument("--password", "-p", help="CMS password") + parser.add_argument( + "--admin-token", "-a", help="CMS contest admin token for full details" + ) parser.add_argument( "--quiet", "-q", action="store_true", help="Disable non-warnings" ) @@ -197,48 +578,110 @@ def main(): with open(args.checks_json, "r") as f: checks = json.load(f) - checker = SolutionChecker(args.url, args.username, args.password) - checker.login() - - time_limit = checker.get_time_limit(args.task) - - submissions = {} - logger.info("Submitting %d solutions...", len(checks)) - for criteria in checks: - sol_path = criteria.get("path") - sub_id = checker.submit(args.task, sol_path) - submissions[sol_path] = (sub_id, criteria) - logger.info("Submitted %s: %s", sol_path.split("/")[-1], sub_id) - - has_failures = False - logger.info("Waiting for evaluations...") - for sol_path, (sub_id, criteria) in submissions.items(): - sol_name = sol_path.split("/")[-1] - status = checker.poll_status(args.task, sol_name, sub_id) - failed = False - if status: - score = status["public_score"] - min_score = criteria["min_score"] - max_score = criteria["max_score"] - if score < min_score - 1e-7 or score > max_score + 1e-7: + try: + checker = SolutionChecker( + args.url, + args.username, + args.password, + admin_token=args.admin_token, + ) + checker.login() + + time_limit = checker.get_time_limit(args.task) + + submissions = {} + logger.info("Submitting %d solutions...", len(checks)) + for criteria in checks: + sol_path = criteria.get("path") + sub_id = checker.submit(args.task, sol_path) + submissions[sol_path] = (sub_id, criteria) + logger.info("Submitted %s: %s", sol_path.split("/")[-1], sub_id) + + results = [] + has_failures = False + logger.info("Waiting for evaluations...") + for sol_path, (sub_id, criteria) in submissions.items(): + sol_name = sol_path.split("/")[-1] + status = checker.poll_status(args.task, sol_name, sub_id) + failed = False + compilation_failed = False + errors = [] + score = 0.0 + details = None + slow = False + + if status: + if status.get("status") == 2: + failed = True + compilation_failed = True + errors.append("Compilation failed.") + else: + score = status.get("public_score", 0.0) + min_score = criteria.get("min_score", 0.0) + max_score = criteria.get("max_score", 100.0) + if score < min_score - 1e-7 or score > max_score + 1e-7: + failed = True + errors.append( + f"score {score} is not in range " + f"[{min_score}, {max_score}]" + ) + + details = checker.get_submission_details(args.task, sub_id) + + if "checks" in criteria and criteria["checks"]: + subtask_errors = checker.check_subtasks( + details, criteria["checks"] + ) + if subtask_errors: + failed = True + errors.extend(subtask_errors) + + if checker.has_slow_testcases(details, time_limit): + slow = True + else: failed = True - error = f"score {score} is not in range [{min_score}, {max_score}]" - else: - failed = True - error = "Evaluation failed." - - if not failed: - logger.info("%20s: check successful", sol_name) - if checker.has_slow_testcases(args.task, sub_id, time_limit): - logger.warning( - "%20s: some testcases took > 50%% of time limit", - sol_name, - ) - else: - has_failures = True - logger.error("%20s: %s", sol_name, error) + errors.append("Evaluation failed.") - return 1 if has_failures else 0 + if not failed: + logger.info("%20s: check successful", sol_name) + if slow: + logger.warning( + "%20s: some testcases took > 50%% of time limit", + sol_name, + ) + else: + has_failures = True + for err in errors: + logger.error("%20s: %s", sol_name, err) + + results.append( + { + "name": sol_name, + "criteria": criteria, + "score": score, + "compilation_failed": compilation_failed, + "details": details, + "failed": failed, + "slow": slow, + } + ) + + # Print report table + table_str = checker.format_report_table( + results, time_limit, use_color=sys.stdout.isatty() + ) + print("\n" + table_str) + + return 1 if has_failures else 0 + + except (ValueError, RuntimeError) as e: + logger.error("%s", e) + return 1 + except requests.exceptions.HTTPError as e: + err_msg = SolutionChecker._extract_error_message(e) + status_code = e.response.status_code if e.response is not None else "unknown" + logger.error("API error (%s): %s", status_code, err_msg) + return 1 if __name__ == "__main__": diff --git a/cmstestsuite/unit_tests/cmscontrib/SolutionCheckerTest.py b/cmstestsuite/unit_tests/cmscontrib/SolutionCheckerTest.py new file mode 100644 index 0000000000..6f0b266012 --- /dev/null +++ b/cmstestsuite/unit_tests/cmscontrib/SolutionCheckerTest.py @@ -0,0 +1,673 @@ +#!/usr/bin/env python3 + +# Contest Management System - http://cms-dev.github.io/ +# Copyright © 2026 Luca Versari +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as +# published by the Free Software Foundation, either version 3 of the +# License, or (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +"""Unit tests for SolutionChecker.""" + +import unittest +from unittest.mock import MagicMock, patch + +import requests + +from cmscontrib.SolutionChecker import SolutionChecker + + +class TestSolutionChecker(unittest.TestCase): + + def setUp(self): + super().setUp() + self.checker = SolutionChecker( + base_url="http://localhost:8888/contest", + username="user", + password="pwd", + ) + + def test_check_subtasks_accepted(self): + # Full score -> pass + details = [ + { + "idx": 0, + "score": 20.0, + "max_score": 20.0, + "score_fraction": 1.0, + "testcases": [{"outcome": "Correct"}], + } + ] + self.assertEqual(self.checker.check_subtasks(details, ["Accepted"]), []) + + # Partial score -> fail + details_partial = [ + { + "idx": 0, + "score": 10.0, + "max_score": 20.0, + "score_fraction": 0.5, + "testcases": [{"outcome": "Partially correct"}], + } + ] + errors = self.checker.check_subtasks(details_partial, ["Accepted"]) + self.assertEqual(len(errors), 1) + self.assertIn("expected Accepted", errors[0]) + + def test_check_subtasks_zero(self): + # 0 score -> pass + details = [ + { + "idx": 0, + "score": 0.0, + "max_score": 20.0, + "score_fraction": 0.0, + "testcases": [{"outcome": "Not correct"}], + } + ] + self.assertEqual(self.checker.check_subtasks(details, ["Zero"]), []) + + # Positive score -> fail + details_pos = [ + { + "idx": 0, + "score": 5.0, + "max_score": 20.0, + "score_fraction": 0.25, + "testcases": [{"outcome": "Partially correct"}], + } + ] + errors = self.checker.check_subtasks(details_pos, ["Zero"]) + self.assertEqual(len(errors), 1) + self.assertIn("expected Zero", errors[0]) + + def test_check_subtasks_partial_score(self): + details = [ + { + "idx": 0, + "score": 10.0, + "max_score": 20.0, + "score_fraction": 0.5, + } + ] + self.assertEqual(self.checker.check_subtasks(details, ["PartialScore"]), []) + + # 0 score -> fail + details_zero = [ + { + "idx": 0, + "score": 0.0, + "max_score": 20.0, + "score_fraction": 0.0, + } + ] + self.assertEqual( + len(self.checker.check_subtasks(details_zero, ["PartialScore"])), 1 + ) + + # Full score -> fail + details_full = [ + { + "idx": 0, + "score": 20.0, + "max_score": 20.0, + "score_fraction": 1.0, + } + ] + self.assertEqual( + len(self.checker.check_subtasks(details_full, ["PartialScore"])), 1 + ) + + def test_check_subtasks_wrong_answer(self): + details = [ + { + "idx": 0, + "score": 0.0, + "max_score": 20.0, + "testcases": [ + {"outcome": "Correct", "text": ["Output is correct"]}, + { + "outcome": "Not correct", + "text": ["Output isn't correct"], + }, + ], + } + ] + self.assertEqual(self.checker.check_subtasks(details, ["WrongAnswer"]), []) + + # All correct -> fail, reports got statuses ['Accepted'] + details_correct = [ + { + "idx": 0, + "score": 20.0, + "max_score": 20.0, + "testcases": [ + {"outcome": "Correct", "text": ["Output is correct"]}, + ], + } + ] + errors = self.checker.check_subtasks(details_correct, ["WrongAnswer"]) + self.assertEqual(len(errors), 1) + self.assertIn("expected WrongAnswer, got statuses ['Accepted']", errors[0]) + + def test_check_subtasks_time_limit_exceeded(self): + details = [ + { + "idx": 0, + "testcases": [ + { + "outcome": "Not correct", + "text": ["Execution timed out"], + "time_limit_was_exceeded": True, + } + ], + } + ] + self.assertEqual( + self.checker.check_subtasks(details, ["TimeLimitExceeded"]), [] + ) + + # No TLE -> fail, reports got statuses ['WrongAnswer'] + details_no_tle = [ + { + "idx": 0, + "testcases": [ + { + "outcome": "Not correct", + "text": ["Output isn't correct"], + } + ], + } + ] + errors = self.checker.check_subtasks(details_no_tle, ["TimeLimitExceeded"]) + self.assertEqual(len(errors), 1) + self.assertIn( + "expected TimeLimitExceeded, got statuses ['WrongAnswer']", + errors[0], + ) + + def test_check_subtasks_wall_time_limit_exceeded(self): + details = [ + { + "idx": 0, + "testcases": [ + { + "outcome": "Not correct", + "text": ["Execution timed out (wall clock limit exceeded)"], + } + ], + } + ] + self.assertEqual( + self.checker.check_subtasks(details, ["WallTimeLimitExceeded"]), [] + ) + + def test_check_subtasks_runtime_error(self): + details_signal = [ + { + "idx": 0, + "testcases": [ + { + "outcome": "Not correct", + "text": ["Execution killed by signal"], + } + ], + } + ] + self.assertEqual( + self.checker.check_subtasks(details_signal, ["RuntimeError"]), [] + ) + + details_returncode = [ + { + "idx": 0, + "testcases": [ + { + "outcome": "Not correct", + "text": [ + "Execution failed because the return code was " "nonzero" + ], + } + ], + } + ] + self.assertEqual( + self.checker.check_subtasks(details_returncode, ["RuntimeError"]), + [], + ) + + def test_check_subtasks_multiple_statuses(self): + # Subtask with multiple different testcase failures + details = [ + { + "idx": 0, + "score": 0.0, + "max_score": 20.0, + "testcases": [ + {"outcome": "Correct", "text": ["Output is correct"]}, + { + "outcome": "Not correct", + "text": ["Execution killed by signal 11"], + }, + { + "outcome": "Not correct", + "text": ["Output isn't correct"], + }, + ], + } + ] + # Expecting RuntimeError -> pass (since RuntimeError is in the set) + self.assertEqual(self.checker.check_subtasks(details, ["RuntimeError"]), []) + # Expecting WrongAnswer -> pass (since WrongAnswer is in the set) + self.assertEqual(self.checker.check_subtasks(details, ["WrongAnswer"]), []) + # Expecting TimeLimitExceeded -> fail, reports actual statuses + errors = self.checker.check_subtasks(details, ["TimeLimitExceeded"]) + self.assertEqual(len(errors), 1) + self.assertIn( + "expected TimeLimitExceeded, got statuses " + "['Accepted', 'RuntimeError', 'WrongAnswer']", + errors[0], + ) + + def test_check_subtasks_null_and_unknown(self): + details = [{"idx": 0, "score": 10.0}] + # null / None check -> pass + self.assertEqual(self.checker.check_subtasks(details, [None]), []) + + # Unknown check -> fail + errors = self.checker.check_subtasks(details, ["InvalidCheckType"]) + self.assertEqual(len(errors), 1) + self.assertIn("unknown check type", errors[0]) + + def test_has_slow_testcases(self): + time_limit = 1.0 + # Testcase took 0.6s (> 0.5 * time_limit) on positive subtask -> True + details_slow = [ + { + "idx": 0, + "score": 10.0, + "testcases": [{"time": 0.6}], + } + ] + self.assertTrue(self.checker.has_slow_testcases(details_slow, time_limit)) + + # Testcase took 0.4s (<= 0.5 * time_limit) -> False + details_fast = [ + { + "idx": 0, + "score": 10.0, + "testcases": [{"time": 0.4}], + } + ] + self.assertFalse(self.checker.has_slow_testcases(details_fast, time_limit)) + + # Testcase took 0.8s on a 0-score subtask -> False + details_zero_score = [ + { + "idx": 0, + "score": 0.0, + "score_fraction": 0.0, + "testcases": [{"time": 0.8}], + } + ] + self.assertFalse( + self.checker.has_slow_testcases(details_zero_score, time_limit) + ) + # Test that user's 0-score / sample subtask with times does not warn + details_sample = [ + { + "idx": 0, + "score_fraction": 1.0, + "score": 0.0, + "max_score": 0.0, + "testcases": [ + { + "idx": "000", + "outcome": "Correct", + "text": ["Output is correct"], + "time": 0.001, + "time_limit": 1.0, + "time_limit_was_exceeded": False, + "memory": 262144, + } + ], + } + ] + self.assertFalse(self.checker.has_slow_testcases(details_sample, time_limit)) + + # Truly missing times -> warns + details_no_times = [ + { + "idx": 0, + "score": 10.0, + "testcases": [{"outcome": "Correct"}], + } + ] + with self.assertLogs("cmscontrib.SolutionChecker", level="WARNING") as cm: + self.assertFalse( + self.checker.has_slow_testcases(details_no_times, time_limit) + ) + self.assertTrue(any("No testcase times found" in msg for msg in cm.output)) + + # Flat testcase structure (e.g. Sum score type) + details_flat = [{"idx": 0, "time": 0.8}] + self.assertTrue(self.checker.has_slow_testcases(details_flat, time_limit)) + + def test_login_validations(self): + # Password without username + c1 = SolutionChecker(base_url="http://localhost:8888", password="pwd") + with self.assertRaises(ValueError) as ctx: + c1.login() + self.assertIn("Password provided without username", str(ctx.exception)) + + # Username without password (no admin token) + c2 = SolutionChecker(base_url="http://localhost:8888", username="user") + with self.assertRaises(ValueError) as ctx: + c2.login() + self.assertIn("provided without password", str(ctx.exception)) + + # Admin token without username + c3 = SolutionChecker(base_url="http://localhost:8888", admin_token="admintoken") + with self.assertRaises(ValueError) as ctx: + c3.login() + self.assertIn("Admin token requires --username", str(ctx.exception)) + + # No credentials -> IP autologin + c4 = SolutionChecker(base_url="http://localhost:8888") + c4.login() + self.assertEqual(c4.auth_header, {}) + + @patch("requests.Session.post") + def test_login_admin_token(self, mock_post): + mock_response = MagicMock() + mock_response.json.return_value = {"login_data": "signed_cookie"} + mock_post.return_value = mock_response + + checker = SolutionChecker( + base_url="http://localhost:8888", + username="marago", + admin_token="admintoken", + ) + checker.login() + self.assertEqual(checker.auth_header, {"X-CMS-Authorization": "signed_cookie"}) + mock_post.assert_called_with( + "http://localhost:8888/api/login", + data={"admin_token": "admintoken", "username": "marago"}, + ) + + @patch("requests.Session.post") + def test_login_failure(self, mock_post): + mock_response = MagicMock() + mock_response.status_code = 403 + mock_response.json.return_value = {"error": "Invalid credentials"} + http_error = requests.exceptions.HTTPError(response=mock_response) + mock_response.raise_for_status.side_effect = http_error + mock_post.return_value = mock_response + + checker = SolutionChecker( + base_url="http://localhost:8888", + username="user", + password="wrong_password", + ) + with self.assertRaises(RuntimeError) as ctx: + checker.login() + self.assertIn("Invalid credentials", str(ctx.exception)) + + def test_extract_error_message(self): + # JSON response with error field + r1 = MagicMock() + r1.json.return_value = {"error": "The contest is not open"} + err1 = requests.exceptions.HTTPError(response=r1) + self.assertEqual( + SolutionChecker._extract_error_message(err1), + "The contest is not open", + ) + + # Plain text response + r2 = MagicMock() + r2.json.side_effect = Exception("Not JSON") + r2.text = "Forbidden" + err2 = requests.exceptions.HTTPError(response=r2) + self.assertEqual(SolutionChecker._extract_error_message(err2), "Forbidden") + + @patch("requests.Session.get") + def test_get_submission_details_endpoints(self, mock_get): + mock_response = MagicMock() + mock_response.json.return_value = {"details": [{"idx": 0}]} + mock_get.return_value = mock_response + + # Without admin token -> calls /details + checker_user = SolutionChecker( + base_url="http://localhost:8888/contest", + username="u", + password="p", + ) + res = checker_user.get_submission_details("task1", "123") + self.assertEqual(res, [{"idx": 0}]) + mock_get.assert_called_with( + "http://localhost:8888/contest/api/task1/submissions/123/details", + headers={}, + ) + + # With admin token -> calls /full_details + checker_admin = SolutionChecker( + base_url="http://localhost:8888/contest", + username="u", + admin_token="admintoken", + ) + res_admin = checker_admin.get_submission_details("task1", "123") + self.assertEqual(res_admin, [{"idx": 0}]) + mock_get.assert_called_with( + "http://localhost:8888/contest/api/task1/submissions/" "123/full_details", + headers={}, + ) + + def test_is_subtask_slow(self): + time_limit = 1.0 + # Positive score and slow testcase (> 0.5 * 1.0) -> True + st_slow = { + "score": 10.0, + "testcases": [{"time": 0.6}], + } + self.assertTrue(SolutionChecker.is_subtask_slow(st_slow, time_limit)) + + # Positive score and fast testcase -> False + st_fast = { + "score": 10.0, + "testcases": [{"time": 0.4}], + } + self.assertFalse(SolutionChecker.is_subtask_slow(st_fast, time_limit)) + + # Zero score with slow testcase -> False + st_zero = { + "score": 0.0, + "score_fraction": 0.0, + "testcases": [{"time": 0.8}], + } + self.assertFalse(SolutionChecker.is_subtask_slow(st_zero, time_limit)) + + def test_check_single_subtask(self): + st_ac = { + "idx": 1, + "score": 20.0, + "max_score": 20.0, + "score_fraction": 1.0, + "testcases": [{"outcome": "Correct"}], + } + self.assertIsNone(self.checker.check_single_subtask(st_ac, 1, "Accepted")) + self.assertIsNotNone(self.checker.check_single_subtask(st_ac, 1, "WrongAnswer")) + self.assertIsNone(self.checker.check_single_subtask(st_ac, 1, None)) + + def test_format_report_table(self): + time_limit = 1.0 + results = [ + { + "name": "sol_correct.cpp", + "criteria": { + "checks": ["Accepted", "Accepted"], + "min_score": 100.0, + "max_score": 100.0, + }, + "score": 100.0, + "compilation_failed": False, + "details": [ + { + "idx": 0, + "score": 50.0, + "max_score": 50.0, + "testcases": [{"outcome": "Correct", "time": 0.1}], + }, + { + "idx": 1, + "score": 50.0, + "max_score": 50.0, + "testcases": [{"outcome": "Correct", "time": 0.1}], + }, + ], + "failed": False, + "slow": False, + }, + { + "name": "sol_slow.cpp", + "criteria": { + "checks": ["Accepted", "Accepted"], + "min_score": 100.0, + "max_score": 100.0, + }, + "score": 100.0, + "compilation_failed": False, + "details": [ + { + "idx": 0, + "score": 50.0, + "max_score": 50.0, + "testcases": [{"outcome": "Correct", "time": 0.1}], + }, + { + "idx": 1, + "score": 50.0, + "max_score": 50.0, + "testcases": [{"outcome": "Correct", "time": 0.7}], + }, + ], + "failed": False, + "slow": True, + }, + { + "name": "sol_failed.cpp", + "criteria": { + "checks": ["Accepted", "WrongAnswer"], + "min_score": 50.0, + "max_score": 50.0, + }, + "score": 100.0, + "compilation_failed": False, + "details": [ + { + "idx": 0, + "score": 50.0, + "max_score": 50.0, + "testcases": [{"outcome": "Correct", "time": 0.1}], + }, + { + "idx": 1, + "score": 50.0, + "max_score": 50.0, + "testcases": [{"outcome": "Correct", "time": 0.1}], + }, + ], + "failed": True, + "slow": False, + }, + { + "name": "sol_compile_err.cpp", + "criteria": { + "checks": ["Accepted", "Accepted"], + "min_score": 100.0, + "max_score": 100.0, + }, + "score": 0.0, + "compilation_failed": True, + "details": None, + "failed": True, + "slow": False, + }, + { + "name": "arcari.cpp", + "criteria": { + "checks": [None, "Accepted", "Accepted"], + "min_score": 110.0, + "max_score": 120.0, + }, + "score": 71.0, + "compilation_failed": False, + "details": [ + { + "idx": 0, + "score": 0.0, + "max_score": 0.0, + "testcases": [{"outcome": "Correct", "time": 0.001}], + }, + { + "idx": 1, + "score": 20.0, + "max_score": 20.0, + "testcases": [{"outcome": "Correct", "time": 0.001}], + }, + { + "idx": 2, + "score": 51.0, + "max_score": 100.0, + "testcases": [{"outcome": "Correct", "time": 0.001}], + }, + ], + "failed": True, + "slow": False, + }, + ] + + # Formatted with color + table_colored = self.checker.format_report_table( + results, time_limit, use_color=True + ) + self.assertIn("sol_correct.cpp", table_colored) + self.assertIn("sol_slow.cpp", table_colored) + self.assertIn("sol_failed.cpp", table_colored) + self.assertIn("sol_compile_err.cpp", table_colored) + self.assertIn("arcari.cpp", table_colored) + # Verify expected score format + self.assertIn("71 (expected 110-120)", table_colored) + self.assertIn("100 (expected 50)", table_colored) + # Verify short status codes + self.assertIn("AC", table_colored) + self.assertIn("WA", table_colored) + self.assertIn("CE", table_colored) + # Verify ANSI colors + self.assertIn("\x1b[32;1m", table_colored) # Green + self.assertIn("\x1b[33;1m", table_colored) # Yellow + self.assertIn("\x1b[31;1m", table_colored) # Red + + # Formatted plain text (no ANSI codes) + table_plain = self.checker.format_report_table( + results, time_limit, use_color=False + ) + self.assertNotIn("\x1b[", table_plain) + self.assertIn("CE", table_plain) + self.assertIn("71 (expected 110-120)", table_plain) + self.assertIn("100", table_plain) + + # Empty results -> empty string + self.assertEqual(self.checker.format_report_table([], time_limit), "") + + +if __name__ == "__main__": + unittest.main() diff --git a/cmstestsuite/unit_tests/grading/scoretypes/GroupMinTest.py b/cmstestsuite/unit_tests/grading/scoretypes/GroupMinTest.py index 294471c750..082a7d02e5 100755 --- a/cmstestsuite/unit_tests/grading/scoretypes/GroupMinTest.py +++ b/cmstestsuite/unit_tests/grading/scoretypes/GroupMinTest.py @@ -21,8 +21,9 @@ import unittest from cms.grading.scoretypes.GroupMin import GroupMin -from cmstestsuite.unit_tests.grading.scoretypes.scoretypetestutils \ - import ScoreTypeTestMixin +from cmstestsuite.unit_tests.grading.scoretypes.scoretypetestutils import ( + ScoreTypeTestMixin, +) class TestGroupMin(ScoreTypeTestMixin, unittest.TestCase): @@ -88,49 +89,69 @@ def test_max_scores_regexp(self): """Test max score is correct when groups are regexp-defined.""" s1, s2, s3 = 10.5, 30.5, 59 parameters = [[0, "0_*"], [s1, "1_*"], [s2, "2_*"], [s3, "3_*"]] - header = ["Subtask 0 (0)", - "Subtask 1 (10.5)", "Subtask 2 (30.5)", "Subtask 3 (59)"] + header = [ + "Subtask 0 (0)", + "Subtask 1 (10.5)", + "Subtask 2 (30.5)", + "Subtask 3 (59)", + ] # Only group 1_* is public. public_testcases = dict(self._public_testcases) - self.assertEqual(GroupMin(parameters, public_testcases, 2).max_scores(), - (s1 + s2 + s3, s1, header)) + self.assertEqual( + GroupMin(parameters, public_testcases, 2).max_scores(), + (s1 + s2 + s3, s1, header), + ) # All groups are public for testcase in public_testcases.keys(): public_testcases[testcase] = True - self.assertEqual(GroupMin(parameters, public_testcases, 2).max_scores(), - (s1 + s2 + s3, s1 + s2 + s3, header)) + self.assertEqual( + GroupMin(parameters, public_testcases, 2).max_scores(), + (s1 + s2 + s3, s1 + s2 + s3, header), + ) # No groups are public for testcase in public_testcases.keys(): public_testcases[testcase] = False - self.assertEqual(GroupMin(parameters, public_testcases, 2).max_scores(), - (s1 + s2 + s3, 0, header)) + self.assertEqual( + GroupMin(parameters, public_testcases, 2).max_scores(), + (s1 + s2 + s3, 0, header), + ) def test_max_scores_number(self): """Test max score is correct when groups are number-defined.""" s1, s2, s3 = 10.5, 30.5, 59 parameters = [[0, 1], [s1, 2], [s2, 2], [s3, 2]] - header = ["Subtask 0 (0)", - "Subtask 1 (10.5)", "Subtask 2 (30.5)", "Subtask 3 (59)"] + header = [ + "Subtask 0 (0)", + "Subtask 1 (10.5)", + "Subtask 2 (30.5)", + "Subtask 3 (59)", + ] # Only group 1_* is public. public_testcases = dict(self._public_testcases) - self.assertEqual(GroupMin(parameters, public_testcases, 2).max_scores(), - (s1 + s2 + s3, s1, header)) + self.assertEqual( + GroupMin(parameters, public_testcases, 2).max_scores(), + (s1 + s2 + s3, s1, header), + ) # All groups are public for testcase in public_testcases.keys(): public_testcases[testcase] = True - self.assertEqual(GroupMin(parameters, public_testcases, 2).max_scores(), - (s1 + s2 + s3, s1 + s2 + s3, header)) + self.assertEqual( + GroupMin(parameters, public_testcases, 2).max_scores(), + (s1 + s2 + s3, s1 + s2 + s3, header), + ) # No groups are public for testcase in public_testcases.keys(): public_testcases[testcase] = False - self.assertEqual(GroupMin(parameters, public_testcases, 2).max_scores(), - (s1 + s2 + s3, 0.0, header)) + self.assertEqual( + GroupMin(parameters, public_testcases, 2).max_scores(), + (s1 + s2 + s3, 0.0, header), + ) def test_compute_score(self): s1, s2, s3 = 10.5, 30.5, 59 @@ -141,23 +162,21 @@ def test_compute_score(self): # All correct. self.assertComputeScore( gmin.compute_score(sr), - s1 + s2 + s3, s1, [0, s1, s2, s3], [ - {"idx": 0}, - {"idx": 1}, - {"idx": 2}, - {"idx": 3} - ]) + s1 + s2 + s3, + s1, + [0, s1, s2, s3], + [{"idx": 0}, {"idx": 1}, {"idx": 2}, {"idx": 3}], + ) # Some non-public subtask is incorrect. self.set_outcome(sr, "3_1", 0.0) self.assertComputeScore( gmin.compute_score(sr), - s1 + s2, s1, [0, s1, s2, 0], [ - {"idx": 0}, - {"idx": 1}, - {"idx": 2}, - {"idx": 3} - ]) + s1 + s2, + s1, + [0, s1, s2, 0], + [{"idx": 0}, {"idx": 1}, {"idx": 2}, {"idx": 3}], + ) # Also the public subtask is incorrect. self.set_outcome(sr, "1_0", 0.0) @@ -165,24 +184,60 @@ def test_compute_score(self): sr.evaluations[1].outcome = 0.0 self.assertComputeScore( gmin.compute_score(sr), - s2, 0.0, [0, 0, s2, 0], [ - {"idx": 0}, - {"idx": 1}, - {"idx": 2}, - {"idx": 3} - ]) + s2, + 0.0, + [0, 0, s2, 0], + [{"idx": 0}, {"idx": 1}, {"idx": 2}, {"idx": 3}], + ) # Some partial results. self.set_outcome(sr, "3_0", 0.5) self.set_outcome(sr, "3_1", 0.1) self.assertComputeScore( gmin.compute_score(sr), - s2 + s3 * 0.1, 0.0, [0, 0, s2, s3 * 0.1], [ - {"idx": 0}, - {"idx": 1}, - {"idx": 2}, - {"idx": 3} - ]) + s2 + s3 * 0.1, + 0.0, + [0, 0, s2, s3 * 0.1], + [{"idx": 0}, {"idx": 1}, {"idx": 2}, {"idx": 3}], + ) + + def test_get_json_details(self): + from cms import ( + FEEDBACK_LEVEL_FULL, + FEEDBACK_LEVEL_RESTRICTED, + FEEDBACK_LEVEL_OI_RESTRICTED, + ) + + parameters = [[10.0, "1_*"], [20.0, "2_*"]] + gmin = GroupMin(parameters, self._public_testcases, 2) + sr = self.get_submission_result(self._public_testcases) + self.set_outcome(sr, "1_0", 0.0) + self.set_outcome(sr, "1_1", 0.0) + _, subtasks, _, _, _ = gmin.compute_score(sr) + + # FULL feedback level includes all testcases and times/memory + full = gmin.get_json_details(subtasks, FEEDBACK_LEVEL_FULL) + self.assertEqual(len(full), 2) + self.assertEqual(len(full[0]["testcases"]), 2) + self.assertIn("time", full[0]["testcases"][0]) + self.assertIn("memory", full[0]["testcases"][0]) + self.assertIn("outcome", full[0]["testcases"][0]) + + # RESTRICTED feedback level strips time and keeps placeholders for + # hidden testcases + restricted = gmin.get_json_details(subtasks, FEEDBACK_LEVEL_RESTRICTED) + self.assertEqual(len(restricted), 2) + self.assertNotIn("time", restricted[0]["testcases"][0]) + self.assertNotIn("memory", restricted[0]["testcases"][0]) + self.assertEqual(restricted[0]["testcases"][0]["outcome"], "Not correct") + # 1_1 was hidden because 1_0 was lowest, but placeholder exists + self.assertNotIn("outcome", restricted[0]["testcases"][1]) + self.assertEqual(restricted[0]["testcases"][1]["idx"], "1_1") + + # OI_RESTRICTED feedback level omits non-visible testcases + oi = gmin.get_json_details(subtasks, FEEDBACK_LEVEL_OI_RESTRICTED) + self.assertEqual(len(oi[0]["testcases"]), 1) + self.assertEqual(oi[0]["testcases"][0]["idx"], "1_0") if __name__ == "__main__": diff --git a/cmstestsuite/unit_tests/grading/scoretypes/SumTest.py b/cmstestsuite/unit_tests/grading/scoretypes/SumTest.py index 1a243026cf..31907dc867 100755 --- a/cmstestsuite/unit_tests/grading/scoretypes/SumTest.py +++ b/cmstestsuite/unit_tests/grading/scoretypes/SumTest.py @@ -21,8 +21,9 @@ import unittest from cms.grading.scoretypes.Sum import Sum -from cmstestsuite.unit_tests.grading.scoretypes.scoretypetestutils \ - import ScoreTypeTestMixin +from cmstestsuite.unit_tests.grading.scoretypes.scoretypetestutils import ( + ScoreTypeTestMixin, +) class TestSum(ScoreTypeTestMixin, unittest.TestCase): @@ -50,9 +51,10 @@ def test_paramaters_invalid(self): def test_max_scores(self): testcase_score = 10.5 - self.assertEqual(Sum(testcase_score, - self._public_testcases, 2).max_scores(), - (testcase_score * 4, testcase_score, [])) + self.assertEqual( + Sum(testcase_score, self._public_testcases, 2).max_scores(), + (testcase_score * 4, testcase_score, []), + ) def test_compute_score(self): testcase_score = 10.5 @@ -62,45 +64,73 @@ def test_compute_score(self): # All correct. self.assertComputeScore( st.compute_score(sr), - testcase_score * 4, testcase_score, [], [ - {"idx": '0'}, - {"idx": '1'}, - {"idx": '2'}, - {"idx": '3'} - ]) + testcase_score * 4, + testcase_score, + [], + [{"idx": "0"}, {"idx": "1"}, {"idx": "2"}, {"idx": "3"}], + ) # Some non-public subtask is incorrect. self.set_outcome(sr, "3", 0.0) self.assertComputeScore( st.compute_score(sr), - testcase_score * 3, testcase_score, [], [ - {"idx": '0'}, - {"idx": '1'}, - {"idx": '2'}, - {"idx": '3'} - ]) + testcase_score * 3, + testcase_score, + [], + [{"idx": "0"}, {"idx": "1"}, {"idx": "2"}, {"idx": "3"}], + ) # Also the public subtask is incorrect. self.set_outcome(sr, "1", 0.0) self.assertComputeScore( st.compute_score(sr), - testcase_score * 2, 0.0, [], [ - {"idx": '0'}, - {"idx": '1'}, - {"idx": '2'}, - {"idx": '3'} - ]) + testcase_score * 2, + 0.0, + [], + [{"idx": "0"}, {"idx": "1"}, {"idx": "2"}, {"idx": "3"}], + ) # Now the public subtask has some partial scores. self.set_outcome(sr, "1", 0.2) self.assertComputeScore( st.compute_score(sr), - testcase_score * 2.2, testcase_score * 0.2, [], [ - {"idx": '0'}, - {"idx": '1'}, - {"idx": '2'}, - {"idx": '3'} - ]) + testcase_score * 2.2, + testcase_score * 0.2, + [], + [{"idx": "0"}, {"idx": "1"}, {"idx": "2"}, {"idx": "3"}], + ) + + def test_get_json_details(self): + from cms import FEEDBACK_LEVEL_FULL, FEEDBACK_LEVEL_RESTRICTED + + st = Sum(10, self._public_testcases, 2) + sr = self.get_submission_result(self._public_testcases) + _, testcases, _, _, _ = st.compute_score(sr) + + full = st.get_json_details(testcases, FEEDBACK_LEVEL_FULL) + self.assertEqual(len(full), 4) + self.assertIn("time", full[0]) + self.assertIn("memory", full[0]) + + restricted = st.get_json_details(testcases, FEEDBACK_LEVEL_RESTRICTED) + self.assertEqual(len(restricted), 4) + self.assertNotIn("time", restricted[0]) + self.assertNotIn("memory", restricted[0]) + + def test_get_html_details(self): + from cms import FEEDBACK_LEVEL_FULL, FEEDBACK_LEVEL_RESTRICTED + + st = Sum(10, self._public_testcases, 2) + sr = self.get_submission_result(self._public_testcases) + _, testcases, _, _, _ = st.compute_score(sr) + + html_full = st.get_html_details(testcases, FEEDBACK_LEVEL_FULL) + self.assertIn("execution-time", html_full) + self.assertIn("memory-used", html_full) + + html_restricted = st.get_html_details(testcases, FEEDBACK_LEVEL_RESTRICTED) + self.assertNotIn("execution-time", html_restricted) + self.assertNotIn("memory-used", html_restricted) if __name__ == "__main__": diff --git a/cmstestsuite/unit_tests/server/contest/api_test.py b/cmstestsuite/unit_tests/server/contest/api_test.py new file mode 100644 index 0000000000..f3759ef403 --- /dev/null +++ b/cmstestsuite/unit_tests/server/contest/api_test.py @@ -0,0 +1,180 @@ +#!/usr/bin/env python3 + +# Contest Management System - http://cms-dev.github.io/ +# Copyright © 2026 Luca Versari +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as +# published by the Free Software Foundation, either version 3 of the +# License, or (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + +"""Unit tests for submission details API handlers.""" + +import unittest +from unittest.mock import MagicMock + +from cms import FEEDBACK_LEVEL_RESTRICTED, FEEDBACK_LEVEL_FULL +from cms.server.contest.handlers.api import ( + ApiSubmissionDetailsHandler, + ApiSubmissionFullDetailsHandler, +) + + +class BaseApiHandlerTest(unittest.TestCase): + def setUp(self): + super().setUp() + self.task = MagicMock() + self.task.name = "my_task" + self.task.feedback_level = FEEDBACK_LEVEL_RESTRICTED + + self.score_type = MagicMock() + self.dataset = MagicMock() + self.dataset.score_type_object = self.score_type + self.task.active_dataset = self.dataset + + self.submission = MagicMock() + self.submission.opaque_id = 1 + self.submission_result = MagicMock() + self.submission_result.scored.return_value = True + self.submission_result.score_details = [{"idx": 0, "score": 100.0}] + self.submission_result.public_score_details = [{"idx": 0, "score": 0.0}] + self.submission.get_result.return_value = self.submission_result + self.submission.tokened.return_value = False + + def create_handler( + self, handler_cls, impersonated=False, phase=0, current_user=True + ): + handler = handler_cls.__new__(handler_cls) + handler.impersonated_by_admin = impersonated + handler.contest = MagicMock() + handler.contest.name = "test_contest" + handler._current_user = MagicMock() if current_user else None + handler.request = MagicMock() + handler.request.arguments = {} + handler.request.headers = {} + handler.application = MagicMock() + handler.application.service = MagicMock() + handler.application.service.contest_id = 1 + handler.is_multi_contest = lambda: False + handler.r_params = {"actual_phase": phase} + handler.json_data = None + handler.status_code = 200 + + def fake_json(data, status_code=200): + handler.json_data = data + handler.status_code = status_code + + handler.json = fake_json + + def fake_get_task(name): + return self.task if name == self.task.name else None + + def fake_get_submission(task, opaque_id): + if task == self.task and str(opaque_id) == "1": + return self.submission + return None + + handler.get_task = fake_get_task + handler.get_submission = fake_get_submission + return handler + + +class TestApiSubmissionDetailsHandler(BaseApiHandlerTest): + + def test_task_not_found(self): + handler = self.create_handler(ApiSubmissionDetailsHandler) + handler.get("unknown_task", "1") + self.assertEqual(handler.status_code, 404) + self.assertEqual(handler.json_data, {"error": "Task not found"}) + + def test_submission_not_found(self): + handler = self.create_handler(ApiSubmissionDetailsHandler) + handler.get("my_task", "999") + self.assertEqual(handler.status_code, 404) + self.assertEqual(handler.json_data, {"error": "Submission not found"}) + + def test_contestant_restricted_details(self): + self.score_type.get_json_details.return_value = [{"idx": 0, "filtered": True}] + handler = self.create_handler(ApiSubmissionDetailsHandler, phase=0) + handler.get("my_task", "1") + + self.assertEqual(handler.status_code, 200) + self.assertEqual(handler.json_data, {"details": [{"idx": 0, "filtered": True}]}) + self.score_type.get_json_details.assert_called_once_with( + self.submission_result.public_score_details, + FEEDBACK_LEVEL_RESTRICTED, + ) + + def test_contestant_tokened_details(self): + self.submission.tokened.return_value = True + self.score_type.get_json_details.return_value = [{"idx": 0, "tokened": True}] + handler = self.create_handler(ApiSubmissionDetailsHandler, phase=0) + handler.get("my_task", "1") + + self.assertEqual(handler.status_code, 200) + self.assertEqual(handler.json_data, {"details": [{"idx": 0, "tokened": True}]}) + self.score_type.get_json_details.assert_called_once_with( + self.submission_result.score_details, FEEDBACK_LEVEL_RESTRICTED + ) + + def test_analysis_mode_details(self): + self.submission.tokened.return_value = False + self.score_type.get_json_details.return_value = [{"idx": 0, "analysis": True}] + handler = self.create_handler(ApiSubmissionDetailsHandler, phase=3) + handler.get("my_task", "1") + + self.assertEqual(handler.status_code, 200) + self.assertEqual(handler.json_data, {"details": [{"idx": 0, "analysis": True}]}) + self.score_type.get_json_details.assert_called_once_with( + self.submission_result.score_details, FEEDBACK_LEVEL_FULL + ) + + +class TestApiSubmissionFullDetailsHandler(BaseApiHandlerTest): + + def test_not_impersonated_forbidden(self): + handler = self.create_handler( + ApiSubmissionFullDetailsHandler, impersonated=False + ) + handler.get("my_task", "1") + self.assertEqual(handler.status_code, 403) + self.assertEqual(handler.json_data, {"error": "Admin impersonation required"}) + + def test_impersonated_task_not_found(self): + handler = self.create_handler( + ApiSubmissionFullDetailsHandler, impersonated=True + ) + handler.get("unknown_task", "1") + self.assertEqual(handler.status_code, 404) + + def test_impersonated_submission_not_found(self): + handler = self.create_handler( + ApiSubmissionFullDetailsHandler, impersonated=True + ) + handler.get("my_task", "999") + self.assertEqual(handler.status_code, 404) + + def test_impersonated_returns_full_details(self): + self.score_type.get_json_details.return_value = [{"idx": 0, "full": True}] + handler = self.create_handler( + ApiSubmissionFullDetailsHandler, impersonated=True + ) + handler.get("my_task", "1") + + self.assertEqual(handler.status_code, 200) + self.assertEqual(handler.json_data, {"details": [{"idx": 0, "full": True}]}) + self.score_type.get_json_details.assert_called_once_with( + self.submission_result.score_details, FEEDBACK_LEVEL_FULL + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/docs/API.rst b/docs/API.rst index 8e85be0645..61b9e3fdf8 100644 --- a/docs/API.rst +++ b/docs/API.rst @@ -102,6 +102,47 @@ Additional details on the submission's results can be retrieved by making an authenticated ``GET`` request to ``/tasks/{taskname}/submissions/{id}/details``. The endpoint will return an HTML snippet matching what is seen by contestants. +Submission details (JSON) +========================= + +An authenticated ``GET`` request to +``/api/{taskname}/submissions/{id}/details`` will return a JSON object with the +submission details filtered according to the task's feedback level: + +.. sourcecode:: json + + { + "details": [ + { + "idx": 0, + "score": 10.0, + "max_score": 10.0, + "score_fraction": 1.0, + "testcases": [ + { + "idx": "0_0", + "outcome": "Correct", + "text": ["Output is correct"] + } + ] + } + ] + } + +If the task uses a non-subtask score type such as ``Sum``, ``details`` will +contain a list of testcase objects directly. + +Unfiltered submission details (JSON) +==================================== + +An authenticated ``GET`` request to +``/api/{taskname}/submissions/{id}/full_details`` will return a JSON object with +the complete, unfiltered submission details (including all testcase execution +times and memory usage). + +This endpoint requires an impersonated authentication token (see `Impersonation +of users`_ below) and returns ``403 Forbidden`` for regular contestants. + Impersonation of users ======================