Skip to content

Commit

Permalink
Fix type hints; fix some io test
Browse files Browse the repository at this point in the history
  • Loading branch information
Mr-Python-in-China committed Dec 9, 2024
1 parent cf182b9 commit f314965
Show file tree
Hide file tree
Showing 3 changed files with 38 additions and 30 deletions.
2 changes: 1 addition & 1 deletion .pylintrc
Original file line number Diff line number Diff line change
@@ -1,3 +1,3 @@
[MASTER]
py-version=3.5
py-version=3.6
disable=R0902,R0903,R0913,R0917,R0912
33 changes: 17 additions & 16 deletions cyaron/io.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,8 @@ def __init__(
data_id: Optional[int] = None,
disable_output: bool = False,
make_dirs: bool = False,
): ...
):
...

@overload
def __init__(
Expand All @@ -38,7 +39,8 @@ def __init__(
output_suffix: str = ".out",
disable_output: bool = False,
make_dirs: bool = False,
): ...
):
...

def __init__( # type: ignore
self,
Expand Down Expand Up @@ -89,15 +91,14 @@ def __init__( # type: ignore
# if the dir "./io" not found it will be created
"""
self.__closed = False
self.input_file, self.output_file = None, None
self.output_file = None
if file_prefix is not None:
# legacy mode
input_file = "{}{{}}{}".format(
self.__escape_format(file_prefix), self.__escape_format(input_suffix)
)
input_file = "{}{{}}{}".format(self.__escape_format(file_prefix),
self.__escape_format(input_suffix))
output_file = "{}{{}}{}".format(
self.__escape_format(file_prefix), self.__escape_format(output_suffix)
)
self.__escape_format(file_prefix),
self.__escape_format(output_suffix))
self.input_filename, self.output_filename = None, None
self.__input_temp, self.__output_temp = False, False
self.__init_file(input_file, data_id, "i", make_dirs)
Expand Down Expand Up @@ -276,13 +277,11 @@ def input_clear_content(self, pos: int = 0):

self.__clear(self.input_file, pos)

def output_gen(
self,
shell_cmd: Union[str, List[str]],
time_limit: float = None,
*,
replace_EOL: bool = True
):
def output_gen(self,
shell_cmd: Union[str, List[str]],
time_limit: Optional[float] = None,
*,
replace_EOL: bool = True):
"""
Run the command `shell_cmd` (usually the std program) and send it the input file as stdin.
Write its output to the output file.
Expand All @@ -302,7 +301,7 @@ def output_gen(
proc = subprocess.Popen(
shell_cmd,
shell=True,
stdin=self.input_file,
stdin=self.input_file.fileno(),
stdout=subprocess.PIPE,
universal_newlines=replace_EOL,
)
Expand Down Expand Up @@ -354,6 +353,8 @@ def output_clear_content(self, pos: int = 0):
Args:
pos: Where file will truncate
"""
if self.output_file is None:
raise ValueError("Output file is disabled")
self.__clear(self.output_file, pos)

def flush_buffer(self):
Expand Down
33 changes: 20 additions & 13 deletions cyaron/tests/io_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -72,36 +72,43 @@ def test_output_gen(self):

with open("test_gen.out", "rb") as f:
output = f.read()
self.assertEqual(output.strip(b"\n"), b"233")
self.assertEqual(output, b"233\n")

def test_output_gen_time_limit_exceeded(self):
with captured_output() as (out, err):
with open("long_time.py", "w") as f:
f.write("import time, os\nfn = input()\ntime.sleep(0.5)\nos.remove(fn)\n")
with captured_output():
with open("long_time.py", "w", encoding="utf-8") as f:
f.write("import time, os\n"
"fn = input()\n"
"time.sleep(0.1)\n"
"os.remove(fn)\n")

with IO("test_gen.in", "test_gen.out") as test:
fd, input_filename = tempfile.mkstemp()
os.close(fd)
abs_input_filename: str = os.path.abspath(input_filename)
with self.assertRaises(subprocess.TimeoutExpired):
test.input_writeln(abs_input_filename)
test.output_gen(f'"{sys.executable}" long_time.py', time_limit=0.1)
time.sleep(0.5)
test.output_gen(f'"{sys.executable}" long_time.py',
time_limit=0.05)
time.sleep(0.1)
try:
os.remove(input_filename)
except FileNotFoundError:
raise RuntimeError("Child processes have not been terminated.") from None
self.fail("Child processes have not been terminated.")

def test_output_gen_time_limit_not_exceeded(self):
with captured_output() as (out, err):
with open("short_time.py", "w") as f:
f.write("import time\ntime.sleep(0.1)\nprint(1)")
with captured_output():
with open("short_time.py", "w", encoding="utf-8") as f:
f.write("import time\n"
"time.sleep(0.1)\n"
"print(1)")

with IO("test_gen.in", "test_gen.out") as test:
test.output_gen(f'"{sys.executable}" short_time.py', time_limit=0.5)
with open("test_gen.out") as f:
test.output_gen(f'"{sys.executable}" short_time.py',
time_limit=0.5)
with open("test_gen.out", encoding="utf-8") as f:
output = f.read()
self.assertEqual(output.strip("\n"), "1")
self.assertEqual(output, "1\n")

def test_init_overload(self):
with IO(file_prefix="data{", data_id=5) as test:
Expand Down

0 comments on commit f314965

Please sign in to comment.