320 lines
11 KiB
Python
320 lines
11 KiB
Python
"""Tests for bitable view config grouping validation (U5 / R4).
|
|
|
|
Layered tests:
|
|
1. Pure Pydantic validation via ``validate_view_config`` (no DB needed — always runs)
|
|
2. Route-level 422 via a mocked BitableService (no DB needed — always runs)
|
|
3. Round-trip via real PG service (skipped if PG unavailable)
|
|
|
|
The validator lives in ``agentkit.bitable.view_config`` and is called from
|
|
the PATCH /views route before ``service.update_view`` is invoked.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
import httpx
|
|
import pytest
|
|
from fastapi import FastAPI
|
|
from httpx import ASGITransport
|
|
from pydantic import ValidationError
|
|
|
|
from agentkit.bitable.models import View, ViewType
|
|
from agentkit.bitable.view_config import (
|
|
MAX_GROUP_BY_FIELDS,
|
|
GroupByItem,
|
|
ViewConfigSchema,
|
|
ViewConfigValidationError,
|
|
validate_view_config,
|
|
)
|
|
from agentkit.server.routes import bitable as bitable_routes
|
|
from agentkit.server.routes.bitable import require_bitable_auth
|
|
|
|
TEST_USER_ID = "test-user-id"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 1. Pure Pydantic validation — always runs (no DB)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _gb(field_id: str, direction: str = "asc") -> dict[str, str]:
|
|
return {"field_id": field_id, "direction": direction}
|
|
|
|
|
|
def test_validate_accepts_empty_config() -> None:
|
|
"""Empty / None config is valid (no group_by to check)."""
|
|
validate_view_config(None)
|
|
validate_view_config({})
|
|
validate_view_config({"filters": []}) # non-U5 keys ignored
|
|
|
|
|
|
def test_validate_accepts_one_to_three_group_by_fields() -> None:
|
|
"""Spec: max 3 levels. 1, 2, 3 all accepted."""
|
|
for n in (1, 2, 3):
|
|
config = {"group_by": [_gb(f"f{i}") for i in range(n)]}
|
|
validate_view_config(config) # no raise
|
|
|
|
|
|
def test_validate_rejects_more_than_three_group_by_fields() -> None:
|
|
"""group_by with >3 fields raises ValidationError (422 at the route)."""
|
|
config = {"group_by": [_gb(f"f{i}") for i in range(MAX_GROUP_BY_FIELDS + 1)]}
|
|
with pytest.raises(ViewConfigValidationError):
|
|
validate_view_config(config)
|
|
|
|
|
|
def test_validate_rejects_invalid_direction() -> None:
|
|
"""direction must be 'asc' or 'desc' (Literal)."""
|
|
config = {"group_by": [_gb("f1", "sideways")]}
|
|
with pytest.raises(ViewConfigValidationError):
|
|
validate_view_config(config)
|
|
|
|
|
|
def test_validate_rejects_missing_field_id() -> None:
|
|
"""field_id is required."""
|
|
config = {"group_by": [{"direction": "asc"}]}
|
|
with pytest.raises(ViewConfigValidationError):
|
|
validate_view_config(config)
|
|
|
|
|
|
def test_validate_rejects_empty_field_id() -> None:
|
|
"""field_id must be non-empty (min_length=1)."""
|
|
config = {"group_by": [_gb("")]}
|
|
with pytest.raises(ViewConfigValidationError):
|
|
validate_view_config(config)
|
|
|
|
|
|
def test_validate_rejects_extra_keys_in_group_by_item() -> None:
|
|
"""extra='forbid' on GroupByItem — unknown keys rejected."""
|
|
config = {"group_by": [{"field_id": "f1", "direction": "asc", "color": "red"}]}
|
|
with pytest.raises(ViewConfigValidationError):
|
|
validate_view_config(config)
|
|
|
|
|
|
def test_validate_defaults_direction_to_asc() -> None:
|
|
"""direction is optional, defaults to 'asc'."""
|
|
item = GroupByItem(field_id="f1")
|
|
assert item.direction == "asc"
|
|
|
|
|
|
def test_validate_passes_through_non_u5_keys() -> None:
|
|
"""filters / sort / hidden_fields are NOT validated here — pass through."""
|
|
config = {
|
|
"filters": [{"field_id": "f1", "op": "weird-op", "value": None}],
|
|
"sort": {"field": "f1", "order": "asc"},
|
|
"hidden_fields": ["f2"],
|
|
"group_by": [_gb("f1")],
|
|
}
|
|
validate_view_config(config) # no raise — non-U5 keys ignored
|
|
|
|
|
|
def test_schema_max_length_constant() -> None:
|
|
"""MAX_GROUP_BY_FIELDS is 3 (matches spec / Feishu / Twenty UX)."""
|
|
assert MAX_GROUP_BY_FIELDS == 3
|
|
|
|
|
|
def test_schema_round_trips_through_model() -> None:
|
|
"""ViewConfigSchema.model_validate round-trips a valid config."""
|
|
raw = {
|
|
"group_by": [{"field_id": "f1", "direction": "asc"}],
|
|
"conditional_formatting": [],
|
|
}
|
|
schema = ViewConfigSchema.model_validate(raw)
|
|
assert len(schema.group_by) == 1
|
|
assert schema.group_by[0].field_id == "f1"
|
|
dumped = schema.model_dump()
|
|
assert dumped["group_by"][0]["direction"] == "asc"
|
|
|
|
|
|
def test_validation_error_carries_structured_errors() -> None:
|
|
"""ViewConfigValidationError.errors is the Pydantic error list (for 422 body)."""
|
|
config = {"group_by": [_gb("")]}
|
|
with pytest.raises(ViewConfigValidationError) as exc_info:
|
|
validate_view_config(config)
|
|
assert isinstance(exc_info.value.errors, list)
|
|
assert len(exc_info.value.errors) > 0
|
|
|
|
|
|
def test_validation_error_is_value_error_subclass() -> None:
|
|
"""Subclasses ValueError so existing handlers pick it up."""
|
|
config = {"group_by": [_gb("")]}
|
|
with pytest.raises(ValueError):
|
|
validate_view_config(config)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 2. Route-level 422 — uses a fake service (no DB needed)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _make_fake_view(view_id: str = "v1", table_id: str = "t1") -> View:
|
|
return View(
|
|
id=view_id,
|
|
table_id=table_id,
|
|
name="Test View",
|
|
view_type=ViewType.grid,
|
|
config={},
|
|
)
|
|
|
|
|
|
class _FakeService:
|
|
"""Minimal service stub for route-level tests.
|
|
|
|
Only implements the methods the PATCH /views route touches:
|
|
``get_view`` (existence + ownership check) and ``update_view``.
|
|
"""
|
|
|
|
def __init__(self, *, existing_view: View | None = None) -> None:
|
|
self._existing = existing_view
|
|
self.updated_config: dict[str, object] | None = None
|
|
self.update_called = False
|
|
|
|
async def get_view(self, view_id: str) -> View | None:
|
|
if self._existing is None or self._existing.id != view_id:
|
|
return None
|
|
return self._existing
|
|
|
|
async def get_table(self, table_id: str) -> Any:
|
|
# Ownership check passes — return a table owned by the test user.
|
|
class _T:
|
|
owner_user_id = TEST_USER_ID
|
|
|
|
return _T()
|
|
|
|
async def update_view(self, view_id: str, **kwargs: object) -> View | None:
|
|
self.update_called = True
|
|
self.updated_config = kwargs.get("config") # type: ignore[assignment]
|
|
if self._existing is None:
|
|
return None
|
|
config = kwargs.get("config")
|
|
if isinstance(config, dict):
|
|
self._existing = self._existing.model_copy(update={"config": config})
|
|
return self._existing
|
|
|
|
|
|
@pytest.fixture
|
|
def fake_service_app() -> FastAPI:
|
|
"""App with a fake service on state — bypasses PG entirely."""
|
|
app = FastAPI()
|
|
app.state.bitable_service = _FakeService(existing_view=_make_fake_view())
|
|
app.include_router(bitable_routes.router, prefix="/api/v1")
|
|
app.dependency_overrides[require_bitable_auth] = lambda: {
|
|
"user_id": TEST_USER_ID,
|
|
"username": "testuser",
|
|
"role": "member",
|
|
}
|
|
return app
|
|
|
|
|
|
@pytest.fixture
|
|
async def fake_client(fake_service_app: FastAPI) -> httpx.AsyncClient:
|
|
transport = ASGITransport(app=fake_service_app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as c:
|
|
yield c
|
|
|
|
|
|
async def test_patch_view_with_valid_group_by_returns_200(
|
|
fake_client: httpx.AsyncClient,
|
|
) -> None:
|
|
"""Valid group_by is accepted and persisted (round-trip at the route layer)."""
|
|
config = {"group_by": [{"field_id": "f1", "direction": "asc"}]}
|
|
resp = await fake_client.patch(
|
|
"/api/v1/bitable/views/v1",
|
|
json={"config": config},
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
body = resp.json()
|
|
assert body["success"] is True
|
|
assert body["view"]["config"]["group_by"] == config["group_by"]
|
|
|
|
|
|
async def test_patch_view_with_too_many_group_by_returns_422(
|
|
fake_client: httpx.AsyncClient,
|
|
) -> None:
|
|
"""group_by with >3 fields returns 422 (not 500)."""
|
|
config = {"group_by": [{"field_id": f"f{i}"} for i in range(4)]}
|
|
resp = await fake_client.patch(
|
|
"/api/v1/bitable/views/v1",
|
|
json={"config": config},
|
|
)
|
|
assert resp.status_code == 422
|
|
body = resp.json()
|
|
assert "errors" in body["detail"]
|
|
|
|
|
|
async def test_patch_view_with_invalid_direction_returns_422(
|
|
fake_client: httpx.AsyncClient,
|
|
) -> None:
|
|
"""Invalid direction value returns 422."""
|
|
resp = await fake_client.patch(
|
|
"/api/v1/bitable/views/v1",
|
|
json={"config": {"group_by": [{"field_id": "f1", "direction": "sideways"}]}},
|
|
)
|
|
assert resp.status_code == 422
|
|
|
|
|
|
async def test_patch_view_with_non_u5_config_returns_200(
|
|
fake_client: httpx.AsyncClient,
|
|
) -> None:
|
|
"""Non-U5 config keys (filters / sort / hidden_fields) pass through unchanged."""
|
|
resp = await fake_client.patch(
|
|
"/api/v1/bitable/views/v1",
|
|
json={"config": {"hidden_fields": ["f1", "f2"]}},
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 3. Round-trip via real PG — skipped if PG unavailable
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.postgres
|
|
async def test_round_trip_group_by_via_real_service(bitable_service) -> None:
|
|
"""PATCH config with group_by → GET view returns same group_by.
|
|
|
|
Requires PostgreSQL. Verifies the config dict survives the JSONB
|
|
round-trip and the route-layer validation does not strip fields.
|
|
"""
|
|
from agentkit.bitable.service import BitableService
|
|
|
|
assert isinstance(bitable_service, BitableService)
|
|
table = await bitable_service.create_table(name="U5 Round-Trip")
|
|
view = await bitable_service.create_view(table.id, name="V1")
|
|
|
|
config = {
|
|
"group_by": [
|
|
{"field_id": "f1", "direction": "asc"},
|
|
{"field_id": "f2", "direction": "desc"},
|
|
],
|
|
}
|
|
# Validator accepts the config (no exception).
|
|
validate_view_config(config)
|
|
# Service persists it as-is.
|
|
updated = await bitable_service.update_view(view.id, config=config)
|
|
assert updated is not None
|
|
assert updated.config.get("group_by") == config["group_by"]
|
|
|
|
# GET view returns the same group_by.
|
|
fetched = await bitable_service.get_view(view.id)
|
|
assert fetched is not None
|
|
assert fetched.config.get("group_by") == config["group_by"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Direct schema construction (covers Literal type errors at the model layer)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_group_by_item_rejects_invalid_direction_via_model() -> None:
|
|
"""Constructing GroupByItem with an invalid direction raises ValidationError."""
|
|
with pytest.raises(ValidationError):
|
|
GroupByItem(field_id="f1", direction="sideways") # type: ignore[arg-type]
|
|
|
|
|
|
def test_view_config_schema_max_length_enforced() -> None:
|
|
"""ViewConfigSchema.group_by max_length=3 enforced at the model level."""
|
|
too_many = [{"field_id": f"f{i}"} for i in range(MAX_GROUP_BY_FIELDS + 1)]
|
|
with pytest.raises(ValidationError):
|
|
ViewConfigSchema.model_validate({"group_by": too_many})
|