import json
import unittest
import zlib
from datetime import date, datetime
from unittest import mock

import pytest
import requests
from parameterized import parameterized

import posthog.request as request_module
from posthog.test.logging_helpers import capture_message_only_logs
from posthog.request import (
    APIError,
    DatetimeSerializer,
    GetResponse,
    KEEP_ALIVE_SOCKET_OPTIONS,
    QuotaLimitError,
    _mask_tokens_in_url,
    batch_post,
    determine_server_host,
    disable_connection_reuse,
    enable_keep_alive,
    flags,
    get,
    set_socket_options,
)
from posthog.test.test_utils import TEST_API_KEY


@pytest.mark.parametrize(
    "url, expected",
    [
        # Token with params after - masks keeping first 10 chars
        (
            "https://example.com/api/flags?token=phc_abc123xyz789&send_cohorts",
            "https://example.com/api/flags?token=phc_abc123...&send_cohorts",
        ),
        # Token at end of URL
        (
            "https://example.com/api/flags?token=phc_abc123xyz789",
            "https://example.com/api/flags?token=phc_abc123...",
        ),
        # No token - unchanged
        (
            "https://example.com/api/flags?other=value",
            "https://example.com/api/flags?other=value",
        ),
        # Short token (<10 chars) - unchanged
        (
            "https://example.com/api/flags?token=short",
            "https://example.com/api/flags?token=short",
        ),
        # Exactly 10 char token - gets ellipsis
        (
            "https://example.com/api/flags?token=1234567890",
            "https://example.com/api/flags?token=1234567890...",
        ),
    ],
)
def test_mask_tokens_in_url(url, expected):
    assert _mask_tokens_in_url(url) == expected


@pytest.mark.parametrize(
    "key, expected_present",
    [
        ("sent_at", True),
        ("sentAt", False),
    ],
)
def test_post_sends_snake_case_sent_at(key, expected_present):
    mock_response = requests.Response()
    mock_response.status_code = 200
    mock_session = mock.MagicMock()
    mock_session.post.return_value = mock_response

    request_module.post(
        TEST_API_KEY,
        host="https://test.posthog.com",
        path="/batch/",
        session=mock_session,
        batch=[],
    )

    data = json.loads(mock_session.post.call_args.kwargs["data"])
    assert (key in data) is expected_present


def test_post_sends_project_api_key_field():
    mock_response = requests.Response()
    mock_response.status_code = 200
    mock_session = mock.MagicMock()
    mock_session.post.return_value = mock_response

    request_module.post(
        TEST_API_KEY,
        host="https://test.posthog.com",
        path="/batch/",
        session=mock_session,
        batch=[],
    )

    data = json.loads(mock_session.post.call_args.kwargs["data"])
    assert data["api_key"] == TEST_API_KEY
    assert "token" not in data


def test_message_only_debug_logs_include_posthog_prefix():
    mock_response = requests.Response()
    mock_response.status_code = 200
    mock_session = mock.MagicMock()
    mock_session.post.return_value = mock_response

    with capture_message_only_logs() as logs:
        request_module.post(
            TEST_API_KEY,
            host="https://test.posthog.com",
            path="/batch/",
            session=mock_session,
            batch=[],
        )

    assert logs.getvalue().splitlines() == [
        mock.ANY,
        "[PostHog] data uploaded successfully",
    ]
    assert logs.getvalue().splitlines()[0].startswith("[PostHog] making request: ")


class TestRequests(unittest.TestCase):
    def test_valid_request(self):
        res = batch_post(
            TEST_API_KEY,
            batch=[
                {"distinct_id": "distinct_id", "event": "python event", "type": "track"}
            ],
        )
        self.assertEqual(res.status_code, 200)

    def test_invalid_request_error(self):
        self.assertRaises(
            Exception, batch_post, "testsecret", "https://t.posthog.com", False, "[{]"
        )

    def test_invalid_host(self):
        self.assertRaises(
            Exception, batch_post, "testsecret", "t.posthog.com/", batch=[]
        )

    def test_post_without_path_preserves_type_error(self):
        mock_session = mock.MagicMock()

        with self.assertRaises(TypeError):
            request_module.post(
                TEST_API_KEY,
                host="https://test.posthog.com",
                session=mock_session,
            )

        mock_session.post.assert_not_called()

    def test_post_sends_string_payload_without_gzip(self):
        mock_response = requests.Response()
        mock_response.status_code = 200
        mock_session = mock.MagicMock()
        mock_session.post.return_value = mock_response

        request_module.post(
            TEST_API_KEY,
            host="https://test.posthog.com",
            path="/batch/",
            session=mock_session,
            batch=[],
        )

        mock_session.post.assert_called_once()
        url = mock_session.post.call_args.args[0]
        data = mock_session.post.call_args.kwargs["data"]
        self.assertEqual(url, "https://test.posthog.com/batch/")
        self.assertIsInstance(data, str)

    def test_post_sends_bytes_payload_with_gzip(self):
        mock_response = requests.Response()
        mock_response.status_code = 200
        mock_session = mock.MagicMock()
        mock_session.post.return_value = mock_response

        request_module.post(
            TEST_API_KEY,
            host="https://test.posthog.com",
            path="/batch/",
            gzip=True,
            session=mock_session,
            batch=[],
        )

        data = mock_session.post.call_args.kwargs["data"]
        headers = mock_session.post.call_args.kwargs["headers"]
        self.assertIsInstance(data, bytes)
        self.assertEqual(headers["Content-Encoding"], "gzip")

    def test_post_falls_back_to_uncompressed_payload_when_gzip_fails(self):
        for compression_error in [OSError("boom"), zlib.error("boom")]:
            with self.subTest(compression_error=type(compression_error)):
                mock_response = requests.Response()
                mock_response.status_code = 200
                mock_session = mock.MagicMock()
                mock_session.post.return_value = mock_response

                with mock.patch.object(
                    request_module, "GzipFile", side_effect=compression_error
                ):
                    request_module.post(
                        TEST_API_KEY,
                        host="https://test.posthog.com",
                        path="/batch/",
                        gzip=True,
                        session=mock_session,
                        batch=[],
                    )

                data = mock_session.post.call_args.kwargs["data"]
                headers = mock_session.post.call_args.kwargs["headers"]
                self.assertIsInstance(data, str)
                self.assertNotIn("Content-Encoding", headers)

    def test_datetime_serialization(self):
        data = {"created": datetime(2012, 3, 4, 5, 6, 7, 891011)}
        result = json.dumps(data, cls=DatetimeSerializer)
        self.assertEqual(result, '{"created": "2012-03-04T05:06:07.891011"}')

    def test_date_serialization(self):
        today = date.today()
        data = {"created": today}
        result = json.dumps(data, cls=DatetimeSerializer)
        expected = '{"created": "%s"}' % today.isoformat()
        self.assertEqual(result, expected)

    def test_should_not_timeout(self):
        res = batch_post(
            TEST_API_KEY,
            batch=[
                {"distinct_id": "distinct_id", "event": "python event", "type": "track"}
            ],
            timeout=15,
        )
        self.assertEqual(res.status_code, 200)

    def test_should_timeout(self):
        with self.assertRaises(requests.ReadTimeout):
            batch_post(
                "key",
                batch=[
                    {
                        "distinct_id": "distinct_id",
                        "event": "python event",
                        "type": "track",
                    }
                ],
                timeout=0.0001,
            )

    def test_quota_limited_flags_response(self):
        mock_response = requests.Response()
        mock_response.status_code = 200
        mock_response._content = json.dumps(
            {
                "quotaLimited": ["feature_flags"],
                "featureFlags": {},
                "featureFlagPayloads": {},
                "errorsWhileComputingFlags": False,
            }
        ).encode("utf-8")

        with mock.patch(
            "posthog.request._flags_session.post", return_value=mock_response
        ):
            with self.assertRaises(QuotaLimitError) as cm:
                flags("fake_key", "fake_host")

            self.assertEqual(cm.exception.status, 200)
            self.assertEqual(cm.exception.message, "Feature flags quota limited")

    def test_normal_flags_response(self):
        mock_response = requests.Response()
        mock_response.status_code = 200
        mock_response._content = json.dumps(
            {
                "featureFlags": {"flag1": True},
                "featureFlagPayloads": {},
                "errorsWhileComputingFlags": False,
            }
        ).encode("utf-8")

        with mock.patch(
            "posthog.request._flags_session.post", return_value=mock_response
        ):
            response = flags("fake_key", "fake_host")
            self.assertEqual(response["featureFlags"], {"flag1": True})


class TestGet(unittest.TestCase):
    """Unit tests for the get() function HTTP-level behavior."""

    @mock.patch("posthog.request._session.get")
    def test_get_returns_data_and_etag(self, mock_get):
        """Test that get() returns GetResponse with data and etag from headers."""
        mock_response = requests.Response()
        mock_response.status_code = 200
        mock_response.headers["ETag"] = '"abc123"'
        mock_response._content = json.dumps({"flags": [{"key": "test-flag"}]}).encode(
            "utf-8"
        )
        mock_get.return_value = mock_response

        response = get("api_key", "/test-url", host="https://example.com")

        self.assertIsInstance(response, GetResponse)
        self.assertEqual(response.data, {"flags": [{"key": "test-flag"}]})
        self.assertEqual(response.etag, '"abc123"')
        self.assertFalse(response.not_modified)

    @mock.patch("posthog.request._session.get")
    def test_get_sends_if_none_match_header_when_etag_provided(self, mock_get):
        """Test that If-None-Match header is sent when etag parameter is provided."""
        mock_response = requests.Response()
        mock_response.status_code = 200
        mock_response.headers["ETag"] = '"new-etag"'
        mock_response._content = json.dumps({"flags": []}).encode("utf-8")
        mock_get.return_value = mock_response

        get("api_key", "/test-url", host="https://example.com", etag='"previous-etag"')

        call_kwargs = mock_get.call_args[1]
        self.assertEqual(call_kwargs["headers"]["If-None-Match"], '"previous-etag"')

    @mock.patch("posthog.request._session.get")
    def test_get_does_not_send_if_none_match_when_no_etag(self, mock_get):
        """Test that If-None-Match header is not sent when no etag provided."""
        mock_response = requests.Response()
        mock_response.status_code = 200
        mock_response._content = json.dumps({"flags": []}).encode("utf-8")
        mock_get.return_value = mock_response

        get("api_key", "/test-url", host="https://example.com")

        call_kwargs = mock_get.call_args[1]
        self.assertNotIn("If-None-Match", call_kwargs["headers"])

    @mock.patch("posthog.request._session.get")
    def test_get_handles_304_not_modified(self, mock_get):
        """Test that 304 Not Modified response returns not_modified=True with no data."""
        mock_response = requests.Response()
        mock_response.status_code = 304
        mock_response.headers["ETag"] = '"unchanged-etag"'
        mock_get.return_value = mock_response

        response = get(
            "api_key", "/test-url", host="https://example.com", etag='"unchanged-etag"'
        )

        self.assertIsInstance(response, GetResponse)
        self.assertIsNone(response.data)
        self.assertEqual(response.etag, '"unchanged-etag"')
        self.assertTrue(response.not_modified)

    @mock.patch("posthog.request._session.get")
    def test_get_304_without_etag_header_uses_request_etag(self, mock_get):
        """Test that 304 response without ETag header falls back to request etag."""
        mock_response = requests.Response()
        mock_response.status_code = 304
        # Server doesn't return ETag header on 304
        mock_get.return_value = mock_response

        response = get(
            "api_key", "/test-url", host="https://example.com", etag='"original-etag"'
        )

        self.assertTrue(response.not_modified)
        self.assertEqual(response.etag, '"original-etag"')

    @mock.patch("posthog.request._session.get")
    def test_get_200_without_etag_header(self, mock_get):
        """Test that 200 response without ETag header returns None for etag."""
        mock_response = requests.Response()
        mock_response.status_code = 200
        mock_response._content = json.dumps({"flags": []}).encode("utf-8")
        # No ETag header
        mock_get.return_value = mock_response

        response = get("api_key", "/test-url", host="https://example.com")

        self.assertFalse(response.not_modified)
        self.assertIsNone(response.etag)
        self.assertEqual(response.data, {"flags": []})

    @mock.patch("posthog.request._session.get")
    def test_get_error_response_raises_api_error(self, mock_get):
        """Test that error responses raise APIError."""
        mock_response = requests.Response()
        mock_response.status_code = 401
        mock_response._content = json.dumps({"detail": "Unauthorized"}).encode("utf-8")
        mock_get.return_value = mock_response

        with self.assertRaises(APIError) as ctx:
            get("bad_key", "/test-url", host="https://example.com")

        self.assertEqual(ctx.exception.status, 401)
        self.assertEqual(ctx.exception.message, "Unauthorized")

    @mock.patch("posthog.request._session.get")
    def test_get_sends_authorization_header(self, mock_get):
        """Test that Authorization header is sent with Bearer token."""
        mock_response = requests.Response()
        mock_response.status_code = 200
        mock_response._content = json.dumps({}).encode("utf-8")
        mock_get.return_value = mock_response

        get("my-api-key", "/test-url", host="https://example.com")

        call_kwargs = mock_get.call_args[1]
        self.assertEqual(call_kwargs["headers"]["Authorization"], "Bearer my-api-key")

    @mock.patch("posthog.request._session.get")
    def test_get_sends_user_agent_header(self, mock_get):
        """Test that User-Agent header is sent."""
        mock_response = requests.Response()
        mock_response.status_code = 200
        mock_response._content = json.dumps({}).encode("utf-8")
        mock_get.return_value = mock_response

        get("api_key", "/test-url", host="https://example.com")

        call_kwargs = mock_get.call_args[1]
        self.assertIn("User-Agent", call_kwargs["headers"])
        self.assertTrue(
            call_kwargs["headers"]["User-Agent"].startswith("posthog-python/")
        )

    @mock.patch("posthog.request._session.get")
    def test_get_passes_timeout(self, mock_get):
        """Test that timeout parameter is passed to the request."""
        mock_response = requests.Response()
        mock_response.status_code = 200
        mock_response._content = json.dumps({}).encode("utf-8")
        mock_get.return_value = mock_response

        get("api_key", "/test-url", host="https://example.com", timeout=30)

        call_kwargs = mock_get.call_args[1]
        self.assertEqual(call_kwargs["timeout"], 30)

    @mock.patch("posthog.request._session.get")
    def test_get_constructs_full_url(self, mock_get):
        """Test that host and url are combined correctly."""
        mock_response = requests.Response()
        mock_response.status_code = 200
        mock_response._content = json.dumps({}).encode("utf-8")
        mock_get.return_value = mock_response

        get("api_key", "/api/flags", host="https://example.com")

        call_args = mock_get.call_args[0]
        self.assertEqual(call_args[0], "https://example.com/api/flags")

    @mock.patch("posthog.request._session.get")
    def test_get_removes_trailing_slash_from_host(self, mock_get):
        """Test that trailing slash is removed from host."""
        mock_response = requests.Response()
        mock_response.status_code = 200
        mock_response._content = json.dumps({}).encode("utf-8")
        mock_get.return_value = mock_response

        get("api_key", "/api/flags", host="https://example.com/")

        call_args = mock_get.call_args[0]
        self.assertEqual(call_args[0], "https://example.com/api/flags")


@pytest.mark.parametrize(
    "host, expected",
    [
        ("https://t.posthog.com", "https://t.posthog.com"),
        ("https://t.posthog.com/", "https://t.posthog.com/"),
        ("t.posthog.com", "t.posthog.com"),
        ("t.posthog.com/", "t.posthog.com/"),
        ("https://us.posthog.com.rg.proxy.com", "https://us.posthog.com.rg.proxy.com"),
        ("app.posthog.com", "app.posthog.com"),
        ("eu.posthog.com", "eu.posthog.com"),
        ("https://app.posthog.com", "https://us.i.posthog.com"),
        ("https://eu.posthog.com", "https://eu.i.posthog.com"),
        ("https://us.posthog.com", "https://us.i.posthog.com"),
        ("https://app.posthog.com/", "https://us.i.posthog.com"),
        ("https://eu.posthog.com/", "https://eu.i.posthog.com"),
        ("https://us.posthog.com/", "https://us.i.posthog.com"),
        (" \nhttps://eu.posthog.com/\t ", "https://eu.i.posthog.com"),
        (" \n\t ", "https://us.i.posthog.com"),
        (None, "https://us.i.posthog.com"),
    ],
)
def test_routing_to_custom_host(host, expected):
    assert determine_server_host(host) == expected


def test_enable_keep_alive_sets_socket_options():
    try:
        enable_keep_alive()
        from posthog.request import _session

        adapter = _session.get_adapter("https://example.com")
        assert adapter.socket_options == KEEP_ALIVE_SOCKET_OPTIONS
    finally:
        set_socket_options(None)


def test_set_socket_options_clears_with_none():
    try:
        enable_keep_alive()
        set_socket_options(None)
        from posthog.request import _session

        adapter = _session.get_adapter("https://example.com")
        assert adapter.socket_options is None
    finally:
        set_socket_options(None)


def test_disable_connection_reuse_creates_fresh_sessions():
    try:
        disable_connection_reuse()
        session1 = request_module._get_session()
        session2 = request_module._get_session()
        assert session1 is not session2
    finally:
        request_module._pooling_enabled = True


def test_set_socket_options_is_idempotent():
    try:
        enable_keep_alive()
        session1 = request_module._session
        enable_keep_alive()
        session2 = request_module._session
        assert session1 is session2
    finally:
        set_socket_options(None)


class TestFlagsSession(unittest.TestCase):
    """Tests for flags session configuration."""

    def test_flags_session_disables_adapter_retries(self):
        """HTTP adapter retries are disabled; flags() handles bounded retries."""
        from posthog.request import _build_flags_session

        session = _build_flags_session()
        adapter = session.get_adapter("https://test.posthog.com")
        retry = adapter.max_retries

        self.assertEqual(retry.total, 0)
        self.assertEqual(retry.connect, 0)
        self.assertEqual(retry.read, 0)
        self.assertEqual(retry.status, 0)

    @mock.patch("posthog.request._get_flags_session")
    def test_flags_uses_flags_session(self, mock_get_flags_session):
        """flags() uses the dedicated flags session, not the general session."""
        mock_response = requests.Response()
        mock_response.status_code = 200
        mock_response._content = json.dumps(
            {
                "featureFlags": {"test-flag": True},
                "featureFlagPayloads": {},
                "errorsWhileComputingFlags": False,
            }
        ).encode("utf-8")

        mock_session = mock.MagicMock()
        mock_session.post.return_value = mock_response
        mock_get_flags_session.return_value = mock_session

        result = flags("test-key", "https://test.posthog.com", distinct_id="user123")

        self.assertEqual(result["featureFlags"]["test-flag"], True)
        mock_get_flags_session.assert_called_once()
        mock_session.post.assert_called_once()

    @mock.patch("posthog.request._get_flags_session")
    def test_flags_no_retry_on_quota_limit(self, mock_get_flags_session):
        """flags() raises QuotaLimitError without retrying (at application level)."""
        mock_response = requests.Response()
        mock_response.status_code = 200
        mock_response._content = json.dumps(
            {
                "quotaLimited": ["feature_flags"],
                "featureFlags": {},
                "featureFlagPayloads": {},
                "errorsWhileComputingFlags": False,
            }
        ).encode("utf-8")

        mock_session = mock.MagicMock()
        mock_session.post.return_value = mock_response
        mock_get_flags_session.return_value = mock_session

        with self.assertRaises(QuotaLimitError):
            flags("test-key", "https://test.posthog.com", distinct_id="user123")

        # QuotaLimitError is raised after response is received, not retried
        self.assertEqual(mock_session.post.call_count, 1)


class TestFlagsRetries(unittest.TestCase):
    """Tests for /flags retry behavior."""

    @mock.patch("posthog.request.time.sleep")
    @mock.patch("posthog.request._get_flags_session")
    def test_flags_retries_transport_errors_once_by_default(
        self, mock_get_flags_session, mock_sleep
    ):
        mock_response = requests.Response()
        mock_response.status_code = 200
        mock_response._content = json.dumps(
            {
                "featureFlags": {"test-flag": True},
                "featureFlagPayloads": {},
                "errorsWhileComputingFlags": False,
            }
        ).encode("utf-8")

        mock_session = mock.MagicMock()
        mock_session.post.side_effect = [
            requests.exceptions.ConnectionError("connection failed"),
            mock_response,
        ]
        mock_get_flags_session.return_value = mock_session

        response = flags("test-key", "https://test.posthog.com", distinct_id="user123")

        self.assertEqual(response["featureFlags"], {"test-flag": True})
        self.assertEqual(mock_session.post.call_count, 2)
        mock_sleep.assert_called_once_with(0.3)

    @mock.patch("posthog.request.time.sleep")
    @mock.patch("posthog.request._get_flags_session")
    def test_flags_retry_count_zero_disables_retries(
        self, mock_get_flags_session, mock_sleep
    ):
        mock_session = mock.MagicMock()
        mock_session.post.side_effect = requests.exceptions.Timeout("timed out")
        mock_get_flags_session.return_value = mock_session

        with self.assertRaises(requests.exceptions.Timeout):
            flags(
                "test-key",
                "https://test.posthog.com",
                max_retries=0,
                distinct_id="user123",
            )

        self.assertEqual(mock_session.post.call_count, 1)
        mock_sleep.assert_not_called()

    @mock.patch("posthog.request.time.sleep")
    @mock.patch("posthog.request._get_flags_session")
    def test_flags_retry_delay_starts_at_300ms_and_doubles(
        self, mock_get_flags_session, mock_sleep
    ):
        mock_response = requests.Response()
        mock_response.status_code = 200
        mock_response._content = json.dumps(
            {
                "featureFlags": {"test-flag": True},
                "featureFlagPayloads": {},
                "errorsWhileComputingFlags": False,
            }
        ).encode("utf-8")

        mock_session = mock.MagicMock()
        mock_session.post.side_effect = [
            requests.exceptions.ConnectionError("connection failed"),
            requests.exceptions.Timeout("timed out"),
            mock_response,
        ]
        mock_get_flags_session.return_value = mock_session

        response = flags(
            "test-key",
            "https://test.posthog.com",
            max_retries=2,
            distinct_id="user123",
        )

        self.assertEqual(response["featureFlags"], {"test-flag": True})
        self.assertEqual(mock_session.post.call_count, 3)
        self.assertEqual(
            [call.args[0] for call in mock_sleep.call_args_list], [0.3, 0.6]
        )

    @parameterized.expand([(502,), (504,)])
    def test_flags_retries_contract_http_status_errors(self, status_code):
        retry_response = requests.Response()
        retry_response.status_code = status_code
        retry_response._content = b'{"detail": "transient error"}'

        success_response = requests.Response()
        success_response.status_code = 200
        success_response._content = json.dumps(
            {
                "featureFlags": {f"transient-{status_code}-flag": True},
                "featureFlagPayloads": {},
                "errorsWhileComputingFlags": False,
            }
        ).encode("utf-8")

        mock_session = mock.MagicMock()
        mock_session.post.side_effect = [retry_response, success_response]

        with (
            mock.patch("posthog.request._get_flags_session", return_value=mock_session),
            mock.patch("posthog.request.time.sleep") as mock_sleep,
        ):
            response = flags(
                "test-key", "https://test.posthog.com", distinct_id="user123"
            )

        self.assertEqual(
            response["featureFlags"], {f"transient-{status_code}-flag": True}
        )
        self.assertEqual(mock_session.post.call_count, 2)
        mock_sleep.assert_called_once_with(0.3)

    @parameterized.expand([(408,), (429,), (500,), (503,)])
    def test_flags_does_not_retry_other_http_status_errors(self, status_code):
        mock_response = requests.Response()
        mock_response.status_code = status_code
        mock_response._content = b'{"detail": "Service unavailable"}'

        mock_session = mock.MagicMock()
        mock_session.post.return_value = mock_response

        with mock.patch(
            "posthog.request._get_flags_session", return_value=mock_session
        ):
            with self.assertRaises(APIError) as cm:
                flags("test-key", "https://test.posthog.com", distinct_id="user123")

        self.assertEqual(cm.exception.status, status_code)
        self.assertEqual(mock_session.post.call_count, 1)