kamidso-back/tests/kamidso/test_datasources.py

import os

import pytest

from kamidso.contracts.errors import KamidsoError, KamidsoErrorCode
from kamidso.contracts.user import CurrentUser
from kamidso.repositories.store import RepositoryStore
from kamidso.resources.datasource_types import (
    catalog,
    connector_engine_type,
    connector_params,
    normalize_params,
    normalize_type,
)
from kamidso.resources.datasources import DatasourceService
from kamidso.security.secrets import SecretStore


def _user() -> CurrentUser:
    return CurrentUser(
        id="usr_test",
        email="test@kamidso.test",
        name="Test",
        role="developer",
        workspace_id="ws_test",
    )


def _service() -> DatasourceService:
    store = RepositoryStore()
    return DatasourceService(store, SecretStore(store))


def test_catalog_includes_requested_engines():
    ids = {item["id"] for item in catalog()}
    assert {
        "mysql",
        "mariadb",
        "oceanbase",
        "postgresql",
        "mssql",
        "oracle",
        "sqlite",
        "duckdb",
        "clickhouse",
        "starrocks",
        "doris",
        "hive",
        "gaussdb",
        "openGauss",
        "vertica",
        "maxcompute",
        "spark",
        "neo4j",
        "mongodb",
    }.issubset(ids)


def test_aliases_and_special_params():
    assert normalize_type("MariaDB") == "mariadb"
    assert normalize_type("opengauss") == "openGauss"
    duck = normalize_params("duckdb", {"path": "C:/data/sales.duckdb"})
    assert duck["type"] == "duckdb"
    assert duck["path"].endswith("sales.duckdb")
    maxc = normalize_params(
        "maxcompute",
        {
            "endpoint": "https://service.cn-hangzhou.maxcompute.aliyun.com/api",
            "project": "sales",
            "access_id": "id",
        },
        password="secret",
    )
    assert maxc["project"] == "sales"
    assert maxc["access_key"] == "secret"
    assert connector_engine_type("mariadb") == "mysql"
    hive = normalize_params("hive", {"host": "hiveserver", "user": "hive"})
    assert hive["database"] == "default"
    hive_payload = connector_params(hive)
    assert hive_payload["username"] == "hive"
    assert "user" not in hive_payload
    spark = connector_params(normalize_params("spark", {"path": "C:/data/sales.parquet"}))
    assert spark["path"].endswith("sales.parquet")


def test_normalize_mysql_params():
    params = normalize_params(
        "MySQL",
        {"host": "127.0.0.1", "user": "root", "database": "sales"},
        password="secret",
    )
    assert params["type"] == "mysql"
    assert params["port"] == 3306
    assert params["password"] == "secret"
    assert params["database"] == "sales"


def test_oracle_requires_service_or_sid():
    with pytest.raises(KamidsoError) as caught:
        normalize_params("oracle", {"host": "db.local", "user": "system"})
    assert caught.value.code == KamidsoErrorCode.VALIDATION_ERROR


def test_sqlite_resolves_walmart_docker_path(tmp_path, monkeypatch):
    monkeypatch.setenv("KAMIDSO_STORAGE_PATH", str(tmp_path))
    monkeypatch.setenv("KAMIDSO_DATA_ROOT", str(tmp_path))
    params = normalize_params("sqlite", {"path": "/app/pilot/examples/Walmart_Sales.db"})
    assert os.path.isfile(params["path"])
    assert params["path"].endswith("Walmart_Sales.db")
    import sqlite3

    conn = sqlite3.connect(params["path"])
    try:
        count = conn.execute("SELECT COUNT(*) FROM walmart_sales").fetchone()[0]
        assert count > 100
    finally:
        conn.close()


def test_sqlite_missing_file_is_not_http_not_found():
    with pytest.raises(KamidsoError) as caught:
        normalize_params("sqlite", {"path": "C:/missing/no-such-file.db"})
    assert caught.value.code == KamidsoErrorCode.VALIDATION_ERROR
    assert "Not Found" not in str(caught.value)
    assert "پیدا نشد" in str(caught.value)


@pytest.mark.asyncio
async def test_create_persists_connector_fields_without_live_test():
    service = _service()
    doc = await service.create(
        _user(),
        name="فروش",
        ds_type="postgresql",
        config={"host": "db.internal", "port": 5432, "user": "app", "database": "sales", "schema": "public"},
        password="pw",
        test=False,
    )
    assert doc["_id"].startswith("ds_")
    assert doc["db_type"] == "postgresql"
    assert doc["db_name"] == "sales"
    assert doc["db_host"] == "db.internal"
    assert doc["db_user"] == "app"
    assert doc["db_pwd"] == "pw"
    assert doc["ext_config"]["schema"] == "public"
    public = DatasourceService.to_public(doc)
    assert public["id"] == doc["_id"]
    assert "password" not in (public.get("config") or {})
    assert public["type"] == "postgresql"


@pytest.mark.asyncio
async def test_sqlite_live_test_and_delete():
    service = _service()
    doc = await service.create(
        _user(),
        name="local sqlite",
        ds_type="sqlite",
        config={"path": ":memory:"},
        test=True,
    )
    assert ":memory:" in str(doc["db_path"])
    listed = await service.list(_user())
    assert listed[0]["_id"] == doc["_id"]
    await service.delete(_user(), doc["_id"])
    assert await service.list(_user()) == []