279 lines
11 KiB
Python
279 lines
11 KiB
Python
import unittest
|
|
import threading
|
|
import subprocess
|
|
import time
|
|
import os
|
|
import signal
|
|
import requests
|
|
import unittest
|
|
|
|
server_process = None
|
|
API_URL = "http://localhost:7000/api"
|
|
DB_PATH = "./data/TEST.sqlite"
|
|
|
|
users = {
|
|
"admin": "admin",
|
|
"max": "12345"
|
|
}
|
|
|
|
def setUpModule():
|
|
"""Start the API server in a separate process before running tests"""
|
|
def run_server():
|
|
global server_process
|
|
# Needet to get python3 dir
|
|
env = os.environ.copy()
|
|
env.update({
|
|
"DATABASE_URL": f"sqlite:///{DB_PATH}",
|
|
"CREATE_OPTIONAL_DEFAULT_USER": "true"
|
|
})
|
|
|
|
server_process = subprocess.Popen(
|
|
["python3", "-m", "simple_chat_api"],
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
env=env
|
|
)
|
|
|
|
# Start server in a separate thread
|
|
server_thread = threading.Thread(target=run_server)
|
|
server_thread.daemon = True
|
|
server_thread.start()
|
|
|
|
# Wait for server to start
|
|
time.sleep(2)
|
|
|
|
def tearDownModule():
|
|
global server_process
|
|
"""Kill the API server after all tests have executed"""
|
|
if server_process:
|
|
server_process.send_signal(signal.SIGTERM)
|
|
server_process.wait()
|
|
os.remove(DB_PATH)
|
|
|
|
def build_session(user: str, password: str) -> requests.Session:
|
|
session = requests.Session()
|
|
response = session.post(f"{API_URL}/user/token", json={
|
|
"user": user,
|
|
"password": password
|
|
})
|
|
if response.status_code != 200:
|
|
raise ValueError(f"Failed to get token for user {user}; {response.text}")
|
|
return [session, response.json()]
|
|
|
|
|
|
class TestServer(unittest.TestCase):
|
|
def test_online(self):
|
|
"""Test if the server is running"""
|
|
response = requests.get(API_URL)
|
|
self.assertEqual(response.status_code, 404, "Server is not running or not reachable")
|
|
|
|
|
|
class TestAuthEndpoints(unittest.TestCase):
|
|
userSessions = {}
|
|
def __init__(self, methodName = "runTest"):
|
|
super().__init__(methodName)
|
|
|
|
def test_1_get_token(self):
|
|
"""Test the /token endpoint"""
|
|
for user,password in users.items():
|
|
with self.subTest(user=user):
|
|
try:
|
|
session, data = build_session(user, password)
|
|
self.userSessions[user] = session
|
|
except ValueError as e:
|
|
self.fail(str(e))
|
|
excepted = {
|
|
"sub": {
|
|
"user": user,
|
|
"role": "admin" if user == "admin" else "user"
|
|
},
|
|
}
|
|
self.assertEqual(data["sub"], excepted["sub"], f"Token content mismatch for user {user}")
|
|
|
|
def test_2_get_user(self):
|
|
if self.userSessions == {}:
|
|
self.skipTest("No user sessions available. Run test_get_token first.")
|
|
for user,session in self.userSessions.items():
|
|
with self.subTest(user=user):
|
|
response = session.get(f"{API_URL}/user")
|
|
self.assertEqual(response.status_code, 200, f"Failed to get user info for {user}; {response.text}")
|
|
data = response.json()
|
|
excepted = {
|
|
"name": user,
|
|
}
|
|
self.assertEqual(data, excepted, f"User info mismatch for {user}")
|
|
|
|
def test_3_get_all(self):
|
|
if self.userSessions == {}:
|
|
self.skipTest("No user sessions available. Run test_get_token first.")
|
|
|
|
for user,session in self.userSessions.items():
|
|
with self.subTest(user=user):
|
|
response = session.get(f"{API_URL}/user/getAll")
|
|
if user == "admin":
|
|
self.assertEqual(response.status_code, 200, f"Failed to get all users for admin; {response.text}")
|
|
data = response.json()
|
|
excepted = [
|
|
{"name": "admin", "role": "admin"},
|
|
{"name": "max", "role": "user"}
|
|
]
|
|
self.assertTrue(all(item in data for item in excepted), f"User list mismatch for admin")
|
|
else:
|
|
self.assertEqual(response.status_code, 401, f"Non-admin user {user} should not access /getAll; {response.text}")
|
|
|
|
def test_4_create_user(self):
|
|
if self.userSessions == {}:
|
|
self.skipTest("No user sessions available. Run test_get_token first.")
|
|
|
|
new_users = [{
|
|
"new_user": "newuser",
|
|
"new_password": "newpass",
|
|
"new_admin": False
|
|
},
|
|
{
|
|
"new_user": "admin2",
|
|
"new_password": "adminpass",
|
|
"new_admin": True
|
|
},
|
|
{
|
|
"new_user": "invalid/user",
|
|
"new_password": "pass",
|
|
"new_admin": False
|
|
}
|
|
]
|
|
|
|
for user,session in self.userSessions.items():
|
|
with self.subTest(user=user):
|
|
if user == "admin":
|
|
for new_user in new_users:
|
|
if "invalid" in new_user["new_user"]:
|
|
response = session.post(f"{API_URL}/user/add", json=new_user)
|
|
self.assertEqual(response.status_code, 400, f"Creating user with invalid name should fail; {response.text}")
|
|
continue
|
|
response = session.post(f"{API_URL}/user/add", json=new_user)
|
|
self.assertEqual(response.status_code, 200, f"Admin failed to create new user; {response.text}")
|
|
data = response.json()
|
|
self.assertIn("created successfully", data["message"], f"Created user info mismatch")
|
|
|
|
response = session.post(f"{API_URL}/user/add", json=new_user)
|
|
self.assertEqual(response.status_code, 400, f"Creating duplicate user should fail; {response.text}")
|
|
else:
|
|
self.assertEqual(response.status_code, 400, f"Non-admin user {user} should not create users; {response.text}")
|
|
|
|
def test_5_delete_user(self):
|
|
if self.userSessions == {}:
|
|
self.skipTest("No user sessions available. Run test_get_token first.")
|
|
|
|
for user,session in self.userSessions.items():
|
|
with self.subTest(user=user):
|
|
if user == "admin":
|
|
response = session.post(f"{API_URL}/user/delete/kim")
|
|
self.assertEqual(response.status_code, 200, f"Admin failed to delete user kim; {response.text}")
|
|
data = response.json()
|
|
self.assertIn("deleted successfully", data["message"], f"Deleted user info mismatch")
|
|
|
|
response = session.post(f"{API_URL}/user/delete/nonexistent")
|
|
self.assertEqual(response.status_code, 400, f"Deleting non-existent user should fail; {response.text}")
|
|
else:
|
|
response = session.post(f"{API_URL}/user/delete/max")
|
|
self.assertEqual(response.status_code, 401, f"Non-admin user {user} should not delete users; {response.text}")
|
|
|
|
def test_6_change_password(self):
|
|
if self.userSessions == {}:
|
|
self.skipTest("No user sessions available. Run test_get_token first.")
|
|
|
|
for user,session in self.userSessions.items():
|
|
with self.subTest(user=user):
|
|
pyload = {
|
|
"new_password": "newadminpass"
|
|
}
|
|
response = session.post(f"{API_URL}/user/changePassword/", json=pyload)
|
|
self.assertEqual(response.status_code, 200, f"Admin failed to change password; {response.text}")
|
|
|
|
|
|
class TestMessage(unittest.TestCase):
|
|
last_msg = None
|
|
userSessions = {}
|
|
messages = [
|
|
{"content": "Admin here, managing things."},
|
|
{"content": "Hello, this is max!"},
|
|
]
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
for user, password in users.items():
|
|
cls.userSessions[user], _ = build_session(user, password)
|
|
|
|
def test_1_post_message(self):
|
|
if not self.userSessions:
|
|
self.skipTest("No user sessions available. Run test_get_token first.")
|
|
|
|
|
|
|
|
for i, (user, session) in enumerate(self.userSessions.items()):
|
|
with self.subTest(user=user):
|
|
response = session.post(f"{API_URL}/messages/general", json=self.messages[i])
|
|
self.assertEqual(
|
|
response.status_code,
|
|
200,
|
|
f"User {user} failed to post message; {response.text}"
|
|
)
|
|
|
|
clear_json = response.json()
|
|
|
|
msg_id, data = list(clear_json.items())[-1]
|
|
|
|
TestMessage.last_msg = {
|
|
"message_id": msg_id,
|
|
"timestamp": data["timestamp"],
|
|
}
|
|
|
|
self.assertIn(
|
|
self.messages[i]["content"],
|
|
data["content"],
|
|
f"Posted message response missing content for user {user}"
|
|
)
|
|
time.sleep(1) # Ensure different timestamps
|
|
|
|
def test_2_get_messages_since(self):
|
|
if not self.userSessions or self.last_msg is None:
|
|
self.skipTest("No user sessions available or no messages posted. Run previous tests first.")
|
|
|
|
for user, session in self.userSessions.items():
|
|
with self.subTest(user=user):
|
|
response = session.get(f"{API_URL}/messages/general?since={self.last_msg['timestamp']}")
|
|
self.assertEqual(
|
|
response.status_code,
|
|
200,
|
|
f"User {user} failed to get messages since timestamp; {response.text}"
|
|
)
|
|
|
|
first_key = list(response.json().keys())[0]
|
|
|
|
self.assertEqual(
|
|
first_key,
|
|
self.last_msg["message_id"], # match full message dict
|
|
f"User {user} did not receive the expected message"
|
|
)
|
|
|
|
def test_3_get_all_messages(self):
|
|
if not self.userSessions or self.last_msg is None:
|
|
self.skipTest("No user sessions available or no messages posted. Run previous tests first.")
|
|
|
|
for user, session in self.userSessions.items():
|
|
with self.subTest(user=user):
|
|
response = session.get(f"{API_URL}/messages/general")
|
|
self.assertEqual(
|
|
response.status_code,
|
|
200,
|
|
f"User {user} failed to get all messages; {response.text}"
|
|
)
|
|
|
|
messages_data = response.json()
|
|
message_contents = [msg["content"] for _, msg in messages_data.items()]
|
|
|
|
expected_contents = [msg["content"] for msg in self.messages]
|
|
for content in expected_contents:
|
|
self.assertIn(content, message_contents, f"Message with content '{content}' not found")
|
|
|
|
self.assertEqual(len(messages_data), len(self.messages), "Number of messages doesn't match expected")
|
|
|