drive unittest directly

This commit is contained in:
Paul Gauthier 2023-06-29 10:26:33 -07:00
parent 81ec0e1608
commit 24589b543e

View file

@ -1,14 +1,14 @@
#!/usr/bin/env python #!/usr/bin/env python
import datetime import datetime
import io
import json import json
import os import os
import random import random
import re import re
import shutil import shutil
import subprocess
import sys
import time import time
import unittest
from collections import defaultdict from collections import defaultdict
from json.decoder import JSONDecodeError from json.decoder import JSONDecodeError
from pathlib import Path from pathlib import Path
@ -89,8 +89,6 @@ def main(
summarize_results(dirname) summarize_results(dirname)
return return
check_docker()
if clean and dirname.exists(): if clean and dirname.exists():
print("Cleaning up and replacing", dirname) print("Cleaning up and replacing", dirname)
dir_files = set(fn.name for fn in dirname.glob("*")) dir_files = set(fn.name for fn in dirname.glob("*"))
@ -323,11 +321,7 @@ def run_test(
if no_unit_tests: if no_unit_tests:
break break
try: errors = run_unit_tests(testdir, history_fname)
errors = run_unit_tests(testdir, history_fname)
except subprocess.TimeoutExpired:
errors = f"Tests in {testdir} timed out!"
timeout = True
if errors: if errors:
test_outcomes.append(False) test_outcomes.append(False)
@ -369,44 +363,24 @@ def run_test(
def run_unit_tests(testdir, history_fname): def run_unit_tests(testdir, history_fname):
test_files = [file for file in testdir.glob("*") if file.name.endswith("_test.py")] suite = unittest.defaultTestLoader.discover(
assert len(test_files) str(testdir),
pattern="*_test.py",
top_level_dir=str(testdir),
)
timeout = 60 stream = io.StringIO()
for test_file in test_files: runner = unittest.TextTestRunner(stream=stream)
dump(test_file) result = runner.run(suite)
command = [ res = stream.getvalue()
"docker",
"run",
"-it",
"--rm",
"--interactive=false",
"-v",
f"{test_file.parent.absolute()}:/app",
"python:3.8-slim",
"bash",
"-c",
f"cd /app && python -m unittest {test_file.name}",
]
print(" ".join(command))
result = subprocess.run( with history_fname.open("a") as fh:
command, fh.write(f"```\n{res}\n```")
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
timeout=timeout,
)
res = cleanup_test_output(result.stdout) if not result.wasSuccessful():
print(f"Tests in {testdir} failed")
with history_fname.open("a") as fh: return res
fh.write(f"```\n{res}\n```")
if result.returncode != 0:
print(f"Test {test_file} failed")
return res
def cleanup_test_output(output): def cleanup_test_output(output):
@ -432,21 +406,5 @@ def cleanup_test_output(output):
return res return res
def check_docker():
command = [
"docker",
"run",
"-it",
"--rm",
"--interactive=false",
"python:3.8-slim",
"/bin/true",
]
result = subprocess.run(command, stdout=subprocess.PIPE, text=True)
if result.returncode:
print("Can't run: " + " ".join(command))
sys.exit(-1)
if __name__ == "__main__": if __name__ == "__main__":
app() app()