"""
Unit tests for PipelineRunner behavior.
Asserts correct event handling, error propagation, and user invocation logic.
Primary collaborators: PipelineRunner, InvokeFrom, GraphRunFailedEvent, UserFrom, and mocked dependencies.
Cross-references: core.app.apps.pipeline.pipeline_runner, core.app.entities.app_invoke_entities.
"""

"""Unit tests for PipelineRunner behavior.

This module validates core control-flow outcomes for
``core.app.apps.pipeline.pipeline_runner``: app/workflow lookup, graph
initialization guards, invoke-source to user-source resolution, and failed-run
event handling. Invariants asserted here include strict graph-config
validation, correct ``InvokeFrom`` to ``UserFrom`` mapping, and publishing
error paths driven by ``GraphRunFailedEvent`` through mocked collaborators.
Primary collaborators include ``PipelineRunner``,
``core.app.entities.app_invoke_entities.InvokeFrom``, ``GraphRunFailedEvent``,
``UserFrom``, and patched DB/runtime dependencies used by the runner.
"""

from types import SimpleNamespace
from unittest.mock import MagicMock

import pytest
from pytest_mock import MockerFixture

import core.app.apps.pipeline.pipeline_runner as module
from core.app.apps.pipeline.pipeline_runner import PipelineRunner
from core.app.entities.app_invoke_entities import InvokeFrom, UserFrom
from graphon.graph_events import GraphRunFailedEvent


def _build_app_generate_entity() -> SimpleNamespace:
    app_config = SimpleNamespace(app_id="pipe", workflow_id="wf", tenant_id="tenant")
    return SimpleNamespace(
        app_config=app_config,
        invoke_from=InvokeFrom.WEB_APP,
        user_id="user",
        trace_manager=MagicMock(),
        inputs={"input1": "v1"},
        files=[],
        workflow_execution_id="run",
        document_id="doc",
        original_document_id=None,
        batch="batch",
        dataset_id="ds",
        datasource_type="local_file",
        datasource_info={"name": "file"},
        start_node_id="start",
        call_depth=0,
        single_iteration_run=None,
        single_loop_run=None,
    )


def _patch_create_session(mocker: MockerFixture, session: MagicMock, *, events: list[str] | None = None):
    """Patch create_session() to yield ``session`` inside its ``with`` body and ``begin()`` block.

    The runner now obtains short-lived sessions via ``create_session()`` instead of the
    Flask scoped ``db.session``, so tests patch the module-level ``create_session`` and
    hand back a context manager that yields the mock session.
    """
    session_context = MagicMock()

    def enter_session():
        if events is not None:
            events.append("session_enter")
        return session

    def exit_session(*args):
        if events is not None:
            events.append("session_exit")
        return False

    session_context.__enter__.side_effect = enter_session
    session_context.__exit__.side_effect = exit_session
    session.begin.return_value.__enter__.return_value = session
    session.begin.return_value.__exit__.return_value = False
    return mocker.patch.object(module, "create_session", return_value=session_context)


@pytest.fixture
def runner():
    app_generate_entity = _build_app_generate_entity()
    queue_manager = MagicMock()
    variable_loader = MagicMock()
    workflow = MagicMock()
    workflow_execution_repository = MagicMock()
    workflow_node_execution_repository = MagicMock()

    return PipelineRunner(
        application_generate_entity=app_generate_entity,
        queue_manager=queue_manager,
        variable_loader=variable_loader,
        workflow=workflow,
        system_user_id="sys",
        workflow_execution_repository=workflow_execution_repository,
        workflow_node_execution_repository=workflow_node_execution_repository,
    )


def test_get_app_id(runner):
    assert runner._get_app_id() == "pipe"


def test_get_workflow_returns_workflow(runner):
    pipeline = MagicMock(tenant_id="tenant", id="pipe")
    workflow = MagicMock(id="wf")

    session = MagicMock()
    session.scalar.return_value = workflow

    result = runner.get_workflow(session=session, pipeline=pipeline, workflow_id="wf")

    assert result == workflow


def test_init_rag_pipeline_graph_invalid_config(mocker, runner):
    workflow = MagicMock(id="wf", tenant_id="tenant", graph_dict={})

    with pytest.raises(ValueError):
        runner._init_rag_pipeline_graph(workflow=workflow, graph_runtime_state=MagicMock())

    workflow.graph_dict = {"nodes": "bad", "edges": []}
    with pytest.raises(ValueError):
        runner._init_rag_pipeline_graph(workflow=workflow, graph_runtime_state=MagicMock())

    workflow.graph_dict = {"nodes": [], "edges": "bad"}
    with pytest.raises(ValueError):
        runner._init_rag_pipeline_graph(workflow=workflow, graph_runtime_state=MagicMock())


def test_init_rag_pipeline_graph_not_found(mocker, runner):
    workflow = MagicMock(id="wf", tenant_id="tenant", graph_dict={"nodes": [], "edges": []})
    mocker.patch.object(module.Graph, "init", return_value=None)

    with pytest.raises(ValueError):
        runner._init_rag_pipeline_graph(workflow=workflow, graph_runtime_state=MagicMock())


def test_update_document_status_on_failure(mocker, runner):
    document = MagicMock()
    document_ref = MagicMock()

    session = MagicMock()
    _patch_create_session(mocker, session)
    get_document_by_ref = mocker.patch.object(
        module.DatasetRefService,
        "get_document_by_ref",
        return_value=document,
    )

    event = GraphRunFailedEvent(error="boom")

    runner._update_document_status(event, document_ref)

    get_document_by_ref.assert_called_once_with(document_ref, session=session)
    assert document.indexing_status == "error"
    assert document.error == "boom"
    session.add.assert_called_once_with(document)
    session.begin.assert_called_once()
    session.begin.return_value.__enter__.assert_called_once()
    session.begin.return_value.__exit__.assert_called_once()


def test_update_document_status_skips_when_document_not_found(mocker, runner):
    document_ref = MagicMock()
    session = MagicMock()
    _patch_create_session(mocker, session)
    get_document_by_ref = mocker.patch.object(
        module.DatasetRefService,
        "get_document_by_ref",
        return_value=None,
    )

    runner._update_document_status(GraphRunFailedEvent(error="boom"), document_ref)

    get_document_by_ref.assert_called_once_with(document_ref, session=session)
    session.add.assert_not_called()


def test_update_document_status_skips_without_document_ref(mocker, runner):
    create_session = mocker.patch.object(module, "create_session")

    runner._update_document_status(GraphRunFailedEvent(error="boom"), None)

    create_session.assert_not_called()


def test_run_pipeline_not_found(mocker: MockerFixture):
    app_generate_entity = _build_app_generate_entity()
    app_generate_entity.invoke_from = InvokeFrom.WEB_APP
    app_generate_entity.single_iteration_run = None
    app_generate_entity.single_loop_run = None

    session = MagicMock()
    session.get.side_effect = [None, None]
    _patch_create_session(mocker, session)

    runner = PipelineRunner(
        application_generate_entity=app_generate_entity,
        queue_manager=MagicMock(),
        variable_loader=MagicMock(),
        workflow=MagicMock(),
        system_user_id="sys",
        workflow_execution_repository=MagicMock(),
        workflow_node_execution_repository=MagicMock(),
    )

    with pytest.raises(ValueError):
        runner.run()


def test_run_pipeline_from_other_tenant_is_not_found(mocker: MockerFixture, runner: PipelineRunner):
    pipeline = MagicMock(id="pipe", tenant_id="other-tenant")
    session = MagicMock()
    session.get.side_effect = [None, pipeline]
    _patch_create_session(mocker, session)

    with pytest.raises(ValueError, match="Pipeline not found"):
        runner.run()

    pipeline.retrieve_dataset.assert_not_called()


@pytest.mark.parametrize(
    "dataset",
    [
        pytest.param(None, id="missing"),
        pytest.param(SimpleNamespace(id="ds", tenant_id="other-tenant"), id="other-tenant"),
        pytest.param(SimpleNamespace(id="other-dataset", tenant_id="tenant"), id="other-dataset"),
    ],
)
def test_run_rejects_unowned_pipeline_dataset(
    mocker: MockerFixture,
    runner: PipelineRunner,
    dataset: SimpleNamespace | None,
):
    pipeline = MagicMock(id="pipe", tenant_id="tenant")
    pipeline.retrieve_dataset.return_value = dataset
    session = MagicMock()
    session.get.side_effect = [None, pipeline]
    _patch_create_session(mocker, session)
    runner.get_workflow = MagicMock()

    with pytest.raises(ValueError, match="Pipeline dataset not found"):
        runner.run()

    pipeline.retrieve_dataset.assert_called_once_with(session)
    runner.get_workflow.assert_not_called()


def test_run_rejects_document_outside_pipeline_dataset_after_async_boundary(
    mocker: MockerFixture,
    runner: PipelineRunner,
):
    runner.application_generate_entity.document_id = "foreign-doc"
    runner.application_generate_entity.original_document_id = "foreign-doc"
    pipeline = MagicMock(id="pipe", tenant_id="tenant")
    pipeline.retrieve_dataset.return_value = SimpleNamespace(id="ds", tenant_id="tenant")
    session = MagicMock()
    session.get.side_effect = [None, pipeline]
    _patch_create_session(mocker, session)
    runner.get_workflow = MagicMock()
    get_document_by_ref = mocker.patch.object(
        module.DatasetRefService,
        "get_document_by_ref",
        return_value=None,
    )

    with pytest.raises(ValueError, match="Pipeline document not found"):
        runner.run()

    document_ref = get_document_by_ref.call_args.args[0]
    assert document_ref.dataset.tenant_id == "tenant"
    assert document_ref.dataset.dataset_id == "ds"
    assert document_ref.document_id == "foreign-doc"
    get_document_by_ref.assert_called_once_with(document_ref, session=session)
    runner.get_workflow.assert_not_called()


def test_run_rejects_original_document_outside_pipeline_dataset_after_async_boundary(
    mocker: MockerFixture,
    runner: PipelineRunner,
):
    runner.application_generate_entity.document_id = "doc"
    runner.application_generate_entity.original_document_id = "foreign-doc"
    pipeline = MagicMock(id="pipe", tenant_id="tenant")
    pipeline.retrieve_dataset.return_value = SimpleNamespace(id="ds", tenant_id="tenant")
    session = MagicMock()
    session.get.side_effect = [None, pipeline]
    _patch_create_session(mocker, session)
    runner.get_workflow = MagicMock()
    get_document_by_ref = mocker.patch.object(
        module.DatasetRefService,
        "get_document_by_ref",
        side_effect=[SimpleNamespace(id="doc"), None],
    )

    with pytest.raises(ValueError, match="Pipeline original document not found"):
        runner.run()

    document_refs = [call.args[0] for call in get_document_by_ref.call_args_list]
    assert [document_ref.document_id for document_ref in document_refs] == ["doc", "foreign-doc"]
    for document_ref in document_refs:
        assert document_ref.dataset.tenant_id == "tenant"
        assert document_ref.dataset.dataset_id == "ds"
    runner.get_workflow.assert_not_called()


def test_run_workflow_not_initialized(mocker: MockerFixture):
    app_generate_entity = _build_app_generate_entity()

    pipeline = MagicMock(id="pipe", tenant_id="tenant")
    pipeline.retrieve_dataset.return_value = SimpleNamespace(id="ds", tenant_id="tenant")

    session = MagicMock()
    session.get.side_effect = [None, pipeline]
    _patch_create_session(mocker, session)

    runner = PipelineRunner(
        application_generate_entity=app_generate_entity,
        queue_manager=MagicMock(),
        variable_loader=MagicMock(),
        workflow=MagicMock(),
        system_user_id="sys",
        workflow_execution_repository=MagicMock(),
        workflow_node_execution_repository=MagicMock(),
    )
    runner.get_workflow = MagicMock(return_value=None)

    with pytest.raises(ValueError):
        runner.run()


def test_run_single_iteration_path(mocker: MockerFixture):
    app_generate_entity = _build_app_generate_entity()
    app_generate_entity.single_iteration_run = MagicMock()

    pipeline = MagicMock(id="pipe", tenant_id="tenant")
    dataset = SimpleNamespace(id="ds", tenant_id="tenant")
    pipeline.retrieve_dataset.return_value = dataset
    session = MagicMock()
    session.get.return_value = pipeline
    _patch_create_session(mocker, session)

    runner = PipelineRunner(
        application_generate_entity=app_generate_entity,
        queue_manager=MagicMock(),
        variable_loader=MagicMock(),
        workflow=MagicMock(),
        system_user_id="sys",
        workflow_execution_repository=MagicMock(),
        workflow_node_execution_repository=MagicMock(),
    )

    runner._resolve_user_from = MagicMock(return_value=UserFrom.ACCOUNT)
    runner.get_workflow = MagicMock(
        return_value=MagicMock(
            id="wf",
            tenant_id="tenant",
            app_id="pipe",
            graph_dict={},
            type="rag-pipeline",
            version="v1",
        )
    )
    runner._prepare_single_node_execution = MagicMock(return_value=("graph", "pool", "state"))
    runner._update_document_status = MagicMock()
    runner._handle_event = MagicMock()

    dataset_ref = MagicMock()
    document_ref = MagicMock()
    create_dataset_ref = mocker.patch.object(
        module.DatasetRefService,
        "create_dataset_ref",
        return_value=dataset_ref,
    )
    create_document_ref_from_id = mocker.patch.object(
        module.DatasetRefService,
        "create_document_ref_from_id",
        return_value=document_ref,
    )

    event = MagicMock()
    workflow_entry = MagicMock()
    workflow_entry.graph_engine = MagicMock()
    workflow_entry.run.return_value = [event]
    mocker.patch.object(module, "WorkflowEntry", return_value=workflow_entry)

    mocker.patch.object(module, "WorkflowPersistenceLayer", return_value=MagicMock())

    runner.run()

    create_dataset_ref.assert_called_once_with(dataset)
    create_document_ref_from_id.assert_called_once_with(dataset_ref, "doc")
    runner._prepare_single_node_execution.assert_called_once()
    runner._update_document_status.assert_called_once_with(event, document_ref)
    runner._handle_event.assert_called()


def test_run_normal_path_builds_graph(mocker: MockerFixture):
    app_generate_entity = _build_app_generate_entity()

    pipeline = MagicMock(id="pipe", tenant_id="tenant")
    pipeline.retrieve_dataset.return_value = SimpleNamespace(id="ds", tenant_id="tenant")
    end_user = MagicMock(session_id="sess")
    events = []

    session = MagicMock()
    session.get.side_effect = [end_user, pipeline]
    _patch_create_session(mocker, session, events=events)

    workflow = MagicMock(
        id="wf",
        tenant_id="tenant",
        app_id="pipe",
        graph_dict={"nodes": [], "edges": []},
        environment_variables=[],
        rag_pipeline_variables=[{"variable": "input1", "belong_to_node_id": "start"}],
        type="rag-pipeline",
        version="v1",
    )

    runner = PipelineRunner(
        application_generate_entity=app_generate_entity,
        queue_manager=MagicMock(),
        variable_loader=MagicMock(),
        workflow=workflow,
        system_user_id="sys",
        workflow_execution_repository=MagicMock(),
        workflow_node_execution_repository=MagicMock(),
    )

    runner._resolve_user_from = MagicMock(return_value=UserFrom.ACCOUNT)
    runner.get_workflow = MagicMock(return_value=workflow)
    runner._init_rag_pipeline_graph = MagicMock(return_value="graph")
    runner._update_document_status = MagicMock()
    runner._handle_event = MagicMock()

    mocker.patch.object(
        module.RAGPipelineVariable,
        "model_validate",
        return_value=SimpleNamespace(belong_to_node_id="start", variable="input1"),
    )
    mocker.patch.object(module, "RAGPipelineVariableInput", side_effect=lambda **kwargs: SimpleNamespace(**kwargs))

    class FakeVariablePool:
        def add(self, selector, value):
            return None

    mocker.patch.object(module, "VariablePool", return_value=FakeVariablePool())

    workflow_entry = MagicMock()
    workflow_entry.graph_engine = MagicMock()
    workflow_entry.run.side_effect = lambda: events.append("workflow_run") or []
    mocker.patch.object(module, "WorkflowEntry", return_value=workflow_entry)
    mocker.patch.object(module, "WorkflowPersistenceLayer", return_value=MagicMock())

    runner.run()

    assert events == ["session_enter", "session_exit", "workflow_run"]
    runner._init_rag_pipeline_graph.assert_called_once()
