feat: add tests on /user and /info

src/client/event_handling/reply_single_b.c

@@ -28,7 +28,7 @@ int reply_s_user(connection_t *connect)
         client_print_subscribed(user_uuid, team_uuid);
     if (connect->last_cmd == unsubscribe)
         client_print_unsubscribed(user_uuid, team_uuid);
-    if (connect->last_cmd == info_cmd)
+    if (connect->last_cmd == info_cmd || connect->last_cmd == other)
         client_print_user(user_uuid, info.name, info.status);
     return SUCCESS;
 }

src/server/cmds/cmd_info.c

@@ -55,7 +55,6 @@ static void info_user(server_t *server, client_t *client)
     user_t *user = get_user_by_uuid(server, client->user);
     team_t *team = get_team_by_uuid(server, client->team);
 
-    printf("send user info\n");
     dprintf(client->fd, "215");
     fsync(client->fd);
     write(client->fd, user_to_info(user, &info, team), sizeof info);

src/server/cmds/cmd_user.c

@@ -35,8 +35,10 @@ void cmd_user(server_t *server, client_t *client)
     if (!is_logged_in(client))
         return;
     arg = get_quoted_arg(client->buffer, 0, NULL);
-    if (arg == NULL)
-        arg = "";
+    if (arg == NULL) {
+        dprintf(client->fd, "502 Syntax error.\n");
+        return;
+    }
     uuid_parse(arg, uuid);
     user = get_user(server, client, uuid);
     if (user == NULL)

src/server/cmds/cmds_utils.c

@@ -65,7 +65,7 @@ bool is_logged_in(client_t *client)
     if (client == NULL)
         return false;
     if (uuid_is_null(client->user)) {
-        dprintf(client->fd, "520\n");
+        dprintf(client->fd, "520");
         return false;
     }
     return true;

tests/server.py

@@ -6,61 +6,70 @@
 
 OUTPUT = list[list[str]]
 
-
-def run_teams(cmds: list[str]) -> tuple[OUTPUT, OUTPUT]:
-    port = find_available_port()
-    print(f"Using port: {port}")
-    path = pathlib.Path(__file__).parent.resolve()
-    so_path = path / "mock"
-
-    # Check if mock libary exists
-    if not (so_path / "libmyteams.so").exists():
-        subprocess.run(["make", "-C", so_path])
-    if not (so_path / "libmyteams.so").exists():
-        raise FileNotFoundError("libmyteams.so")
-
-    envp = os.environ.copy()
-    envp.clear()
-    envp["LD_LIBRARY_PATH"] = str(so_path)
-
-    server_proc = subprocess.Popen(
-        ["./myteams_server", str(port)],
-        env=envp,
-        stdout=subprocess.PIPE,
-        stderr=subprocess.PIPE,
-    )
-
-    client_proc = subprocess.Popen(
-        ["./myteams_cli", "0.0.0.0", str(port)],
-        env=envp,
-        stdin=subprocess.PIPE,
-        stdout=subprocess.PIPE,
-        stderr=subprocess.PIPE,
-    )
-    if client_proc.stdin is None:
-        raise Exception("Failed to open stdin")
-
-    for command in cmds:
-        print(f"Sending command: {command}")
-        client_proc.stdin.write(command.encode())
-        client_proc.stdin.flush()
-
-    # Wait for the server to reply all the events
-    time.sleep(0.5)
-
-    client_proc.stdin.close()
-    if client_proc.wait() != 0:
-        print(f"{client_proc=}")
-        raise Exception("Client failed")
-
-    server_proc.terminate()
-
-    out_server = server_proc.stderr
-    out_client = client_proc.stderr
-    if out_server is None or out_client is None:
-        raise Exception("Server or client failed")
-
-    return decode(out_server.read()), decode(out_client.read())
+class Team:
+    def __init__(self):
+        port = find_available_port()
+        print(f"Using port: {port}")
+        path = pathlib.Path(__file__).parent.resolve()
+        so_path = path / "mock"
+
+        # Check if mock libary exists
+        if not (so_path / "libmyteams.so").exists():
+            subprocess.run(["make", "-C", so_path])
+        if not (so_path / "libmyteams.so").exists():
+            raise FileNotFoundError("libmyteams.so")
+
+        envp = os.environ.copy()
+        envp.clear()
+        envp["LD_LIBRARY_PATH"] = str(so_path)
+
+        self.server_proc = subprocess.Popen(
+            ["./myteams_server", str(port)],
+            env=envp,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+        )
+
+        self.client_proc = subprocess.Popen(
+            ["./myteams_cli", "0.0.0.0", str(port)],
+            env=envp,
+            stdin=subprocess.PIPE,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+        )
+        if self.client_proc.stdin is None:
+            raise Exception("Failed to open stdin")
+        os.set_blocking(self.server_proc.stderr.fileno(), False)
+        os.set_blocking(self.client_proc.stderr.fileno(), False)
+
+    def __del__(self):
+        self.server_proc.terminate()
+        self.client_proc.stdin.close()
+        if self.client_proc.wait() != 0:
+            print(f"{self.client_proc=}")
+            raise Exception("Client failed")
+
+    def run(self, cmds: list[str]) -> tuple[OUTPUT, OUTPUT, OUTPUT]:
+        for command in cmds:
+            if (command[-1] != '\n'):
+                command += '\n'
+            print(f"Sending command: {command}")
+            self.client_proc.stdin.write(command.encode())
+            self.client_proc.stdin.flush()
+
+        # Wait for the server to reply all the events
+        time.sleep(0.5)
+
+        out_server = self.server_proc.stderr
+        out_client = self.client_proc.stderr
+        if out_server is None or out_client is None:
+            raise Exception("Server or client failed")
+
+        if (read_server := out_server.read()):
+            read_server = decode(read_server)
+        if (read_client := out_client.read()):
+            read_client = decode(read_client)
+        return read_server, read_client
 
 
 def find_available_port():

tests/test_info.py

@@ -0,0 +1,27 @@
+from server import Team
+from pprint import pprint
+
+
+def test_info_user():
+    team = Team()
+    name = "toto"
+    serv, cli = team.run([f'/login "{name}"\n'])
+    uuid = serv[0][1]
+    serv, cli = team.run(['/info'])
+    assert serv == None
+    assert cli[0][0] == 'client_print_user'
+    assert cli[0][1] == uuid
+    assert cli[0][2] == 'toto'
+    assert cli[0][3] == '1'
+
+def test_info_errors():
+    team = Team()
+    name = "toto"
+    serv, cli = team.run([f'/login "{name}"\n', "/logout"])
+    uuid = serv[0][1]
+    serv, cli = team.run(['/info'])
+    assert cli[1][0] == 'client_error_unauthorized'
+
+if __name__ == "__main__":
+    # test_info_user()
+    test_info_errors()

tests/test_login.py

@@ -1,10 +1,11 @@
-from server import run_teams
+from server import Team
 from pprint import pprint
 
 
 def test_login():
+    team = Team()
     name = "bob"
-    serv, cli = run_teams([
+    serv, cli = team.run([
         f'/login "{name}"\n',
     ])
     pprint(serv)
@@ -20,10 +21,10 @@ def test_login():
 
 
 def test_logout():
+    team = Team()
     name = "bob"
-    serv, cli = run_teams([
+    serv, cli = team.run([
         f'/login "{name}"\n',
-        '/logout\n',
     ])
     pprint(serv)
     pprint(cli)
@@ -40,15 +41,19 @@ def test_logout():
     assert cli[0][1] == uuid
     assert cli[0][2] == name
 
+    serv, cli = team.run([
+        '/logout\n',
+    ])
     # logout
 
-    assert serv[2][0] == "server_event_user_logged_out"
-    assert serv[2][1] == uuid
+    assert serv[0][0] == "server_event_user_logged_out"
+    assert serv[0][1] == uuid
 
-    assert cli[1][0] == "client_event_logged_out"
-    assert cli[1][1] == uuid
-    assert cli[1][2] == name
+    assert cli[0][0] == "client_event_logged_out"
+    assert cli[0][1] == uuid
+    assert cli[0][2] == name
 
 
 if __name__ == "__main__":
-    test_logout()
+    test_login()
+    # test_logout()

tests/test_user.py

@@ -0,0 +1,109 @@
+from server import Team
+from pprint import pprint
+
+
+def test_user_single():
+    team = Team()
+    name = "toto"
+    serv, cli = team.run([
+        f'/login "{name}"\n',
+    ])
+    pprint(serv)
+    pprint(cli)
+
+    #login
+    assert serv[0][0] == "server_event_user_created"
+    uuid = serv[0][1]
+    assert serv[0][2] == name
+
+    assert cli[0][0] == "client_event_logged_in"
+    assert cli[0][1] == uuid
+    assert cli[0][2] == name
+
+    #user
+    serv, cli = team.run([
+        f'/user "{uuid}"\n',
+    ])
+    assert serv == None
+    assert cli[0][0] == "client_print_user"
+    assert cli[0][1] == uuid
+    assert cli[0][2] == name
+    assert cli[0][3] == '1'
+    pprint(serv)
+    pprint(cli)
+
+def test_user_not_logged():
+    team = Team()
+    name = "toto"
+    serv, cli = team.run([
+        f'/login "{name}"\n',
+        f'/logout\n'
+    ])
+    pprint(serv)
+    pprint(cli)
+
+    #login
+    assert serv[0][0] == "server_event_user_created"
+    uuid = serv[0][1]
+    assert serv[0][2] == name
+
+    assert cli[0][0] == "client_event_logged_in"
+    assert cli[0][1] == uuid
+    assert cli[0][2] == name
+
+    #user
+    serv, cli = team.run([
+        f'/user "{uuid}"\n',
+    ])
+    pprint(serv)
+    pprint(cli)
+
+    assert serv[0][0] == "server_event_user_logged_out"
+    assert serv[0][1] == uuid
+
+    assert cli[0][0] == "client_event_logged_out"
+    assert cli[0][1] == uuid
+    assert cli[0][2] == name
+
+    assert cli[1][0] == "client_error_unauthorized"
+
+def test_user_wrong():
+    team = Team()
+    name = "toto"
+    serv, cli = team.run([
+        f'/login "{name}"\n',
+    ])
+    pprint(serv)
+    pprint(cli)
+
+    #login
+    assert serv[0][0] == "server_event_user_created"
+    uuid = serv[0][1]
+    assert serv[0][2] == name
+
+    assert cli[0][0] == "client_event_logged_in"
+    assert cli[0][1] == uuid
+    assert cli[0][2] == name
+
+    #user
+    serv, cli = team.run([
+        f'/user "{"wrong uuid"}"\n',
+    ])
+    pprint(serv)
+    pprint(cli)
+    assert serv == None
+    assert cli[0][0] == "client_error_unknown_user"
+    assert cli[0][1] == "00000000-0000-0000-0000-000000000000"
+
+    serv, cli = team.run([
+        f'/user\n',
+    ])
+    pprint(serv)
+    pprint(cli)
+    assert serv == None
+    assert cli == None
+
+if __name__ == "__main__":
+    # test_user_single()
+    # test_user_not_logged()
+    test_user_wrong()