"""Unit tests for read_delta tool: models, service, and tool.""" from unittest.mock import AsyncMock, MagicMock, patch import httpx import pytest from pydantic import ValidationError as PydanticValidationError from src.models.read_delta import ReadDeltaRequest from src.services.duckdb_delta import read_delta from src.utils.errors import AuthServiceError # Required params for ReadDeltaRequest in tests _REQ = {"cluster_id": "c1"} def _mock_httpx_get_response(response: httpx.Response): mock_client = MagicMock() mock_client.get = AsyncMock(return_value=response) mock_cm = MagicMock() mock_cm.__aenter__ = AsyncMock(return_value=mock_client) mock_cm.__aexit__ = AsyncMock(return_value=None) return mock_cm class TestReadDeltaRequest: """Validation of ReadDeltaRequest model.""" @pytest.fixture(autouse=True) def _mock_gs_base_url(self): with patch( "src.services.observability_resolve.get_observability_global_service_url", return_value="https://gs.example.com", ): yield def test_valid_local_path(self): req = ReadDeltaRequest.model_validate( {"table_path": "file:///tmp/delta", "vendor": "local", **_REQ} ) assert req.table_path == "file:///tmp/delta" def test_valid_s3_path(self): ReadDeltaRequest.model_validate( {"table_path": "s3://bucket/key", "vendor": "aws", **_REQ} ) def test_valid_s3_path_for_alicloud(self): ReadDeltaRequest.model_validate( {"table_path": "s3://bucket/key", "vendor": "alicloud", **_REQ} ) def test_valid_gs_path_for_gcp(self): ReadDeltaRequest.model_validate( {"table_path": "gs://bucket/key", "vendor": "gcp", **_REQ} ) def test_empty_table_path_rejected(self): with pytest.raises(PydanticValidationError): ReadDeltaRequest.model_validate({"table_path": "", "vendor": "local", **_REQ}) with pytest.raises(PydanticValidationError): ReadDeltaRequest.model_validate( {"table_path": " ", "vendor": "local", **_REQ} ) def test_path_scheme_must_match_cloud(self): # Dedicated/premium (non-local) must be s3:// with pytest.raises(PydanticValidationError): ReadDeltaRequest.model_validate({"table_path": "file:///x", "vendor": "aws", **_REQ}) with pytest.raises(PydanticValidationError): ReadDeltaRequest.model_validate({"table_path": "s3://b/k", "vendor": "gcp", **_REQ}) # Local must be file:/// with pytest.raises(PydanticValidationError): ReadDeltaRequest.model_validate({"table_path": "s3://b/k", "vendor": "local", **_REQ}) def test_path_must_include_bucket_or_path(self): with pytest.raises(PydanticValidationError): ReadDeltaRequest.model_validate({"table_path": "s3://", "vendor": "aws", **_REQ}) with pytest.raises(PydanticValidationError): ReadDeltaRequest.model_validate({"table_path": "s3://bucket", "vendor": "aws", **_REQ}) def test_limit_bounds(self): req = ReadDeltaRequest.model_validate( {"table_path": "file:///x", "vendor": "local", "limit": 1, **_REQ} ) assert req.limit == 1 req = ReadDeltaRequest.model_validate( {"table_path": "file:///x", "vendor": "local", "limit": 100_000, **_REQ} ) assert req.limit == 100_000 with pytest.raises(PydanticValidationError): ReadDeltaRequest.model_validate( {"table_path": "file:///x", "vendor": "local", "limit": 0, **_REQ} ) with pytest.raises(PydanticValidationError): ReadDeltaRequest.model_validate( {"table_path": "file:///x", "vendor": "local", "limit": 100_001, **_REQ} ) class TestReadDeltaService: """DuckDB Delta service with mocked DuckDB.""" @pytest.fixture(autouse=True) def _mock_gs_base_url(self): with patch( "src.services.observability_resolve.get_observability_global_service_url", return_value="https://gs.example.com", ): yield @patch("src.services.duckdb_delta.duckdb.connect") def test_read_delta_local_returns_rows(self, mock_connect): mock_conn = MagicMock() mock_df = MagicMock() mock_df.empty = False mock_df.to_dict.return_value = [{"id": 1, "name": "a"}] mock_conn.execute.return_value.fetchdf.return_value = mock_df mock_connect.return_value = mock_conn with patch("src.services.duckdb_delta.duckdb.connect", mock_connect): result = read_delta("file:///tmp/delta", vendor="local", limit=10) assert result == [{"id": 1, "name": "a"}] @patch("src.services.duckdb_delta.duckdb.connect") def test_read_delta_empty_result(self, mock_connect): mock_conn = MagicMock() mock_df = MagicMock() mock_df.empty = True mock_conn.execute.return_value.fetchdf.return_value = mock_df mock_connect.return_value = mock_conn with patch("src.services.duckdb_delta.duckdb.connect", mock_connect): result = read_delta("file:///tmp/delta", vendor="local") assert result == [] class TestReadDeltaTool: """read_delta_tool MCP tool.""" @pytest.fixture(autouse=True) def _mock_gs_base_url(self): with patch( "src.services.observability_resolve.get_observability_global_service_url", return_value="https://gs.example.com", ): yield @pytest.mark.asyncio async def test_local_reads_do_not_require_server_auth(self): from src.tools.read_delta import read_delta_tool gs_payload = { "code": 0, "message": "", "data": { "clusters": [ { "id": "019cd1e0-a8ff-7b6f-9e90-209d4946bb14", "tenantID": "1369847559691397282", "internalReadUri": ( "https://www.ds.us-west-2-staging.aws.observability.tidbcloud.com/" "internal/metrics/019cd1e0-a8ff-7b6f-9e90-209d4946bb14/api/v1/" ), "vendor": "aws", "region": "us-west-2", } ] }, } gs_response = httpx.Response(200, json=gs_payload) with patch( "src.services.observability_resolve.get_server_auth0_token", return_value="auth0-token", ): with patch( "src.services.observability_resolve.httpx.AsyncClient", return_value=_mock_httpx_get_response(gs_response), ): with patch( "src.tools.read_delta.get_valid_server_auth", new_callable=AsyncMock, side_effect=AuthServiceError("server auth missing"), ) as mock_auth: with patch("src.tools.read_delta.read_delta") as mock_read: mock_read.return_value = [{"x": 1}] result = await read_delta_tool( table_path="file:///tmp/delta", env="dev", deploy_type="local", cluster_id="", ) mock_auth.assert_not_awaited() assert len(result) == 1 assert result[0]["type"] == "text" assert '"x": 1' in result[0]["text"] @pytest.mark.asyncio async def test_missing_server_auth_returns_error_for_cloud_read(self): from src.tools.read_delta import read_delta_tool gs_payload = { "code": 0, "message": "", "data": { "clusters": [ { "id": "019cd1e0-a8ff-7b6f-9e90-209d4946bb14", "tenantID": "1369847559691397282", "internalReadUri": ( "https://www.ds.us-west-2-staging.aws.observability.tidbcloud.com/" "internal/metrics/019cd1e0-a8ff-7b6f-9e90-209d4946bb14/api/v1/" ), "vendor": "aws", "region": "us-west-2", } ] }, } gs_response = httpx.Response(200, json=gs_payload) with patch( "src.services.observability_resolve.get_server_auth0_token", return_value="auth0-token", ): with patch( "src.services.observability_resolve.httpx.AsyncClient", return_value=_mock_httpx_get_response(gs_response), ): with patch( "src.tools.read_delta.get_valid_server_auth", new_callable=AsyncMock, side_effect=AuthServiceError("server auth missing"), ) as mock_auth: with patch("src.tools.read_delta.read_delta") as mock_read: result = await read_delta_tool( table_path="s3://bucket/key", env="staging", deploy_type="dedicated", cluster_id="c1", ) mock_auth.assert_awaited_once() mock_read.assert_not_called() assert len(result) == 1 assert result[0]["type"] == "text" assert "server auth missing" in result[0]["text"] @pytest.mark.asyncio async def test_validation_error_returns_text_message(self): from src.tools.read_delta import read_delta_tool gs_payload = { "code": 0, "message": "", "data": { "clusters": [ { "id": "019cd1e0-a8ff-7b6f-9e90-209d4946bb14", "tenantID": "1369847559691397282", "internalReadUri": ( "https://www.ds.us-west-2-staging.aws.observability.tidbcloud.com/" "internal/metrics/019cd1e0-a8ff-7b6f-9e90-209d4946bb14/api/v1/" ), "vendor": "aws", "region": "us-west-2", } ] }, } gs_response = httpx.Response(200, json=gs_payload) with patch( "src.services.observability_resolve.get_server_auth0_token", return_value="auth0-token", ): with patch( "src.services.observability_resolve.httpx.AsyncClient", return_value=_mock_httpx_get_response(gs_response), ): with patch("src.tools.read_delta.get_valid_server_auth", new_callable=AsyncMock): # Passing file:/// with dedicated (which GS resolves as aws) should trigger validation error result = await read_delta_tool( table_path="file:///b/k", env="staging", deploy_type="dedicated", cluster_id="c1", ) assert len(result) == 1 assert result[0]["type"] == "text" assert "Validation" in result[0]["text"] or "table_path" in result[0]["text"] @pytest.mark.asyncio async def test_success_returns_json_payload(self): from src.tools.read_delta import read_delta_tool gs_payload = { "code": 0, "message": "", "data": { "clusters": [ { "id": "019cd1e0-a8ff-7b6f-9e90-209d4946bb14", "tenantID": "1369847559691397282", "internalReadUri": ( "https://www.ds.us-west-2-staging.aws.observability.tidbcloud.com/" "internal/metrics/019cd1e0-a8ff-7b6f-9e90-209d4946bb14/api/v1/" ), "vendor": "aws", "region": "us-west-2", } ] }, } gs_response = httpx.Response(200, json=gs_payload) with patch( "src.services.observability_resolve.get_server_auth0_token", return_value="auth0-token", ): with patch( "src.services.observability_resolve.httpx.AsyncClient", return_value=_mock_httpx_get_response(gs_response), ): with patch("src.tools.read_delta.get_valid_server_auth", new_callable=AsyncMock): with patch( "src.tools.read_delta.get_delta_credentials", new_callable=AsyncMock, return_value=None, ): with patch("src.tools.read_delta.read_delta") as mock_read: mock_read.return_value = [{"x": 1}, {"x": 2}] result = await read_delta_tool( table_path="file:///tmp/delta", env="dev", deploy_type="local", cluster_id="", limit=10, ) assert len(result) == 1 assert result[0]["type"] == "text" assert '"x": 1' in result[0]["text"] @pytest.mark.asyncio async def test_delta_error_returns_text_message(self): from src.tools.read_delta import read_delta_tool from src.utils.errors import DeltaError gs_payload = { "code": 0, "message": "", "data": { "clusters": [ { "id": "019cd1e0-a8ff-7b6f-9e90-209d4946bb14", "tenantID": "1369847559691397282", "internalReadUri": ( "https://www.ds.us-west-2-staging.aws.observability.tidbcloud.com/" "internal/metrics/019cd1e0-a8ff-7b6f-9e90-209d4946bb14/api/v1/" ), "vendor": "aws", "region": "us-west-2", } ] }, } gs_response = httpx.Response(200, json=gs_payload) with patch( "src.services.observability_resolve.get_server_auth0_token", return_value="auth0-token", ): with patch( "src.services.observability_resolve.httpx.AsyncClient", return_value=_mock_httpx_get_response(gs_response), ): with patch("src.tools.read_delta.get_valid_server_auth", new_callable=AsyncMock): with patch( "src.tools.read_delta.get_delta_credentials", new_callable=AsyncMock, return_value=None, ): with patch("src.tools.read_delta.read_delta") as mock_read: mock_read.side_effect = DeltaError("Table not found") result = await read_delta_tool( table_path="file:///tmp/delta", env="dev", deploy_type="local", cluster_id="", ) assert len(result) == 1 assert result[0]["type"] == "text" assert "Delta Error" in result[0]["text"] assert "Table not found" in result[0]["text"] @pytest.mark.asyncio async def test_alicloud_cluster_passes_vendor_and_credentials_to_reader(self): from src.tools.read_delta import read_delta_tool with patch( "src.tools.read_delta.get_cluster_info", new_callable=AsyncMock, return_value=({"vendor": "alicloud", "region": "cn-hangzhou"}, "auth0-token"), ): with patch("src.tools.read_delta.get_valid_server_auth", new_callable=AsyncMock): with patch( "src.tools.read_delta.get_delta_credentials", new_callable=AsyncMock, return_value={ "access_key_id": "ak", "secret_access_key": "sk", "session_token": "st", "provider": "alicloud", }, ): with patch("src.tools.read_delta.read_delta") as mock_read: mock_read.return_value = [{"x": 1}] result = await read_delta_tool( table_path="s3://bucket/path", env="prod", deploy_type="premium", cluster_id="c1", ) mock_read.assert_called_once() assert mock_read.call_args.args[1] == "alicloud" assert mock_read.call_args.args[4]["region"] == "cn-hangzhou" assert len(result) == 1 assert result[0]["type"] == "text" @pytest.mark.asyncio async def test_gcp_cluster_uses_access_token_parquet_reader(self): from src.tools.read_delta import read_delta_tool with patch( "src.tools.read_delta.get_cluster_info", new_callable=AsyncMock, return_value=({"vendor": "gcp", "region": "us-central1"}, "auth0-token"), ): with patch("src.tools.read_delta.get_valid_server_auth", new_callable=AsyncMock): with patch( "src.tools.read_delta.get_gcp_access_token", new_callable=AsyncMock, return_value="ya29.test-token", ): with patch( "src.tools.read_delta.read_gcp_delta_parquet" ) as mock_parquet_read: mock_parquet_read.return_value = { "local_cache_path": "file:///tmp/o11y-gcp-cache", "rows": [{"x": 1}], } result = await read_delta_tool( table_path="gs://bucket/path", env="prod", deploy_type="premium", cluster_id="c1", ) mock_parquet_read.assert_called_once() assert mock_parquet_read.call_args.args[0] == "gs://bucket/path" assert mock_parquet_read.call_args.args[1] == "ya29.test-token" assert len(result) == 1 assert result[0]["type"] == "text" assert "local_cache_path" in result[0]["text"]