Skip to content

Commit

Permalink
Refactor TestFile
Browse files Browse the repository at this point in the history
  • Loading branch information
@JustinMeimar
JustinMeimar committed Oct 18, 2024
1 parent e5116cd commit 89c1f74
Show file tree
Hide file tree
Showing 2 changed files with 33 additions and 85 deletions.
4 changes: 2 additions & 2 deletions dragon_runner/runner.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand Down
114 changes: 31 additions & 83 deletions dragon_runner/testfile.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -7,37 +7,51 @@
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:
"""
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):
collection.add(self.input_stream)
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
"""
Expand All @@ -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.
Expand All @@ -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:
Expand All @@ -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
Expand All @@ -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}")

0 comments on commit 89c1f74

Please sign in to comment.