add: python tester

.gitignore

@@ -1,2 +1,4 @@
 build
 obj
+log
+*.pyc

Makefile

@@ -29,5 +29,10 @@ re:
 	$(MAKE) fclean 
 	$(MAKE) all
 
-.PHONY: all clean fclean re
+tests:
+	$(MAKE) all
+	mkdir -p log
+	python3 test/test.py build/$(NAME)
+
+.PHONY: all clean fclean re tests
 -include $(DEP)

src/main.c

@@ -90,6 +90,8 @@ static int check_reply(struct sockaddr_in const *me, struct sockaddr_in const *s
 
 int main(int ac, char **av)
 {
+    signal(SIGINT, signal_handler);
+
     (void) ac;
     struct statistics stats;
     struct setting settings;
@@ -104,7 +106,6 @@ int main(int ac, char **av)
     size_t packet_size = sizeof(struct icmphdr) + settings.payload_size;
     size_t recv_packet_size = packet_size + sizeof(struct iphdr);
 
-    signal(SIGINT, signal_handler);
 
     if (dns_lookup(&settings.dest))
         goto error0;

test/exectution.py

@@ -0,0 +1,4 @@
+
+
+def test_excution(excutable: str, tests_category: dict[str, list[dict[str, str]]], log_path: str):
+    pass

test/parsing.py

@@ -0,0 +1,24 @@
+import subprocess
+import os
+import signal
+import time
+
+import thread_manager
+from print import print_test
+
+def _test(title: str, excutable: str, args: str, exit_code: int, log_path: str) -> None:
+    if not os.path.exists(log_path):
+        os.makedirs(log_path)
+    f = open(f"{log_path}/{title}.log", "w")
+    p = subprocess.Popen(f"{excutable} {args}".split(" "), stdout=f, stderr=f)
+    time.sleep(0.05)
+    p.send_signal(signal.SIGINT)
+    value = p.wait()
+    f.close()
+    print_test(exit_code, value, title, args)
+
+def test_parsing(excutable: str, tests_category: dict[str, list[dict[str, str]]], log_path: str):
+    for category_name, tests in tests_category.items():
+        print(category_name, end="\n" * 2)
+        for test in tests:
+            thread_manager.add_to_queu(_test, (test["title"], excutable, test['args'], test['exit'], f"{log_path}/{category_name}"))

test/print.py

@@ -0,0 +1,8 @@
+
+
+def print_test(expected_value, value, title: str, desc: str) -> None:
+    print(title, end="\t")
+    if expected_value == value:
+        print("[OK]")
+        return
+    print(f"[ERR]: {desc} ({expected_value} != {value})")

test/test.py

@@ -0,0 +1,24 @@
+from collections.abc import Callable
+import json
+import sys
+import os
+import shutil
+
+import thread_manager
+import parsing
+import exectution
+
+excutable: str = sys.argv[1]
+
+with open('test/tests.json', 'r') as file:
+    tests: dict[str, dict] = json.load(file)
+
+tester: dict[str, Callable[[str, dict], None]] = {"parsing": parsing.test_parsing, "excution": exectution.test_excution} 
+
+shutil.rmtree("log")
+
+for category_name, tests in tests.items():
+    print(category_name, end="\n" * 3)
+    tester[category_name](excutable, tests, f"log/{category_name}")
+    thread_manager.wait_pool()
+

test/tests.json

@@ -0,0 +1,47 @@
+{
+    "parsing": {
+        "working": [
+            {
+                "title": "normal",
+                "args": "1.1.1.1",
+                "exit": 0
+            },
+            {
+                "title": "dns",
+                "args": "google.com",
+                "exit": 0
+            },
+            {
+                "title": "ttl",
+                "args": "1.1.1.1 --ttl 128",
+                "exit": 0
+            },
+            {
+                "title": "preload",
+                "args": "1.1.1.1 -p 1",
+                "exit": 0
+            },
+            {
+                "title": "payload size",
+                "args": "1.1.1.1 -s 1",
+                "exit": 0
+            }
+        ]
+    },
+    "excution": {
+        "payload_size": [
+            {
+                "title": "normal",
+                "args": "1.1.1.1 -s 56"
+            },
+            {
+                "title": "empty payload",
+                "args": "1.1.1.1 -s 0"
+            },
+            {
+                "title": "too high",
+                "args": "1.1.1.1 -s "
+            }
+        ]
+    }
+}

test/thread_manager.py

@@ -0,0 +1,11 @@
+import concurrent.futures
+
+_pool = concurrent.futures.ThreadPoolExecutor(max_workers=10)
+
+def add_to_queu(func: callable, args: tuple):
+    global _pool
+    _pool.submit(func, *args)
+
+def wait_pool():
+    global _pool
+    _pool.shutdown(wait=True)