replypier/backend/tests/api/test_widget_chat_workflow.py

"""
Copyright 2024-2026 ReplyPier

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""

import pytest
from unittest.mock import Mock, MagicMock, patch, AsyncMock
from app.models.agent import Agent, AgentType
from app.models.chat_history import ChatHistory
from app.models.workflow import Workflow, WorkflowStatus
from app.models.workflow_node import WorkflowNode, NodeType
from app.models.organization import Organization
from app.models.user import User
from uuid import UUID, uuid4
from app.services.conversation_token import VERIFIED_SCOPE


@pytest.fixture
def test_organization(db):
    """Create a test organization"""
    organization = Organization(
        id=uuid4(),
        name="Test Organization",
        domain="test.com"
    )
    db.add(organization)
    db.commit()
    db.refresh(organization)
    return organization


@pytest.fixture
def test_user(db, test_organization):
    """Create a test user"""
    user = User(
        id=uuid4(),
        email="test@example.com",
        hashed_password="hashed_password",
        is_active=True,
        organization_id=test_organization.id,
        full_name="Test User"
    )
    db.add(user)
    db.commit()
    db.refresh(user)
    return user


@pytest.fixture
def test_agent(db, test_organization):
    """Create a test agent with workflow enabled"""
    agent = Agent(
        id=uuid4(),
        name="Test Agent",
        display_name="Test Agent",
        description="Test agent with workflow",
        agent_type=AgentType.CUSTOMER_SUPPORT,
        instructions=["Test instruction"],
        is_active=True,
        organization_id=test_organization.id,
        use_workflow=True
    )
    db.add(agent)
    db.commit()
    db.refresh(agent)
    return agent


@pytest.fixture
def test_workflow(db, test_organization, test_agent, test_user):
    """Create a test workflow"""
    workflow = Workflow(
        id=uuid4(),
        name="Test Workflow",
        description="Test workflow",
        status=WorkflowStatus.PUBLISHED,
        version=1,
        is_template=False,
        default_language="en",
        canvas_data={
            "nodes": [
                {
                    "id": "start",
                    "type": "start",
                    "data": {"label": "Start"}
                }
            ]
        },
        settings={},
        organization_id=test_organization.id,
        agent_id=test_agent.id,
        created_by=test_user.id
    )
    db.add(workflow)
    db.commit()
    db.refresh(workflow)
    
    # Update agent to use this workflow
    test_agent.active_workflow_id = workflow.id
    db.commit()
    
    return workflow


class TestWidgetChatWorkflow:
    
    def test_agent_has_workflow_enabled(self, test_agent, test_workflow):
        """Test that the agent has workflow properly configured"""
        assert test_agent.use_workflow is True
        assert test_agent.active_workflow_id == test_workflow.id
    
    def test_workflow_is_published(self, test_workflow):
        """Test that the workflow is in published status"""
        assert test_workflow.status == WorkflowStatus.PUBLISHED
        assert test_workflow.canvas_data is not None
        assert "nodes" in test_workflow.canvas_data
    
    @pytest.mark.asyncio
    async def test_workflow_execution_service_mock(self, test_agent, test_workflow):
        """Test that WorkflowExecutionService can be mocked properly"""
        from app.services.workflow_execution import WorkflowExecutionService
        
        with patch.object(WorkflowExecutionService, '__init__', return_value=None), \
             patch.object(WorkflowExecutionService, 'execute_workflow', new_callable=AsyncMock) as mock_execute:
            
            mock_execute.return_value = {
                "success": True,
                "message": "Workflow executed",
                "next_node": None
            }
            
            # Create service instance
            service = WorkflowExecutionService(None)
            
            # Call execute_workflow
            result = await service.execute_workflow(
                session_id="test_session",
                message="Hello",
                agent_id=test_agent.id,
                workflow_id=test_workflow.id
            )
            
            # Verify the mock was called and returned expected result
            mock_execute.assert_called_once()
            assert result["success"] is True
            assert result["message"] == "Workflow executed"
    
    def test_workflow_data_structure(self, test_workflow):
        """Test that workflow has proper data structure"""
        assert isinstance(test_workflow.canvas_data, dict)
        assert "nodes" in test_workflow.canvas_data
        assert len(test_workflow.canvas_data["nodes"]) >= 1
        
        # Check first node structure
        first_node = test_workflow.canvas_data["nodes"][0]
        assert "id" in first_node
        assert "type" in first_node
        assert "data" in first_node
    
    def test_agent_workflow_relationship(self, test_agent, test_workflow):
        """Test the relationship between agent and workflow"""
        assert test_agent.active_workflow_id == test_workflow.id
        assert test_workflow.agent_id == test_agent.id
        assert test_agent.organization_id == test_workflow.organization_id


@pytest.fixture
def one_message_session(db, test_workflow, test_agent, test_customer, test_ai_config, monkeypatch):
    """A widget session on a workflow that is a single MESSAGE node."""
    from app.api import widget_chat
    from app.services import message_delivery
    from app.repositories.session_to_agent import SessionToAgentRepository

    db.add(WorkflowNode(
        id=uuid4(),
        workflow_id=test_workflow.id,
        node_type=NodeType.MESSAGE,
        name="Greeting",
        config={"message_text": "Hello from the workflow"},
    ))
    db.commit()

    session_id = uuid4()
    active_session = SessionToAgentRepository(db).create_session(
        session_id=session_id,
        agent_id=test_agent.id,
        customer_id=test_customer.id,
        organization_id=test_agent.organization_id,
    )
    assert active_session.workflow_id == test_workflow.id

    sio = MagicMock()
    sio.emit = AsyncMock()
    sio.get_session = AsyncMock(return_value={
        "widget_id": str(uuid4()),
        "org_id": str(test_agent.organization_id),
        "agent_id": str(test_agent.id),
        "customer_id": str(test_customer.id),
        "session_id": str(session_id),
        "ai_config": test_ai_config,
        # A keyless session is resumed by a token the embedding app vouched for.
        "conversation_scope": VERIFIED_SCOPE,
    })
    monkeypatch.setattr(widget_chat, "sio", sio)
    monkeypatch.setattr(message_delivery, "sio", sio)
    monkeypatch.setattr(widget_chat, "get_db", lambda: iter([db]))
    monkeypatch.setattr(widget_chat, "notify_new_chat", AsyncMock())
    return sio, str(session_id)


def _chat_responses(sio):
    return [c.args[1]["message"] for c in sio.emit.call_args_list if c.args[0] == "chat_response"]


def _stored_bot_messages(db, session_id):
    return [m.message for m in db.query(ChatHistory).filter(
        ChatHistory.session_id == UUID(session_id), ChatHistory.message_type == "bot").all()]


class TestOneMessageWorkflowIsSentOnce:
    """A workflow ending on a MESSAGE node used to send and store that message
    twice: once as an intermediate message and once as the final reply."""

    @pytest.mark.asyncio
    async def test_on_load(self, db, one_message_session):
        from app.api import widget_chat
        sio, session_id = one_message_session

        await widget_chat.handle_get_workflow_state("test_sid")

        assert _chat_responses(sio) == ["Hello from the workflow"]
        states = [c.args[1] for c in sio.emit.call_args_list if c.args[0] == "workflow_state"]
        assert [(s["type"], s["message"]) for s in states] == [("message", "Hello from the workflow")]
        assert _stored_bot_messages(db, session_id) == ["Hello from the workflow"]

    @pytest.mark.asyncio
    async def test_on_first_message(self, db, one_message_session):
        from app.api import widget_chat
        sio, session_id = one_message_session

        await widget_chat.handle_widget_chat("test_sid", {"message": "hi"})

        assert _chat_responses(sio) == ["Hello from the workflow"]
        assert _stored_bot_messages(db, session_id) == ["Hello from the workflow"]

# Synthetic values are deliberately recognizable in a rendered log message.
_SYNTHETIC_KEY = "sk-f01-synthetic-key-do-not-log"
_SYNTHETIC_NOTE = "f01-private-form-answer"
_SYNTHETIC_EMAIL = "f01-private@example.test"
_SYNTHETIC_STATE = "f01-private-previous-state"


def _rendered_log(caplog):
    import logging

    formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
    return "\n".join(formatter.format(record) for record in caplog.records)


@pytest.fixture
def form_logging_session(db, test_workflow, test_agent, test_customer, test_ai_config, monkeypatch):
    """A real SQLite session: FORM -> CONDITION -> GUARDRAILS -> MESSAGE -> LLM."""
    from app.api import widget_chat
    from app.core.security import encrypt_api_key
    from app.models.session_to_agent import SessionToAgent, SessionStatus
    from app.models.workflow_connection import WorkflowConnection
    from app.services import socket_rate_limit

    form = WorkflowNode(id=uuid4(), workflow_id=test_workflow.id, node_type=NodeType.FORM,
                        name="Contact Form", config={"form_fields": [
                            {"name": "email", "type": "email", "label": "Email"},
                            {"name": "note", "type": "text", "label": "Note"},
                        ]})
    condition = WorkflowNode(id=uuid4(), workflow_id=test_workflow.id, node_type=NodeType.CONDITION,
                             name="Condition", config={"condition_expression": "{{contact_form_note}} > 1"})
    guard = WorkflowNode(id=uuid4(), workflow_id=test_workflow.id, node_type=NodeType.GUARDRAILS,
                         name="Guard", config={"enabled_guardrails": ["pii"], "pii_action": "redact",
                                               "text_source": "contact_form_note"})
    message = WorkflowNode(id=uuid4(), workflow_id=test_workflow.id, node_type=NodeType.MESSAGE,
                           name="Message", config={"message_text": "Noted {{contact_form_note}}"})
    model = WorkflowNode(id=uuid4(), workflow_id=test_workflow.id, node_type=NodeType.LLM,
                         name="Model", config={"system_prompt": "Use {{contact_form_email}}",
                                               "exit_condition": "continuous_execution"})
    db.add_all([form, condition, guard, message, model])
    for source, target, label in [(form, condition, None), (condition, guard, "false"),
                                  (guard, message, "pass"), (message, model, None)]:
        db.add(WorkflowConnection(id=uuid4(), workflow_id=test_workflow.id,
                                  source_node_id=source.id, target_node_id=target.id, label=label))
    test_customer.email = "anonymous@noemail.com"
    test_ai_config.encrypted_api_key = encrypt_api_key(_SYNTHETIC_KEY)
    test_ai_config.settings = {"base_url": "https://model.example.test/v1"}
    active = SessionToAgent(session_id=uuid4(), organization_id=test_agent.organization_id,
                            customer_id=test_customer.id, agent_id=test_agent.id,
                            workflow_id=test_workflow.id, current_node_id=form.id,
                            workflow_state={"form_state": "waiting", "variables": {
                                "earlier": {"private": _SYNTHETIC_STATE}}},
                            workflow_history=[], status=SessionStatus.OPEN)
    db.add(active)
    db.commit()
    sio = MagicMock()
    sio.emit = AsyncMock()
    sio.get_session = AsyncMock(return_value={
        "widget_id": str(uuid4()), "org_id": str(test_agent.organization_id),
        "agent_id": str(test_agent.id), "customer_id": str(test_customer.id),
        "session_id": str(active.session_id), "ai_config": test_ai_config,
        "conversation_scope": VERIFIED_SCOPE, "source": "widget",
    })
    monkeypatch.setattr(widget_chat, "sio", sio)
    monkeypatch.setattr(socket_rate_limit, "sio", sio)
    monkeypatch.setattr(widget_chat, "get_db", lambda: iter([db]))
    return active, form, model, sio


@pytest.mark.asyncio
@pytest.mark.parametrize("duplicate_email", [False, True])
async def test_form_submission_keeps_secrets_out_of_rendered_logs(
    db, test_organization, test_customer, test_ai_config, form_logging_session, duplicate_email,
    caplog, monkeypatch,
):
    import logging
    from app.api import widget_chat
    from app.models.customer import Customer
    from app.models.schemas.chat import ChatResponse
    from app.services import workflow_execution

    active, form, model, sio = form_logging_session
    if duplicate_email:
        db.add(Customer(id=uuid4(), organization_id=test_organization.id,
                        email=_SYNTHETIC_EMAIL, full_name="Other"))
        db.commit()
    agent = MagicMock()
    agent._get_llm_response_only = AsyncMock(return_value=ChatResponse(
        message=f"Reply {_SYNTHETIC_NOTE}", transfer_to_human=False, end_chat=False,
        request_rating=False, create_ticket=False,
    ))
    agent.safe_cleanup_mcp_tools = AsyncMock()
    create_agent = AsyncMock(return_value=agent)
    monkeypatch.setattr(workflow_execution.ChatAgent, "create_async", create_agent)
    session_id = str(active.session_id)
    caplog.set_level(logging.DEBUG)
    caplog.clear()

    await widget_chat.handle_form_submission("sid-f01", {"form_data": {
        "email": _SYNTHETIC_EMAIL, "note": _SYNTHETIC_NOTE,
    }})

    assert _chat_responses(sio) == [f"Noted {_SYNTHETIC_NOTE}", f"Reply {_SYNTHETIC_NOTE}"]
    assert sorted(_stored_bot_messages(db, session_id)) == sorted(_chat_responses(sio))
    assert any(c.args[0] == "form_submitted" and c.args[1]["success"] for c in sio.emit.call_args_list)
    db.refresh(active)
    db.refresh(test_customer)
    assert test_customer.email == ("anonymous@noemail.com" if duplicate_email else _SYNTHETIC_EMAIL)
    assert active.current_node_id == model.id
    assert active.workflow_state["variables"][str(form.id)] == {
        "contact_form_email": _SYNTHETIC_EMAIL, "contact_form_note": _SYNTHETIC_NOTE,
    }
    assert active.workflow_history[-1]["data"] == {"email": _SYNTHETIC_EMAIL, "note": _SYNTHETIC_NOTE}
    kwargs = create_agent.await_args.kwargs
    assert kwargs["api_key"] == _SYNTHETIC_KEY
    assert kwargs["model_name"] == test_ai_config.model_name
    assert kwargs["model_type"] == test_ai_config.model_type
    assert kwargs["custom_system_prompt"] == f"Use {_SYNTHETIC_EMAIL}"
    assert kwargs["session_id"] == session_id
    assert kwargs["base_url"] == "https://model.example.test/v1"
    assert kwargs["org_id"] == str(test_organization.id)
    assert kwargs["customer_id"] == str(test_customer.id)
    assert kwargs["agent_id"] == str(active.agent_id)
    assert kwargs["source"] == "widget"
    assert _SYNTHETIC_NOTE in agent._get_llm_response_only.await_args.kwargs["message"]
    agent.safe_cleanup_mcp_tools.assert_awaited_once()

    logs = _rendered_log(caplog)
    for secret in (_SYNTHETIC_KEY, _SYNTHETIC_NOTE, _SYNTHETIC_EMAIL, _SYNTHETIC_STATE):
        assert secret not in logs
    assert session_id in logs and str(form.id) in logs and str(model.id) in logs
    assert "form" in logs.lower()


@pytest.mark.asyncio
@pytest.mark.parametrize("case", ["missing", "unexpected", "exception"])
async def test_form_submission_failure_logs_identifiers_not_sensitive_inputs(case, caplog):
    import logging
    from app.services.workflow_execution import WorkflowExecutionService

    service = WorkflowExecutionService(Mock())
    node_id = uuid4()
    session_id = "f01-failure-session"
    workflow_id = uuid4()
    error = f"database failed while handling {_SYNTHETIC_NOTE} and {_SYNTHETIC_KEY}"
    service.session_repo.get_session = Mock(
        side_effect=RuntimeError(error) if case == "exception" else None,
        return_value=Mock(workflow_state={"form_state": "not-waiting", "private": _SYNTHETIC_STATE},
                          current_node_id=node_id) if case == "unexpected" else None,
    )
    service.execute_workflow = AsyncMock()
    caplog.set_level(logging.DEBUG)
    caplog.clear()
    result = await service.submit_form(session_id, {"note": _SYNTHETIC_NOTE}, workflow_id,
                                       api_key=_SYNTHETIC_KEY)
    expected = {"missing": "Session not found", "unexpected": "No form submission expected",
                "exception": error}[case]
    assert result.success is False and result.error == expected
    service.execute_workflow.assert_not_awaited()
    logs = _rendered_log(caplog)
    for secret in (_SYNTHETIC_KEY, _SYNTHETIC_NOTE, _SYNTHETIC_STATE):
        assert secret not in logs
    assert session_id in logs and str(workflow_id) in logs
    assert any(record.levelno >= logging.WARNING for record in caplog.records)
    if case == "unexpected":
        assert str(node_id) in logs
    if case == "exception":
        assert "RuntimeError" in logs


def test_landing_page_interpolation_does_not_log_submitted_values(caplog):
    import logging
    from app.services.workflow_execution import WorkflowExecutionService

    service = WorkflowExecutionService(Mock())
    node = Mock(spec=WorkflowNode)
    node.id = uuid4()
    node.config = {"landing_page_heading": "Hello", "landing_page_content": "Saved {{note}}"}
    state = {"variables": {"form": {"note": _SYNTHETIC_NOTE}}}
    caplog.set_level(logging.DEBUG)
    caplog.clear()
    result = service._execute_landing_page_node(node, state)
    assert result.landing_page_data == {"heading": "Hello", "content": f"Saved {_SYNTHETIC_NOTE}"}
    logs = _rendered_log(caplog)
    assert _SYNTHETIC_NOTE not in logs
    assert str(node.id) in logs


@pytest.mark.parametrize("operation", ["update", "history"])
def test_form_state_persistence_failure_logs_no_database_parameters(
    db, form_logging_session, caplog, monkeypatch, operation,
):
    import logging
    from app.repositories.session_to_agent import SessionToAgentRepository

    active, form, _, _ = form_logging_session
    session_id = str(active.session_id)
    repository = SessionToAgentRepository(db)
    monkeypatch.setattr(db, "commit", Mock(side_effect=RuntimeError(f"database error: {_SYNTHETIC_NOTE}")))
    caplog.set_level(logging.DEBUG)
    caplog.clear()
    if operation == "update":
        result = repository.update_workflow_state(session_id, form.id, {"secret": _SYNTHETIC_NOTE})
    else:
        result = repository.add_workflow_history_entry(session_id, form.id, "form_submission", {"note": _SYNTHETIC_NOTE})
    assert result is False
    logs = _rendered_log(caplog)
    assert _SYNTHETIC_NOTE not in logs
    assert session_id in logs and "RuntimeError" in logs


@pytest.mark.asyncio
async def test_model_failure_does_not_log_form_data_or_api_key(
    db, form_logging_session, caplog, monkeypatch,
):
    import logging
    from app.api import widget_chat
    from app.services import workflow_execution

    active, form, model, sio = form_logging_session
    error = f"model refused {_SYNTHETIC_NOTE} with {_SYNTHETIC_KEY}"
    monkeypatch.setattr(workflow_execution.ChatAgent, "create_async", AsyncMock(side_effect=RuntimeError(error)))
    session_id = str(active.session_id)
    caplog.set_level(logging.DEBUG)
    caplog.clear()
    await widget_chat.handle_form_submission("sid-f01", {"form_data": {
        "email": _SYNTHETIC_EMAIL, "note": _SYNTHETIC_NOTE,
    }})
    errors = [c.args[1] for c in sio.emit.call_args_list if c.args[0] == "error"]
    assert errors == [{"error": error, "type": "form_error"}]
    assert not any(c.args[0] == "form_submitted" for c in sio.emit.call_args_list)
    logs = _rendered_log(caplog)
    for secret in (_SYNTHETIC_KEY, _SYNTHETIC_NOTE, _SYNTHETIC_EMAIL, _SYNTHETIC_STATE):
        assert secret not in logs
    assert session_id in logs and str(form.id) in logs and str(model.id) in logs
    assert "RuntimeError" in logs


def test_form_email_lookup_database_failure_does_not_log_parameters(caplog):
    import logging
    from app.repositories.customer import CustomerRepository

    db = Mock()
    db.query.side_effect = RuntimeError(f"database lookup failed: {_SYNTHETIC_EMAIL}")
    caplog.set_level(logging.DEBUG)
    caplog.clear()
    assert CustomerRepository(db).get_customer_by_email(_SYNTHETIC_EMAIL, uuid4()) is None
    logs = _rendered_log(caplog)
    assert _SYNTHETIC_EMAIL not in logs and "RuntimeError" in logs


@pytest.mark.asyncio
@pytest.mark.parametrize("error_type", [ValueError, RuntimeError])
async def test_widget_form_exception_logs_type_without_exception_contents(
    form_logging_session, caplog, monkeypatch, error_type,
):
    import logging
    from app.api import widget_chat

    _, _, _, sio = form_logging_session
    detail = f"failed with {_SYNTHETIC_KEY} and {_SYNTHETIC_NOTE}"
    monkeypatch.setattr(widget_chat, "_own_session", Mock(side_effect=error_type(detail)))
    caplog.set_level(logging.DEBUG)
    caplog.clear()
    await widget_chat.handle_form_submission("sid-f01", {"form_data": {"note": _SYNTHETIC_NOTE}})

    emitted = [call.args[1] for call in sio.emit.call_args_list if call.args[0] == "error"]
    assert emitted == [{"error": detail if error_type is ValueError else "Failed to submit form",
                        "type": "form_error"}]
    logs = _rendered_log(caplog)
    assert _SYNTHETIC_KEY not in logs and _SYNTHETIC_NOTE not in logs
    assert "sid-f01" in logs and error_type.__name__ in logs