graders: add types to BridgedInteractiveGrader

dmoj/graders/bridged.py

@@ -1,29 +1,43 @@
 import os
 import shlex
 import subprocess
+from typing import TYPE_CHECKING
 
+from dmoj.checkers import CheckerOutput
+from dmoj.config import ConfigNode
 from dmoj.contrib import contrib_modules
 from dmoj.error import InternalError
+from dmoj.executors.base_executor import BaseExecutor
 from dmoj.graders.standard import StandardGrader
 from dmoj.judgeenv import env, get_problem_root
+from dmoj.problem import Problem, TestCase
+from dmoj.result import Result
 from dmoj.utils.helper_files import compile_with_auxiliary_files, mktemp
 from dmoj.utils.unicode import utf8text
 
+if TYPE_CHECKING:
+    from dmoj.judge import JudgeWorker
+
 
 class BridgedInteractiveGrader(StandardGrader):
-    def __init__(self, judge, problem, language, source):
+    handler_data: ConfigNode
+    interactor_binary: BaseExecutor
+    contrib_type: str
+
+    def __init__(self, judge: 'JudgeWorker', problem: Problem, language: str, source: bytes) -> None:
         super().__init__(judge, problem, language, source)
         self.handler_data = self.problem.config.interactive
         self.interactor_binary = self._generate_interactor_binary()
         self.contrib_type = self.handler_data.get('type', 'default')
 
         if self.contrib_type not in contrib_modules:
-            raise InternalError('%s is not a valid contrib module' % self.contrib_type)
+            raise InternalError(f'{self.contrib_type} is not a valid contrib module')
 
-    def check_result(self, case, result):
+    def check_result(self, case: TestCase, result: Result) -> CheckerOutput:
         # We parse the return code first in case the grader crashed, so it can raise the IE.
         # Usually a grader crash will result in IR/RTE/TLE,
         # so checking submission return code first will cover up the grader fail.
+        assert self._interactor.stderr is not None
         stderr = self._interactor.stderr.read()
         parsed_result = contrib_modules[self.contrib_type].ContribModule.parse_return_code(
             self._interactor,
@@ -38,7 +52,7 @@ def check_result(self, case, result):
 
         return (not result.result_flag) and parsed_result
 
-    def _launch_process(self, case):
+    def _launch_process(self, case: TestCase) -> None:
         self._interactor_stdin_pipe, submission_stdout_pipe = os.pipe()
         submission_stdin_pipe, self._interactor_stdout_pipe = os.pipe()
         self._current_proc = self.binary.launch(
@@ -53,7 +67,10 @@ def _launch_process(self, case):
         os.close(submission_stdin_pipe)
         os.close(submission_stdout_pipe)
 
-    def _interact_with_process(self, case, result, input):
+    def _interact_with_process(self, case: TestCase, result: Result, input: bytes) -> bytes:
+        assert self._current_proc is not None
+        assert self._current_proc.stderr is not None
+
         judge_output = case.output_data()
         # Give TL + 2s by default, so we do not race (and incorrectly throw IE) if submission gets TLE
         self._interactor_time_limit = (self.handler_data.preprocessing_time or 2) + self.problem.time_limit
@@ -92,7 +109,7 @@ def _interact_with_process(self, case, result, input):
 
             return self._current_proc.stderr.read()
 
-    def _generate_interactor_binary(self):
+    def _generate_interactor_binary(self) -> BaseExecutor:
         files = self.handler_data.files
         if isinstance(files, str):
             filenames = [files]