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", "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_separate_display_names_migrate_without_losing_accounts(self):
        old_path = Path(self.tempdir.name) / "separate-names.sqlite3"
        current_db_path = server.DB_PATH
        connection = sqlite3.connect(old_path)
        try:
            connection.execute("""CREATE TABLE users (
                user_id INTEGER PRIMARY KEY, login_username TEXT NOT NULL,
                login_key TEXT NOT NULL UNIQUE, display_name TEXT NOT NULL,
                created_at TEXT NOT NULL)""")
            connection.execute("""CREATE TABLE sessions (
                token_hash TEXT PRIMARY KEY, user_id INTEGER NOT NULL REFERENCES users(user_id) ON DELETE CASCADE,
                expires_at TEXT NOT NULL)""")
            connection.execute("""CREATE TABLE preferences (
                user_id INTEGER NOT NULL REFERENCES users(user_id) ON DELETE CASCADE,
                property_id TEXT NOT NULL, preference_json TEXT NOT NULL,
                updated_at TEXT NOT NULL, PRIMARY KEY (user_id, property_id))""")
            connection.execute("INSERT INTO users VALUES (1, 'Aly', 'aly', 'Other Name', 'today')")
            connection.execute("INSERT INTO sessions VALUES ('token', 1, 'later')")
            connection.execute("INSERT INTO preferences VALUES (1, 'role', '[\"Tank\"]', 'today')")
            connection.commit()
            server.DB_PATH = old_path
            server.init_db()
            self.assertEqual(
                [row[1] for row in connection.execute("PRAGMA table_info(users)")],
                ["user_id", "login_username", "login_key", "created_at"],
            )
            self.assertEqual(connection.execute("SELECT login_username FROM users").fetchone()[0], "Aly")
            self.assertEqual(connection.execute("SELECT user_id FROM sessions").fetchone()[0], 1)
            self.assertEqual(connection.execute("SELECT preference_json FROM preferences").fetchone()[0], '["Tank"]')
        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.assertNotIn("displayName", data["user"])
        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"}, 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["loginUsername"] for person in data["people"]], ["Aly", "Bex"])
        self.assertTrue(all("displayName" not in person for person in data["people"]))
        self.assertEqual(len({person["userId"] for person in data["people"]}), 2)

        status, data, _ = self.request("PUT", "/api/me", {"loginUsername": "BEX"}, cookie)
        self.assertEqual(status, 409)
        status, data, _ = self.request("PUT", "/api/me", {"loginUsername": "AlyNew"}, 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.assertNotIn("displayName", data["user"])
        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"])

        for typed_name in ("AlyNew", "alynew"):
            status, data, _ = self.request("POST", "/api/username-suggestions", {"loginUsername": typed_name})
            self.assertEqual(status, 200)
            self.assertNotIn("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", {"confirmLoginUsername": "meridian"}, cookie)
        self.assertEqual(status, 400)
        status, _, _ = self.request("DELETE", "/api/me", {"confirmLoginUsername": "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()