Add problem validation for dmoj-cli; #335

dmoj/commands/__init__.py

@@ -10,6 +10,7 @@
 from dmoj.commands.show import ShowCommand
 from dmoj.commands.submissions import ListSubmissionsCommand
 from dmoj.commands.submit import SubmitCommand
+from dmoj.commands.test import TestCommand
 
 all_commands: List[Type[Command]] = [
     ListProblemsCommand,
@@ -18,6 +19,7 @@
     ResubmitCommand,
     RejudgeCommand,
     DifferenceCommand,
+    TestCommand,
     ShowCommand,
     HelpCommand,
     QuitCommand,

dmoj/commands/test.py

@@ -0,0 +1,102 @@
+import sys
+import traceback
+from operator import itemgetter
+
+from dmoj.commands.base_command import Command
+from dmoj.error import InvalidCommandException
+from dmoj.judgeenv import get_problem_root, get_supported_problems
+from dmoj.problem import ProblemConfig, ProblemDataManager
+from dmoj.testsuite import Tester
+from dmoj.utils.ansi import ansi_style, print_ansi
+
+
+class ProblemTester(Tester):
+    def test_problem(self, problem_id):
+        self.output(ansi_style('Testing problem #ansi[%s](cyan|bold)...') % problem_id)
+
+        config = ProblemConfig(ProblemDataManager(get_problem_root(problem_id)))
+
+        if not config or 'tests' not in config or not config['tests']:
+            self.output(ansi_style('\t#ansi[Skipped](magenta|bold) - No tests found'))
+            return 0
+
+        fails = 0
+        for test in config['tests']:
+            # Do this check here as we need some way to identify the test
+            if 'source' not in test:
+                self.output(ansi_style('\t[Skipped](magenta|bold) - No source found for test'))
+                continue
+
+            test_name = test.get('label', test['source'])
+            self.output(ansi_style('\tRunning test #ansi[%s](yellow|bold)') % test_name)
+            try:
+                test_fails = self.run_test(problem_id, test)
+            except Exception:
+                fails += 1
+                self.output(ansi_style('\t#ansi[Test failed with exception:](red|bold)'))
+                self.output(traceback.format_exc())
+            else:
+                self.output(
+                    ansi_style('\tResult of test #ansi[%s](yellow|bold): ') % test_name
+                    + ansi_style(['#ansi[Failed](red|bold)', '#ansi[Success](green|bold)'][not test_fails])
+                )
+                fails += test_fails
+
+        return fails
+
+    def _check_targets(targets):
+        if 'posix' in targets:
+            return True
+        if 'freebsd' in sys.platform:
+            if 'freebsd' in targets:
+                return True
+            if not sys.platform.startswith('freebsd') and 'kfreebsd' in targets:
+                return True
+        elif sys.platform.startswith('linux') and 'linux' in targets:
+            return True
+        return False
+
+    def run_test(self, problem_id, config):
+        if 'targets' in config and not self._check_targets(config['targets']):
+            return 0
+
+        return self._run_test_case(problem_id, get_problem_root(problem_id), config)
+
+
+class TestCommand(Command):
+    name = 'test'
+    help = 'Runs tests on problems.'
+
+    def _populate_parser(self):
+        self.arg_parser.add_argument('problem_ids', nargs='+', help='id of problems to test')
+
+    def execute(self, line):
+        args = self.arg_parser.parse_args(line)
+
+        problem_ids = args.problem_ids
+        supported_problems = set(get_supported_problems())
+
+        unknown_problems = filter(lambda x: x not in supported_problems, problem_ids)
+        unknown_problems = ', '.join(map(lambda x: "'%s'" % x, unknown_problems))
+        if unknown_problems:
+            raise InvalidCommandException("unknown problem(s) %s" % unknown_problems)
+
+        tester = ProblemTester()
+        total_fails = 0
+        for problem_id in problem_ids:
+            fails = tester.test_problem(problem_id)
+            if fails:
+                print_ansi('Problem #ansi[%s](cyan|bold) #ansi[failed %d case(s)](red|bold).' % (problem_id, fails))
+            else:
+                print_ansi('Problem #ansi[%s](cyan|bold) passed with flying colours.' % problem_id)
+            print()
+            total_fails += fails
+
+        print()
+        print('Test complete')
+        if fails:
+            print_ansi('#ansi[A total of %d test(s) failed](red|bold)' % fails)
+        else:
+            print_ansi('#ansi[All tests passed.](green|bold)')
+
+        return fails

dmoj/problem.py

@@ -31,31 +31,13 @@ def __init__(self, problem_id, time_limit, memory_limit, meta):
 
         # Cache root dir so that we don't need to scan all roots (potentially very slow on networked mount).
         self.root_dir = get_problem_root(problem_id)
-        self.problem_data = ProblemDataManager(self)
+        self.problem_data = ProblemDataManager(self.root_dir)
 
         # Checkers modules must be stored in a dict, for the duration of execution,
         # lest globals be deleted with the module.
         self._checkers = {}
 
-        try:
-            doc = yaml.safe_load(self.problem_data['init.yml'])
-            if not doc:
-                raise InvalidInitException('I find your lack of content disturbing.')
-            self.config = ConfigNode(
-                doc,
-                defaults={
-                    'wall_time_factor': 3,
-                    'output_prefix_length': 0 if 'signature_grader' in doc else 64,
-                    'output_limit_length': 25165824,
-                    'binary_data': False,
-                    'short_circuit': True,
-                    'points': 1,
-                    'symlinks': {},
-                    'meta': meta,
-                },
-            )
-        except (IOError, KeyError, ParserError, ScannerError) as e:
-            raise InvalidInitException(str(e))
+        self.config = ProblemConfig(self.problem_data, meta)
 
         self.problem_data.archive = self._resolve_archive_files()
         self._resolve_test_cases()
@@ -188,27 +170,50 @@ def _resolve_archive_files(self):
 
 
 class ProblemDataManager(dict):
-    def __init__(self, problem, **kwargs):
+    def __init__(self, problem_root_dir, **kwargs):
         super().__init__(**kwargs)
-        self.problem = problem
+        self.problem_root_dir = problem_root_dir
         self.archive = None
 
     def __missing__(self, key):
         try:
-            with open(os.path.join(self.problem.root_dir, key), 'rb') as f:
+            with open(os.path.join(self.problem_root_dir, key), 'rb') as f:
                 return f.read()
         except IOError:
             if self.archive:
                 zipinfo = self.archive.getinfo(key)
                 with self.archive.open(zipinfo) as f:
                     return f.read()
-            raise KeyError('file "%s" could not be found in "%s"' % (key, self.problem.root_dir))
+            raise KeyError('file "%s" could not be found in "%s"' % (key, self.problem_root_dir))
 
     def __del__(self):
         if self.archive:
             self.archive.close()
 
 
+class ProblemConfig(ConfigNode):
+    def __init__(self, problem_data, meta={}):
+        try:
+            doc = yaml.safe_load(problem_data['init.yml'])
+            if not doc:
+                raise InvalidInitException('I find your lack of content disturbing.')
+            super().__init__(
+                doc,
+                defaults={
+                    'wall_time_factor': 3,
+                    'output_prefix_length': 0 if 'signature_grader' in doc else 64,
+                    'output_limit_length': 25165824,
+                    'binary_data': False,
+                    'short_circuit': True,
+                    'points': 1,
+                    'symlinks': {},
+                    'meta': meta,
+                },
+            )
+        except (IOError, KeyError, ParserError, ScannerError) as e:
+            raise InvalidInitException(str(e))
+
+
 class BatchedTestCase:
     def __init__(self, batch_no, config, problem, cases):
         self.config = config

dmoj/testsuite.py

@@ -23,12 +23,12 @@ def fail(self, message):
         self.output('\t\t' + message.replace('\r\n', '\n').replace('\n', '\r\n\t\t'))
         self.failed = True
 
-    def set_expected(self, codes_all, codes_cases, points_all, points_cases, feedback_all, feedback_cases):
+    def set_expected(self, codes_all, codes_cases, score_all, score_cases, feedback_all, feedback_cases):
         self.failed = False
         self.codes_all = codes_all
         self.codes_cases = codes_cases
-        self.points_all = points_all
-        self.points_cases = points_cases
+        self.score_all = score_all
+        self.score_cases = score_cases
         self.feedback_all = feedback_all
         self.feedback_cases = feedback_cases
 
@@ -49,15 +49,15 @@ def test_case_status_packet(self, position, result):
         elif code not in self.codes_all:
             self.fail('Unexpected global code: %s, expecting %s' % (code, ', '.join(self.codes_all)))
 
-        if position in self.points_cases:
-            if result.points not in self.points_cases[position]:
+        if position in self.score_cases:
+            if result.points not in self.score_cases[position]:
                 self.fail(
-                    'Unexpected points for case %d: %s, expecting %s'
-                    % (position, result.points, ', '.join(self.points_cases[position]))
+                    'Unexpected score for case %d: %s, expecting %s'
+                    % (position, result.points, ', '.join(self.score_cases[position]))
                 )
-        elif self.points_all is not None and result.points not in self.points_all:
-            self.fail('Unexpected global points: %s, expecting %s' % (
-                result.points, ', '.join(map(str, self.points_all))))
+        elif self.score_all is not None and result.points not in self.score_all:
+            self.fail('Unexpected global score: %s, expecting %s' % (
+                result.points, ', '.join(map(str, self.score_all))))
 
         feedback = self.feedback_all
         if position in self.feedback_cases:
@@ -191,9 +191,12 @@ def run_test_case(self, problem, case, case_dir):
                 pass
 
         if not config:
-            self.output(ansi_style('\t\t#ansi[Skipped](magenta|bold) - No usable test.yml'))
+            self.output(ansi_style('\t\t#ansi[Skipped](magenta|bold) - No usable test config'))
             return 0
 
+        return self._run_test_case(problem, case_dir, config)
+
+    def _run_test_case(self, problem, case_dir, config):
         if 'skip' in config and config['skip']:
             self.output(ansi_style('\t\t#ansi[Skipped](magenta|bold) - Unsupported on current platform'))
             return 0
@@ -215,8 +218,8 @@ def run_test_case(self, problem, case, case_dir):
         codes_all, codes_cases = self.parse_expect(
             config.get('expect', 'AC'), config.get('cases', {}), self.parse_expected_codes
         )
-        points_all, points_cases = self.parse_expect(
-            config.get('points'), config.get('points_cases', {}), self.parse_points
+        score_all, score_cases = self.parse_expect(
+            config.get('score'), config.get('score_cases', {}), self.parse_score
         )
         feedback_all, feedback_cases = self.parse_expect(
             config.get('feedback'), config.get('feedback_cases', {}), self.parse_feedback
@@ -228,7 +231,7 @@ def output_case(data):
         fails = 0
         for source in sources:
             self.sub_id += 1
-            self.manager.set_expected(codes_all, codes_cases, points_all, points_cases, feedback_all, feedback_cases)
+            self.manager.set_expected(codes_all, codes_cases, score_all, score_cases, feedback_all, feedback_cases)
             self.judge.begin_grading(
                 Submission(self.sub_id, problem, language, source, time, memory, False, {}),
                 blocking=True,
@@ -257,13 +260,13 @@ def parse_expected_codes(self, codes):
             assert not (result - self.all_codes)
             return result
 
-    def parse_points(self, points):
-        if points is None or points == '*':
+    def parse_score(self, score):
+        if score is None or score == '*':
             return None
-        elif isinstance(points, (str, int)):
-            return {int(points)}
+        elif isinstance(score, (str, int)):
+            return {int(score)}
         else:
-            return set(map(int, points))
+            return set(map(int, score))
 
     def parse_feedback(self, feedback):
         if feedback is None or feedback == '*':