"""Unit tests for human-input test delivery with SQLite-backed member lookup."""

from __future__ import annotations

from unittest.mock import MagicMock, patch
from uuid import uuid4

import pytest
from flask import Flask
from sqlalchemy.engine import Engine
from sqlalchemy.orm import Session

from configs import dify_config
from core.workflow.human_input_adapter import (
    EmailDeliveryConfig,
    EmailDeliveryMethod,
    EmailRecipients,
    ExternalRecipient,
    MemberRecipient,
)
from graphon.runtime import VariablePool
from models.account import Account, TenantAccountJoin
from models.engine import db
from services import human_input_delivery_test_service as service_module
from services.feature_service import FeatureModel
from services.human_input_delivery_test_service import (
    DeliveryTestContext,
    DeliveryTestEmailRecipient,
    DeliveryTestError,
    DeliveryTestRegistry,
    DeliveryTestResult,
    DeliveryTestStatus,
    DeliveryTestUnsupportedError,
    EmailDeliveryTestHandler,
    HumanInputDeliveryTestService,
    _build_form_link,
)


def _make_valid_email_config():
    return EmailDeliveryConfig(
        recipients=EmailRecipients(include_bound_group=False, items=[]),
        subject="Subj",
        body="Body",
    )


def test_build_form_link():
    with patch.object(dify_config, "APP_WEB_URL", "http://example.com/"):
        assert _build_form_link("token123") == "http://example.com/form/token123"

    with patch.object(dify_config, "APP_WEB_URL", "http://example.com"):
        assert _build_form_link("token123") == "http://example.com/form/token123"

    assert _build_form_link(None) is None

    with patch.object(dify_config, "APP_WEB_URL", None):
        assert _build_form_link("token123") is None


class TestDeliveryTestRegistry:
    def test_register(self):
        registry = DeliveryTestRegistry()
        assert len(registry._handlers) == 0
        handler = MagicMock()
        registry.register(handler)
        assert len(registry._handlers) == 1
        assert registry._handlers[0] == handler

    def test_register_and_dispatch(self):
        handler = MagicMock()
        handler.supports.return_value = True
        handler.send_test.return_value = DeliveryTestResult(status=DeliveryTestStatus.OK)

        registry = DeliveryTestRegistry([handler])
        context = MagicMock(spec=DeliveryTestContext)
        method = MagicMock()

        result = registry.dispatch(context=context, method=method)

        assert result.status == DeliveryTestStatus.OK
        handler.supports.assert_called_once_with(method)
        handler.send_test.assert_called_once_with(context=context, method=method)

    def test_dispatch_unsupported(self):
        handler = MagicMock()
        handler.supports.return_value = False

        registry = DeliveryTestRegistry([handler])
        context = MagicMock(spec=DeliveryTestContext)
        method = MagicMock()

        with pytest.raises(DeliveryTestUnsupportedError, match="Delivery method does not support test send."):
            registry.dispatch(context=context, method=method)

    def test_default(self) -> None:
        app = Flask(__name__)
        app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///:memory:"
        db.init_app(app)

        with app.app_context():
            registry = DeliveryTestRegistry.default()

        assert len(registry._handlers) == 1
        assert isinstance(registry._handlers[0], EmailDeliveryTestHandler)


def test_human_input_delivery_test_service():
    registry = MagicMock(spec=DeliveryTestRegistry)
    service = HumanInputDeliveryTestService(registry=registry)
    context = MagicMock(spec=DeliveryTestContext)
    method = MagicMock()

    service.send_test(context=context, method=method)
    registry.dispatch.assert_called_once_with(context=context, method=method)


class TestEmailDeliveryTestHandler:
    def test_init_with_engine(self):
        engine = MagicMock(spec=Engine)
        handler = EmailDeliveryTestHandler(session_factory=engine)
        assert handler._session_factory.kw["bind"] == engine

    def test_supports(self, sqlite_engine: Engine) -> None:
        handler = EmailDeliveryTestHandler(session_factory=sqlite_engine)
        method = EmailDeliveryMethod(config=_make_valid_email_config())
        assert handler.supports(method) is True
        assert handler.supports(MagicMock()) is False

    def test_send_test_unsupported_method(self, sqlite_engine: Engine) -> None:
        handler = EmailDeliveryTestHandler(session_factory=sqlite_engine)
        with pytest.raises(DeliveryTestUnsupportedError):
            handler.send_test(context=MagicMock(), method=MagicMock())

    def test_send_test_feature_disabled(self, monkeypatch: pytest.MonkeyPatch, sqlite_engine: Engine) -> None:
        monkeypatch.setattr(
            service_module.FeatureService,
            "get_features",
            lambda _tenant_id, **_kwargs: FeatureModel(human_input_email_delivery_enabled=False),
        )
        handler = EmailDeliveryTestHandler(session_factory=sqlite_engine)
        context = DeliveryTestContext(
            tenant_id="t1", app_id="a1", node_id="n1", node_title="title", rendered_content="content"
        )
        method = EmailDeliveryMethod(config=_make_valid_email_config())

        with pytest.raises(DeliveryTestError, match="Email delivery is not available"):
            handler.send_test(context=context, method=method)

    def test_send_test_mail_not_inited(self, monkeypatch: pytest.MonkeyPatch, sqlite_engine: Engine) -> None:
        monkeypatch.setattr(
            service_module.FeatureService,
            "get_features",
            lambda _id, **_kwargs: FeatureModel(human_input_email_delivery_enabled=True),
        )
        monkeypatch.setattr(service_module.mail, "is_inited", lambda: False)

        handler = EmailDeliveryTestHandler(session_factory=sqlite_engine)
        context = DeliveryTestContext(
            tenant_id="t1", app_id="a1", node_id="n1", node_title="title", rendered_content="content"
        )
        method = EmailDeliveryMethod(config=_make_valid_email_config())

        with pytest.raises(DeliveryTestError, match="Mail client is not initialized."):
            handler.send_test(context=context, method=method)

    def test_send_test_no_recipients(self, monkeypatch: pytest.MonkeyPatch, sqlite_engine: Engine) -> None:
        monkeypatch.setattr(
            service_module.FeatureService,
            "get_features",
            lambda _id, **_kwargs: FeatureModel(human_input_email_delivery_enabled=True),
        )
        monkeypatch.setattr(service_module.mail, "is_inited", lambda: True)

        handler = EmailDeliveryTestHandler(session_factory=sqlite_engine)
        handler._resolve_recipients = MagicMock(return_value=[])

        context = DeliveryTestContext(
            tenant_id="t1", app_id="a1", node_id="n1", node_title="title", rendered_content="content"
        )
        method = EmailDeliveryMethod(config=_make_valid_email_config())

        with pytest.raises(DeliveryTestError, match="No recipients configured"):
            handler.send_test(context=context, method=method)

    def test_send_test_success(self, monkeypatch: pytest.MonkeyPatch, sqlite_engine: Engine) -> None:
        monkeypatch.setattr(
            service_module.FeatureService,
            "get_features",
            lambda _id, **_kwargs: FeatureModel(human_input_email_delivery_enabled=True),
        )
        monkeypatch.setattr(service_module.mail, "is_inited", lambda: True)
        mock_mail_send = MagicMock()
        monkeypatch.setattr(service_module.mail, "send", mock_mail_send)
        monkeypatch.setattr(service_module, "render_email_template", lambda t, s: f"RENDERED_{t}")

        handler = EmailDeliveryTestHandler(session_factory=sqlite_engine)
        handler._resolve_recipients = MagicMock(return_value=["test@example.com"])

        variable_pool = VariablePool()
        context = DeliveryTestContext(
            tenant_id="t1",
            app_id="a1",
            node_id="n1",
            node_title="title",
            rendered_content="content",
            variable_pool=variable_pool,
            recipients=[DeliveryTestEmailRecipient(email="test@example.com", form_token="token123")],
        )

        method = EmailDeliveryMethod(config=_make_valid_email_config())

        result = handler.send_test(context=context, method=method)

        assert result.status == DeliveryTestStatus.OK
        assert result.delivered_to == ["test@example.com"]
        mock_mail_send.assert_called_once()
        args, kwargs = mock_mail_send.call_args
        assert kwargs["to"] == "test@example.com"
        assert "RENDERED_Subj" in kwargs["subject"]

    def test_send_test_sanitizes_subject(self, monkeypatch: pytest.MonkeyPatch, sqlite_engine: Engine) -> None:
        monkeypatch.setattr(
            service_module.FeatureService,
            "get_features",
            lambda _id, **_kwargs: FeatureModel(human_input_email_delivery_enabled=True),
        )
        monkeypatch.setattr(service_module.mail, "is_inited", lambda: True)
        mock_mail_send = MagicMock()
        monkeypatch.setattr(service_module.mail, "send", mock_mail_send)
        monkeypatch.setattr(
            service_module,
            "render_email_template",
            lambda template, substitutions: template.replace("{{ recipient_email }}", substitutions["recipient_email"]),
        )

        handler = EmailDeliveryTestHandler(session_factory=sqlite_engine)
        handler._resolve_recipients = MagicMock(return_value=["test@example.com"])

        context = DeliveryTestContext(
            tenant_id="t1",
            app_id="a1",
            node_id="n1",
            node_title="title",
            rendered_content="content",
            recipients=[DeliveryTestEmailRecipient(email="test@example.com", form_token="token123")],
        )
        method = EmailDeliveryMethod(
            config=EmailDeliveryConfig(
                recipients=EmailRecipients(whole_workspace=False, items=[]),
                subject="<b>Notice</b>\r\nBCC:{{ recipient_email }}",
                body="Body",
            )
        )

        handler.send_test(context=context, method=method)

        _, kwargs = mock_mail_send.call_args
        assert kwargs["subject"] == "Notice BCC:test@example.com"

    def test_resolve_recipients_external(self, sqlite_engine: Engine) -> None:
        handler = EmailDeliveryTestHandler(session_factory=sqlite_engine)
        method = EmailDeliveryMethod(
            config=EmailDeliveryConfig(
                recipients=EmailRecipients(
                    items=[ExternalRecipient(email="ext@example.com")], include_bound_group=False
                ),
                subject="",
                body="",
            )
        )
        assert handler._resolve_recipients(tenant_id="t1", method=method) == ["ext@example.com"]

    @pytest.mark.parametrize("sqlite_session", [(Account, TenantAccountJoin)], indirect=True)
    def test_resolve_recipients_member(self, sqlite_engine: Engine, sqlite_session: Session) -> None:
        tenant_id = str(uuid4())
        account = Account(name="Test User", email="member@example.com")
        sqlite_session.add(account)
        sqlite_session.commit()

        join = TenantAccountJoin(tenant_id=tenant_id, account_id=account.id)
        sqlite_session.add(join)
        sqlite_session.commit()

        handler = EmailDeliveryTestHandler(session_factory=sqlite_engine)
        method = EmailDeliveryMethod(
            config=EmailDeliveryConfig(
                recipients=EmailRecipients(items=[MemberRecipient(reference_id=account.id)], include_bound_group=False),
                subject="",
                body="",
            )
        )
        assert handler._resolve_recipients(tenant_id=tenant_id, method=method) == ["member@example.com"]

    @pytest.mark.parametrize("sqlite_session", [(Account, TenantAccountJoin)], indirect=True)
    def test_resolve_recipients_whole_workspace(self, sqlite_engine: Engine, sqlite_session: Session) -> None:
        tenant_id = str(uuid4())
        account1 = Account(name="User 1", email=f"u1-{uuid4()}@example.com")
        account2 = Account(name="User 2", email=f"u2-{uuid4()}@example.com")
        sqlite_session.add_all([account1, account2])
        sqlite_session.commit()

        for acc in [account1, account2]:
            join = TenantAccountJoin(tenant_id=tenant_id, account_id=acc.id)
            sqlite_session.add(join)
        sqlite_session.commit()

        handler = EmailDeliveryTestHandler(session_factory=sqlite_engine)
        method = EmailDeliveryMethod(
            config=EmailDeliveryConfig(
                recipients=EmailRecipients(items=[], include_bound_group=True),
                subject="",
                body="",
            )
        )
        recipients = handler._resolve_recipients(tenant_id=tenant_id, method=method)
        assert set(recipients) == {account1.email, account2.email}

    def test_query_workspace_member_emails_empty_ids(self, sqlite_engine: Engine) -> None:
        handler = EmailDeliveryTestHandler(session_factory=sqlite_engine)
        assert handler._query_workspace_member_emails(tenant_id="t1", user_ids=[]) == {}

    def test_build_substitutions(self):
        context = DeliveryTestContext(
            tenant_id="t1",
            app_id="a1",
            node_id="n1",
            node_title="title",
            rendered_content="content",
            template_vars={"custom": "var"},
            recipients=[DeliveryTestEmailRecipient(email="test@example.com", form_token="token123")],
        )

        with patch.object(dify_config, "APP_WEB_URL", "http://example.com"):
            subs = EmailDeliveryTestHandler._build_substitutions(context=context, recipient_email="test@example.com")

        assert subs["node_title"] == "title"
        assert subs["form_content"] == "content"
        assert subs["recipient_email"] == "test@example.com"
        assert subs["custom"] == "var"
        assert subs["form_token"] == "token123"
        assert "form/token123" in subs["form_link"]

        subs_no_match = EmailDeliveryTestHandler._build_substitutions(
            context=context, recipient_email="other@example.com"
        )
        assert subs_no_match["form_token"] == ""
        assert subs_no_match["form_link"] == ""
