feat: API for semantic layers

This commit is contained in:
Beto Dealmeida
2026-02-11 16:02:37 -05:00
parent 913259299e
commit 6f93e1cbb1
15 changed files with 1719 additions and 9 deletions

View File

@@ -18,7 +18,11 @@
import pytest
from marshmallow import ValidationError
from superset.semantic_layers.schemas import SemanticViewPutSchema
from superset.semantic_layers.schemas import (
SemanticLayerPostSchema,
SemanticLayerPutSchema,
SemanticViewPutSchema,
)
def test_semantic_view_put_schema_both_fields() -> None:
@@ -77,3 +81,128 @@ def test_semantic_view_put_schema_unknown_field() -> None:
with pytest.raises(ValidationError) as exc_info:
schema.load({"unknown_field": "value"})
assert "unknown_field" in exc_info.value.messages
# =============================================================================
# SemanticLayerPostSchema tests
# =============================================================================
def test_post_schema_all_fields() -> None:
"""Test loading all fields."""
schema = SemanticLayerPostSchema()
result = schema.load(
{
"name": "My Layer",
"description": "A layer",
"type": "snowflake",
"configuration": {"account": "test"},
"cache_timeout": 300,
}
)
assert result["name"] == "My Layer"
assert result["type"] == "snowflake"
assert result["configuration"] == {"account": "test"}
assert result["cache_timeout"] == 300
def test_post_schema_required_fields_only() -> None:
"""Test loading with only required fields."""
schema = SemanticLayerPostSchema()
result = schema.load(
{
"name": "My Layer",
"type": "snowflake",
"configuration": {"account": "test"},
}
)
assert result["name"] == "My Layer"
assert "description" not in result
assert "cache_timeout" not in result
def test_post_schema_missing_name() -> None:
"""Test that missing name raises ValidationError."""
schema = SemanticLayerPostSchema()
with pytest.raises(ValidationError) as exc_info:
schema.load({"type": "snowflake", "configuration": {}})
assert "name" in exc_info.value.messages
def test_post_schema_missing_type() -> None:
"""Test that missing type raises ValidationError."""
schema = SemanticLayerPostSchema()
with pytest.raises(ValidationError) as exc_info:
schema.load({"name": "My Layer", "configuration": {}})
assert "type" in exc_info.value.messages
def test_post_schema_missing_configuration() -> None:
"""Test that missing configuration raises ValidationError."""
schema = SemanticLayerPostSchema()
with pytest.raises(ValidationError) as exc_info:
schema.load({"name": "My Layer", "type": "snowflake"})
assert "configuration" in exc_info.value.messages
def test_post_schema_null_description() -> None:
"""Test that description accepts None."""
schema = SemanticLayerPostSchema()
result = schema.load(
{
"name": "My Layer",
"type": "snowflake",
"configuration": {},
"description": None,
}
)
assert result["description"] is None
# =============================================================================
# SemanticLayerPutSchema tests
# =============================================================================
def test_put_schema_all_fields() -> None:
"""Test loading all fields."""
schema = SemanticLayerPutSchema()
result = schema.load(
{
"name": "Updated",
"description": "New desc",
"configuration": {"account": "new"},
"cache_timeout": 600,
}
)
assert result["name"] == "Updated"
assert result["configuration"] == {"account": "new"}
def test_put_schema_empty() -> None:
"""Test loading empty payload."""
schema = SemanticLayerPutSchema()
result = schema.load({})
assert result == {}
def test_put_schema_name_only() -> None:
"""Test loading with only name."""
schema = SemanticLayerPutSchema()
result = schema.load({"name": "New Name"})
assert result == {"name": "New Name"}
def test_put_schema_configuration_only() -> None:
"""Test loading with only configuration."""
schema = SemanticLayerPutSchema()
result = schema.load({"configuration": {"key": "value"}})
assert result == {"configuration": {"key": "value"}}
def test_put_schema_unknown_field() -> None:
"""Test that unknown fields raise ValidationError."""
schema = SemanticLayerPutSchema()
with pytest.raises(ValidationError) as exc_info:
schema.load({"unknown_field": "value"})
assert "unknown_field" in exc_info.value.messages