Merge pull request #327 from MartinFillon/316-functionnal-test

316 functionnal test

.gitignore

@@ -21,3 +21,4 @@ gui/doc
 *.so
 unit_tests
 !tests/server/*.so
+__pycache__

tests/functionnal_test/client_test.py

@@ -0,0 +1,49 @@
+##
+## EPITECH PROJECT, 2024
+## Zappy
+## File description:
+## client_test
+##
+
+import subprocess
+from time import sleep
+from config import ConfigTest, team_connection, cli_simple_cmds, connect_nbr_cmds
+from utils import ignore_timeout, init_socket_client
+
+def client_output_test(config: ConfigTest):
+    """
+        Function that launch a server and start a client with command
+        specified in `config`
+    """
+
+    print(config.name)
+    args = config.config.args
+    server = subprocess.Popen(args=args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+    sleep(1)
+    cli = init_socket_client(config.cliConf.port)
+
+    for cmd in config.cliConf.cmds:
+        cli.send(cmd.encode())
+        sleep(1)
+
+    buff = cli.recv(1024)
+    ignore_timeout(server, 2.0)
+    server.kill()
+
+    if buff.decode() == config.cliConf.output:
+        print("\t", config.success)
+    else:
+        print("\t", config.failure)
+    print("Output:\n", buff.decode(), sep="")
+    print("VS:\n", config.cliConf.output, sep="")
+
+
+def client_tests():
+    """"
+        Here we test multiple configurations:
+            - `team_connection`
+            - `cli_simple_cmds`
+    """
+    client_output_test(team_connection)
+    client_output_test(cli_simple_cmds)
+    client_output_test(connect_nbr_cmds)

tests/functionnal_test/config.py

@@ -0,0 +1,219 @@
+##
+## EPITECH PROJECT, 2024
+## Zappy
+## File description:
+## config
+##
+
+class Config:
+    def __init__(self, args: list, timeout=None |float) -> None:
+        self.args = args
+        self.timeout = timeout
+
+class ConfigClient:
+    def __init__(
+        self,
+        port: int,
+        cmd_line: None | list[str],
+        cmds: None | list[str],
+        output: None | str,
+        host: str="127.0.0.1"
+    ) -> None:
+        self.port = port
+        self.host = host
+        self.cmd_line = cmd_line
+        self.cmds = cmds
+        self.output = output
+
+class ConfigTest:
+    def __init__(
+        self,
+        name: str,
+        config: Config,
+        success: str,
+        failure: str,
+        cliConf: ConfigClient,
+    ) -> None:
+        self.name = name
+        self.config = config
+        self.success = success
+        self.failure = failure
+        self.cliConf = cliConf
+
+### CONFIGS ###
+
+timeout_config = ConfigTest(
+    name = "[TEST Timeout] ",
+    config = Config(
+        args=[
+            "./zappy_server",
+            "-p", "4242",
+            "-x", "10",
+            "-y", "10",
+            "-n", "team1", "team2", "team3",
+            "-c", "4",
+            "-f", "100",
+            "-l", "DEBUG"
+        ],
+        timeout=2
+    ),
+    success = "Test has timeout: OK",
+    failure = "Test has not timeout: KO",
+    cliConf=ConfigClient(
+        port=4242,
+        cmd_line= None,
+        cmds= None,
+        output= None,
+    )
+)
+
+
+simple_connection = ConfigTest(
+    name = "[TEST Client connect] ",
+    config = Config(
+        args=[
+            "./zappy_server",
+            "-p", "4242",
+            "-x", "10",
+            "-y", "10",
+            "-n", "team1", "team2", "team3",
+            "-c", "4",
+            "-f", "100",
+            "-l", "DEBUG"
+        ]
+    ),
+    success = "Test connection handled: OK",
+    failure = "Test client not handled: KO",
+
+    cliConf=ConfigClient(
+        port=4242,
+        cmd_line= None,
+        cmds= None,
+        output= "WELCOME\n",
+    )
+)
+
+team_connection = ConfigTest(
+    name = "[TEST Team connect] ",
+    config = Config(
+        args=[
+            "./zappy_server",
+            "-p", "8080",
+            "-x", "10",
+            "-y", "10",
+            "-n", "team1", "team2", "team3",
+            "-c", "4",
+            "-f", "100",
+            "-l", "DEBUG"
+        ]
+    ),
+    success = "Test Team handled: OK",
+    failure = "Test Team not handled: KO",
+
+    cliConf=ConfigClient(
+        port=8080,
+        cmd_line= None,
+        cmds= ["team1\n"],
+        output= "WELCOME\n3\n10 10\n",
+    )
+)
+
+cli_simple_cmds = ConfigTest(
+    name = "[TEST Simple commands] ",
+    config = Config(
+        args=[
+            "./zappy_server",
+            "-p", "8080",
+            "-x", "10",
+            "-y", "10",
+            "-n", "team1", "team2", "team3",
+            "-c", "4",
+            "-f", "100",
+            "-l", "DEBUG"
+        ]
+    ),
+    success = "Test Simple commands handled: OK",
+    failure = "Test Simple commands not handled: KO",
+
+    cliConf=ConfigClient(
+        port=8080,
+        cmd_line= ["echo", "-ne", "team1\nForward\nRight\nLeft"],
+        cmds= ["team1\n", "Forward\n", "Left\n", "Right\n"],
+        output= "WELCOME\n3\n10 10\nok\nok\nok\n"
+    )
+)
+
+connect_nbr_cmds = ConfigTest(
+    name = "[TEST Connect_nbr] ",
+    config = Config(
+        args=[
+            "./zappy_server",
+            "-p", "8080",
+            "-x", "10",
+            "-y", "10",
+            "-n", "team1", "team2", "team3",
+            "-c", "4",
+            "-f", "100",
+            "-l", "DEBUG"
+        ]
+    ),
+    success = "Test Connect_nbr handled: OK",
+    failure = "Test Connect_nbr not handled: KO",
+
+    cliConf=ConfigClient(
+        port=8080,
+        cmd_line= None,
+        cmds= ["team1\n", "Connect_nbr\n"],
+        output= "WELCOME\n3\n10 10\n3\n"
+    )
+)
+
+broadcast_cmds = ConfigTest(
+    name = "[TEST Broadcast] ",
+    config = Config(
+        args=[
+            "./zappy_server",
+            "-p", "4242",
+            "-x", "10",
+            "-y", "10",
+            "-n", "team1", "team2", "team3",
+            "-c", "4",
+            "-f", "100",
+            "-l", "DEBUG"
+        ]
+    ),
+    success = "Test Broadcast handled: OK",
+    failure = "Test Broadcast not handled: KO",
+
+    cliConf=ConfigClient(
+        port=4242,
+        cmd_line= None,
+        cmds= ["team1\n", "Broadcast Hello\n"],
+        output= "WELCOME\n2\n10 10\nok\n"
+    )
+)
+
+hard_test = ConfigTest(
+    name = "[TEST Hard input] ",
+    config = Config(
+        args=[
+            "./zappy_server",
+            "-p", "8080",
+            "-x", "10",
+            "-y", "10",
+            "-n", "team1", "team2", "team3",
+            "-c", "4",
+            "-f", "100",
+            "-l", "DEBUG"
+        ]
+    ),
+    success = "Test Hard input handled: OK",
+    failure = "Test Hard input handled: KO",
+
+    cliConf=ConfigClient(
+        port=8080,
+        cmd_line= ["cat", "/dev/urandom"],
+        cmds= None,
+        output= "WELCOME\n"
+    )
+)

tests/functionnal_test/hard_test.py

@@ -0,0 +1,69 @@
+##
+## EPITECH PROJECT, 2024
+## Zappy
+## File description:
+## hard_test
+##
+
+import subprocess
+from time import sleep
+from config import hard_test, broadcast_cmds
+from utils import ignore_timeout, init_socket_client
+
+def hard_test_urandom():
+    """
+        Hard test `cat /dev/urandom | nc localhost 8080`,
+        to test server integrity with huge inputs
+        NOTE: To test this command, the easiest way is to use subprocess
+        (other solution is socket + subprocess)
+    """
+    print(hard_test.name)
+    args = hard_test.config.args
+    server = subprocess.Popen(args=args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+    urandom = subprocess.Popen(args=hard_test.cliConf.cmd_line, stdout=subprocess.PIPE)
+    cli = subprocess.Popen(
+        args=["nc", "localhost", "8080"],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+        stdin=urandom.stdout
+    )
+
+    try:
+        server.wait(4.0)
+        cli.wait(4.0)
+    except subprocess.TimeoutExpired:
+        print("\t", hard_test.success, server.returncode)
+    except subprocess.SubprocessError:
+        print("\t SUBPROCESS ERROR: ", hard_test.failure)
+
+def broadcast_test():
+    """
+        Test of the broadcast by connecting 2 clients
+    """
+    print(broadcast_cmds.name)
+    args = broadcast_cmds.config.args
+    server = subprocess.Popen(args=args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+
+    sender = init_socket_client(broadcast_cmds.cliConf.port)
+    receiver = init_socket_client(broadcast_cmds.cliConf.port)
+
+    receiver.send(b"team1\n")
+    for cmd in broadcast_cmds.cliConf.cmds:
+        sender.send(cmd.encode())
+        sleep(1)
+
+    data_receiver = receiver.recv(1024)
+    data_sender = sender.recv(1024)
+    ignore_timeout(server, 2.0)
+    server.kill()
+    if data_receiver.decode().find("Hello") == -1:
+        print(broadcast_cmds.failure, "Broadcast not received")
+        return
+    if data_sender.decode() == broadcast_cmds.cliConf.output:
+        print("\t", broadcast_cmds.success)
+    else:
+        print("\t", broadcast_cmds.failure)
+    print("Output:\n", data_sender.decode(), sep="")
+    print("VS:\n", broadcast_cmds.cliConf.output, sep="")
+    sender.close()
+    receiver.close()

tests/functionnal_test/simple_test.py

@@ -0,0 +1,39 @@
+##
+## EPITECH PROJECT, 2024
+## Zappy
+## File description:
+## timeout_test
+##
+
+import subprocess
+from time import sleep
+from config import timeout_config, simple_connection
+from utils import ignore_timeout, init_socket_client
+
+def test_timeout():
+    """" Here we test the timeout of the server after `timeout` seconds """
+
+    print(timeout_config.name)
+    try:
+        timeout = timeout_config.config.timeout
+        args = timeout_config.config.args
+        subprocess.run(args=args, timeout=timeout, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+        print("\t", timeout_config.failure)
+    except subprocess.TimeoutExpired:
+        print("\t", timeout_config.success)
+
+def test_simple_connection():
+    """" Here we test a simple connection with the server """
+
+    print(simple_connection.name)
+    args = simple_connection.config.args
+    server = subprocess.Popen(args=args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+    sleep(1)
+    cli = init_socket_client(simple_connection.cliConf.port)
+    ignore_timeout(server, 2.0)
+
+    data = cli.recv(1024)
+    if data.decode() == simple_connection.cliConf.output:
+        print("\t", simple_connection.success)
+    else:
+        print("\t", simple_connection.failure)