from __future__ import annotations

import importlib.util
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
import threading
import unittest
from urllib.error import HTTPError
from urllib.request import Request

from scripts import apply_deploy_seed


REPO_ROOT = Path(__file__).resolve().parents[1]
SCRIPT_PATH = REPO_ROOT / "scripts" / "apply_connected_apps.py"

SPEC = importlib.util.spec_from_file_location("apply_connected_apps", SCRIPT_PATH)
assert SPEC is not None and SPEC.loader is not None
apply_connected_apps = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(apply_connected_apps)


class ConnectedAppsBootstrapOAuthTests(unittest.TestCase):
    def test_seed_dispatch_forwards_programmatic_oauth_binding(self) -> None:
        environment = {
            "PATH": "/usr/bin:/bin",
            "STAGING_INSPECTION_OAUTH_CLIENT_ID": "salesforce-compat-client",
            "STAGING_INSPECTION_OAUTH_CLIENT_SECRET": "programmatic-secret",
        }

        outer_dispatch_environment = (
            apply_connected_apps.deploy_offline_env.deploy_seed_dispatch_env(
                environment,
                profile="staging",
                nexuscrm_worktree=None,
            )
        )
        dispatcher_environment = apply_deploy_seed._dispatcher_environment(
            "staging",
            outer_dispatch_environment,
        )
        adapter_environment = apply_deploy_seed._adapter_environment(
            "connected_apps",
            dispatcher_environment,
        )

        self.assertEqual(
            "salesforce-compat-client",
            adapter_environment["STAGING_INSPECTION_OAUTH_CLIENT_ID"],
        )
        self.assertEqual(
            "programmatic-secret",
            adapter_environment["STAGING_INSPECTION_OAUTH_CLIENT_SECRET"],
        )


class ConnectedAppsProfileResolutionTests(unittest.TestCase):
    EXISTING_PROFILE_NAME = "Alliance & Channel"


    def test_reviewed_seed_declares_existing_profile_semantically(self) -> None:
        seed_file = (
            REPO_ROOT
            / "deployments"
            / "offline"
            / "runtime"
            / "seeds"
            / "connected-apps.seed.yaml"
        )

        spec = apply_connected_apps.load_connected_apps(seed_file)

        user = spec["users"]["connected_apps_zeno_user"]
        self.assertEqual(user["existing_profile_name"], self.EXISTING_PROFILE_NAME)
        self.assertEqual(user["create_fields"]["ProfileId"], "system_admin")

        tony = spec["users"]["connected_apps_tony_user"]
        self.assertEqual(tony["existing_profile_name"], "Pre-Sales")
        self.assertEqual(tony["create_fields"]["ProfileId"], "system_admin")
        self.assertEqual(
            tony["create_fields"]["Username"], "chenhao.huang@pingcap.com"
        )

        apps_by_key = {app["consumer_key"]: app for app in spec["connected_apps"]}
        self.assertIn("oXCHlm77BgAh6nJGfCqew41Mq9gdyPyGo6FzA2viLGw", apps_by_key)
        tony_app = apps_by_key["xV6EN2In3lh8Co3MLUmgib4iW3KtSF3RW3n-yIVi5DU"]
        self.assertEqual(
            tony_app["fields"]["ConsumerSecret"],
            "{{ env.CONNECTED_APPS_TONY_CONSUMER_SECRET }}",
        )
        self.assertEqual(tony_app["run_as_user_key"], "connected_apps_tony_user")


class ConnectedAppsClientCredentialsVerificationTests(unittest.TestCase):
    def test_verification_http_client_refuses_redirects(self) -> None:
        redirected_requests = 0

        class RedirectHandler(BaseHTTPRequestHandler):
            def do_GET(self) -> None:
                nonlocal redirected_requests
                if self.path == "/protected":
                    self.send_response(302)
                    self.send_header("Location", "/redirected")
                    self.end_headers()
                    return
                if self.path == "/redirected":
                    redirected_requests += 1
                    self.send_response(200)
                    self.end_headers()
                    return
                self.send_response(404)
                self.end_headers()

            def log_message(self, _format, *_args) -> None:
                return

        server = ThreadingHTTPServer(("127.0.0.1", 0), RedirectHandler)
        thread = threading.Thread(target=server.serve_forever, daemon=True)
        thread.start()
        try:
            request = Request(
                f"http://127.0.0.1:{server.server_port}/protected",
                headers={"Authorization": "Bearer issued-access-token"},
            )
            with self.assertRaises(HTTPError) as raised:
                apply_connected_apps.open_verification_request(request, timeout=5)
            self.assertEqual(302, raised.exception.code)
            self.assertEqual(0, redirected_requests)
        finally:
            server.shutdown()
            server.server_close()
            thread.join(timeout=5)


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