"""
Unit tests for server/oauth_login.py — client-portal social sign-in.
The actual HTTP exchange with Google is monkeypatched (_post_form/_get_json)
so these tests never make a real network call; what's exercised is the
state/CSRF handling, credential CRUD, and the "must match an existing,
active client_portal_user" security rule.

Run with: cd server && ../venv/bin/python3 -m unittest tests.test_oauth_login -v
"""
import os
import sys
import tempfile
import unittest

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))


class OAuthLoginTests(unittest.TestCase):
    def setUp(self):
        self._tmpdir = tempfile.mkdtemp()
        os.environ["CA_DATA_DIR"] = self._tmpdir
        for mod in ("db", "oauth_login"):
            sys.modules.pop(mod, None)
        import db as db_mod
        global db, oauth_login
        db = db_mod
        import oauth_login as oauth_login_mod
        oauth_login = oauth_login_mod
        db.init_db()

    def test_start_raises_not_configured_without_a_key(self):
        with self.assertRaises(oauth_login.OAuthError) as ctx:
            oauth_login.start("google", 1, "http://localhost:8000")
        self.assertEqual(str(ctx.exception), "not_configured")

    def test_start_raises_unknown_provider(self):
        with self.assertRaises(oauth_login.OAuthError) as ctx:
            oauth_login.start("bogus", 1, "http://localhost:8000")
        self.assertEqual(str(ctx.exception), "unknown_provider")

    def test_set_key_then_start_builds_a_real_authorize_url(self):
        oauth_login.set_key(1, "google", "test-client-id", "test-secret")
        url = oauth_login.start("google", 1, "http://localhost:8000")
        self.assertIn("accounts.google.com", url)
        self.assertIn("client_id=test-client-id", url)
        self.assertIn("state=", url)
        self.assertIn("redirect_uri=", url)

    def test_start_records_a_state_row(self):
        oauth_login.set_key(1, "google", "cid", "secret")
        oauth_login.start("google", 1, "http://localhost:8000")
        rows = db.query("SELECT * FROM oauth_state")
        self.assertEqual(len(rows), 1)
        self.assertEqual(rows[0]["provider"], "google")

    def test_complete_rejects_unknown_state(self):
        oauth_login.set_key(1, "google", "cid", "secret")
        with self.assertRaises(oauth_login.OAuthError) as ctx:
            oauth_login.complete("google", "some-code", "not-a-real-state", "http://localhost:8000")
        self.assertEqual(str(ctx.exception), "invalid_state")

    def test_complete_rejects_expired_state(self):
        oauth_login.set_key(1, "google", "cid", "secret")
        db.execute("INSERT INTO oauth_state (state, provider, firm_id, created_at)"
                  " VALUES ('old-state', 'google', 1, datetime('now', '-20 minutes'))")
        with self.assertRaises(oauth_login.OAuthError) as ctx:
            oauth_login.complete("google", "some-code", "old-state", "http://localhost:8000")
        self.assertEqual(str(ctx.exception), "state_expired")

    def test_state_is_single_use(self):
        oauth_login.set_key(1, "google", "cid", "secret")
        url = oauth_login.start("google", 1, "http://localhost:8000")
        state = url.split("state=")[1].split("&")[0]
        oauth_login._post_form = lambda *a, **k: {"access_token": "tok"}
        oauth_login._get_json = lambda *a, **k: {"email": "nobody@example.com"}
        with self.assertRaises(oauth_login.OAuthError):
            oauth_login.complete("google", "code", state, "http://localhost:8000")
        # second attempt with the same state must also fail (already consumed)
        with self.assertRaises(oauth_login.OAuthError) as ctx:
            oauth_login.complete("google", "code", state, "http://localhost:8000")
        self.assertEqual(str(ctx.exception), "invalid_state")

    def test_complete_rejects_email_with_no_matching_portal_account(self):
        oauth_login.set_key(1, "google", "cid", "secret")
        url = oauth_login.start("google", 1, "http://localhost:8000")
        state = url.split("state=")[1].split("&")[0]
        oauth_login._post_form = lambda *a, **k: {"access_token": "tok"}
        oauth_login._get_json = lambda *a, **k: {"email": "stranger@example.com", "name": "Stranger"}
        with self.assertRaises(oauth_login.OAuthError) as ctx:
            oauth_login.complete("google", "code", state, "http://localhost:8000")
        self.assertEqual(str(ctx.exception), "no_portal_account")

    def test_complete_matches_an_existing_active_portal_user_by_email(self):
        cid = db.execute("INSERT INTO client (firm_id, code, name) VALUES (1, 'CL-1', 'Test Client')")
        db.execute("INSERT INTO client_portal_user (client_id, name, email, pwd_hash, pwd_salt, is_active)"
                  " VALUES (?, 'Ram Sharma', 'ram@example.com', 'x', 'y', 1)", (cid,))
        oauth_login.set_key(1, "google", "cid", "secret")
        url = oauth_login.start("google", 1, "http://localhost:8000")
        state = url.split("state=")[1].split("&")[0]
        oauth_login._post_form = lambda *a, **k: {"access_token": "tok"}
        oauth_login._get_json = lambda *a, **k: {"email": "Ram@Example.com", "name": "Ram Sharma"}
        pu = oauth_login.complete("google", "code", state, "http://localhost:8000")
        self.assertEqual(pu["email"], "ram@example.com")

    def test_complete_rejects_inactive_portal_user(self):
        cid = db.execute("INSERT INTO client (firm_id, code, name) VALUES (1, 'CL-1', 'Test Client')")
        db.execute("INSERT INTO client_portal_user (client_id, name, email, pwd_hash, pwd_salt, is_active)"
                  " VALUES (?, 'Ram Sharma', 'ram@example.com', 'x', 'y', 0)", (cid,))
        oauth_login.set_key(1, "google", "cid", "secret")
        url = oauth_login.start("google", 1, "http://localhost:8000")
        state = url.split("state=")[1].split("&")[0]
        oauth_login._post_form = lambda *a, **k: {"access_token": "tok"}
        oauth_login._get_json = lambda *a, **k: {"email": "ram@example.com", "name": "Ram Sharma"}
        with self.assertRaises(oauth_login.OAuthError) as ctx:
            oauth_login.complete("google", "code", state, "http://localhost:8000")
        self.assertEqual(str(ctx.exception), "no_portal_account")

    def test_list_keys_never_exposes_the_secret(self):
        oauth_login.set_key(1, "google", "cid", "top-secret-value")
        rows = oauth_login.list_keys(1)
        self.assertEqual(len(rows), 1)
        self.assertNotIn("client_secret_enc", rows[0])
        import json as json_mod
        self.assertNotIn("top-secret-value", json_mod.dumps(rows[0]))

    def test_set_key_requires_client_id_and_secret(self):
        with self.assertRaises(oauth_login.OAuthError):
            oauth_login.set_key(1, "google", "", "secret")
        with self.assertRaises(oauth_login.OAuthError):
            oauth_login.set_key(1, "google", "cid", "")

    def test_replacing_credentials_keeps_secret_if_blank(self):
        oauth_login.set_key(1, "google", "cid-1", "secret-1")
        oauth_login.set_key(1, "google", "cid-2", "")   # blank secret = keep existing
        row = db.query("SELECT client_id FROM oauth_provider_key WHERE provider='google'", one=True)
        self.assertEqual(row["client_id"], "cid-2")

    def test_delete_key_removes_it(self):
        oauth_login.set_key(1, "google", "cid", "secret")
        oauth_login.delete_key(1, "google")
        self.assertEqual(oauth_login.list_keys(1), [])

    def test_configured_providers_empty_until_a_key_is_saved(self):
        self.assertEqual(oauth_login.configured_providers(1), [])
        oauth_login.set_key(1, "google", "cid", "secret")
        self.assertEqual(oauth_login.configured_providers(1), [{"provider": "google", "label": "Google"}])

    def test_configured_providers_excludes_inactive(self):
        oauth_login.set_key(1, "google", "cid", "secret", is_active=False)
        self.assertEqual(oauth_login.configured_providers(1), [])


if __name__ == "__main__":
    unittest.main()
