243 lines
12 KiB
Python
243 lines
12 KiB
Python
import http.client
|
|
import json
|
|
import sqlite3
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
from pathlib import Path
|
|
from http.server import ThreadingHTTPServer
|
|
|
|
import server
|
|
|
|
|
|
class ServerTest(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.tempdir = tempfile.TemporaryDirectory()
|
|
cls.original_db_path = server.DB_PATH
|
|
server.DB_PATH = Path(cls.tempdir.name) / "preferences.sqlite3"
|
|
server.init_db()
|
|
cls.httpd = ThreadingHTTPServer(("127.0.0.1", 0), server.Handler)
|
|
cls.thread = threading.Thread(target=cls.httpd.serve_forever, daemon=True)
|
|
cls.thread.start()
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
cls.httpd.shutdown()
|
|
cls.httpd.server_close()
|
|
cls.thread.join()
|
|
server.DB_PATH = cls.original_db_path
|
|
cls.tempdir.cleanup()
|
|
|
|
def request(self, method, path, body=None, cookie=None):
|
|
connection = http.client.HTTPConnection("127.0.0.1", self.httpd.server_port)
|
|
headers = {}
|
|
if body is not None:
|
|
headers.update({"Content-Type": "application/json", "X-Requested-With": "PreferencesSPA"})
|
|
if cookie:
|
|
headers["Cookie"] = cookie
|
|
connection.request(method, path, body=json.dumps(body) if body is not None else None, headers=headers)
|
|
response = connection.getresponse()
|
|
payload = response.read()
|
|
result = response.status, json.loads(payload) if payload else None, response.getheader("Set-Cookie")
|
|
connection.close()
|
|
return result
|
|
|
|
def test_schema_uses_stable_user_ids(self):
|
|
with server.db() as connection:
|
|
columns = {
|
|
table: [row["name"] for row in connection.execute(f"PRAGMA table_info({table})")]
|
|
for table in ("users", "sessions", "preferences")
|
|
}
|
|
foreign_keys = {
|
|
table: [(row["from"], row["to"]) for row in connection.execute(f"PRAGMA foreign_key_list({table})")]
|
|
for table in ("sessions", "preferences")
|
|
}
|
|
self.assertEqual(columns["users"], ["user_id", "login_username", "login_key", "display_name", "created_at"])
|
|
self.assertEqual(columns["sessions"], ["token_hash", "user_id", "expires_at"])
|
|
self.assertEqual(columns["preferences"], ["user_id", "property_id", "preference_json", "updated_at"])
|
|
self.assertEqual(foreign_keys["sessions"], [("user_id", "user_id")])
|
|
self.assertEqual(foreign_keys["preferences"], [("user_id", "user_id")])
|
|
|
|
def test_old_schema_requires_a_fresh_database(self):
|
|
old_path = Path(self.tempdir.name) / "old.sqlite3"
|
|
current_db_path = server.DB_PATH
|
|
connection = sqlite3.connect(old_path)
|
|
try:
|
|
connection.execute("CREATE TABLE users (username_key TEXT PRIMARY KEY, username TEXT, created_at TEXT)")
|
|
connection.execute("INSERT INTO users VALUES ('aly', 'Aly', 'today')")
|
|
connection.commit()
|
|
server.DB_PATH = old_path
|
|
with self.assertRaisesRegex(RuntimeError, "Wipe the preproduction database"):
|
|
server.init_db()
|
|
self.assertEqual(connection.execute("SELECT username FROM users").fetchone()[0], "Aly")
|
|
finally:
|
|
server.DB_PATH = current_db_path
|
|
connection.close()
|
|
|
|
def test_login_save_and_group_view(self):
|
|
status, data, _ = self.request("GET", "/api/bootstrap")
|
|
self.assertEqual(status, 200)
|
|
self.assertIsNone(data["user"])
|
|
self.assertEqual(len(data["properties"][0]["categories"]), 9)
|
|
|
|
status, _, _ = self.request("POST", "/api/login", {"loginUsername": "Aly", "password": "wrong"})
|
|
self.assertEqual(status, 401)
|
|
|
|
status, data, cookie_header = self.request("POST", "/api/login", {"loginUsername": "Aly", "password": "galnik"})
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(data["user"]["loginUsername"], "Aly")
|
|
self.assertEqual(data["user"]["displayName"], "Aly")
|
|
aly_id = data["user"]["userId"]
|
|
cookie = cookie_header.split(";", 1)[0]
|
|
|
|
bad_ranking = ["Tank", "DPS", "DPS"]
|
|
status, _, _ = self.request("PUT", "/api/preferences/role", {"ranking": bad_ranking}, cookie)
|
|
self.assertEqual(status, 400)
|
|
|
|
ranking = ["Healer", "DPS", "Tank"]
|
|
status, _, _ = self.request("PUT", "/api/preferences/role", {"ranking": ranking}, cookie)
|
|
self.assertEqual(status, 200)
|
|
|
|
class_ranking = list(reversed(server.PROPERTIES[0]["categories"]))
|
|
status, _, _ = self.request("PUT", "/api/preferences/class", {"ranking": class_ranking}, cookie)
|
|
self.assertEqual(status, 200)
|
|
|
|
content_categories = server.PROPERTY_BY_ID["content"]["categories"]
|
|
content_choices = {category: "Yes" for category in content_categories}
|
|
content_choices["Open World PvP"] = "Maybe"
|
|
content_choices["40-Man Raids"] = "No"
|
|
status, _, _ = self.request("PUT", "/api/preferences/content", {"choices": {"Group Questing": "Yes"}}, cookie)
|
|
self.assertEqual(status, 400)
|
|
status, _, _ = self.request("PUT", "/api/preferences/content", {"choices": {**content_choices, "40-Man Raids": "Later"}}, cookie)
|
|
self.assertEqual(status, 400)
|
|
status, _, _ = self.request("PUT", "/api/preferences/content", {"choices": content_choices}, cookie)
|
|
self.assertEqual(status, 200)
|
|
|
|
status, _, _ = self.request("POST", "/api/login", {"loginUsername": "Bex", "password": "galnik"})
|
|
self.assertEqual(status, 200)
|
|
|
|
status, data, _ = self.request("GET", "/api/bootstrap", cookie=cookie)
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(data["preferences"]["role"]["ranking"], ranking)
|
|
self.assertEqual(data["preferences"]["class"]["ranking"], class_ranking)
|
|
self.assertEqual(data["preferences"]["content"]["choices"], content_choices)
|
|
self.assertEqual(len(data["people"]), 2)
|
|
self.assertEqual(data["people"][0]["preferences"]["role"]["ranking"], ranking)
|
|
self.assertEqual(data["people"][0]["preferences"]["content"]["choices"], content_choices)
|
|
self.assertEqual(data["people"][1]["preferences"], {})
|
|
|
|
status, _, second_cookie_header = self.request("POST", "/api/login", {"loginUsername": "aly", "password": "galnik"})
|
|
self.assertEqual(status, 200)
|
|
second_cookie = second_cookie_header.split(";", 1)[0]
|
|
status, data, _ = self.request("PUT", "/api/me", {"loginUsername": "Aly", "displayName": "Bex"}, cookie)
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(data["user"]["userId"], aly_id)
|
|
status, data, _ = self.request("GET", "/api/bootstrap", cookie=cookie)
|
|
self.assertEqual(len(data["people"]), 2)
|
|
self.assertEqual([person["displayName"] for person in data["people"]], ["Bex", "Bex"])
|
|
self.assertEqual(len({person["userId"] for person in data["people"]}), 2)
|
|
|
|
status, data, _ = self.request("PUT", "/api/me", {"loginUsername": "BEX", "displayName": "Bex"}, cookie)
|
|
self.assertEqual(status, 409)
|
|
status, data, _ = self.request("PUT", "/api/me", {"loginUsername": "AlyNew", "displayName": "Bex"}, cookie)
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(data["user"]["userId"], aly_id)
|
|
status, data, _ = self.request("GET", "/api/bootstrap", cookie=second_cookie)
|
|
self.assertEqual(data["user"]["loginUsername"], "AlyNew")
|
|
self.assertEqual(data["user"]["displayName"], "Bex")
|
|
self.assertEqual(data["preferences"]["role"]["ranking"], ranking)
|
|
self.assertEqual(data["preferences"]["content"]["choices"], content_choices)
|
|
status, data, _ = self.request("POST", "/api/login", {"loginUsername": "alynew", "password": "galnik"})
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(data["user"]["userId"], aly_id)
|
|
status, data, _ = self.request("POST", "/api/username-suggestions", {"loginUsername": "AlyNe"})
|
|
self.assertEqual(status, 200)
|
|
self.assertIn("AlyNew", data["suggestions"])
|
|
|
|
status, _, _ = self.request("POST", "/api/logout", {}, cookie)
|
|
self.assertEqual(status, 204)
|
|
status, data, _ = self.request("GET", "/api/bootstrap", cookie=cookie)
|
|
self.assertIsNone(data["user"])
|
|
|
|
def test_partial_and_empty_rankings(self):
|
|
status, _, cookie_header = self.request(
|
|
"POST", "/api/login", {"loginUsername": "Sparse", "password": "galnik"}
|
|
)
|
|
self.assertEqual(status, 200)
|
|
cookie = cookie_header.split(";", 1)[0]
|
|
for ranking in (["Paladin", "Paladin"], ["Paladin", "Unknown"], [7], {"Paladin": 1}):
|
|
status, _, _ = self.request("PUT", "/api/preferences/class", {"ranking": ranking}, cookie)
|
|
self.assertEqual(status, 400)
|
|
|
|
status, data, _ = self.request("PUT", "/api/preferences/class", {"ranking": ["Paladin", "Mage"]}, cookie)
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(data["ranking"], ["Paladin", "Mage"])
|
|
status, data, _ = self.request("PUT", "/api/preferences/role", {"ranking": []}, cookie)
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(data["ranking"], [])
|
|
status, data, _ = self.request("GET", "/api/bootstrap", cookie=cookie)
|
|
self.assertEqual(data["preferences"]["class"]["ranking"], ["Paladin", "Mage"])
|
|
self.assertEqual(data["preferences"]["role"]["ranking"], [])
|
|
|
|
def test_username_suggestions_and_account_deletion(self):
|
|
status, data, _ = self.request(
|
|
"POST", "/api/username-suggestions", {"loginUsername": "Merid"}
|
|
)
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(data["suggestions"], [])
|
|
status, _, _ = self.request(
|
|
"POST", "/api/login", {"loginUsername": "Meridian", "password": "gálnik"}
|
|
)
|
|
self.assertEqual(status, 401)
|
|
|
|
status, data, cookie_header = self.request(
|
|
"POST", "/api/login", {"loginUsername": "Meridian", "password": "galnik"}
|
|
)
|
|
self.assertEqual(status, 200)
|
|
meridian_id = data["user"]["userId"]
|
|
cookie = cookie_header.split(";", 1)[0]
|
|
status, data, _ = self.request(
|
|
"POST", "/api/username-suggestions", {"loginUsername": "Meridain"}
|
|
)
|
|
self.assertEqual(status, 200)
|
|
self.assertIn("Meridian", data["suggestions"])
|
|
|
|
status, data, _ = self.request(
|
|
"POST", "/api/login", {"loginUsername": "Meridain", "password": "galnik"}
|
|
)
|
|
self.assertEqual(status, 409)
|
|
self.assertIn("Meridian", data["suggestions"])
|
|
|
|
status, data, _ = self.request(
|
|
"POST", "/api/login", {"loginUsername": "mERIDIAN", "password": "galnik"}
|
|
)
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(data["user"]["loginUsername"], "Meridian")
|
|
|
|
status, _, _ = self.request(
|
|
"POST", "/api/login", {"loginUsername": "Meridain", "password": "galnik", "createNew": True}
|
|
)
|
|
self.assertEqual(status, 200)
|
|
|
|
status, _, _ = self.request(
|
|
"PUT", "/api/preferences/role", {"ranking": ["DPS", "Tank", "Healer"]}, cookie
|
|
)
|
|
self.assertEqual(status, 200)
|
|
status, _, _ = self.request("DELETE", "/api/me", {"confirmDisplayName": "meridian"}, cookie)
|
|
self.assertEqual(status, 400)
|
|
status, _, _ = self.request("DELETE", "/api/me", {"confirmDisplayName": "Meridian"}, cookie)
|
|
self.assertEqual(status, 204)
|
|
|
|
status, _, _ = self.request("PUT", "/api/preferences/role", {"ranking": ["Tank", "Healer", "DPS"]}, cookie)
|
|
self.assertEqual(status, 401)
|
|
with server.db() as connection:
|
|
self.assertIsNone(connection.execute("SELECT 1 FROM users WHERE login_key = ?", ("meridian",)).fetchone())
|
|
self.assertIsNone(connection.execute("SELECT 1 FROM preferences WHERE user_id = ?", (meridian_id,)).fetchone())
|
|
self.assertIsNone(connection.execute("SELECT 1 FROM sessions WHERE user_id = ?", (meridian_id,)).fetchone())
|
|
self.assertIsNotNone(connection.execute("SELECT 1 FROM users WHERE login_key = ?", ("meridain",)).fetchone())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|