mirror of
https://github.com/apache/superset.git
synced 2026-05-07 08:54:23 +00:00
297 lines
10 KiB
Python
297 lines
10 KiB
Python
# Licensed to the Apache Software Foundation (ASF) under one
|
|
# or more contributor license agreements. See the NOTICE file
|
|
# distributed with this work for additional information
|
|
# regarding copyright ownership. The ASF licenses this file
|
|
# to you under the Apache License, Version 2.0 (the
|
|
# "License"); you may not use this file except in compliance
|
|
# with the License. You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing,
|
|
# software distributed under the License is distributed on an
|
|
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
|
|
# KIND, either express or implied. See the License for the
|
|
# specific language governing permissions and limitations
|
|
# under the License.
|
|
|
|
import pytest
|
|
from marshmallow import ValidationError
|
|
|
|
from superset.semantic_layers.schemas import (
|
|
SemanticLayerPostSchema,
|
|
SemanticLayerPutSchema,
|
|
SemanticViewPostSchema,
|
|
SemanticViewPutSchema,
|
|
)
|
|
|
|
|
|
def test_semantic_view_put_schema_both_fields() -> None:
|
|
"""Test loading both description and cache_timeout."""
|
|
schema = SemanticViewPutSchema()
|
|
result = schema.load({"description": "A description", "cache_timeout": 300})
|
|
assert result == {"description": "A description", "cache_timeout": 300}
|
|
|
|
|
|
def test_semantic_view_put_schema_description_only() -> None:
|
|
"""Test loading with only description."""
|
|
schema = SemanticViewPutSchema()
|
|
result = schema.load({"description": "Just a description"})
|
|
assert result == {"description": "Just a description"}
|
|
|
|
|
|
def test_semantic_view_put_schema_cache_timeout_only() -> None:
|
|
"""Test loading with only cache_timeout."""
|
|
schema = SemanticViewPutSchema()
|
|
result = schema.load({"cache_timeout": 600})
|
|
assert result == {"cache_timeout": 600}
|
|
|
|
|
|
def test_semantic_view_put_schema_empty() -> None:
|
|
"""Test loading empty payload."""
|
|
schema = SemanticViewPutSchema()
|
|
result = schema.load({})
|
|
assert result == {}
|
|
|
|
|
|
def test_semantic_view_put_schema_null_description() -> None:
|
|
"""Test that description accepts None."""
|
|
schema = SemanticViewPutSchema()
|
|
result = schema.load({"description": None})
|
|
assert result == {"description": None}
|
|
|
|
|
|
def test_semantic_view_put_schema_null_cache_timeout() -> None:
|
|
"""Test that cache_timeout accepts None."""
|
|
schema = SemanticViewPutSchema()
|
|
result = schema.load({"cache_timeout": None})
|
|
assert result == {"cache_timeout": None}
|
|
|
|
|
|
def test_semantic_view_put_schema_invalid_cache_timeout() -> None:
|
|
"""Test that non-integer cache_timeout raises ValidationError."""
|
|
schema = SemanticViewPutSchema()
|
|
with pytest.raises(ValidationError) as exc_info:
|
|
schema.load({"cache_timeout": "not_a_number"})
|
|
assert "cache_timeout" in exc_info.value.messages
|
|
|
|
|
|
def test_semantic_view_put_schema_unknown_field() -> None:
|
|
"""Test that unknown fields raise ValidationError."""
|
|
schema = SemanticViewPutSchema()
|
|
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
|
|
|
|
|
|
# =============================================================================
|
|
# SemanticViewPostSchema tests
|
|
# =============================================================================
|
|
|
|
|
|
def test_semantic_view_post_schema_all_fields() -> None:
|
|
"""Test loading all SemanticViewPostSchema fields."""
|
|
schema = SemanticViewPostSchema()
|
|
result = schema.load(
|
|
{
|
|
"name": "Orders View",
|
|
"semantic_layer_uuid": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
|
|
"configuration": {"table": "orders"},
|
|
"description": "View over orders",
|
|
"cache_timeout": 120,
|
|
}
|
|
)
|
|
assert result == {
|
|
"name": "Orders View",
|
|
"semantic_layer_uuid": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
|
|
"configuration": {"table": "orders"},
|
|
"description": "View over orders",
|
|
"cache_timeout": 120,
|
|
}
|
|
|
|
|
|
def test_semantic_view_post_schema_required_fields_with_default_configuration() -> None:
|
|
"""Test required fields and configuration load_default behavior."""
|
|
schema = SemanticViewPostSchema()
|
|
result = schema.load(
|
|
{
|
|
"name": "Orders View",
|
|
"semantic_layer_uuid": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
|
|
}
|
|
)
|
|
assert result == {
|
|
"name": "Orders View",
|
|
"semantic_layer_uuid": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
|
|
"configuration": {},
|
|
}
|
|
|
|
|
|
def test_semantic_view_post_schema_missing_name() -> None:
|
|
"""Test missing name validation."""
|
|
schema = SemanticViewPostSchema()
|
|
with pytest.raises(ValidationError) as exc_info:
|
|
schema.load({"semantic_layer_uuid": "abc", "configuration": {}})
|
|
assert "name" in exc_info.value.messages
|
|
|
|
|
|
def test_semantic_view_post_schema_missing_semantic_layer_uuid() -> None:
|
|
"""Test missing semantic_layer_uuid validation."""
|
|
schema = SemanticViewPostSchema()
|
|
with pytest.raises(ValidationError) as exc_info:
|
|
schema.load({"name": "Orders View", "configuration": {}})
|
|
assert "semantic_layer_uuid" in exc_info.value.messages
|
|
|
|
|
|
def test_semantic_view_post_schema_null_optional_fields() -> None:
|
|
"""Test optional nullable fields accept None."""
|
|
schema = SemanticViewPostSchema()
|
|
result = schema.load(
|
|
{
|
|
"name": "Orders View",
|
|
"semantic_layer_uuid": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
|
|
"description": None,
|
|
"cache_timeout": None,
|
|
}
|
|
)
|
|
assert result["description"] is None
|
|
assert result["cache_timeout"] is None
|
|
|
|
|
|
def test_semantic_view_post_schema_unknown_field() -> None:
|
|
"""Test unknown field validation for SemanticViewPostSchema."""
|
|
schema = SemanticViewPostSchema()
|
|
with pytest.raises(ValidationError) as exc_info:
|
|
schema.load(
|
|
{
|
|
"name": "Orders View",
|
|
"semantic_layer_uuid": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
|
|
"unknown_field": "value",
|
|
}
|
|
)
|
|
assert "unknown_field" in exc_info.value.messages
|