refactor(system): Adds reserved namespaces and logging to tag filtering

Signed-off-by: Roel <75250264+RoelBollens-TomTom@users.noreply.github.com>

Author: RoelBollens-TomTom

packages/overture-schema-core/src/overture/schema/core/tag_providers.py

@@ -37,7 +37,7 @@ def theme_provider(
     model_class: type[BaseModel], key: ModelKey, tags: set[str]
 ) -> set[str]:
     for tp in _extract_types(model_class):
-        if issubclass(tp, OvertureFeature):
+        if isinstance(tp, type) and issubclass(tp, OvertureFeature):
             tags.add(
                 "overture:theme=" + get_args(tp.model_fields["theme"].annotation)[0]
             )

packages/overture-schema-system/src/overture/schema/system/discovery.py

@@ -5,7 +5,7 @@
 import re
 from collections.abc import Callable
 from dataclasses import dataclass, replace
-from typing import Annotated, Any, Literal, Union, get_args, get_origin
+from typing import Annotated, Any, Literal, TypeAlias, Union, get_args, get_origin
 
 from pydantic import BaseModel
 
@@ -18,6 +18,11 @@
     "overture": {"overture-schema-core"},
     "feature": {"overture-schema-system"},
 }
+RESERVED_NAMESPACES: dict[str, set[str]] = {
+    "overture": {"overture-schema-core"},
+    "system": {"overture-schema-system"},
+}
+
 TAG = r"[a-z0-9][a-z0-9_-]*"
 NAMESPACE_TAG = r"[a-z0-9]+:[a-z0-9]+(?:=[a-z0-9_.-]+)?"
 TAG_RE = re.compile(rf"^(?:{TAG}|{NAMESPACE_TAG})$")
@@ -61,13 +66,19 @@ class TagProviderKey:
     package_name: str  # distribution package name
 
 
-TagProvider = Callable[[type[BaseModel], ModelKey, set[str]], set[str]]
+TagProvider: TypeAlias = Callable[[type[BaseModel], ModelKey, set[str]], set[str]]
+
+ModelDict: TypeAlias = dict[ModelKey, type[BaseModel]]
+
+TagProviderDict: TypeAlias = dict[TagProviderKey, TagProvider]
+
+ModelKeyFilter: TypeAlias = Callable[[ModelKey], bool]
 
 
 def generate_tags(
     model_class: type[BaseModel],
     key: ModelKey,
-    providers: dict[TagProviderKey, TagProvider],
+    providers: TagProviderDict,
 ) -> set[str]:
     tags: set[str] = set()
 
@@ -85,16 +96,53 @@ def generate_tags(
 
 
 def _filter_tags(tags: set[str], provider: TagProviderKey) -> set[str]:
-    reserved_tags = tuple(
-        tag for tag, dist in RESERVED_TAGS.items() if provider.package_name not in dist
-    )
+    """Filter tags, removing invalid, reserved, or namespace-restricted tags for a provider."""
+    filtered_tags: set[str] = set()
+    reserved_tags: set[str] = {
+        tag for tag, pkgs in RESERVED_TAGS.items() if provider.package_name not in pkgs
+    }
+    reserved_namespaces: set[str] = {
+        ns
+        for ns, pkgs in RESERVED_NAMESPACES.items()
+        if provider.package_name not in pkgs
+    }
+
+    for tag in tags:
+        # Validate tag format
+        if not TAG_RE.fullmatch(tag):
+            logger.debug(
+                f"Tag provider '{provider.name}' (package '{provider.package_name}') attempted to set '{tag}' as tag. "
+                f"This tag does not match the required format."
+            )
+            continue
+
+        # Reserved tag check
+        if tag in reserved_tags:
+            allowed_pkgs = RESERVED_TAGS.get(tag, set())
+            logger.debug(
+                f"Tag provider '{provider.name}' (package '{provider.package_name}') attempted to set reserved tag '{tag}'. "
+                f"This tag can only be set by packages from: {allowed_pkgs}."
+            )
+            continue
+
+        # Reserved namespace check
+        tag_ns = namespace(tag)
+        if tag_ns and tag_ns in reserved_namespaces:
+            allowed_pkgs = RESERVED_NAMESPACES.get(tag_ns, set())
+            logger.debug(
+                f"Tag provider '{provider.name}' (package '{provider.package_name}') attempted to set tag '{tag}' in reserved namespace '{tag_ns}'. "
+                f"This namespace can only be set by packages from: {allowed_pkgs}."
+            )
+            continue
 
-    return {tag for tag in tags if TAG_RE.match(tag) and tag not in reserved_tags}
+        filtered_tags.add(tag)
+
+    return filtered_tags
 
 
 def discover_tag_providers(
     tag_providers_group: str = "overture.tag_providers",
-) -> dict[TagProviderKey, TagProvider]:
+) -> TagProviderDict:
     tag_providers = {}
 
     try:
@@ -123,7 +171,7 @@ def discover_tag_providers(
 
 def discover_models(
     model_group: str = "overture.models",
-) -> dict[ModelKey, type[BaseModel]]:
+) -> ModelDict:
     """Discover all registered Overture models via entry points.
 
     Parameters
@@ -173,13 +221,13 @@ def discover_models(
 
 
 def filter_models(
-    models: dict[ModelKey, type[BaseModel]],
+    models: ModelDict,
     tags: tuple[str, ...] = (),
     excluded_tags: tuple[str, ...] = (),
     type_names: tuple[str, ...] = (),
-) -> dict[ModelKey, type[BaseModel]]:
+) -> ModelDict:
     """Filter models to those that contain all required tags."""
-    filters = []
+    filters: list[ModelKeyFilter] = []
 
     if tags:
         filters.append(lambda key: all(tag in key.tags for tag in tags))
@@ -221,6 +269,15 @@ def get_registered_model(feature_type: str) -> type[BaseModel] | None:
     return None
 
 
+def namespace(tag: str) -> str:
+    """Extract the namespace from a tag, or return an empty string if there is no namespace."""
+    if not TAG_RE.fullmatch(tag):
+        raise ValueError(f"Invalid tag format: {tag}")
+    if ":" in tag:
+        return tag.split(":")[0]
+    return ""
+
+
 def tags_by_key(tags: frozenset[str] | set[str], key: str) -> set[str]:
     """Extract values for k/v tags with the given key.
 
@@ -244,7 +301,10 @@ def tags_by_namespace(tags: frozenset[str] | set[str], namespace: str) -> set[st
 def feature_provider(
     model_class: type[BaseModel], key: ModelKey, tags: set[str]
 ) -> set[str]:
-    if any(issubclass(tp, Feature) for tp in _extract_types(model_class)):
+    if any(
+        isinstance(tp, type) and issubclass(tp, Feature)
+        for tp in _extract_types(model_class)
+    ):
         tags.add("feature")
     return tags
 

packages/overture-schema-system/tests/test_tag_providers.py

@@ -0,0 +1,117 @@
+import pytest
+from pydantic import BaseModel
+
+from overture.schema.system.discovery import (
+    ModelKey,
+    TagProviderKey,
+    _filter_tags,
+    feature_provider,
+)
+from overture.schema.system.feature import Feature
+
+
+@pytest.fixture
+def core_tag_provider() -> TagProviderKey:
+    return TagProviderKey(
+        name="core", entry_point="core:Provider", package_name="overture-schema-core"
+    )
+
+
+@pytest.fixture
+def system_tag_provider() -> TagProviderKey:
+    return TagProviderKey(
+        name="system",
+        entry_point="system:Provider",
+        package_name="overture-schema-system",
+    )
+
+
+@pytest.fixture
+def other_tag_provider() -> TagProviderKey:
+    return TagProviderKey(
+        name="other", entry_point="other:Provider", package_name="other-package"
+    )
+
+
+@pytest.fixture
+def feature() -> type[Feature]:
+    class SomeFeature(Feature):
+        pass
+
+    return SomeFeature
+
+
+@pytest.fixture
+def not_a_feature() -> type[BaseModel]:
+    class NotAFeature(BaseModel):
+        pass
+
+    return NotAFeature
+
+
+def test_valid_tags(other_tag_provider: TagProviderKey) -> None:
+    tags = {"valid", "other:valid", "other:valid=true"}
+    filtered = _filter_tags(tags, other_tag_provider)
+    assert filtered == tags
+
+
+def test_invalid_tag(other_tag_provider: TagProviderKey) -> None:
+    tags = {"InvalidTag"}
+    filtered = _filter_tags(tags, other_tag_provider)
+    assert filtered == set()
+
+
+def test_reserved_tag(other_tag_provider: TagProviderKey) -> None:
+    tags = {"overture", "feature", "valid"}
+    filtered = _filter_tags(tags, other_tag_provider)
+    assert "valid" in filtered
+    assert "overture" not in filtered
+    assert "feature" not in filtered
+
+
+def test_allowed_reserved_tag(
+    core_tag_provider: TagProviderKey, system_tag_provider: TagProviderKey
+) -> None:
+    assert "overture" in _filter_tags({"overture"}, core_tag_provider)
+    assert "feature" in _filter_tags({"feature"}, system_tag_provider)
+
+
+def test_reserved_namespace(other_tag_provider: TagProviderKey) -> None:
+    tags = {"overture:feature", "system:feature", "valid:tag"}
+    filtered = _filter_tags(tags, other_tag_provider)
+    assert "valid:tag" in filtered
+    assert "overture:feature" not in filtered
+    assert "system:feature" not in filtered
+
+
+def test_allowed_reserved_namespace(
+    core_tag_provider: TagProviderKey, system_tag_provider: TagProviderKey
+) -> None:
+    assert "overture:feature" in _filter_tags({"overture:feature"}, core_tag_provider)
+    assert "system:feature" in _filter_tags({"system:feature"}, system_tag_provider)
+
+
+def test_empty_tags(other_tag_provider: TagProviderKey) -> None:
+    assert _filter_tags(set(), other_tag_provider) == set()
+
+
+def test_mixed_tags(other_tag_provider: TagProviderKey) -> None:
+    tags = {"valid", "feature", "overture:feature", "InvalidTag"}
+    filtered = _filter_tags(tags, other_tag_provider)
+    assert filtered == {"valid"}
+
+
+def test_feature_provider_adds_feature_tag(feature: type[Feature]) -> None:
+    key = ModelKey(name="feature", entry_point="system:Feature", tags=frozenset())
+    result = feature_provider(feature, key, set())
+    assert "feature" in result
+
+
+def test_feature_provider_does_not_add_feature_tag(
+    not_a_feature: type[BaseModel],
+) -> None:
+    key = ModelKey(
+        name="notafeature", entry_point="system:NotAFeature", tags=frozenset()
+    )
+    result = feature_provider(not_a_feature, key, set())
+    assert "feature" not in result

packages/overture-schema-system/tests/test_tags.py

@@ -1,4 +1,13 @@
-from overture.schema.system.discovery import tags_by_key, tags_by_namespace
+import re
+import unittest
+
+from overture.schema.system.discovery import (
+    NAMESPACE_TAG,
+    TAG,
+    TAG_RE,
+    tags_by_key,
+    tags_by_namespace,
+)
 
 
 def test_tags_by_key_returns_correct_values() -> None:
@@ -41,3 +50,71 @@ def test_tags_by_namespace_handles_empty_tags() -> None:
     namespace = "system"
     result = tags_by_namespace(tags, namespace)
     assert result == set()
+
+
+class TestSimpleTagRegex(unittest.TestCase):
+    def test_valid_simple_tags(self) -> None:
+        valid_tags = [
+            "v",
+            "valid",
+            "valid1",
+            "valid_tag",
+            "valid-tag",
+            "0valid",
+            "42",
+        ]
+        for tag in valid_tags:
+            self.assertTrue(re.fullmatch(TAG, tag), f"Should match: {tag}")
+            self.assertTrue(TAG_RE.fullmatch(tag), f"TAG_RE should match: {tag}")
+
+    def test_invalid_simple_tags(self) -> None:
+        invalid_tags = [
+            "",
+            "_invalid",
+            "-invalid",
+            "Invalid",
+            "invalid!",
+            "invalid ",
+            "in.valid",
+            "3.14",
+        ]
+        for tag in invalid_tags:
+            self.assertFalse(re.fullmatch(TAG, tag), f"Should not match: {tag}")
+            self.assertFalse(TAG_RE.fullmatch(tag), f"TAG_RE should not match: {tag}")
+
+
+class TestNamespaceTagRegex(unittest.TestCase):
+    def test_valid_namespace_tags(self) -> None:
+        valid_tags = [
+            "ns:predicate",
+            "ns:predicate1",
+            "ns:predicate=value",
+            "ns:predicate=value_0",
+            "ns:predicate=value-0",
+            "ns:predicate=value.0",
+            "ns:predicate=value_2-3.4",
+            "ns:predicate=42",
+            "ns:predicate=3.14",
+        ]
+        for tag in valid_tags:
+            self.assertTrue(re.fullmatch(NAMESPACE_TAG, tag), f"Should match: {tag}")
+            self.assertTrue(TAG_RE.fullmatch(tag), f"TAG_RE should match: {tag}")
+
+    def test_invalid_namespace_tags(self) -> None:
+        invalid_tags = [
+            "ns:",
+            ":predicate",
+            "ns:predicate=",
+            "ns:predicate=Value",
+            "ns:predicate=value ",
+            "ns:predicate=value!",
+            "ns:predicate=ns:value",
+            "ns:predicate=predicate=value",
+            "Ns:predicate",
+            "ns:Predicate",
+        ]
+        for tag in invalid_tags:
+            self.assertFalse(
+                re.fullmatch(NAMESPACE_TAG, tag), f"Should not match: {tag}"
+            )
+            self.assertFalse(TAG_RE.fullmatch(tag), f"TAG_RE should not match: {tag}")