297 lines
12 KiB
Python
297 lines
12 KiB
Python
# #region Test.DatasetsMixin [C:3] [TYPE Module] [SEMANTICS test, dataset, superset, crud, detail]
|
|
# @BRIEF Tests for core/superset_client/_datasets.py — SupersetDatasetsMixin.
|
|
# @RELATION BINDS_TO -> [SupersetDatasetsMixin]
|
|
|
|
from pathlib import Path
|
|
import sys
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent.parent / "src"))
|
|
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
|
|
class TestDatasetsMixin:
|
|
"""Test all methods of SupersetDatasetsMixin."""
|
|
|
|
@pytest.fixture
|
|
def mixin(self):
|
|
from src.core.superset_client._datasets import SupersetDatasetsMixin
|
|
|
|
m = SupersetDatasetsMixin()
|
|
m.client = MagicMock()
|
|
m.client.request = AsyncMock()
|
|
m._validate_query_params = MagicMock(return_value={"page": 0, "page_size": 100})
|
|
m._fetch_all_pages = AsyncMock(return_value=[{"id": 1}, {"id": 2}])
|
|
return m
|
|
|
|
# ── get_datasets ──
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_datasets_success(self, mixin):
|
|
total, data = await mixin.get_datasets(query={"columns": ["id"]})
|
|
assert total == 2
|
|
assert len(data) == 2
|
|
mixin._validate_query_params.assert_called_once_with({"columns": ["id"]})
|
|
mixin._fetch_all_pages.assert_awaited_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_datasets_no_query(self, mixin):
|
|
total, data = await mixin.get_datasets()
|
|
assert total == 2
|
|
mixin._validate_query_params.assert_called_once_with(None)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_datasets_empty_result(self, mixin):
|
|
mixin._fetch_all_pages = AsyncMock(return_value=[])
|
|
total, data = await mixin.get_datasets()
|
|
assert total == 0
|
|
assert data == []
|
|
|
|
# ── get_datasets_summary ──
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_datasets_summary(self, mixin):
|
|
mixin.get_datasets = AsyncMock(
|
|
return_value=(
|
|
2,
|
|
[
|
|
{
|
|
"id": 1,
|
|
"table_name": "users",
|
|
"schema": "public",
|
|
"database": {"database_name": "Postgres"},
|
|
},
|
|
{
|
|
"id": 2,
|
|
"table_name": "orders",
|
|
"schema": "public",
|
|
"database": {"database_name": "Postgres"},
|
|
},
|
|
],
|
|
)
|
|
)
|
|
result = await mixin.get_datasets_summary()
|
|
assert len(result) == 2
|
|
assert result[0]["table_name"] == "users"
|
|
assert result[0]["database"] == "Postgres"
|
|
assert result[1]["schema"] == "public"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_datasets_summary_missing_database(self, mixin):
|
|
mixin.get_datasets = AsyncMock(
|
|
return_value=(
|
|
1,
|
|
[{"id": 1, "table_name": "t", "schema": "public"}],
|
|
)
|
|
)
|
|
result = await mixin.get_datasets_summary()
|
|
assert result[0]["database"] == "Unknown"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_datasets_summary_empty(self, mixin):
|
|
mixin.get_datasets = AsyncMock(return_value=(0, []))
|
|
result = await mixin.get_datasets_summary()
|
|
assert result == []
|
|
|
|
# ── get_dataset_linked_dashboard_count ──
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_dataset_linked_dashboard_count_direct_key(self, mixin):
|
|
mixin.client.request.return_value = {"dashboards": [{"id": 1}, {"id": 2}]}
|
|
result = await mixin.get_dataset_linked_dashboard_count(1)
|
|
assert result == 2
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_dataset_linked_dashboard_count_result_key(self, mixin):
|
|
mixin.client.request.return_value = {"result": {"dashboards": [{"id": 1}]}}
|
|
result = await mixin.get_dataset_linked_dashboard_count(1)
|
|
assert result == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_dataset_linked_dashboard_count_empty(self, mixin):
|
|
mixin.client.request.return_value = {"result": {"dashboards": []}}
|
|
result = await mixin.get_dataset_linked_dashboard_count(1)
|
|
assert result == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_dataset_linked_dashboard_count_non_dict(self, mixin):
|
|
mixin.client.request.return_value = "not_a_dict"
|
|
result = await mixin.get_dataset_linked_dashboard_count(1)
|
|
assert result == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_dataset_linked_dashboard_count_error(self, mixin):
|
|
mixin.client.request.side_effect = Exception("API error")
|
|
with patch("src.core.superset_client._datasets.app_logger") as mock_log:
|
|
result = await mixin.get_dataset_linked_dashboard_count(1)
|
|
assert result == 0
|
|
mock_log.warning.assert_called_once()
|
|
|
|
# ── get_dataset_detail ──
|
|
|
|
@pytest.fixture
|
|
def detail_mixin(self):
|
|
from src.core.superset_client._datasets import SupersetDatasetsMixin
|
|
|
|
m = SupersetDatasetsMixin()
|
|
m.client = MagicMock()
|
|
m.client.request = AsyncMock()
|
|
m.get_dataset = AsyncMock()
|
|
return m
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_dataset_detail_full(self, detail_mixin):
|
|
detail_mixin.get_dataset.return_value = {
|
|
"result": {
|
|
"id": 1,
|
|
"table_name": "users",
|
|
"schema": "public",
|
|
"database": {"id": 1, "backend": "postgresql", "database_name": "Postgres"},
|
|
"database_id": 1,
|
|
"description": "User table",
|
|
"columns": [
|
|
{"id": 1, "column_name": "name", "type": "text", "is_dttm": False, "is_active": True, "description": ""},
|
|
{"id": 2, "column_name": "created", "type": "timestamp", "is_dttm": True, "is_active": True, "description": ""},
|
|
],
|
|
"metrics": [
|
|
{"id": 1, "metric_name": "count", "expression": "COUNT(*)", "verbose_name": "Count", "description": "", "metric_type": "count"},
|
|
],
|
|
"sql": "SELECT * FROM users",
|
|
"is_sqllab_view": False,
|
|
"created_on": "2024-01-01",
|
|
"changed_on": "2024-01-02",
|
|
}
|
|
}
|
|
detail_mixin.client.request.return_value = {
|
|
"dashboards": {"result": [{"id": 10, "dashboard_title": "Main", "slug": "main"}]}
|
|
}
|
|
result = await detail_mixin.get_dataset_detail(1)
|
|
assert result["id"] == 1
|
|
assert result["table_name"] == "users"
|
|
assert len(result["columns"]) == 2
|
|
assert result["columns"][0]["name"] == "name"
|
|
assert result["columns"][1]["is_dttm"] is True
|
|
assert len(result["metrics"]) == 1
|
|
assert result["metrics"][0]["metric_name"] == "count"
|
|
assert result["linked_dashboard_count"] == 1
|
|
assert result["database_backend"] == "postgresql"
|
|
assert result["is_sqllab_view"] is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_dataset_detail_no_result_key(self, detail_mixin):
|
|
detail_mixin.get_dataset.return_value = {
|
|
"id": 1,
|
|
"table_name": "no_result",
|
|
"columns": [],
|
|
"metrics": [],
|
|
}
|
|
detail_mixin.client.request.return_value = {}
|
|
result = await detail_mixin.get_dataset_detail(1)
|
|
assert result["table_name"] == "no_result"
|
|
assert result["columns"] == []
|
|
assert result["metrics"] == []
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_dataset_detail_related_objects_error(self, detail_mixin):
|
|
detail_mixin.get_dataset.return_value = {
|
|
"result": {
|
|
"id": 1,
|
|
"table_name": "test",
|
|
"columns": [],
|
|
"metrics": [],
|
|
}
|
|
}
|
|
detail_mixin.client.request.side_effect = Exception("Connection error")
|
|
with patch("src.core.superset_client._datasets.app_logger") as mock_log:
|
|
result = await detail_mixin.get_dataset_detail(1)
|
|
assert result["linked_dashboards"] == []
|
|
mock_log.warning.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_dataset_detail_related_objects_has_dashboards_direct(self, detail_mixin):
|
|
detail_mixin.get_dataset.return_value = {
|
|
"result": {
|
|
"id": 1,
|
|
"table_name": "test",
|
|
"columns": [],
|
|
"metrics": [],
|
|
"database": {},
|
|
}
|
|
}
|
|
detail_mixin.client.request.return_value = {"dashboards": [{"id": 5, "dashboard_title": "Dash"}]}
|
|
result = await detail_mixin.get_dataset_detail(1)
|
|
assert result["linked_dashboard_count"] == 1
|
|
assert result["linked_dashboards"][0]["id"] == 5
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_dataset_detail_related_objects_with_int_list(self, detail_mixin):
|
|
detail_mixin.get_dataset.return_value = {
|
|
"result": {
|
|
"id": 1,
|
|
"table_name": "test",
|
|
"columns": [],
|
|
"metrics": [],
|
|
}
|
|
}
|
|
detail_mixin.client.request.return_value = {"dashboards": [5, 6]}
|
|
result = await detail_mixin.get_dataset_detail(1)
|
|
assert result["linked_dashboard_count"] == 2
|
|
assert result["linked_dashboards"][0]["id"] == 5
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_dataset_detail_as_bool_variants(self, detail_mixin):
|
|
detail_mixin.get_dataset.return_value = {
|
|
"result": {
|
|
"id": 1,
|
|
"table_name": "test",
|
|
"columns": [
|
|
{"id": 1, "column_name": "a", "is_dttm": "1", "is_active": "true"},
|
|
{"id": 2, "column_name": "b", "is_dttm": "yes", "is_active": "on"},
|
|
{"id": 3, "column_name": "c", "is_dttm": None, "is_active": None},
|
|
{"id": 4, "column_name": "d", "is_dttm": 1, "is_active": 0},
|
|
],
|
|
"metrics": [],
|
|
"database": {"backend": "mysql"},
|
|
}
|
|
}
|
|
detail_mixin.client.request.return_value = {}
|
|
result = await detail_mixin.get_dataset_detail(1)
|
|
assert result["columns"][0]["is_dttm"] is True
|
|
assert result["columns"][1]["is_dttm"] is True
|
|
assert result["columns"][2]["is_dttm"] is False
|
|
assert result["columns"][3]["is_dttm"] is True
|
|
assert result["columns"][0]["is_active"] is True
|
|
assert result["columns"][3]["is_active"] is False
|
|
|
|
# ── get_dataset ──
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_dataset_success(self, mixin):
|
|
mixin.client.request.return_value = {"result": {"id": 1, "table_name": "test"}}
|
|
result = await mixin.get_dataset(1)
|
|
assert result["result"]["id"] == 1
|
|
|
|
# ── update_dataset ──
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_dataset_success(self, mixin):
|
|
mixin.client.request.return_value = {"result": "ok"}
|
|
result = await mixin.update_dataset(1, {"columns": []}, override_columns=True)
|
|
mixin.client.request.assert_awaited_once()
|
|
assert result["result"] == "ok"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_dataset_no_override(self, mixin):
|
|
mixin.client.request.return_value = {"result": "ok"}
|
|
result = await mixin.update_dataset(1, {"columns": []})
|
|
assert result["result"] == "ok"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_dataset_with_json_data(self, mixin):
|
|
mixin.client.request.return_value = {"result": "ok"}
|
|
result = await mixin.update_dataset(1, {"table_name": "new_name"}, override_columns=False)
|
|
assert result["result"] == "ok"
|
|
# #endregion Test.DatasetsMixin
|