import socket
from unittest.mock import patch, MagicMock

import pytest
from channels.routing import URLRouter
from channels.testing import WebsocketCommunicator
from asgiref.sync import async_to_sync
from django.contrib.auth import get_user_model
from django.test import override_settings
from rest_framework.test import APIClient
from apps.chat.models import Conversation, ConversationParticipant, Message, MessageReaction
from apps.chat.routing import websocket_urlpatterns
from apps.chat.link_preview import fetch_link_preview, LinkPreviewError
from apps.companies.models import Organization, OrganizationMembership


@pytest.fixture
def organization(db):
    return Organization.objects.create(name="Test Org")


@pytest.fixture
def user(db):
    User = get_user_model()
    return User.objects.create_user(email="chat@example.com", password="pass123")


@pytest.fixture
def user2(db):
    User = get_user_model()
    return User.objects.create_user(email="chat2@example.com", password="pass123")


@pytest.fixture
def conversation(db, user, user2, organization):
    conv = Conversation.objects.create(
        organization=organization,
        created_by=user,
        is_group=False
    )
    ConversationParticipant.objects.create(
        conversation=conv,
        user=user,
        is_admin=True
    )
    ConversationParticipant.objects.create(
        conversation=conv,
        user=user2
    )
    return conv


@pytest.fixture
def message(db, conversation, user):
    return Message.objects.create(
        conversation=conversation,
        sender=user,
        ciphertext="encrypted:Hello, world!"
    )


@pytest.mark.django_db
class TestConversationModel:
    def test_create_conversation(self, user, organization):
        conv = Conversation.objects.create(
            organization=organization,
            created_by=user,
            is_group=False
        )
        assert conv.is_group is False
        assert conv.created_by == user

    def test_create_group_conversation(self, user, organization):
        conv = Conversation.objects.create(
            organization=organization,
            name="Team Chat",
            created_by=user,
            is_group=True
        )
        assert conv.is_group is True
        assert conv.name == "Team Chat"

    def test_add_participants(self, user, user2, organization):
        conv = Conversation.objects.create(organization=organization, created_by=user)
        ConversationParticipant.objects.create(conversation=conv, user=user)
        ConversationParticipant.objects.create(conversation=conv, user=user2)
        assert conv.participants.count() == 2

    def test_conversation_str(self, user, organization):
        conv = Conversation.objects.create(organization=organization, created_by=user, name="Test Chat")
        assert str(conv) == "Test Chat"


@pytest.mark.django_db
class TestMessageModel:
    def test_create_message(self, conversation, user):
        msg = Message.objects.create(
            conversation=conversation,
            sender=user,
            ciphertext="encrypted:Test message"
        )
        assert msg.ciphertext == "encrypted:Test message"
        assert msg.sender == user
        assert msg.is_read is False

    def test_message_ordering(self, conversation, user):
        msg1 = Message.objects.create(conversation=conversation, sender=user, ciphertext="First")
        msg2 = Message.objects.create(conversation=conversation, sender=user, ciphertext="Second")
        messages = list(conversation.messages.all())
        assert messages[0] == msg1
        assert messages[1] == msg2

    def test_message_with_attachment(self, conversation, user):
        msg = Message.objects.create(
            conversation=conversation,
            sender=user,
            ciphertext="Image",
            has_attachments=True,
        )
        assert msg.has_attachments is True

    def test_mark_message_read(self, message):
        message.is_read = True
        message.save()
        message.refresh_from_db()
        assert message.is_read is True


@pytest.mark.django_db
class TestMessageTenantIsolation:
    def test_cannot_send_to_legacy_conversation_with_foreign_participant(self):
        User = get_user_model()
        organization_a = Organization.objects.create(name="Chat Org A")
        organization_b = Organization.objects.create(name="Chat Org B")
        sender = User.objects.create_user(email="sender@chat-security.test", password="pass123")
        foreign_participant = User.objects.create_user(
            email="foreign@chat-security.test", password="pass123"
        )
        OrganizationMembership.objects.create(organization=organization_a, user=sender)
        OrganizationMembership.objects.create(
            organization=organization_b, user=foreign_participant
        )
        legacy_conversation = Conversation.objects.create(
            organization=organization_a, created_by=sender
        )
        ConversationParticipant.objects.create(conversation=legacy_conversation, user=sender)
        ConversationParticipant.objects.create(
            conversation=legacy_conversation, user=foreign_participant
        )
        client = APIClient()
        client.force_authenticate(sender)

        response = client.post(
            "/api/chat/messages/",
            {
                "conversation": str(legacy_conversation.pk),
                "ciphertext": "encrypted",
                "content_nonce": "nonce",
                "key_wraps": [
                    {"recipient": str(sender.pk), "wrapped_key": "sender-key", "wrap_nonce": "a"},
                    {
                        "recipient": str(foreign_participant.pk),
                        "wrapped_key": "foreign-key",
                        "wrap_nonce": "b",
                    },
                ],
            },
            format="json",
        )

        assert response.status_code == 400
        assert Message.objects.filter(conversation=legacy_conversation).count() == 0

    def test_cannot_send_to_conversation_outside_active_organization(self):
        User = get_user_model()
        organization_a = Organization.objects.create(name="Active Chat Org A")
        organization_b = Organization.objects.create(name="Active Chat Org B")
        sender = User.objects.create_user(email="multi-org@chat-security.test", password="pass123")
        OrganizationMembership.objects.create(organization=organization_a, user=sender)
        OrganizationMembership.objects.create(organization=organization_b, user=sender)
        conversation = Conversation.objects.create(organization=organization_a, created_by=sender)
        ConversationParticipant.objects.create(conversation=conversation, user=sender)
        client = APIClient()
        session = client.session
        session["active_org_id"] = str(organization_b.pk)
        session.save()
        client.force_authenticate(sender)

        response = client.post(
            "/api/chat/messages/",
            {
                "conversation": str(conversation.pk),
                "ciphertext": "encrypted",
                "content_nonce": "nonce",
                "key_wraps": [
                    {"recipient": str(sender.pk), "wrapped_key": "sender-key", "wrap_nonce": "a"}
                ],
            },
            format="json",
        )

        assert response.status_code == 400
        assert Message.objects.filter(conversation=conversation).count() == 0

    def test_cannot_wrap_message_key_for_non_participant(self):
        User = get_user_model()
        organization = Organization.objects.create(name="Key Wrap Org")
        sender = User.objects.create_user(email="key-sender@chat-security.test", password="pass123")
        outsider = User.objects.create_user(email="key-outsider@chat-security.test", password="pass123")
        OrganizationMembership.objects.create(organization=organization, user=sender)
        OrganizationMembership.objects.create(organization=organization, user=outsider)
        conversation = Conversation.objects.create(organization=organization, created_by=sender)
        ConversationParticipant.objects.create(conversation=conversation, user=sender)
        client = APIClient()
        client.force_authenticate(sender)

        response = client.post(
            "/api/chat/messages/",
            {
                "conversation": str(conversation.pk),
                "ciphertext": "encrypted",
                "content_nonce": "nonce",
                "key_wraps": [
                    {"recipient": str(sender.pk), "wrapped_key": "sender-key", "wrap_nonce": "a"},
                    {"recipient": str(outsider.pk), "wrapped_key": "outsider-key", "wrap_nonce": "b"},
                ],
            },
            format="json",
        )

        assert response.status_code == 400
        assert Message.objects.filter(conversation=conversation).count() == 0


@pytest.mark.django_db(transaction=True, serialized_rollback=True)
@override_settings(
    CHANNEL_LAYERS={"default": {"BACKEND": "channels.layers.InMemoryChannelLayer"}}
)
def test_chat_socket_rejects_conversation_outside_active_organization():
    User = get_user_model()
    organization_a = Organization.objects.create(name="Socket Org A")
    organization_b = Organization.objects.create(name="Socket Org B")
    participant = User.objects.create_user(
        email="multi-org@chat-socket-security.test", password="pass123"
    )
    OrganizationMembership.objects.create(organization=organization_a, user=participant)
    OrganizationMembership.objects.create(organization=organization_b, user=participant)
    conversation = Conversation.objects.create(
        organization=organization_a, created_by=participant
    )
    ConversationParticipant.objects.create(conversation=conversation, user=participant)
    async def exercise_socket():
        communicator = WebsocketCommunicator(
            URLRouter(websocket_urlpatterns), f"/ws/chat/{conversation.pk}/"
        )
        communicator.scope["user"] = participant
        communicator.scope["session"] = {"active_org_id": str(organization_b.pk)}
        connected, _ = await communicator.connect()
        if connected:
            await communicator.disconnect()
        return connected

    connected = async_to_sync(exercise_socket)()

    assert connected is False

@pytest.mark.django_db
class TestMessageReactionModel:
    def test_add_reaction(self, message, user):
        reaction = MessageReaction.objects.create(
            message=message,
            user=user,
            reaction="👍"
        )
        assert reaction.reaction == "👍"
        assert message.reactions.count() == 1

    def test_unique_reaction_per_user(self, message, user):
        MessageReaction.objects.create(message=message, user=user, reaction="👍")
        with pytest.raises(Exception):
            MessageReaction.objects.create(message=message, user=user, reaction="👍")


class TestLinkPreviewSSRFGuards:
    @pytest.fixture(autouse=True)
    def _no_live_dns(self):
        """Resolve public hostnames offline.

        `_resolve_and_validate_host` calls `socket.getaddrinfo` before any
        request, so mocking `requests.get` alone still leaves a real DNS lookup
        for `example.com`. That made these tests network-dependent and flaky
        under full-suite load. Literal IPs and the `.invalid` TLD are passed
        through to the real resolver so the guard's own logic — private-range
        rejection and NXDOMAIN handling — is still exercised for real.
        """
        real = socket.getaddrinfo

        def fake(host, *args, **kwargs):
            if host == "example.com":
                return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 0))]
            return real(host, *args, **kwargs)

        with patch("apps.chat.link_preview.socket.getaddrinfo", side_effect=fake):
            yield

    def test_rejects_non_http_scheme(self):
        with pytest.raises(LinkPreviewError):
            fetch_link_preview("file:///etc/passwd")

    def test_rejects_localhost(self):
        with pytest.raises(LinkPreviewError):
            fetch_link_preview("http://localhost/")

    def test_rejects_loopback_ip(self):
        with pytest.raises(LinkPreviewError):
            fetch_link_preview("http://127.0.0.1/")

    def test_rejects_link_local_metadata_ip(self):
        # 169.254.169.254 is the cloud-provider metadata endpoint — the
        # canonical SSRF target this guard exists to block.
        with pytest.raises(LinkPreviewError):
            fetch_link_preview("http://169.254.169.254/latest/meta-data/")

    def test_rejects_private_network_ip(self):
        with pytest.raises(LinkPreviewError):
            fetch_link_preview("http://10.0.0.5/")

    def test_rejects_unresolvable_host(self):
        with pytest.raises(LinkPreviewError):
            fetch_link_preview("http://this-domain-should-not-resolve.invalid/")

    @patch("apps.chat.link_preview.requests.get")
    def test_rejects_redirect_to_private_ip(self, mock_get):
        # Even if the original host is public, a redirect landing on a
        # private address must still be blocked.
        mock_resp = MagicMock()
        mock_resp.url = "http://127.0.0.1/internal"
        mock_get.return_value = mock_resp
        with pytest.raises(LinkPreviewError):
            fetch_link_preview("http://example.com/redirect")

    @patch("apps.chat.link_preview.requests.get")
    def test_rejects_non_html_content_type(self, mock_get):
        mock_resp = MagicMock()
        mock_resp.url = "http://example.com/file.pdf"
        mock_resp.headers = {"Content-Type": "application/pdf"}
        mock_get.return_value = mock_resp
        with pytest.raises(LinkPreviewError):
            fetch_link_preview("http://example.com/file.pdf")

    @patch("apps.chat.link_preview.requests.get")
    def test_parses_og_tags(self, mock_get):
        html = b"""
        <html><head>
          <title>Fallback Title</title>
          <meta property="og:title" content="Real Title">
          <meta property="og:description" content="A description">
          <meta property="og:image" content="https://example.com/img.png">
          <meta property="og:site_name" content="Example Site">
        </head></html>
        """
        mock_resp = MagicMock()
        mock_resp.url = "http://example.com/page"
        mock_resp.headers = {"Content-Type": "text/html; charset=utf-8"}
        mock_resp.encoding = "utf-8"
        mock_resp.iter_content.return_value = [html]
        mock_get.return_value = mock_resp

        preview = fetch_link_preview("http://example.com/page")
        assert preview["title"] == "Real Title"
        assert preview["description"] == "A description"
        assert preview["image"] == "https://example.com/img.png"
        assert preview["site_name"] == "Example Site"

    @patch("apps.chat.link_preview.requests.get")
    def test_falls_back_to_title_tag(self, mock_get):
        html = b"<html><head><title>Just a Title</title></head></html>"
        mock_resp = MagicMock()
        mock_resp.url = "http://example.com/page"
        mock_resp.headers = {"Content-Type": "text/html"}
        mock_resp.encoding = "utf-8"
        mock_resp.iter_content.return_value = [html]
        mock_get.return_value = mock_resp

        preview = fetch_link_preview("http://example.com/page")
        assert preview["title"] == "Just a Title"


@pytest.mark.django_db
class TestLinkPreviewAPI:
    def test_requires_authentication(self):
        client = APIClient()
        resp = client.get("/api/chat/link-preview/", {"url": "http://example.com/"})
        assert resp.status_code == 403 or resp.status_code == 401

    def test_requires_url_param(self, user):
        client = APIClient()
        client.force_authenticate(user=user)
        resp = client.get("/api/chat/link-preview/")
        assert resp.status_code == 400

    def test_blocks_ssrf_target(self, user):
        client = APIClient()
        client.force_authenticate(user=user)
        resp = client.get("/api/chat/link-preview/", {"url": "http://169.254.169.254/"})
        assert resp.status_code == 400

    @patch("apps.chat.link_preview.requests.get")
    def test_returns_preview_for_valid_url(self, mock_get, user):
        html = b'<html><head><meta property="og:title" content="Hi"></head></html>'
        mock_resp = MagicMock()
        mock_resp.url = "http://example.com/"
        mock_resp.headers = {"Content-Type": "text/html"}
        mock_resp.encoding = "utf-8"
        mock_resp.iter_content.return_value = [html]
        mock_get.return_value = mock_resp

        client = APIClient()
        client.force_authenticate(user=user)
        resp = client.get("/api/chat/link-preview/", {"url": "http://example.com/"})
        assert resp.status_code == 200
        assert resp.data["title"] == "Hi"


@pytest.mark.django_db
class TestConversationPaginationOrdering:
    """BE-005: ConversationViewSet's queryset previously had no explicit
    ordering at all (combined with .distinct() and a Max() annotation),
    which Django/DRF pagination warns about. `-last_message_at_anno, -pk` is
    now the explicit order — last-active-first with a unique tie-breaker."""

    def _client_for(self, user, organization):
        from apps.companies.models import OrganizationMembership
        OrganizationMembership.objects.get_or_create(organization=organization, user=user)
        client = APIClient()
        client.force_authenticate(user=user)
        return client

    def test_no_unordered_pagination_warning(self, user, user2, organization):
        Conversation.objects.create(organization=organization, created_by=user, is_group=False)
        Conversation.objects.create(organization=organization, created_by=user, is_group=False)
        client = self._client_for(user, organization)
        import warnings
        from django.core.paginator import UnorderedObjectListWarning

        with warnings.catch_warnings():
            warnings.simplefilter("error", UnorderedObjectListWarning)
            resp = client.get("/api/chat/conversations/")
        assert resp.status_code == 200

    def test_stable_across_identical_pages(self, user, user2, organization):
        for _ in range(3):
            conv = Conversation.objects.create(organization=organization, created_by=user, is_group=False)
            ConversationParticipant.objects.create(conversation=conv, user=user, is_admin=True)
        client = self._client_for(user, organization)
        first = client.get("/api/chat/conversations/").json()
        second = client.get("/api/chat/conversations/").json()
        first_rows = first["results"] if isinstance(first, dict) and "results" in first else first
        second_rows = second["results"] if isinstance(second, dict) and "results" in second else second
        assert [r["id"] for r in first_rows] == [r["id"] for r in second_rows]


@pytest.mark.django_db
class TestChatWebSocketGating:
    """WEBSOCKET_ENABLED=False suppresses the chat broadcast. The open thread's
    5s poll is what actually delivers messages on the WSGI-only host."""

    def _viewset(self):
        from apps.chat.views import MessageViewSet
        view = MessageViewSet()
        view.request = MagicMock()
        return view

    def test_broadcast_skipped_when_disabled(self, settings):
        settings.WEBSOCKET_ENABLED = False
        with patch("channels.layers.get_channel_layer") as layer:
            self._viewset()._broadcast_new_message(MagicMock())
        layer.assert_not_called()

    def test_broadcast_attempted_when_enabled(self, settings):
        settings.WEBSOCKET_ENABLED = True
        with patch("channels.layers.get_channel_layer") as layer:
            layer.return_value = None  # stop before serializing a mock message
            self._viewset()._broadcast_new_message(MagicMock())
        layer.assert_called()