You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@superset.apache.org by mi...@apache.org on 2026/10/07 10:38:01 UTC
(superset) branch master updated: feat(semantic): add optional metadata refresh SDK contract (#44834)
This is an automated email from the ASF dual-hosted git repository.
aminghadersohi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/superset.git
The following commit(s) were added to refs/heads/master by this push:
new d526ee4bd1f feat(semantic): add optional metadata refresh SDK contract (#44834)
d526ee4bd1f is described below
commit d526ee4bd1f38ee28bd2cbb82be9bbd57e1bf30c
Author: Mike Bridge <mi...@preset.io>
AuthorDate: Wed Oct 7 04:37:54 2026 -0600
feat(semantic): add optional metadata refresh SDK contract (#44834)
Signed-off-by: Mike Bridge <mi...@preset.io>
---
.../extensions/contribution-types.md | 3 +
docs/developer_docs/semantic-metadata-contract.md | 187 ++++++
.../src/superset_core/semantic_layers/layer.py | 15 +
.../src/superset_core/semantic_layers/metadata.py | 196 ++++++
.../src/superset_core/semantic_layers/view.py | 11 +
.../semantic_layers/metadata_contract_test.py | 694 +++++++++++++++++++++
6 files changed, 1106 insertions(+)
diff --git a/docs/developer_docs/extensions/contribution-types.md b/docs/developer_docs/extensions/contribution-types.md
index 4c8fe44c781..350137326a6 100644
--- a/docs/developer_docs/extensions/contribution-types.md
+++ b/docs/developer_docs/extensions/contribution-types.md
@@ -245,6 +245,9 @@ See [MCP Integration](./mcp.md) for implementation details.
### Semantic Layers
+For optional catalog acquisition and host publication, see the
+[semantic provider metadata contract](../semantic-metadata-contract.md).
+
Extensions can register custom semantic layer implementations that allow Superset to connect to external data modeling frameworks. Each semantic layer defines how to authenticate, discover semantic views (tables/metrics/dimensions), and execute queries against the external system.
```python
diff --git a/docs/developer_docs/semantic-metadata-contract.md b/docs/developer_docs/semantic-metadata-contract.md
new file mode 100644
index 00000000000..40be00dce05
--- /dev/null
+++ b/docs/developer_docs/semantic-metadata-contract.md
@@ -0,0 +1,187 @@
+---
+title: Semantic provider metadata contract
+---
+
+<!--
+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.
+-->
+
+# Semantic provider metadata contract
+
+The optional interfaces in `superset_core.semantic_layers.metadata` let a provider
+use a host-owned catalog store. This SDK defines the boundary; it does not supply
+storage, authorization, endpoints or a refresh UI. Existing providers need no new
+overrides.
+
+## Opting in
+
+`SemanticLayer.supports_metadata_refresh(configuration)` defaults to `False`.
+Overrides must determine support without construction, network requests or cache
+writes. `SemanticLayer.metadata_refresh` defaults to `None`; an opted-in provider
+returns a `MetadataRefreshAdapter`. The property must return the same instance
+for the layer's lifetime; views and runtime-schema discovery use that bound
+instance. If support is declared but the instance is absent, an enabled host must
+reject the configuration rather than silently using legacy cache keys.
+
+The host authorizes and resolves the stored connection before constructing the
+provider. It calls `adapter.bind(store, deadline=deadline)` once, before discovery. Rebinding raises
+`MetadataRefreshError("configuration")`; an unbound refresh raises
+`MetadataRefreshError("unsupported")`. Invalid or exhausted bind deadlines raise
+`MetadataRefreshError("deadline")` before binding and leave the adapter unbound.
+
+Full and custom views use the bound store. The adapter's instance
+`get_runtime_schema(runtime_data=None)` returns the existing schema shape using
+that same store, avoiding the legacy classmethod's unbound construction path.
+Unsaved configuration previews cannot publish into a stored connection namespace.
+A bound store failure must not silently fall back to provider-local caching.
+Existing unbound preview/classmethod discovery remains a provider responsibility;
+it never publishes into the stored connection namespace. The in-memory example
+below demonstrates the bound path, not a complete preview implementation.
+
+One bound scope contains one catalog covering that connection's supported runtime
+configurations. A provider needing different catalogs per runtime database/schema
+must not opt in until its scope can satisfy that requirement.
+
+Discovery must be identical for every user of that connection scope. Providers
+whose discovery depends on per-user OAuth credentials, impersonation or another
+user-specific identity must not declare `supports_metadata_refresh`; the shared
+catalog has no per-user dimension. Query authorization and result-cache user/RLS
+identity remain separate host responsibilities.
+
+## Data exchanged
+
+| Type | Fields / signature | Meaning |
+| --- | --- | --- |
+| `CatalogLoader` | `Callable[[float], str]` | Acquisition within a host-supplied absolute monotonic deadline returning provider-validated canonical JSON; valid empty data is allowed |
+| `CatalogSnapshot` | `payload: str`, `cache_token: str`, `observed_at: str` | Immutable observation and its captured scope-qualified identity |
+| `MetadataRefreshResult` | `status: Literal["changed", "unchanged"]`, `snapshot: CatalogSnapshot` | Confirmed publication, not just completed acquisition |
+| `MetadataSnapshotStore.read` | `(fetch: CatalogLoader, *, deadline: float) -> CatalogSnapshot` | Valid observation or bounded coordinated acquisition |
+| `MetadataRefreshAdapter.bind` | `(store: MetadataSnapshotStore, *, deadline: float) -> None` | Bind once and capture the host operation's discovery deadline |
+| `MetadataRefreshAdapter.refresh` | `(*, deadline: float) -> MetadataRefreshResult` | Pass the caller's budget to the bound store |
+| `MetadataSnapshotStore.refresh` | `(fetch: CatalogLoader, *, deadline: float) -> MetadataRefreshResult` | Bypass a hit and confirm a fresh publication |
+
+The provider validates and interprets its payload. The host owns scope, expiry,
+authorization and publication. These records do not validate vendor JSON or
+implement coordination. `observed_at` is a UTC RFC3339 source-observation time,
+not entry creation, expiry or an ordering authority. `CatalogSnapshot` rejects
+non-string fields and empty tokens/timestamps with fixed messages that do not echo
+input. Timestamp syntax remains the producer's responsibility; this SDK record
+does not parse RFC3339 or validate canonical JSON.
+
+Every successful publication gets a new nonempty opaque `cache_token`, including
+unchanged discovery. Cache hits retain the captured token. Equal discovery data
+cannot prove equal metric definitions if the upstream selector omits complete
+query semantics. `changed` includes installation without a prior observation.
+
+`SemanticView.metadata_cache_token` defaults to `None` for legacy views. A bound
+opted-in view returns the token captured with its parsed members. Hosts reject
+missing tokens in that mode and include the captured token in metadata-dependent
+compatibility and result-cache keys, preserving view/query/access identities.
+An old result must never be labelled with a token fetched later. Host/provider
+implementations must enforce and test these obligations.
+
+## Failures and concurrency
+
+`MetadataRefreshErrorCategory` is the typed set accepted by `MetadataRefreshError`:
+`unsupported`, `configuration`, `in_progress`, `configuration_changed`, `upstream`,
+`invalid_payload`, `deadline`, `unavailable` and `indeterminate`. Producers pass
+these categories, not raw vendor messages or credentials; the constructor rejects
+unknown values with a fixed error that does not echo them. Hosts must not expose
+exception cause chains. Providers translating sensitive vendor errors should use
+`raise MetadataRefreshError("upstream") from None`. HTTP status codes and
+localized text belong to the host.
+
+- `configuration_changed`: the authorized connection scope/configuration changed during the operation; discard the candidate rather than publish under obsolete authority.
+- `invalid_payload`: acquisition or a stored observation cannot satisfy the provider's catalog schema; never publish partial metadata or an error as an empty catalog.
+- `unavailable`: host-owned metadata storage or coordination cannot be reached or used; `upstream` instead identifies failure to acquire metadata from the provider's remote service.
+
+The host establishes one finite absolute `float` deadline using
+`time.monotonic()` in the calling process before waiting or acquisition. It passes
+that value explicitly to `adapter.bind(store, deadline=deadline)`,
+`adapter.refresh(deadline=deadline)`,
+`store.read(fetch, deadline=deadline)` and `store.refresh(fetch, deadline=deadline)`.
+The adapter forwards it to the store, and the store passes it to `fetch(deadline)`
+unchanged. Multiple calls within the same operation share this deadline through
+publication; neither the store nor nested acquisition may mint a new budget.
+Implementations use `remaining_budget` to reject non-finite, expired or
+implausibly large deadlines before I/O, including cache reads. The protocol
+signatures do not execute these checks on behalf of implementations.
+The provider checks
+the remaining time before each upstream operation and uses the tighter of that
+budget and its own transport limit. Exhaustion raises
+`MetadataRefreshError("deadline")` before further I/O; nested calls must not restart
+the budget. Checking only after an unbounded request is insufficient. Actual
+transport and publication budget enforcement require host/provider tests.
+
+The existing layer/view and runtime-schema discovery signatures stay compatible.
+Ordinary `from_configuration(configuration)` construction remains deadline-free.
+The host hands the operation's deadline to `bind`; the adapter captures it for
+full/custom-view and runtime-schema discovery, passing it unchanged to store
+reads. A new operation requires a fresh provider/adapter instance and a new bind,
+not rebinding or renewing an existing instance's discovery budget. The explicit
+store/refresh parameters remain required. The example exercises normal provider
+construction followed by this bind-time handoff.
+
+Use `remaining_budget(deadline)` before I/O and publication. It returns the
+positive remaining seconds and raises `MetadataRefreshError("deadline")` for
+non-finite, expired or implausibly large budgets. The SDK ceiling
+`MAX_METADATA_BUDGET_SECONDS = 300.0` permits slow metadata discovery while
+catching common `time.time()`/monotonic clock confusion; it is not a default or a
+reason to extend the caller's budget. Longer requests must be split into bounded
+operations. Hosts/providers may use tighter limits. For a host with an injected
+monotonic clock, `remaining_budget(deadline, now=clock())` uses that same clock
+without introducing host types into the SDK.
+
+This value is neither a duration nor a UTC timestamp. Do not serialize it across
+processes, use it as cache expiry, or use it to order publications. A deadline
+cannot cancel arbitrary synchronous code and never replaces a publication fence.
+
+An active explicit refresh can report `in_progress`. Cold readers wait/re-read
+within a finite deadline or fail safely. A delayed writer cannot replace a later
+accepted observation. Failure never installs an error as an empty catalog or
+renews freshness. Unconfirmed publication reports `indeterminate`; a timeout proves
+neither success nor rollback. These guarantees require real host backend tests.
+
+## Separate host cache controls and diagnostics
+
+Catalog invalidation, independent compatibility invalidation, selected-query
+result refresh and read-only cache inspection are host responsibilities. No extra
+provider methods are required. Catalog invalidation retires both older writer
+authority and derived-cache validity. Compatibility-only invalidation need not
+fetch metadata; result refresh does not imply catalog refresh.
+
+Inspection reports recorded entry creation and expiry separately from source
+observation. It must not fetch, fill or renew entries. Missing legacy timestamps
+remain unknown. Cache payloads, credentials and raw keys are not diagnostic
+fields. Storage instrumentation, access enforcement and UI presentation are not
+functionality supplied by these SDK records.
+
+## Compatibility and verification
+
+No new abstract methods are added to the existing layer or view bases. Providers
+importing this API must require a released SDK containing it: a disabled feature
+cannot make a missing import safe. Fleet-wide freshness requires a compatible
+host/provider fleet and a coordinated store. Upstream query-version pinning and
+vendor-cache clearing are not promised.
+
+`tests/unit_tests/semantic_layers/metadata_contract_test.py` contains a minimal legacy
+provider and an opted-in in-memory example. It checks captured observations,
+duplicate-label/raw-ID preservation, immutable records, field validation, explicit shared deadlines, error categories and
+imports without host dependencies in a fresh isolated subprocess. The test file
+is collected by the standard unit-test CI lane. The sequential example does not establish
+multiworker or backend correctness.
diff --git a/superset-core/src/superset_core/semantic_layers/layer.py b/superset-core/src/superset_core/semantic_layers/layer.py
index ba81d90b1f6..4ab6d6004a5 100644
--- a/superset-core/src/superset_core/semantic_layers/layer.py
+++ b/superset-core/src/superset_core/semantic_layers/layer.py
@@ -22,6 +22,7 @@ from typing import Any, Generic, TypeVar
from pydantic import BaseModel
+from superset_core.semantic_layers.metadata import MetadataRefreshAdapter
from superset_core.semantic_layers.view import SemanticView
ConfigT = TypeVar("ConfigT", bound=BaseModel)
@@ -35,6 +36,20 @@ class SemanticLayer(ABC, Generic[ConfigT, SemanticViewT]):
configuration_class: type[BaseModel]
+ @classmethod
+ def supports_metadata_refresh(cls, configuration: dict[str, Any]) -> bool:
+ """Declare opt-in support without construction, discovery or other I/O."""
+ return False
+
+ @property
+ def metadata_refresh(self) -> MetadataRefreshAdapter | None:
+ """Return a stable adapter instance, or None for legacy metadata behavior.
+
+ An opted-in layer returns the same adapter for its lifetime. Its views
+ and runtime schema use the instance bound by the host.
+ """
+ return None
+
@classmethod
@abstractmethod
def from_configuration(
diff --git a/superset-core/src/superset_core/semantic_layers/metadata.py b/superset-core/src/superset_core/semantic_layers/metadata.py
new file mode 100644
index 00000000000..7cdfce0613a
--- /dev/null
+++ b/superset-core/src/superset_core/semantic_layers/metadata.py
@@ -0,0 +1,196 @@
+# 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.
+
+"""Optional provider metadata acquisition and host publication interfaces."""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from collections.abc import Callable
+from dataclasses import dataclass, field
+from math import isfinite
+from time import monotonic
+from typing import Any, get_args, Literal, Protocol, TypeAlias
+
+# Five minutes permits slow discovery while rejecting accidental wall-clock deadlines.
+# This is a sanity ceiling, not a default; hosts/providers should use tighter budgets.
+MAX_METADATA_BUDGET_SECONDS: float = 300.0
+
+# The host passes a finite absolute time.monotonic() deadline in this process.
+CatalogLoader: TypeAlias = Callable[[float], str]
+MetadataRefreshErrorCategory: TypeAlias = Literal[
+ "unsupported",
+ "configuration",
+ "in_progress",
+ "configuration_changed",
+ "upstream",
+ "invalid_payload",
+ "deadline",
+ "unavailable",
+ "indeterminate",
+]
+
+
+@dataclass(frozen=True)
+class CatalogSnapshot:
+ """Carry one validated observation and its captured cache identity.
+
+ The provider supplies canonical JSON in payload. The host assigns a nonempty,
+ opaque, scope-qualified cache_token for each successful publication, including
+ unchanged discovery. observed_at is a UTC RFC3339 source-observation time for
+ display, not a cache creation/expiry time or an ordering authority. The producer
+ is responsible for RFC3339 formatting; this record checks type/nonemptiness.
+ """
+
+ payload: str = field(repr=False)
+ cache_token: str = field(repr=False)
+ observed_at: str
+
+ def __post_init__(self) -> None:
+ if not isinstance(self.payload, str):
+ raise TypeError("Catalog payload must be a string")
+ if not isinstance(self.cache_token, str):
+ raise TypeError("Catalog cache_token must be a string")
+ if not isinstance(self.observed_at, str):
+ raise TypeError("Catalog observed_at must be a string")
+ if not self.cache_token:
+ raise ValueError("Catalog cache token must not be empty")
+ if not self.observed_at:
+ raise ValueError("Catalog observed_at must not be empty")
+
+
+@dataclass(frozen=True)
+class MetadataRefreshResult:
+ """Report a confirmed publication and its immutable observation.
+
+ Unchanged compares discovery payloads, not complete upstream metric
+ definitions. Changed includes installation without a prior observation.
+ An indeterminate publication must raise an error instead of returning success.
+ """
+
+ status: Literal["changed", "unchanged"]
+ snapshot: CatalogSnapshot
+
+ def __post_init__(self) -> None:
+ if self.status not in ("changed", "unchanged"):
+ raise ValueError("Unknown metadata refresh result status")
+ if not isinstance(self.snapshot, CatalogSnapshot):
+ raise TypeError("Metadata refresh snapshot must be a CatalogSnapshot")
+
+
+class MetadataRefreshError(Exception):
+ """Carry a validated failure category without vendor messages or credentials.
+
+ Hosts must not expose exception cause chains. Providers translating sensitive
+ upstream exceptions should suppress their displayed chain with ``from None``.
+ """
+
+ def __init__(self, category: MetadataRefreshErrorCategory) -> None:
+ if category not in get_args(MetadataRefreshErrorCategory):
+ raise ValueError("Unknown metadata refresh error category")
+ self.category: MetadataRefreshErrorCategory = category
+ super().__init__(category)
+
+
+def remaining_budget(deadline: float, *, now: float | None = None) -> float:
+ """Return bounded seconds left or reject an invalid monotonic deadline.
+
+ ``now`` supports hosts with an injected monotonic clock; it must use the same
+ process-local clock as deadline. Neither value is a wall-clock timestamp.
+ The sanity ceiling catches epoch timestamps without silently clamping them.
+ """
+ remaining: float = deadline - (monotonic() if now is None else now)
+ if not isfinite(remaining) or not 0 < remaining <= MAX_METADATA_BUDGET_SECONDS:
+ raise MetadataRefreshError("deadline")
+ return remaining
+
+
+class MetadataSnapshotStore(Protocol):
+ """Define host-owned publication for one authorized connection scope.
+
+ The loader receives a finite absolute time.monotonic() deadline and acquires
+ and validates canonical JSON within that budget and any tighter provider limit.
+ The caller sets one deadline before waiting or acquisition and passes it to
+ every store/adapter call; implementations must not reset it for nested calls.
+ Use remaining_budget to reject non-finite, expired or implausibly large budgets
+ with the deadline category before further I/O, including cache reads.
+ The deadline is process-local, never serialized or used for cache expiry or
+ publication ordering; it does not replace writer fencing.
+ The host owns expiry, writer fencing and publication confirmation. This
+ protocol does not provide storage, authorization or distributed coordination.
+ Host-only invalidation and cache inspection are outside the provider interface.
+ """
+
+ def read(self, fetch: CatalogLoader, *, deadline: float) -> CatalogSnapshot:
+ """Read a valid observation or perform bounded coordinated acquisition.
+
+ Cold readers wait/re-read within a finite deadline or fail safely; they
+ must not silently use an obsolete local copy. Cache hits retain identity
+ and expiry. Return the payload and its identity from the same observation.
+ Pass the caller's deadline unchanged to nested refresh and loader calls.
+ """
+ ...
+
+ def refresh(
+ self, fetch: CatalogLoader, *, deadline: float
+ ) -> MetadataRefreshResult:
+ """Bypass a hit and confirm publication with a fresh cache token.
+
+ A busy explicit refresh reports in_progress. Failure must not renew
+ freshness. An unconfirmed outcome reports indeterminate rather than
+ retrying publication blindly or claiming success or rollback.
+ Acquisition and publication share the caller's deadline without renewal.
+ """
+ ...
+
+
+class MetadataRefreshAdapter(ABC):
+ """Opt a provider into host-coordinated metadata discovery and refresh."""
+
+ @abstractmethod
+ def bind(self, store: MetadataSnapshotStore, *, deadline: float) -> None:
+ """Bind once before discovery and reject rebinding as configuration error.
+
+ Participating full/custom views and runtime schema must use this store.
+ The host supplies trusted scope and checks authority before construction.
+ Construction stays deadline-free. Capture the caller's finite absolute
+ monotonic deadline here and pass it unchanged to discovery store reads.
+ Validate with remaining_budget before binding; failure leaves the adapter
+ unbound. A new host operation needs
+ a fresh adapter, not a renewed discovery budget on an existing binding.
+ """
+
+ @abstractmethod
+ def refresh(self, *, deadline: float) -> MetadataRefreshResult:
+ """Acquire validated metadata and request confirmed host publication.
+
+ An unbound adapter reports unsupported. Do not return success after only
+ fetching upstream data or silently fall back to provider-local caching.
+ Pass the caller's finite absolute monotonic deadline unchanged to the
+ store. Validate with remaining_budget before I/O; nested calls never create a
+ new budget.
+ """
+
+ @abstractmethod
+ def get_runtime_schema(
+ self, runtime_data: dict[str, Any] | None = None
+ ) -> dict[str, Any]:
+ """Discover runtime fields through the bound store used by provider views.
+
+ Return the existing runtime-schema shape. This instance hook avoids the
+ legacy classmethod path bypassing the bound observation.
+ """
diff --git a/superset-core/src/superset_core/semantic_layers/view.py b/superset-core/src/superset_core/semantic_layers/view.py
index bf542a6d6dd..7491d312b60 100644
--- a/superset-core/src/superset_core/semantic_layers/view.py
+++ b/superset-core/src/superset_core/semantic_layers/view.py
@@ -70,6 +70,17 @@ class SemanticView(ABC):
# implementations are not required to add a formal @abstractmethod.
name: str
+ @property
+ def metadata_cache_token(self) -> str | None:
+ """Return the identity captured with these members, or None for legacy views.
+
+ A provider using a bound metadata store must return its observation's
+ nonempty token. The host must reject a missing token in that mode rather
+ than silently using legacy cache keys. Never look up a later identity
+ independently of the data used for discovery or compatibility.
+ """
+ return None
+
@abstractmethod
def uid(self) -> str:
"""
diff --git a/tests/unit_tests/semantic_layers/metadata_contract_test.py b/tests/unit_tests/semantic_layers/metadata_contract_test.py
new file mode 100644
index 00000000000..f0352ac99f7
--- /dev/null
+++ b/tests/unit_tests/semantic_layers/metadata_contract_test.py
@@ -0,0 +1,694 @@
+# 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.
+
+from __future__ import annotations
+
+import inspect
+import json # noqa: TID251 -- SDK examples must not import the host JSON utility.
+import subprocess
+import sys
+from collections.abc import Callable
+from dataclasses import FrozenInstanceError
+from pathlib import Path
+from time import monotonic
+from typing import Any, cast, get_args, get_type_hints, Literal
+
+import pyarrow as pa
+import pytest
+from pydantic import BaseModel
+from superset_core.semantic_layers import metadata as metadata_sdk
+from superset_core.semantic_layers.layer import SemanticLayer
+from superset_core.semantic_layers.metadata import (
+ CatalogLoader,
+ CatalogSnapshot,
+ MetadataRefreshAdapter,
+ MetadataRefreshError,
+ MetadataRefreshErrorCategory,
+ MetadataRefreshResult,
+ MetadataSnapshotStore,
+ remaining_budget,
+)
+from superset_core.semantic_layers.types import (
+ Dimension,
+ Filter,
+ Metric,
+ SemanticQuery,
+ SemanticResult,
+)
+from superset_core.semantic_layers.view import SemanticView
+
+
+@pytest.fixture
+def clock(monkeypatch: pytest.MonkeyPatch) -> list[float]:
+ """Control the SDK monotonic clock used by the examples."""
+ times: list[float] = [100.0]
+ monkeypatch.setattr(metadata_sdk, "monotonic", lambda: times[0])
+ return times
+
+
+class LegacyView(SemanticView):
+ name: str = "legacy"
+
+ def uid(self) -> str:
+ return "legacy"
+
+ def get_dimensions(self) -> set[Dimension]:
+ return set()
+
+ def get_metrics(self) -> set[Metric]:
+ return {Metric("orders", "Orders", pa.int64(), "orders")}
+
+ def get_values(
+ self, dimension: Dimension, filters: set[Filter] | None = None
+ ) -> SemanticResult:
+ return SemanticResult([], pa.table({"value": [17]}))
+
+ def get_table(self, query: SemanticQuery) -> SemanticResult:
+ return SemanticResult([], pa.table({"value": [17]}))
+
+ def get_row_count(self, query: SemanticQuery) -> SemanticResult:
+ return SemanticResult([], pa.table({"count": [1]}))
+
+ def get_compatible_metrics(
+ self, selected_metrics: set[Metric], selected_dimensions: set[Dimension]
+ ) -> set[Metric]:
+ return self.get_metrics()
+
+ def get_compatible_dimensions(
+ self, selected_metrics: set[Metric], selected_dimensions: set[Dimension]
+ ) -> set[Dimension]:
+ return self.get_dimensions()
+
+
+class LegacyLayer(SemanticLayer[BaseModel, LegacyView]):
+ configuration_class: type[BaseModel] = BaseModel
+
+ @classmethod
+ def from_configuration(cls, configuration: dict[str, Any]) -> LegacyLayer:
+ return cls()
+
+ @classmethod
+ def get_configuration_schema(
+ cls, configuration: BaseModel | None = None
+ ) -> dict[str, Any]:
+ return {"type": "object"}
+
+ @classmethod
+ def get_runtime_schema(
+ cls, configuration: BaseModel, runtime_data: dict[str, Any] | None = None
+ ) -> dict[str, Any]:
+ return {"type": "object"}
+
+ def get_semantic_views(
+ self, runtime_configuration: dict[str, Any]
+ ) -> set[LegacyView]:
+ return {LegacyView()}
+
+ def get_semantic_view(
+ self, name: str, additional_configuration: dict[str, Any]
+ ) -> LegacyView:
+ return LegacyView()
+
+
+class SnapshotView(LegacyView):
+ def __init__(self, snapshot: CatalogSnapshot) -> None:
+ self.snapshot: CatalogSnapshot = snapshot
+
+ @property
+ def metadata_cache_token(self) -> str:
+ return self.snapshot.cache_token
+
+ def get_metrics(self) -> set[Metric]:
+ members: list[str] = json.loads(self.snapshot.payload)
+ return {
+ Metric(member, member, pa.int64(), member, verbose_name="Shared label")
+ for member in members
+ }
+
+
+class MemoryStore:
+ """Sequential example, without waiting or distributed guarantees."""
+
+ def __init__(self) -> None:
+ self.snapshot: CatalogSnapshot | None = None
+ self.publications: int = 0
+
+ def read(self, fetch: CatalogLoader, *, deadline: float) -> CatalogSnapshot:
+ remaining_budget(deadline)
+ return (
+ self.snapshot
+ if self.snapshot is not None
+ else self.refresh(fetch, deadline=deadline).snapshot
+ )
+
+ def refresh(
+ self, fetch: CatalogLoader, *, deadline: float
+ ) -> MetadataRefreshResult:
+ remaining_budget(deadline)
+ payload: str = fetch(deadline)
+ remaining_budget(deadline)
+ previous: CatalogSnapshot | None = self.snapshot
+ self.publications += 1
+ self.snapshot = CatalogSnapshot(
+ payload, f"connection:{self.publications}", "2026-09-30T12:00:00Z"
+ )
+ return MetadataRefreshResult(
+ "unchanged"
+ if previous is not None and previous.payload == payload
+ else "changed",
+ self.snapshot,
+ )
+
+
+class MemoryAdapter(MetadataRefreshAdapter):
+ """Operation-scoped provider example sharing the host's discovery budget."""
+
+ def __init__(self) -> None:
+ self.deadline: float | None = None
+ self.store: MetadataSnapshotStore | None = None
+ self.payload: str = '["orders", "revenue"]'
+ self.fetches: int = 0
+
+ def bind(self, store: MetadataSnapshotStore, *, deadline: float) -> None:
+ if self.store is not None:
+ raise MetadataRefreshError("configuration")
+ remaining_budget(deadline)
+ self.store = store
+ self.deadline = deadline
+
+ def fetch(self, deadline: float) -> str:
+ remaining_budget(deadline)
+ self.fetches += 1
+ return self.payload
+
+ def snapshot(self) -> CatalogSnapshot:
+ if self.store is None or self.deadline is None:
+ raise MetadataRefreshError("unsupported")
+ return self.store.read(self.fetch, deadline=self.deadline)
+
+ def refresh(self, *, deadline: float) -> MetadataRefreshResult:
+ if self.store is None:
+ raise MetadataRefreshError("unsupported")
+ return self.store.refresh(self.fetch, deadline=deadline)
+
+ def get_runtime_schema(
+ self, runtime_data: dict[str, Any] | None = None
+ ) -> dict[str, Any]:
+ return {"enum": json.loads(self.snapshot().payload)}
+
+
+class OptedInLayer(LegacyLayer):
+ def __init__(self) -> None:
+ self.adapter: MemoryAdapter = MemoryAdapter()
+
+ @classmethod
+ def supports_metadata_refresh(cls, configuration: dict[str, Any]) -> bool:
+ return True
+
+ @property
+ def metadata_refresh(self) -> MemoryAdapter:
+ return self.adapter
+
+ def get_semantic_view(
+ self, name: str, additional_configuration: dict[str, Any]
+ ) -> SnapshotView:
+ return SnapshotView(self.adapter.snapshot())
+
+ def get_semantic_views(
+ self, runtime_configuration: dict[str, Any]
+ ) -> set[LegacyView]:
+ return {self.get_semantic_view("example", {})}
+
+
+def test_legacy_provider_needs_no_metadata_overrides() -> None:
+ layer: LegacyLayer = LegacyLayer.from_configuration({})
+ view: LegacyView = layer.get_semantic_view("legacy", {})
+ assert not layer.supports_metadata_refresh({})
+ assert layer.metadata_refresh is None
+ assert view.metadata_cache_token is None
+ assert {metric.id for metric in view.get_compatible_metrics(set(), set())} == {
+ "orders"
+ }
+ assert view.get_table(SemanticQuery([], [])).results.to_pydict() == {"value": [17]}
+
+
+def test_provider_and_host_exchange_one_captured_observation() -> None:
+ layer: OptedInLayer = OptedInLayer()
+ store: MemoryStore = MemoryStore()
+ assert layer.supports_metadata_refresh({})
+ assert layer.adapter.fetches == 0
+ assert layer.metadata_refresh is layer.metadata_refresh
+ deadline: float = monotonic() + 30.0
+ layer.metadata_refresh.bind(store, deadline=deadline)
+ old_view: SnapshotView = layer.get_semantic_view("example", {})
+ assert layer.adapter.get_runtime_schema() == {"enum": ["orders", "revenue"]}
+ assert layer.adapter.fetches == 1
+ assert {metric.id for metric in old_view.get_metrics()} == {"orders", "revenue"}
+ assert {metric.verbose_name for metric in old_view.get_metrics()} == {
+ "Shared label"
+ }
+
+ layer.adapter.payload = '["orders", "revenue", "customers"]'
+ result: MetadataRefreshResult = layer.metadata_refresh.refresh(deadline=deadline)
+ new_view: SnapshotView = layer.get_semantic_view("example", {})
+ assert result.status == "changed"
+ assert new_view.metadata_cache_token == result.snapshot.cache_token
+ assert old_view.metadata_cache_token != new_view.metadata_cache_token
+ assert {metric.id for metric in old_view.get_metrics()} == {"orders", "revenue"}
+ assert {metric.id for metric in new_view.get_compatible_metrics(set(), set())} == {
+ "orders",
+ "revenue",
+ "customers",
+ }
+ assert layer.adapter.get_runtime_schema() == {
+ "enum": ["orders", "revenue", "customers"]
+ }
+ assert layer.adapter.fetches == 2
+
+
+def test_unchanged_discovery_can_have_a_new_cache_identity() -> None:
+ deadline: float = monotonic() + 30.0
+ adapter: MemoryAdapter = MemoryAdapter()
+ adapter.bind(MemoryStore(), deadline=deadline)
+ first: MetadataRefreshResult = adapter.refresh(deadline=deadline)
+ second: MetadataRefreshResult = adapter.refresh(deadline=deadline)
+ assert second.status == "unchanged"
+ assert first.snapshot.payload == second.snapshot.payload
+ assert first.snapshot.cache_token != second.snapshot.cache_token
+
+
+def test_adapter_example_rejects_use_before_binding_and_rebinding() -> None:
+ deadline: float = monotonic() + 30.0
+ adapter: MemoryAdapter = MemoryAdapter()
+ with pytest.raises(MetadataRefreshError, match="unsupported"):
+ adapter.refresh(deadline=deadline)
+ adapter.bind(MemoryStore(), deadline=deadline)
+ with pytest.raises(MetadataRefreshError, match="configuration"):
+ adapter.bind(MemoryStore(), deadline=deadline)
+ assert adapter.fetches == 0
+
+
+@pytest.mark.parametrize("field", ["payload", "cache_token", "observed_at"])
+def test_snapshot_cannot_change_after_capture(field: str) -> None:
+ snapshot: CatalogSnapshot = CatalogSnapshot("[]", "scope:1", "2026-09-30T12:00:00Z")
+ with pytest.raises(FrozenInstanceError):
+ setattr(snapshot, field, "replacement")
+
+
+@pytest.mark.parametrize("field", ["status", "snapshot"])
+def test_refresh_result_cannot_change_after_publication(field: str) -> None:
+ snapshot: CatalogSnapshot = CatalogSnapshot("[]", "scope:1", "2026-09-30T12:00:00Z")
+ result: MetadataRefreshResult = MetadataRefreshResult("changed", snapshot)
+ with pytest.raises(FrozenInstanceError):
+ setattr(result, field, "replacement")
+
+
+@pytest.mark.parametrize(
+ "category",
+ [
+ "unsupported",
+ "configuration",
+ "in_progress",
+ "configuration_changed",
+ "upstream",
+ "invalid_payload",
+ "deadline",
+ "unavailable",
+ "indeterminate",
+ ],
+)
+def test_declared_errors_cross_the_boundary_without_vendor_details(
+ category: MetadataRefreshErrorCategory,
+) -> None:
+ error: MetadataRefreshError = MetadataRefreshError(category)
+ assert error.category == category
+ assert str(error) == category
+ assert error.args == (category,)
+
+
+def test_sdk_metadata_imports_without_host_dependencies() -> None:
+ source: Path = Path(__file__).resolve().parents[3] / "superset-core" / "src"
+ script: str = """
+import importlib.abc
+import sys
+from collections.abc import Sequence
+from importlib.machinery import ModuleSpec
+from types import ModuleType
+
+class RejectHostImports(importlib.abc.MetaPathFinder):
+ def find_spec(
+ self, fullname: str, path: Sequence[str] | None = None,
+ target: ModuleType | None = None,
+ ) -> ModuleSpec | None:
+ if fullname.split('.')[0] in {'superset', 'flask', 'redis', 'flask_appbuilder'}:
+ raise AssertionError('Host dependency imported: ' + fullname)
+ return None
+
+sys.meta_path.insert(0, RejectHostImports())
+sys.path.insert(0, sys.argv[1])
+from superset_core.semantic_layers import metadata as metadata_sdk
+from superset_core.semantic_layers.layer import SemanticLayer
+from superset_core.semantic_layers.metadata import CatalogSnapshot
+from superset_core.semantic_layers.view import SemanticView
+assert metadata_sdk.remaining_budget(25.0, now=0.0) == 25.0
+assert not SemanticLayer.supports_metadata_refresh({})
+assert CatalogSnapshot('[]', 'scope:1', '2026-09-30T12:00:00Z').cache_token == 'scope:1'
+"""
+ # The isolated import probe uses a fixed script/interpreter and no shell.
+ completed: subprocess.CompletedProcess[str] = subprocess.run( # noqa: S603
+ [sys.executable, "-I", "-c", script, str(source)],
+ capture_output=True,
+ text=True,
+ check=False,
+ timeout=30,
+ )
+ assert completed.returncode == 0, completed.stderr
+
+
+@pytest.mark.parametrize("category", ["", "vendor detail: private upstream response"])
+def test_unknown_error_category_is_rejected_without_echoing_it(category: str) -> None:
+ with pytest.raises(ValueError, match="^Unknown metadata refresh error category$"):
+ MetadataRefreshError(cast(MetadataRefreshErrorCategory, category))
+
+
+def test_empty_snapshot_identity_is_rejected() -> None:
+ with pytest.raises(ValueError, match="^Catalog cache token must not be empty$"):
+ CatalogSnapshot("[]", "", "2026-09-30T12:00:00Z")
+
+
+def test_optional_adapter_annotation_resolves_for_sdk_introspection() -> None:
+ descriptor: property = inspect.getattr_static(SemanticLayer, "metadata_refresh")
+ assert descriptor.fget is not None
+ hints: dict[str, Any] = get_type_hints(descriptor.fget)
+ assert hints["return"] == MetadataRefreshAdapter | None
+
+
+@pytest.mark.parametrize("status", ["", "failed", "indeterminate"])
+def test_unconfirmed_refresh_status_cannot_construct_success(status: str) -> None:
+ snapshot: CatalogSnapshot = CatalogSnapshot("[]", "scope:1", "2026-09-30T12:00:00Z")
+ with pytest.raises(ValueError, match="^Unknown metadata refresh result status$"):
+ MetadataRefreshResult(cast(Literal["changed", "unchanged"], status), snapshot)
+
+
+@pytest.mark.parametrize("as_result", [False, True])
+def test_snapshot_representations_omit_catalog_and_cache_identity(
+ as_result: bool,
+) -> None:
+ snapshot: CatalogSnapshot = CatalogSnapshot(
+ '["private_catalog_member"]', "private_cache_identity", "2026-09-30T12:00:00Z"
+ )
+ value: CatalogSnapshot | MetadataRefreshResult = (
+ MetadataRefreshResult("changed", snapshot) if as_result else snapshot
+ )
+ representation: str = repr(value)
+ assert "private_catalog_member" not in representation
+ assert "private_cache_identity" not in representation
+ assert "2026-09-30T12:00:00Z" in representation
+
+
+def test_catalog_loader_accepts_one_absolute_deadline() -> None:
+ assert get_args(CatalogLoader) == ([float], str)
+
+
+def test_expired_loader_deadline_rejects_before_acquisition() -> None:
+ adapter: MemoryAdapter = MemoryAdapter()
+ with pytest.raises(MetadataRefreshError, match="^deadline$"):
+ adapter.fetch(monotonic() - 1.0)
+ assert adapter.fetches == 0
+
+
+def test_store_passes_one_deadline_and_cache_hits_do_not_acquire(
+ clock: list[float],
+) -> None:
+ deadlines: list[float] = []
+
+ def fetch(deadline: float) -> str:
+ deadlines.append(deadline)
+ return "[]"
+
+ store: MemoryStore = MemoryStore()
+ first: CatalogSnapshot = store.read(fetch, deadline=125.0)
+ assert store.read(fetch, deadline=125.0) is first
+ refreshed: MetadataRefreshResult = store.refresh(fetch, deadline=125.0)
+ assert refreshed.status == "unchanged"
+ assert deadlines == [125.0, 125.0]
+
+
+@pytest.mark.parametrize("field", ["payload", "cache_token", "observed_at"])
+@pytest.mark.parametrize("value", [None, 17, True, ["private upstream detail"]])
+def test_snapshot_rejects_non_string_fields(field: str, value: object) -> None:
+ """Malformed fields cannot cross the SDK boundary or appear in errors."""
+ fields: dict[str, str] = {
+ "payload": "[]",
+ "cache_token": "scope:1",
+ "observed_at": "2026-09-30T12:00:00Z",
+ }
+ fields[field] = cast(str, value)
+ with pytest.raises(TypeError, match=f"^Catalog {field} must be a string$"):
+ CatalogSnapshot(**fields)
+
+
+def test_snapshot_rejects_empty_observation_time() -> None:
+ """An observation always carries a source timestamp for display."""
+ with pytest.raises(ValueError, match="^Catalog observed_at must not be empty$"):
+ CatalogSnapshot("[]", "scope:1", "")
+
+
+@pytest.mark.parametrize(
+ "method",
+ [
+ MetadataSnapshotStore.read,
+ MetadataSnapshotStore.refresh,
+ MetadataRefreshAdapter.bind,
+ MetadataRefreshAdapter.refresh,
+ ],
+)
+def test_metadata_operations_require_explicit_caller_deadline(
+ method: Callable[..., object],
+) -> None:
+ """All metadata operation seams accept the same SDK-native deadline type."""
+ parameter: inspect.Parameter = inspect.signature(method).parameters["deadline"]
+ assert parameter.default is inspect.Parameter.empty
+ assert parameter.kind is inspect.Parameter.KEYWORD_ONLY
+ assert get_type_hints(method)["deadline"] is float
+
+
+@pytest.mark.parametrize("deadline", [float("nan"), float("inf"), -float("inf"), -1.0])
+@pytest.mark.parametrize("operation", ["read", "refresh", "adapter", "loader"])
+def test_example_rejects_invalid_deadlines_without_acquisition(
+ deadline: float,
+ operation: str,
+) -> None:
+ """Invalid or exhausted budgets cannot fetch or publish metadata."""
+ store: MemoryStore = MemoryStore()
+ adapter: MemoryAdapter = MemoryAdapter()
+ adapter.bind(store, deadline=monotonic() + 30.0)
+ operations: dict[str, Callable[[], object]] = {
+ "read": lambda: store.read(adapter.fetch, deadline=deadline),
+ "refresh": lambda: store.refresh(adapter.fetch, deadline=deadline),
+ "adapter": lambda: adapter.refresh(deadline=deadline),
+ "loader": lambda: adapter.fetch(deadline),
+ }
+ with pytest.raises(MetadataRefreshError, match="^deadline$"):
+ operations[operation]()
+ assert adapter.fetches == 0
+ assert store.publications == 0
+
+
+def test_sequential_store_calls_share_remaining_budget(
+ clock: list[float],
+) -> None:
+ """Time spent in one acquisition reduces the next call's remaining budget."""
+ deadlines: list[float] = []
+
+ def fetch(deadline: float) -> str:
+ """Consume part of the operation's budget."""
+ deadlines.append(deadline)
+ clock[0] += 20.0
+ return "[]"
+
+ store: MemoryStore = MemoryStore()
+ store.read(fetch, deadline=150.0)
+ store.refresh(fetch, deadline=150.0)
+ assert deadlines == [150.0, 150.0]
+ before: CatalogSnapshot | None = store.snapshot
+ clock[0] = 150.0
+ with pytest.raises(MetadataRefreshError, match="^deadline$"):
+ store.refresh(fetch, deadline=150.0)
+ with pytest.raises(MetadataRefreshError, match="^deadline$"):
+ store.read(fetch, deadline=150.0)
+ assert len(deadlines) == 2
+ assert store.publications == 2
+ assert store.snapshot is before
+
+
+def test_provider_factory_accepts_discovery_deadline_only_at_bind(
+ clock: list[float],
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ """Normal construction stays deadline-free; bind carries the host budget."""
+ deadlines: list[float] = []
+
+ def fetch(self: MemoryAdapter, deadline: float) -> str:
+ """Capture the unchanged budget passed to provider acquisition."""
+ deadlines.append(deadline)
+ return self.payload
+
+ monkeypatch.setattr(MemoryAdapter, "fetch", fetch)
+ layer: LegacyLayer = OptedInLayer.from_configuration({})
+ assert isinstance(layer, OptedInLayer)
+ assert layer.adapter.fetches == 0
+ assert layer.adapter.store is None
+ assert deadlines == []
+ with pytest.raises(MetadataRefreshError, match="^unsupported$"):
+ layer.get_semantic_view("example", {})
+ store: MemoryStore = MemoryStore()
+ layer.adapter.bind(store, deadline=125.0)
+ assert layer.get_semantic_views({})
+ clock[0] = 120.0
+ assert layer.adapter.get_runtime_schema() == {"enum": ["orders", "revenue"]}
+ layer.adapter.refresh(deadline=125.0)
+ assert deadlines == [125.0, 125.0]
+ clock[0] = 125.0
+ with pytest.raises(MetadataRefreshError, match="^deadline$"):
+ layer.adapter.get_runtime_schema()
+ assert store.publications == 2
+
+
+@pytest.mark.parametrize("deadline", [float("nan"), float("inf"), -float("inf"), 100.0])
+def test_invalid_bind_deadline_does_not_partially_bind(
+ clock: list[float], deadline: float
+) -> None:
+ """Invalid binding leaves the adapter unbound and permits a valid first bind."""
+
+ layer: LegacyLayer = OptedInLayer.from_configuration({})
+ assert isinstance(layer, OptedInLayer)
+ store: MemoryStore = MemoryStore()
+ with pytest.raises(MetadataRefreshError, match="^deadline$"):
+ layer.adapter.bind(store, deadline=deadline)
+ assert layer.adapter.store is None
+ assert layer.adapter.fetches == 0
+ with pytest.raises(MetadataRefreshError, match="^unsupported$"):
+ layer.adapter.snapshot()
+ layer.adapter.bind(store, deadline=125.0)
+ assert layer.adapter.snapshot().payload == layer.adapter.payload
+ with pytest.raises(MetadataRefreshError, match="^configuration$"):
+ layer.adapter.bind(MemoryStore(), deadline=150.0)
+
+
+def test_discovery_and_adapter_refresh_preserve_host_deadline(
+ clock: list[float],
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ """View/schema discovery and explicit refresh retain one caller budget."""
+ deadlines: list[float] = []
+
+ def fetch(self: MemoryAdapter, deadline: float) -> str:
+ """Record the deadline reaching the provider through each path."""
+ deadlines.append(deadline)
+ return self.payload
+
+ monkeypatch.setattr(MemoryAdapter, "fetch", fetch)
+ layer: OptedInLayer = OptedInLayer()
+ store: MemoryStore = MemoryStore()
+ layer.adapter.bind(store, deadline=125.0)
+ layer.get_semantic_view("example", {})
+ clock[0] = 120.0
+ assert layer.adapter.get_runtime_schema() == {"enum": ["orders", "revenue"]}
+ layer.adapter.refresh(deadline=125.0)
+ assert deadlines == [125.0, 125.0]
+ clock[0] = 125.0
+ with pytest.raises(MetadataRefreshError, match="^deadline$"):
+ layer.get_semantic_views({})
+ with pytest.raises(MetadataRefreshError, match="^deadline$"):
+ layer.adapter.get_runtime_schema()
+ with pytest.raises(MetadataRefreshError, match="^deadline$"):
+ layer.adapter.refresh(deadline=125.0)
+ assert deadlines == [125.0, 125.0]
+ assert store.publications == 2
+
+
+@pytest.mark.parametrize("warm", [False, True])
+def test_acquisition_exhausting_budget_cannot_publish(
+ clock: list[float], warm: bool
+) -> None:
+ """A completed fetch cannot publish or replace a snapshot after its deadline."""
+
+ def fetch(deadline: float) -> str:
+ """Exhaust the budget during acquisition."""
+ clock[0] = deadline
+ return '["new"]'
+
+ store: MemoryStore = MemoryStore()
+ if warm:
+ store.snapshot = CatalogSnapshot("[]", "scope:1", "2026-09-30T12:00:00Z")
+ store.publications = 1
+ before: CatalogSnapshot | None = store.snapshot
+ publications: int = store.publications
+ with pytest.raises(MetadataRefreshError, match="^deadline$"):
+ store.refresh(fetch, deadline=150.0)
+ assert store.publications == publications
+ assert store.snapshot is before
+
+
+@pytest.mark.parametrize(
+ "deadline",
+ [
+ float("nan"),
+ float("inf"),
+ -float("inf"),
+ 0.0,
+ 99.0,
+ 100.0,
+ 401.0,
+ 1_790_000_000.0,
+ ],
+)
+def test_remaining_budget_rejects_invalid_or_implausible_deadline(
+ clock: list[float], deadline: float
+) -> None:
+ with pytest.raises(MetadataRefreshError, match="^deadline$"):
+ remaining_budget(deadline)
+
+
+def test_remaining_budget_decreases_and_accepts_ceiling(clock: list[float]) -> None:
+ assert metadata_sdk.MAX_METADATA_BUDGET_SECONDS == 300.0
+ assert remaining_budget(400.0) == 300.0
+ clock[0] = 110.0
+ assert remaining_budget(400.0) == 290.0
+ clock[0] = 400.0
+ with pytest.raises(MetadataRefreshError, match="^deadline$"):
+ remaining_budget(400.0)
+
+
+@pytest.mark.parametrize("snapshot", [None, "not a snapshot", {}, 17, True])
+def test_refresh_result_rejects_non_snapshot(snapshot: object) -> None:
+ with pytest.raises(
+ TypeError, match="^Metadata refresh snapshot must be a CatalogSnapshot$"
+ ):
+ MetadataRefreshResult("changed", cast(CatalogSnapshot, snapshot))
+
+
+@pytest.mark.parametrize("now", [float("nan"), float("inf"), -float("inf"), 400.0])
+def test_remaining_budget_rejects_invalid_injected_clock(now: float) -> None:
+ with pytest.raises(MetadataRefreshError, match="^deadline$"):
+ remaining_budget(400.0, now=now)
+
+
+def test_remaining_budget_uses_injected_monotonic_clock() -> None:
+ assert remaining_budget(125.0, now=100.0) == 25.0
+ assert remaining_budget(25.0, now=0.0) == 25.0