conftest.py•5.93 kB
from typing import Any
import pytest
from mcp.server.fastmcp import Context
from pytest_mock import MockerFixture
from keboola_mcp_server.clients.client import KeboolaClient
@pytest.fixture
def mock_components() -> list[dict[str, Any]]:
"""Mock result of `component_list`"""
return [
{
'id': 'keboola.ex-aws-s3',
'name': 'AWS S3 Extractor',
'type': 'extractor',
'description': 'Extract data from AWS S3',
'version': '1',
},
{
'id': 'keboola.wr-google-drive',
'name': 'Google Drive Writer',
'type': 'writer',
'description': 'Write data to Google Drive',
'version': '1',
},
{
'id': 'keboola.app-google-drive',
'name': 'Google Drive Application',
'type': 'application',
'description': 'Application for Google Drive',
'version': '1',
},
{
'id': 'keboola.snowflake-transformation',
'name': 'Snowflake Transformation',
'type': 'transformation',
'description': 'Snowflake SQL transformation',
'version': '1',
},
]
@pytest.fixture
def mock_configurations() -> list[dict[str, Any]]:
"""Mock result of `configuration_list`"""
return [
{
'id': '123',
'name': 'My Config',
'description': 'Test configuration',
'created': '2024-01-01T00:00:00Z',
'isDisabled': False,
'isDeleted': False,
'version': 1,
'configuration': {},
},
{
'id': '456',
'name': 'My Config 2',
'description': 'Test configuration 2',
'created': '2024-01-01T00:00:00Z',
'isDisabled': True,
'isDeleted': True,
'version': 2,
'configuration': {},
},
]
@pytest.fixture
def mock_component() -> dict[str, Any]:
"""Mock result of `component_detail`"""
return {
'id': 'keboola.ex-aws-s3',
'name': 'AWS S3 Extractor',
'type': 'extractor',
'description': 'Extract data from AWS S3',
'longDescription': 'Extract data from AWS S3 looooooooong',
'categories': ['extractor'],
'version': 1,
'created': '2024-01-01T00:00:00Z',
'data': {'data1': 'data1', 'data2': 'data2'},
'component_flags': ['flag1', 'flag2'],
'configurationSchema': {},
'configurationDescription': 'Extract data from AWS S3',
'emptyConfiguration': {},
'rootConfigurationExamples': [{'foo': 'root'}],
'rowConfigurationExamples': [{'foo': 'row'}],
}
@pytest.fixture
def mock_tf_component() -> dict[str, Any]:
"""Mock result of `component_detail` for a transformation component"""
return {
'componentId': 'keboola.google-bigquery-transformation',
'componentType': 'transformation',
'componentName': 'Google BigQuery',
'componentCategories': [],
'description': "BigQuery is Google's fully managed, serverless data warehouse",
'longDescription': 'Application which runs KBC transformations',
'documentationUrl': 'https://help.keboola.com/transformations/bigquery',
'documentation': '---\ntitle: Google BigQuery Transformation\npermalink: /transformations/bigquery/\n---',
'configurationSchema': {},
'configurationRowSchema': {},
'configurationDescription': None,
'rootConfigurationExamples': [],
'rowConfigurationExamples': [],
'componentFlags': [
'genericDockerUI',
'genericDockerUI-tableOutput',
'genericCodeBlocksUI',
'genericVariablesUI',
'genericDockerUI-tableInput',
],
}
@pytest.fixture
def mock_configuration() -> dict[str, Any]:
"""Mock mock_configuration tool."""
return {
'id': '123',
'name': 'My Config',
'description': 'Test configuration',
'created': '2024-01-01T00:00:00Z',
'isDisabled': False,
'isDeleted': False,
'version': 1,
'configuration': {},
'rows': [{'id': '1', 'name': 'Row 1', 'version': 1}, {'id': '2', 'name': 'Row 2', 'version': 1}],
}
@pytest.fixture
def mock_tf_configuration() -> dict[str, Any]:
"""Mock mock_configuration tool."""
return {
'id': '124',
'name': 'My Transformation',
'description': 'Test transformation configuration',
'created': '2024-01-01T00:00:00Z',
'isDisabled': False,
'isDeleted': False,
'version': 1,
'configuration': {
'parameters': {
'blocks': [
{
'name': 'Blocks',
'codes': [{'name': 'Code 1', 'script': ['SELECT * FROM customers;', 'SELECT * FROM orders;']}],
},
],
},
'storage': {
'input': {'tables': []},
'output': {'tables': [{'source': 'customers', 'destination': 'out.c-my-transformation.customers'}]},
},
},
}
@pytest.fixture
def mock_metadata() -> list[dict[str, Any]]:
"""Mock mock_component_configuration tool."""
return [
{
'id': '1',
'key': 'test-key',
'value': 'test-value',
'provider': 'user',
'timestamp': '2024-01-01T00:00:00Z',
}
]
@pytest.fixture
def mock_branch_id() -> str:
return 'default'
@pytest.fixture
def mcp_context_components_configs(mocker: MockerFixture, mcp_context_client: Context, mock_branch_id: str) -> Context:
keboola_client = mcp_context_client.session.state[KeboolaClient.STATE_KEY]
keboola_client.storage_client.branch_id = mock_branch_id
return mcp_context_client