#
#  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.
#

import importlib.util
import sys
from pathlib import Path
from types import ModuleType, SimpleNamespace

import pytest


class _DummyManager:
    def route(self, *_args, **_kwargs):
        def decorator(func):
            return func

        return decorator


class _ExprField:
    def __init__(self, name):
        self.name = name

    def __eq__(self, other):
        return (self.name, other)


class _DummyAPITokenModel:
    tenant_id = _ExprField("tenant_id")
    token = _ExprField("token")


@pytest.fixture(scope="session")
def auth():
    return "unit-auth"


@pytest.fixture(scope="session", autouse=True)
def set_tenant_info():
    return None


def _load_system_module(monkeypatch):
    repo_root = Path(__file__).resolve().parents[4]

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

    apps_mod = ModuleType("api.apps")
    apps_mod.__path__ = [str(repo_root / "api" / "apps")]
    apps_mod.login_required = lambda fn: fn
    apps_mod.current_user = SimpleNamespace(id="user-1")
    monkeypatch.setitem(sys.modules, "api.apps", apps_mod)

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

    settings_mod = ModuleType("common.settings")
    settings_mod.docStoreConn = SimpleNamespace(health=lambda: {"type": "doc", "status": "green"})
    settings_mod.STORAGE_IMPL = SimpleNamespace(health=lambda: True)
    settings_mod.STORAGE_IMPL_TYPE = "MINIO"
    settings_mod.DATABASE_TYPE = "MYSQL"
    settings_mod.REGISTER_ENABLED = True
    settings_mod.DISABLE_PASSWORD_LOGIN = False
    common_pkg.settings = settings_mod
    monkeypatch.setitem(sys.modules, "common.settings", settings_mod)

    versions_mod = ModuleType("common.versions")
    versions_mod.get_ragflow_version = lambda: "0.0.0-unit"
    monkeypatch.setitem(sys.modules, "common.versions", versions_mod)

    time_utils_mod = ModuleType("common.time_utils")
    time_utils_mod.current_timestamp = lambda: 111
    time_utils_mod.datetime_format = lambda _dt: "2026-01-01 00:00:00"
    monkeypatch.setitem(sys.modules, "common.time_utils", time_utils_mod)

    api_utils_mod = ModuleType("api.utils.api_utils")
    api_utils_mod.get_json_result = lambda data=None, message="success", code=0: {
        "code": code,
        "message": message,
        "data": data,
    }
    api_utils_mod.get_data_error_result = lambda message="", code=102, data=None: {
        "code": code,
        "message": message,
        "data": data,
    }
    api_utils_mod.server_error_response = lambda exc: {
        "code": 100,
        "message": repr(exc),
        "data": None,
    }
    api_utils_mod.generate_confirmation_token = lambda: "ragflow-abcdefghijklmnopqrstuvwxyz0123456789"
    monkeypatch.setitem(sys.modules, "api.utils.api_utils", api_utils_mod)

    api_service_mod = ModuleType("api.db.services.api_service")
    api_service_mod.APITokenService = SimpleNamespace(
        save=lambda **_kwargs: True,
        query=lambda **_kwargs: [],
        filter_update=lambda *_args, **_kwargs: True,
        filter_delete=lambda *_args, **_kwargs: True,
    )
    monkeypatch.setitem(sys.modules, "api.db.services.api_service", api_service_mod)

    kb_service_mod = ModuleType("api.db.services.knowledgebase_service")
    kb_service_mod.KnowledgebaseService = SimpleNamespace(get_by_id=lambda _kb_id: True)
    monkeypatch.setitem(sys.modules, "api.db.services.knowledgebase_service", kb_service_mod)

    user_service_mod = ModuleType("api.db.services.user_service")
    user_service_mod.UserTenantService = SimpleNamespace(query=lambda **_kwargs: [SimpleNamespace(role="owner", tenant_id="tenant-1")])
    monkeypatch.setitem(sys.modules, "api.db.services.user_service", user_service_mod)

    db_models_mod = ModuleType("api.db.db_models")
    db_models_mod.APIToken = _DummyAPITokenModel
    monkeypatch.setitem(sys.modules, "api.db.db_models", db_models_mod)

    rag_pkg = ModuleType("rag")
    rag_pkg.__path__ = []
    monkeypatch.setitem(sys.modules, "rag", rag_pkg)

    rag_utils_pkg = ModuleType("rag.utils")
    rag_utils_pkg.__path__ = []
    monkeypatch.setitem(sys.modules, "rag.utils", rag_utils_pkg)

    redis_mod = ModuleType("rag.utils.redis_conn")
    redis_mod.REDIS_CONN = SimpleNamespace(
        health=lambda: True,
        smembers=lambda *_args, **_kwargs: set(),
        zrangebyscore=lambda *_args, **_kwargs: [],
    )
    monkeypatch.setitem(sys.modules, "rag.utils.redis_conn", redis_mod)

    health_utils_mod = ModuleType("api.utils.health_utils")
    health_utils_mod.run_health_checks = lambda: ({"status": "ok"}, True)
    health_utils_mod.get_oceanbase_status = lambda: {"status": "alive"}
    monkeypatch.setitem(sys.modules, "api.utils.health_utils", health_utils_mod)

    quart_mod = ModuleType("quart")
    quart_mod.jsonify = lambda payload: payload
    monkeypatch.setitem(sys.modules, "quart", quart_mod)

    module_path = repo_root / "api" / "apps" / "restful_apis" / "system_api.py"
    spec = importlib.util.spec_from_file_location("test_system_routes_unit_module", module_path)
    module = importlib.util.module_from_spec(spec)
    module.manager = _DummyManager()
    monkeypatch.setitem(sys.modules, "test_system_routes_unit_module", module)
    spec.loader.exec_module(module)
    return module


@pytest.mark.p2
def test_status_branch_matrix_unit(monkeypatch):
    module = _load_system_module(monkeypatch)

    monkeypatch.setattr(module.settings, "docStoreConn", SimpleNamespace(health=lambda: {"type": "es", "status": "green"}))
    monkeypatch.setattr(module.settings, "STORAGE_IMPL", SimpleNamespace(health=lambda: True))
    monkeypatch.setattr(module.KnowledgebaseService, "get_by_id", lambda _kb_id: True)
    monkeypatch.setattr(module.REDIS_CONN, "health", lambda: True)
    monkeypatch.setattr(module.REDIS_CONN, "smembers", lambda _key: {"executor-1"})
    monkeypatch.setattr(module.REDIS_CONN, "zrangebyscore", lambda *_args, **_kwargs: ['{"beat": 1}'])

    res = module.status()
    assert res["code"] == 0
    assert res["data"]["doc_engine"]["status"] == "green"
    assert res["data"]["storage"]["status"] == "green"
    assert res["data"]["database"]["status"] == "green"
    assert res["data"]["redis"]["status"] == "green"
    assert res["data"]["task_executor_heartbeats"]["executor-1"][0]["beat"] == 1

    monkeypatch.setattr(
        module.settings,
        "docStoreConn",
        SimpleNamespace(health=lambda: (_ for _ in ()).throw(RuntimeError("doc down"))),
    )
    monkeypatch.setattr(
        module.settings,
        "STORAGE_IMPL",
        SimpleNamespace(health=lambda: (_ for _ in ()).throw(RuntimeError("storage down"))),
    )
    monkeypatch.setattr(
        module.KnowledgebaseService,
        "get_by_id",
        lambda _kb_id: (_ for _ in ()).throw(RuntimeError("db down")),
    )
    monkeypatch.setattr(module.REDIS_CONN, "health", lambda: False)
    monkeypatch.setattr(module.REDIS_CONN, "smembers", lambda _key: (_ for _ in ()).throw(RuntimeError("hb down")))

    res = module.status()
    assert res["code"] == 0
    assert res["data"]["doc_engine"]["status"] == "red"
    assert "doc down" in res["data"]["doc_engine"]["error"]
    assert res["data"]["storage"]["status"] == "red"
    assert "storage down" in res["data"]["storage"]["error"]
    assert res["data"]["database"]["status"] == "red"
    assert "db down" in res["data"]["database"]["error"]
    assert res["data"]["redis"]["status"] == "red"
    assert "Lost connection!" in res["data"]["redis"]["error"]
    assert res["data"]["task_executor_heartbeats"] == {}


@pytest.mark.p2
def test_get_config_returns_register_enabled_unit(monkeypatch):
    module = _load_system_module(monkeypatch)
    monkeypatch.setattr(module.settings, "REGISTER_ENABLED", False)
    res = module.get_config()
    assert res["code"] == 0
    assert res["data"]["registerEnabled"] is False
