diff --git a/dragon_runner/runner.py b/dragon_runner/runner.py index d167869..a7a5e0b 100644 --- a/dragon_runner/runner.py +++ b/dragon_runner/runner.py @@ -76,7 +76,7 @@ def log(self, file=sys.stdout, args: CLIArgs=None): log(Fore.RED + fail_msg + Fore.RESET + f"{self.test.file}", indent=3, file=file) level = 3 if self.did_pass else 2 - log(f"==> Expected Out ({self.test.expected_out_bytes} bytes):", indent=5, level=level) + log(f"==> Expected Out ({len(self.test.expected_out)} bytes):", indent=5, level=level) log_multiline(self.test.expected_out, level=level, indent=6) log(f"==> Generated Out ({len(self.gen_output)} bytes):", indent=5, level=level) log_multiline(self.gen_output, level=level, indent=6) @@ -135,7 +135,7 @@ def run(self, test: TestFile, exe: Executable) -> TestResult: for index, step in enumerate(self.tc): last_step = index == len(self.tc) - 1 - input_stream = test.get_input_stream() if step.uses_ins else b'' + input_stream = test.input_stream if step.uses_ins else b'' output_file = self.resolve_output_file(step) command = self.resolve_command(step, MagicParams(exe.exe_path, input_file, output_file)) command_result = self.run_command(command, input_stream) diff --git a/dragon_runner/testfile.py b/dragon_runner/testfile.py index 901cb6d..b1450d4 100644 --- a/dragon_runner/testfile.py +++ b/dragon_runner/testfile.py @@ -7,15 +7,14 @@ class TestFile(Verifiable): __test__ = False def __init__(self, test_path, input_dir="input", input_stream_dir="input-stream", - output_dir="output", - comment_syntax="//"): - self.path = test_path - self.stem, self.extension = os.path.splitext(os.path.basename(test_path)) - self.file = self.stem + self.extension - self.input_dir = input_dir - self.input_stream_dir = input_stream_dir - self.output_dir = output_dir - self.comment_syntax = comment_syntax # default C99 // + output_dir="output", comment_syntax="//"): + self.path = test_path + self.stem, self.extension = os.path.splitext(os.path.basename(test_path)) + self.file = self.stem + self.extension + self.input_dir = input_dir + self.input_stream_dir = input_stream_dir + self.output_dir = output_dir + self.comment_syntax = comment_syntax # default C99 // self.verify() def verify(self) -> ErrorCollection: @@ -23,10 +22,10 @@ def verify(self) -> ErrorCollection: Ensure the paths supplied in CHECK_FILE and INPUT_FILE exist """ collection = ErrorCollection() - self.expected_out = self.get_expected_out() - self.input_stream = self.get_input_stream() + self.expected_out = self._get_content("CHECK:", "CHECK_FILE:") + self.input_stream = self._get_content("INPUT:", "INPUT_FILE:") - # If a parse and read of a tests input or output fails, propogate here + # If a parse and read of a tests input or output fails, propagate here if isinstance(self.expected_out, TestFileError): collection.add(self.expected_out) if isinstance(self.input_stream, TestFileError): @@ -34,10 +33,25 @@ def verify(self) -> ErrorCollection: if collection.has_errors(): return collection - self.expected_out_bytes = len(self.expected_out) - self.input_stream_bytes = len(self.input_stream) + def _get_content(self, inline_directive: str, file_directive: str) -> Union[bytes, TestFileError]: + """ + Generic method to get content based on directives + """ + content = self._get_directive_contents(inline_directive) + if content: + return content + + file_path = self._get_directive_contents(file_directive) + if file_path: + test_dir = os.path.dirname(self.path) + full_path = os.path.join(test_dir, bytes_to_str(file_path)) + if not os.path.exists(full_path): + return TestFileError(f"Failed to locate path supplied to {file_directive}: {full_path}") + return self._get_file_bytes(full_path) + + return b'' # default to empty content - def get_file_bytes(self, file_path: str) -> Optional[bytes]: + def _get_file_bytes(self, file_path: str) -> Optional[bytes]: """ Get file contents in bytes """ @@ -48,10 +62,8 @@ def get_file_bytes(self, file_path: str) -> Optional[bytes]: return file_bytes except FileNotFoundError: return None - except: - return None - def get_directive_contents(self, directive_prefix: str) -> Optional[bytes]: + def _get_directive_contents(self, directive_prefix: str) -> Optional[bytes]: """ Look into the testfile itself for contents defined in directives. Directives can appear anywhere in a line, as long as they're preceded by a comment syntax. @@ -60,7 +72,6 @@ def get_directive_contents(self, directive_prefix: str) -> Optional[bytes]: first_match = True with open(self.path, 'r') as test_file: for line in test_file: - comment_index = line.find(self.comment_syntax) directive_index = line.find(directive_prefix) if comment_index == -1 or directive_index == -1 or comment_index > directive_index: @@ -77,69 +88,7 @@ def get_directive_contents(self, directive_prefix: str) -> Optional[bytes]: first_match = False contents.seek(0) - if contents: - content_bytes = contents.getvalue() - return content_bytes - return None - - def get_file_contents(self, file_suffix, symmetric_dir) -> Optional[bytes]: - """ - Look into a symetric directory and current directory for a file with an - identical file path but differnt suffix. - """ - sym_path = self.path.replace(self.input_dir, f"/{symmetric_dir}/")\ - .replace(self.extension, file_suffix) - if os.path.exists(sym_path): - return self.get_file_bytes(sym_path) - - same_dir_path = self.path.replace(self.extension, file_suffix) - if os.path.exists(same_dir_path): - return self.get_file_bytes(same_dir_path) - return None - - def get_expected_out(self) -> Union[bytes, TestFileError]: - """ - Load the expected output for a test into a byte stream - """ - out_bytes = self.get_file_contents(".out", self.output_dir) - if out_bytes: - return out_bytes - - out_bytes = self.get_directive_contents("CHECK:") - if out_bytes: - return out_bytes - - check_file = self.get_directive_contents("CHECK_FILE:") - if check_file: - test_dir = os.path.dirname(self.path) - check_file_path = os.path.join(test_dir, bytes_to_str(check_file)) - if not os.path.exists(check_file_path): - return TestFileError( - f"Failed to locate path supplied to CHECK_FILE: {check_file_path}") - return self.get_file_bytes(check_file_path) - return b''# default expect empty output - - def get_input_stream(self) -> Union[bytes, TestFileError]: - """ - Load the input stream for a test into a byte stream - """ - out_bytes = self.get_file_contents(".ins", self.input_stream_dir) - if out_bytes: - return out_bytes - - out_bytes = self.get_directive_contents("INPUT:") - if out_bytes: - return out_bytes - - input_file = self.get_directive_contents("INPUT_FILE:") - if input_file: - test_dir = os.path.dirname(self.path) - input_file_path = os.path.join(test_dir, bytes_to_str(input_file)) - if not os.path.exists(input_file_path): - return TestFileError( - f"Failed to locate path supplied to INPUT_FILE: {input_file_path}") - return self.get_file_bytes(input_file_path) - return b''# default no input stream + return contents.getvalue() if contents else None def __repr__(self): max_test_name_length = 30 @@ -150,4 +99,3 @@ def __repr__(self): return (f"{test_name:<{max_test_name_length}}" f"{len(self.expected_out.getvalue()):>4}\t" f"{len(self.input_stream.getvalue()):>4}") -