Chris Coutinho
327 lines
10 KiB
Python
327 lines
10 KiB
Python
import logging
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from nextcloud_mcp_server.client.tables import TablesClient
|
|
from tests.client.conftest import (
|
|
create_mock_error_response,
|
|
create_mock_response,
|
|
create_mock_table_row_ocs_response,
|
|
create_mock_table_row_response,
|
|
create_mock_table_schema_response,
|
|
create_mock_tables_list_response,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Mark all tests in this module as unit tests
|
|
pytestmark = pytest.mark.unit
|
|
|
|
|
|
async def test_tables_list_tables(mocker):
|
|
"""Test that list_tables correctly parses the API response (OCS format)."""
|
|
mock_response = create_mock_tables_list_response(
|
|
tables=[
|
|
{"id": 1, "title": "Table 1"},
|
|
{"id": 2, "title": "Table 2"},
|
|
]
|
|
)
|
|
|
|
mock_client = mocker.AsyncMock(spec=httpx.AsyncClient)
|
|
mock_make_request = mocker.patch.object(
|
|
TablesClient, "_make_request", return_value=mock_response
|
|
)
|
|
|
|
client = TablesClient(mock_client, "testuser")
|
|
tables = await client.list_tables()
|
|
|
|
assert isinstance(tables, list)
|
|
assert len(tables) == 2
|
|
assert tables[0]["id"] == 1
|
|
assert tables[0]["title"] == "Table 1"
|
|
|
|
mock_make_request.assert_called_once()
|
|
|
|
|
|
async def test_tables_get_schema(mocker):
|
|
"""Test that get_table_schema correctly parses the API response."""
|
|
mock_response = create_mock_table_schema_response(
|
|
table_id=123,
|
|
columns=[
|
|
{"id": 1, "title": "Name", "type": "text"},
|
|
{"id": 2, "title": "Age", "type": "number"},
|
|
{"id": 3, "title": "Email", "type": "text"},
|
|
],
|
|
)
|
|
|
|
mock_client = mocker.AsyncMock(spec=httpx.AsyncClient)
|
|
mock_make_request = mocker.patch.object(
|
|
TablesClient, "_make_request", return_value=mock_response
|
|
)
|
|
|
|
client = TablesClient(mock_client, "testuser")
|
|
schema = await client.get_table_schema(table_id=123)
|
|
|
|
assert isinstance(schema, dict)
|
|
assert "columns" in schema
|
|
assert len(schema["columns"]) == 3
|
|
assert schema["columns"][0]["title"] == "Name"
|
|
|
|
mock_make_request.assert_called_once()
|
|
assert "/tables/123/scheme" in mock_make_request.call_args[0][1]
|
|
|
|
|
|
async def test_tables_get_rows(mocker):
|
|
"""Test that get_table_rows correctly parses the API response."""
|
|
mock_response = create_mock_response(
|
|
status_code=200,
|
|
json_data=[
|
|
{
|
|
"id": 1,
|
|
"tableId": 123,
|
|
"data": [
|
|
{"columnId": 1, "value": "John"},
|
|
{"columnId": 2, "value": 30},
|
|
],
|
|
"createdBy": "testuser",
|
|
"createdAt": "2024-01-01T00:00:00+00:00",
|
|
"lastEditBy": "testuser",
|
|
"lastEditAt": "2024-01-01T00:00:00+00:00",
|
|
},
|
|
{
|
|
"id": 2,
|
|
"tableId": 123,
|
|
"data": [
|
|
{"columnId": 1, "value": "Jane"},
|
|
{"columnId": 2, "value": 25},
|
|
],
|
|
"createdBy": "testuser",
|
|
"createdAt": "2024-01-01T00:00:00+00:00",
|
|
"lastEditBy": "testuser",
|
|
"lastEditAt": "2024-01-01T00:00:00+00:00",
|
|
},
|
|
],
|
|
)
|
|
|
|
mock_client = mocker.AsyncMock(spec=httpx.AsyncClient)
|
|
mock_make_request = mocker.patch.object(
|
|
TablesClient, "_make_request", return_value=mock_response
|
|
)
|
|
|
|
client = TablesClient(mock_client, "testuser")
|
|
rows = await client.get_table_rows(table_id=123)
|
|
|
|
assert isinstance(rows, list)
|
|
assert len(rows) == 2
|
|
assert rows[0]["id"] == 1
|
|
assert rows[0]["tableId"] == 123
|
|
|
|
mock_make_request.assert_called_once()
|
|
|
|
|
|
async def test_tables_get_rows_with_pagination(mocker):
|
|
"""Test that get_table_rows correctly handles pagination parameters."""
|
|
mock_response = create_mock_response(
|
|
status_code=200,
|
|
json_data=[
|
|
{
|
|
"id": 1,
|
|
"tableId": 123,
|
|
"data": [],
|
|
"createdBy": "testuser",
|
|
"createdAt": "2024-01-01T00:00:00+00:00",
|
|
"lastEditBy": "testuser",
|
|
"lastEditAt": "2024-01-01T00:00:00+00:00",
|
|
},
|
|
],
|
|
)
|
|
|
|
mock_client = mocker.AsyncMock(spec=httpx.AsyncClient)
|
|
mock_make_request = mocker.patch.object(
|
|
TablesClient, "_make_request", return_value=mock_response
|
|
)
|
|
|
|
client = TablesClient(mock_client, "testuser")
|
|
rows = await client.get_table_rows(table_id=123, limit=5, offset=10)
|
|
|
|
assert isinstance(rows, list)
|
|
|
|
# Verify pagination parameters were passed
|
|
call_args = mock_make_request.call_args
|
|
assert call_args[1]["params"]["limit"] == 5
|
|
assert call_args[1]["params"]["offset"] == 10
|
|
|
|
|
|
async def test_tables_create_row(mocker):
|
|
"""Test that create_row correctly parses the API response (OCS format)."""
|
|
mock_response = create_mock_table_row_ocs_response(
|
|
row_id=456,
|
|
table_id=123,
|
|
data=[
|
|
{"columnId": 1, "value": "Test Name"},
|
|
{"columnId": 2, "value": 99},
|
|
],
|
|
)
|
|
|
|
mock_client = mocker.AsyncMock(spec=httpx.AsyncClient)
|
|
mock_make_request = mocker.patch.object(
|
|
TablesClient, "_make_request", return_value=mock_response
|
|
)
|
|
|
|
client = TablesClient(mock_client, "testuser")
|
|
test_data = {1: "Test Name", 2: 99}
|
|
created_row = await client.create_row(table_id=123, data=test_data)
|
|
|
|
assert isinstance(created_row, dict)
|
|
assert created_row["id"] == 456
|
|
assert created_row["tableId"] == 123
|
|
|
|
# Verify the data was transformed to string keys
|
|
call_args = mock_make_request.call_args
|
|
assert call_args[1]["json"]["data"]["1"] == "Test Name"
|
|
assert call_args[1]["json"]["data"]["2"] == 99
|
|
|
|
|
|
async def test_tables_update_row(mocker):
|
|
"""Test that update_row correctly parses the API response."""
|
|
mock_response = create_mock_table_row_response(
|
|
row_id=456,
|
|
table_id=123,
|
|
data=[
|
|
{"columnId": 1, "value": "Updated Name"},
|
|
{"columnId": 2, "value": 100},
|
|
],
|
|
)
|
|
|
|
mock_client = mocker.AsyncMock(spec=httpx.AsyncClient)
|
|
mock_make_request = mocker.patch.object(
|
|
TablesClient, "_make_request", return_value=mock_response
|
|
)
|
|
|
|
client = TablesClient(mock_client, "testuser")
|
|
update_data = {1: "Updated Name", 2: 100}
|
|
updated_row = await client.update_row(row_id=456, data=update_data)
|
|
|
|
assert isinstance(updated_row, dict)
|
|
assert updated_row["id"] == 456
|
|
|
|
# Verify the PUT request was made
|
|
call_args = mock_make_request.call_args
|
|
assert call_args[0][0] == "PUT"
|
|
assert "/rows/456" in call_args[0][1]
|
|
|
|
|
|
async def test_tables_delete_row(mocker):
|
|
"""Test that delete_row correctly parses the API response."""
|
|
mock_response = create_mock_response(
|
|
status_code=200, json_data={"message": "Row deleted"}
|
|
)
|
|
|
|
mock_client = mocker.AsyncMock(spec=httpx.AsyncClient)
|
|
mock_make_request = mocker.patch.object(
|
|
TablesClient, "_make_request", return_value=mock_response
|
|
)
|
|
|
|
client = TablesClient(mock_client, "testuser")
|
|
result = await client.delete_row(row_id=456)
|
|
|
|
assert isinstance(result, dict)
|
|
|
|
# Verify the DELETE request was made
|
|
call_args = mock_make_request.call_args
|
|
assert call_args[0][0] == "DELETE"
|
|
assert "/rows/456" in call_args[0][1]
|
|
|
|
|
|
async def test_tables_delete_nonexistent_row(mocker):
|
|
"""Test that deleting a non-existent row raises HTTPStatusError."""
|
|
error_response = create_mock_error_response(404, "Row not found")
|
|
|
|
mock_client = mocker.AsyncMock(spec=httpx.AsyncClient)
|
|
mock_make_request = mocker.patch.object(TablesClient, "_make_request")
|
|
mock_make_request.side_effect = httpx.HTTPStatusError(
|
|
"404 Not Found",
|
|
request=httpx.Request("DELETE", "http://test.local"),
|
|
response=error_response,
|
|
)
|
|
|
|
client = TablesClient(mock_client, "testuser")
|
|
|
|
with pytest.raises(httpx.HTTPStatusError) as excinfo:
|
|
await client.delete_row(row_id=999999999)
|
|
|
|
assert excinfo.value.response.status_code == 404
|
|
|
|
|
|
async def test_tables_get_nonexistent_schema(mocker):
|
|
"""Test that getting schema for non-existent table raises HTTPStatusError."""
|
|
error_response = create_mock_error_response(404, "Table not found")
|
|
|
|
mock_client = mocker.AsyncMock(spec=httpx.AsyncClient)
|
|
mock_make_request = mocker.patch.object(TablesClient, "_make_request")
|
|
mock_make_request.side_effect = httpx.HTTPStatusError(
|
|
"404 Not Found",
|
|
request=httpx.Request("GET", "http://test.local"),
|
|
response=error_response,
|
|
)
|
|
|
|
client = TablesClient(mock_client, "testuser")
|
|
|
|
with pytest.raises(httpx.HTTPStatusError) as excinfo:
|
|
await client.get_table_schema(table_id=999999999)
|
|
|
|
assert excinfo.value.response.status_code == 404
|
|
|
|
|
|
def test_tables_transform_row_data():
|
|
"""Test the transform_row_data utility method (synchronous)."""
|
|
# This is a pure function, no mocking needed
|
|
client = TablesClient(None, "testuser") # Client not used for this method
|
|
|
|
raw_rows = [
|
|
{
|
|
"id": 1,
|
|
"tableId": 123,
|
|
"createdBy": "testuser",
|
|
"createdAt": "2024-01-01T00:00:00+00:00",
|
|
"lastEditBy": "testuser",
|
|
"lastEditAt": "2024-01-01T00:00:00+00:00",
|
|
"data": [
|
|
{"columnId": 1, "value": "John Doe"},
|
|
{"columnId": 2, "value": 30},
|
|
{"columnId": 3, "value": "john@example.com"},
|
|
],
|
|
},
|
|
{
|
|
"id": 2,
|
|
"tableId": 123,
|
|
"createdBy": "testuser",
|
|
"createdAt": "2024-01-01T00:00:00+00:00",
|
|
"lastEditBy": "testuser",
|
|
"lastEditAt": "2024-01-01T00:00:00+00:00",
|
|
"data": [
|
|
{"columnId": 1, "value": "Jane Smith"},
|
|
{"columnId": 2, "value": 25},
|
|
{"columnId": 3, "value": "jane@example.com"},
|
|
],
|
|
},
|
|
]
|
|
|
|
columns = [
|
|
{"id": 1, "title": "Name", "type": "text"},
|
|
{"id": 2, "title": "Age", "type": "number"},
|
|
{"id": 3, "title": "Email", "type": "text"},
|
|
]
|
|
|
|
transformed = client.transform_row_data(raw_rows, columns)
|
|
|
|
assert len(transformed) == 2
|
|
assert transformed[0]["id"] == 1
|
|
assert transformed[0]["data"]["Name"] == "John Doe"
|
|
assert transformed[0]["data"]["Age"] == 30
|
|
assert transformed[0]["data"]["Email"] == "john@example.com"
|
|
|
|
assert transformed[1]["data"]["Name"] == "Jane Smith"
|
|
assert transformed[1]["data"]["Age"] == 25
|