34 lines
1.4 KiB
Python
34 lines
1.4 KiB
Python
import time
|
|
import subprocess
|
|
import os
|
|
import signal
|
|
|
|
import thread_manager
|
|
|
|
def _cmp_header(expected: str, value: str):
|
|
pass
|
|
|
|
def _test(title: str, excutable: str, args: str, exit_code: int, delay: float, log_path: str) -> None:
|
|
log_path = f"{log_path}/{title}"
|
|
if not os.path.exists(log_path):
|
|
os.makedirs(log_path)
|
|
normal_file = open(f"{log_path}/normal.log", "w")
|
|
your_file = open(f"{log_path}/your.log", "w")
|
|
your_program = subprocess.Popen(f"{excutable} {args}".split(" "), stdout=your_file, stderr=your_file)
|
|
normal_program = subprocess.Popen(f"ping {args}".split(" "), stdout=normal_file, stderr=normal_file)
|
|
time.sleep(delay)
|
|
your_program.send_signal(signal.SIGINT)
|
|
normal_program.send_signal(signal.SIGINT)
|
|
your_exit = your_program.wait()
|
|
normal_exit = normal_program.wait()
|
|
your_file.close()
|
|
normal_file.close()
|
|
|
|
def test_excution(excutable: str, tests_category: dict[str, list[dict[str, str]]], log_path: str):
|
|
for category_name, tests in tests_category.items():
|
|
print(category_name)
|
|
for test in tests:
|
|
_test(test["title"], excutable, test['args'], test.get("exit", 0), test.get("delay", 1), f"{log_path}/{category_name}")
|
|
thread_manager.add_to_queu(_test, (test["title"], excutable, test['args'], test.get("exit", 0), test.get("delay", 1), f"{log_path}/{category_name}"))
|
|
thread_manager.wait_pool()
|
|
print("\n" * 1) |