#
#  Copyright 2026 The InfiniFlow Authors. All Rights Reserved.
#
#  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.
#

"""
Unit tests for the Invoke component's header variable interpolation.

These tests exercise the real Invoke._invoke method, verifying that
{variable} placeholders in HTTP header values are resolved via canvas
variable lookup (issue #13277).
"""

import importlib.util
import json
import sys
from pathlib import Path
from types import ModuleType, SimpleNamespace
from unittest.mock import MagicMock

import pytest


def _load_invoke_module(monkeypatch):
    """Load the real Invoke class with monkeypatched stubs that are
    automatically cleaned up after each test."""
    repo_root = Path(__file__).resolve().parents[4]

    # -- lightweight stubs (auto-restored by monkeypatch) --------------------

    quart = ModuleType("quart")
    quart.make_response = lambda *a, **kw: None
    quart.jsonify = lambda *a, **kw: None
    monkeypatch.setitem(sys.modules, "quart", quart)

    pd = ModuleType("pandas")
    pd.DataFrame = type("DataFrame", (), {})
    monkeypatch.setitem(sys.modules, "pandas", pd)

    deepdoc = ModuleType("deepdoc")
    deepdoc.__path__ = []
    monkeypatch.setitem(sys.modules, "deepdoc", deepdoc)
    deepdoc_parser = ModuleType("deepdoc.parser")
    deepdoc_parser.HtmlParser = MagicMock
    monkeypatch.setitem(sys.modules, "deepdoc.parser", deepdoc_parser)
    monkeypatch.setitem(sys.modules, "xgboost", ModuleType("xgboost"))

    # -- common package and submodules ---------------------------------------

    common_pkg = ModuleType("common")
    common_pkg.__path__ = [str(repo_root / "common")]
    monkeypatch.setitem(sys.modules, "common", common_pkg)

    constants = ModuleType("common.constants")

    class _RetCode:
        SUCCESS = 0
        EXCEPTION_ERROR = 100

    constants.RetCode = _RetCode
    monkeypatch.setitem(sys.modules, "common.constants", constants)

    conn_spec = importlib.util.spec_from_file_location("common.connection_utils", repo_root / "common" / "connection_utils.py")
    conn_mod = importlib.util.module_from_spec(conn_spec)
    monkeypatch.setitem(sys.modules, "common.connection_utils", conn_mod)
    conn_spec.loader.exec_module(conn_mod)

    misc_spec = importlib.util.spec_from_file_location("common.misc_utils", repo_root / "common" / "misc_utils.py")
    misc_mod = importlib.util.module_from_spec(misc_spec)
    monkeypatch.setitem(sys.modules, "common.misc_utils", misc_mod)
    misc_spec.loader.exec_module(misc_mod)

    # -- agent package (bare stubs to skip __init__ auto-import) -------------

    agent_pkg = ModuleType("agent")
    agent_pkg.__path__ = [str(repo_root / "agent")]
    monkeypatch.setitem(sys.modules, "agent", agent_pkg)

    agent_settings = ModuleType("agent.settings")
    agent_settings.FLOAT_ZERO = 1e-8
    agent_settings.PARAM_MAXDEPTH = 5
    monkeypatch.setitem(sys.modules, "agent.settings", agent_settings)

    component_pkg = ModuleType("agent.component")
    component_pkg.__path__ = [str(repo_root / "agent" / "component")]
    monkeypatch.setitem(sys.modules, "agent.component", component_pkg)

    # -- load the real base.py and invoke.py ---------------------------------

    base_spec = importlib.util.spec_from_file_location("agent.component.base", repo_root / "agent" / "component" / "base.py")
    base_mod = importlib.util.module_from_spec(base_spec)
    monkeypatch.setitem(sys.modules, "agent.component.base", base_mod)
    base_spec.loader.exec_module(base_mod)

    invoke_spec = importlib.util.spec_from_file_location("agent.component.invoke", repo_root / "agent" / "component" / "invoke.py")
    invoke_mod = importlib.util.module_from_spec(invoke_spec)
    monkeypatch.setitem(sys.modules, "agent.component.invoke", invoke_mod)
    invoke_spec.loader.exec_module(invoke_mod)

    return invoke_mod


def _make_invoke(module, *, url="http://example.com", method="get", headers="", variables=None, proxy="", timeout_sec=60, clean_html=False, datatype="json", variable_values=None):
    """Build an Invoke instance with a mocked canvas."""
    variable_values = variable_values or {}

    canvas = MagicMock()
    canvas.get_variable_value = MagicMock(side_effect=lambda k: variable_values.get(k, ""))
    canvas.is_canceled = MagicMock(return_value=False)

    param = module.InvokeParam.__new__(module.InvokeParam)
    param.url = url
    param.method = method
    param.headers = headers
    param.variables = variables or []
    param.proxy = proxy
    param.timeout = timeout_sec
    param.clean_html = clean_html
    param.datatype = datatype
    param.max_retries = 0
    param.delay_after_error = 0
    param.outputs = {}
    param.inputs = {}

    inst = module.Invoke.__new__(module.Invoke)
    inst._canvas = canvas
    inst._param = param
    inst._id = "invoke_test"

    return inst


@pytest.mark.p2
def test_header_single_variable(monkeypatch):
    module = _load_invoke_module(monkeypatch)
    invoke = _make_invoke(
        module,
        headers=json.dumps({"Authorization": "Bearer {auth_token}"}),
        variable_values={"auth_token": "secret123"},
    )
    mock_get = MagicMock(return_value=SimpleNamespace(text="ok"))
    monkeypatch.setattr(module.requests, "get", mock_get)
    invoke._invoke()
    assert mock_get.call_args[1]["headers"]["Authorization"] == "Bearer secret123"


@pytest.mark.p2
def test_header_multiple_variables(monkeypatch):
    module = _load_invoke_module(monkeypatch)
    invoke = _make_invoke(
        module,
        headers=json.dumps(
            {
                "Authorization": "Bearer {token}",
                "X-Request-Id": "{req_id}",
                "Content-Type": "application/json",
            }
        ),
        variable_values={"token": "tok_abc", "req_id": "id-42"},
    )
    mock_get = MagicMock(return_value=SimpleNamespace(text="ok"))
    monkeypatch.setattr(module.requests, "get", mock_get)
    invoke._invoke()
    h = mock_get.call_args[1]["headers"]
    assert h["Authorization"] == "Bearer tok_abc"
    assert h["X-Request-Id"] == "id-42"
    assert h["Content-Type"] == "application/json"


@pytest.mark.p2
def test_header_no_variables_unchanged(monkeypatch):
    module = _load_invoke_module(monkeypatch)
    invoke = _make_invoke(
        module,
        headers=json.dumps({"Content-Type": "application/json"}),
    )
    mock_get = MagicMock(return_value=SimpleNamespace(text="ok"))
    monkeypatch.setattr(module.requests, "get", mock_get)
    invoke._invoke()
    assert mock_get.call_args[1]["headers"]["Content-Type"] == "application/json"


@pytest.mark.p2
def test_header_empty(monkeypatch):
    module = _load_invoke_module(monkeypatch)
    invoke = _make_invoke(module, headers="")
    mock_get = MagicMock(return_value=SimpleNamespace(text="ok"))
    monkeypatch.setattr(module.requests, "get", mock_get)
    invoke._invoke()
    assert mock_get.call_args[1]["headers"] == {}


@pytest.mark.p2
def test_header_component_ref_variable(monkeypatch):
    module = _load_invoke_module(monkeypatch)
    invoke = _make_invoke(
        module,
        headers=json.dumps({"Authorization": "Bearer {begin@token}"}),
        variable_values={"begin@token": "my_token"},
    )
    mock_get = MagicMock(return_value=SimpleNamespace(text="ok"))
    monkeypatch.setattr(module.requests, "get", mock_get)
    invoke._invoke()
    assert mock_get.call_args[1]["headers"]["Authorization"] == "Bearer my_token"


@pytest.mark.p2
def test_header_env_variable(monkeypatch):
    module = _load_invoke_module(monkeypatch)
    invoke = _make_invoke(
        module,
        headers=json.dumps({"Authorization": "Bearer {env.api_key}"}),
        variable_values={"env.api_key": "env_secret"},
    )
    mock_get = MagicMock(return_value=SimpleNamespace(text="ok"))
    monkeypatch.setattr(module.requests, "get", mock_get)
    invoke._invoke()
    assert mock_get.call_args[1]["headers"]["Authorization"] == "Bearer env_secret"


@pytest.mark.p2
def test_header_missing_variable_becomes_empty(monkeypatch):
    module = _load_invoke_module(monkeypatch)
    invoke = _make_invoke(
        module,
        headers=json.dumps({"Authorization": "Bearer {nonexistent}"}),
        variable_values={},
    )
    mock_get = MagicMock(return_value=SimpleNamespace(text="ok"))
    monkeypatch.setattr(module.requests, "get", mock_get)
    invoke._invoke()
    assert mock_get.call_args[1]["headers"]["Authorization"] == "Bearer "


@pytest.mark.p2
def test_header_variable_with_post(monkeypatch):
    module = _load_invoke_module(monkeypatch)
    invoke = _make_invoke(
        module,
        method="post",
        headers=json.dumps({"Authorization": "Bearer {token}"}),
        variable_values={"token": "post_token"},
    )
    mock_post = MagicMock(return_value=SimpleNamespace(text="ok"))
    monkeypatch.setattr(module.requests, "post", mock_post)
    invoke._invoke()
    assert mock_post.call_args[1]["headers"]["Authorization"] == "Bearer post_token"


@pytest.mark.p2
def test_header_variable_with_put(monkeypatch):
    module = _load_invoke_module(monkeypatch)
    invoke = _make_invoke(
        module,
        method="put",
        headers=json.dumps({"Authorization": "Bearer {token}"}),
        variable_values={"token": "put_token"},
    )
    mock_put = MagicMock(return_value=SimpleNamespace(text="ok"))
    monkeypatch.setattr(module.requests, "put", mock_put)
    invoke._invoke()
    assert mock_put.call_args[1]["headers"]["Authorization"] == "Bearer put_token"


@pytest.mark.p2
def test_invoke_blocks_loopback_url_with_ssrf_guard(monkeypatch):
    """Invoke must use the shared SSRF guard before requests.* (issue Invoke SSRF)."""
    module = _load_invoke_module(monkeypatch)
    invoke = _make_invoke(module, url="http://127.0.0.1:8123/api")
    mock_get = MagicMock(return_value=SimpleNamespace(text="ok"))
    monkeypatch.setattr(module.requests, "get", mock_get)
    result = invoke._invoke()
    mock_get.assert_not_called()
    assert result == "Http request error: URL not valid"
    assert invoke.output("_ERROR") == "URL not valid"


@pytest.mark.p2
def test_invoke_blocks_metadata_ip(monkeypatch):
    module = _load_invoke_module(monkeypatch)
    invoke = _make_invoke(module, url="http://169.254.169.254/latest/meta-data/")
    mock_get = MagicMock(return_value=SimpleNamespace(text="should not run"))
    monkeypatch.setattr(module.requests, "get", mock_get)
    result = invoke._invoke()
    mock_get.assert_not_called()
    assert "URL not valid" in result
    assert invoke.output("_ERROR") == "URL not valid"


@pytest.mark.p2
def test_invoke_url_without_scheme_gets_scheme_then_validated(monkeypatch):
    """Bare hostnames are prefixed with http:// before SSRF validation."""
    module = _load_invoke_module(monkeypatch)
    invoke = _make_invoke(module, url="127.0.0.1:9380/")
    mock_get = MagicMock(return_value=SimpleNamespace(text="should not run"))
    monkeypatch.setattr(module.requests, "get", mock_get)
    result = invoke._invoke()
    mock_get.assert_not_called()
    assert "URL not valid" in result


@pytest.mark.p2
def test_invoke_blocks_loopback_proxy(monkeypatch):
    module = _load_invoke_module(monkeypatch)
    invoke = _make_invoke(module, url="http://example.com", proxy="http://127.0.0.1:8080")
    mock_get = MagicMock(return_value=SimpleNamespace(text="should not run"))
    monkeypatch.setattr(module.requests, "get", mock_get)
    result = invoke._invoke()
    mock_get.assert_not_called()
    assert "URL not valid" in result
    assert invoke.output("_ERROR") == "URL not valid"


@pytest.mark.p2
def test_invoke_disables_redirect_following(monkeypatch):
    module = _load_invoke_module(monkeypatch)
    invoke = _make_invoke(module, url="http://example.com")
    mock_get = MagicMock(return_value=SimpleNamespace(text="ok"))
    monkeypatch.setattr(module.requests, "get", mock_get)
    invoke._invoke()
    assert mock_get.call_args[1]["allow_redirects"] is False


@pytest.mark.p2
def test_invoke_pins_dns_and_disables_redirects(monkeypatch):
    module = _load_invoke_module(monkeypatch)
    invoke = _make_invoke(module, url="http://example.com")
    pin_ctx = MagicMock()
    pin_ctx.__enter__ = MagicMock(return_value=None)
    pin_ctx.__exit__ = MagicMock(return_value=None)
    monkeypatch.setattr(module, "assert_url_is_safe", MagicMock(return_value=("example.com", "93.184.216.34")))
    monkeypatch.setattr(module, "pin_dns", MagicMock(return_value=pin_ctx))
    mock_get = MagicMock(return_value=SimpleNamespace(text="ok"))
    monkeypatch.setattr(module.requests, "get", mock_get)

    invoke._invoke()

    module.pin_dns.assert_called_once_with("example.com", "93.184.216.34")
    assert mock_get.call_args[1]["allow_redirects"] is False
