From aeb5195d916c81080bc2f1379dc26362646552a6 Mon Sep 17 00:00:00 2001 From: Philip Z Date: Tue, 20 Oct 2026 05:54:20 +0800 Subject: [PATCH] Type Python Context extension contracts and validate wiring early --- README.md | 23 +++ examples/typed-extension-spi/README.md | 16 ++ examples/typed-extension-spi/main.py | 146 ++++++++++++++ scripts/verify-examples.sh | 3 +- src/teaql/data_service/__init__.py | 15 -- src/teaql/runtime/__init__.py | 8 + src/teaql/runtime/context.py | 91 +++++---- src/teaql/runtime/extensions.py | 108 ++++++++++ tests/provider/sqlite/test_sqlite.py | 4 +- tests/runtime/test_typed_extension_spi.py | 228 ++++++++++++++++++++++ 10 files changed, 591 insertions(+), 51 deletions(-) create mode 100644 examples/typed-extension-spi/README.md create mode 100644 examples/typed-extension-spi/main.py create mode 100644 src/teaql/runtime/extensions.py create mode 100644 tests/runtime/test_typed_extension_spi.py diff --git a/README.md b/README.md index 5434791..c7547a9 100644 --- a/README.md +++ b/README.md @@ -285,6 +285,29 @@ atomic allocation and floor advancement. See the runnable [Internal ID example](examples/internal-id/README.md). The old synchronous `generate_id`/`next_id` context calls are unsupported; use the async capability. +## Typed Context extension points + +Trusted application wiring installs runtime capabilities through typed Context +methods. Malformed providers fail during registration, before a query or Save +reaches the database. The contracts are importable from `teaql.runtime`: + +| Capability | Install on Context | Minimum Protocol | +| --- | --- | --- | +| Query authorization | `with_request_policy` | Callable query policy or `RequestPolicy.apply` | +| Checker/Fix | `with_checker_registry` | `CheckerRegistry.checker` returning a `Checker` | +| Physical schema | `with_schema_provider` | Async `SchemaCapability._ensure_schema` | +| Application audit | `with_app_audit_event_sink` | `SafeAuditEventSink.on_safe_event` | +| Runtime telemetry | `with_runtime_telemetry` | `RuntimeTelemetryAdapter.start`, or diagnostic-only `record_diagnostic` | +| Internal IDs | `with_internal_id_allocator` | Async `InternalIdAllocator.next_id` and `ensure_id_floor` | + +The schema capability is independent of the query/mutation data service; +`context.ensure_schema()` remains the application entry point and runs +generated bootstrap after physical schema reconciliation. Existing callable +query policies still work. Telemetry start and diagnostic failures remain +fail-open, whereas query-policy and Checker denials remain fail-closed. The +[typed extension example](examples/typed-extension-spi/README.md) wires all +six contracts into one small SQLite application. + ## Optional standard OpenTelemetry module The local standard-module changes are not yet released. Install the optional diff --git a/examples/typed-extension-spi/README.md b/examples/typed-extension-spi/README.md new file mode 100644 index 0000000..9e8da76 --- /dev/null +++ b/examples/typed-extension-spi/README.md @@ -0,0 +1,16 @@ +# Typed Context extension SPI example + +This focused SQLite example installs six independent Context capabilities: +request policy, Checker/Fix registry, schema-only provider, safe application +audit sink, runtime telemetry, and Internal ID allocator. The schema provider +has no query/mutation methods. The example verifies accepted and denied +operations, twice-run schema ordering, and a persisted ID from the installed +allocator. + +```bash +PYTHONPATH=src python examples/typed-extension-spi/main.py +``` + +Extensions are trusted application wiring. Registration rejects missing +methods before business execution; a policy or Checker denial still fails +closed. Optional telemetry adapter failures do not change a business result. diff --git a/examples/typed-extension-spi/main.py b/examples/typed-extension-spi/main.py new file mode 100644 index 0000000..0d14f05 --- /dev/null +++ b/examples/typed-extension-spi/main.py @@ -0,0 +1,146 @@ +"""Small, complete demonstration of six typed Context extension contracts.""" + +import asyncio +from pathlib import Path +from tempfile import TemporaryDirectory + +import aiosqlite + +from teaql.core.meta import EntityDescriptor, PropertyDescriptor +from teaql.core.mutation import InsertCommand, MutationRequest +from teaql.core.query import SelectQuery +from teaql.core.value import DataType +from teaql.data_service import QueryRequest +from teaql.provider.sqlite import SimpleSchemaProvider, create_sqlite_service +from teaql.runtime import RuntimeModule + + +class SchoolPolicy: + def __init__(self): + self.deny = False + self.calls = 0 + + def apply(self, query): + self.calls += 1 + if self.deny: + raise PermissionError("school query denied") + + +class SchoolChecker: + def check_and_fix(self, context, record, location, results): + if not record.get("name") or not record["name"].val: + raise ValueError("school name required") + + +class SchoolCheckerRegistry: + def checker(self, entity): + return SchoolChecker() if entity == "School" else None + + +class SchemaOnly: + def __init__(self, service): + self.service = service + self.calls = 0 + + async def _ensure_schema(self, context, capability): + self.calls += 1 + await self.service._ensure_schema(context, capability) + + +class AuditSink: + def __init__(self): + self.events = [] + + def on_safe_event(self, context, event): + self.events.append(event) + + +class Telemetry: + def __init__(self): + self.operations = [] + + def start(self, operation): + self.operations.append(operation.name) + + class Scope: + def success(self, attributes=None): pass + def failure(self, error): pass + + return Scope() + + +class InternalIds: + def __init__(self): + self.level = 699 + self.lock = asyncio.Lock() + + async def next_id(self, entity): + async with self.lock: + self.level += 1 + return self.level + + async def ensure_id_floor(self, entity, floor): + async with self.lock: + self.level = max(self.level, floor) + + +async def run(): + with TemporaryDirectory() as directory: + path = Path(directory) / "typed-spi.db" + school = (EntityDescriptor("School").table_name("school_data") + .property(PropertyDescriptor("id", DataType.I64).is_id()) + .property(PropertyDescriptor("version", DataType.I64).is_version()) + .property(PropertyDescriptor("name", DataType.Text))) + provider = SimpleSchemaProvider() + provider.register_entity(school) + service = create_sqlite_service(str(path), provider) + policy, audit, telemetry = SchoolPolicy(), AuditSink(), Telemetry() + schema, ids = SchemaOnly(service), InternalIds() + context = (RuntimeModule.new().entity(school).into_context() + .insert_resource("dataService", service) + .with_schema_provider(schema) + .with_request_policy(policy) + .with_checker_registry(SchoolCheckerRegistry()) + .with_app_audit_event_sink(audit) + .with_runtime_telemetry(telemetry) + .with_internal_id_allocator(ids)) + context.disable_sql_log() + assert not hasattr(schema, "query") and not hasattr(schema, "mutate") + await context.ensure_schema() + await context.ensure_schema() + assert schema.calls == 2 + + await service.mutate(context, MutationRequest( + InsertCommand.new("School").value("name", "Riverside"), + comment="create reviewed school")) + request = QueryRequest(SelectQuery("School").project("id", "name"), + _comment="load schools", _purpose="verify accepted school") + rows = (await service.query(context, context.prepare_query_request(request))).rows + assert [(row["id"], row["name"]) for row in rows] == [(700, "Riverside")] + assert policy.calls == 1 and len(audit.events) == 1 + assert telemetry.operations + + policy.deny = True + try: + context.prepare_query_request(request) + except PermissionError: + pass + else: + raise AssertionError("denied policy must block query preparation") + try: + await service.mutate(context, MutationRequest( + InsertCommand.new("School").value("name", ""), + comment="reject nameless school")) + except ValueError: + pass + else: + raise AssertionError("checker must reject invalid school") + async with aiosqlite.connect(path) as db: + async with db.execute("SELECT id FROM school_data") as cursor: + assert [row[0] for row in await cursor.fetchall()] == [700] + await service.close() + print("PASS Python typed Context extension SPI example") + + +if __name__ == "__main__": + asyncio.run(run()) diff --git a/scripts/verify-examples.sh b/scripts/verify-examples.sh index d47868c..c3f191e 100755 --- a/scripts/verify-examples.sh +++ b/scripts/verify-examples.sh @@ -2,7 +2,7 @@ set -euo pipefail repo="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" -expected=(business-clock business-id conformance facet-trace internal-id mutation-policy object-storage opaque-entity-reference order-management query-policy school-management task_board trace-chain) +expected=(business-clock business-id conformance facet-trace internal-id mutation-policy object-storage opaque-entity-reference order-management query-policy school-management task_board trace-chain typed-extension-spi) mapfile -t actual < <(find "$repo/examples" -mindepth 1 -maxdepth 1 -type d -printf '%f\n' | sort) if [[ "${actual[*]}" != "${expected[*]}" ]]; then echo "example inventory changed; update scripts/verify-examples.sh: ${actual[*]}" >&2 @@ -30,6 +30,7 @@ PYTHONPATH="$repo/src" python "$repo/examples/mutation-policy/main.py" PYTHONPATH="$repo/src" python "$repo/examples/business-clock/main.py" PYTHONPATH="$repo/src" python "$repo/examples/business-id/main.py" PYTHONPATH="$repo/src" python "$repo/examples/internal-id/main.py" +PYTHONPATH="$repo/src" python "$repo/examples/typed-extension-spi/main.py" PYTHONPATH="$repo/src" python "$repo/examples/query-policy/main.py" PYTHONPATH="$repo/src" python "$repo/examples/opaque-entity-reference/main.py" order_management_tmp="$(mktemp -d)" diff --git a/src/teaql/data_service/__init__.py b/src/teaql/data_service/__init__.py index f2edc1b..8ae4c17 100644 --- a/src/teaql/data_service/__init__.py +++ b/src/teaql/data_service/__init__.py @@ -328,21 +328,6 @@ async def begin(self) -> Transaction: ... -@dataclass -class _SchemaRequest: - entity_name: str - - -@dataclass -class _SchemaResult: - changed: bool - - -class _SchemaExecutor(DataServiceExecutor, Protocol): - async def _ensure_schema_request(self, request: _SchemaRequest) -> _SchemaResult: - ... - - class IdGeneratorExecutor(DataServiceExecutor, Protocol): async def next_id(self, entity: str) -> int: ... diff --git a/src/teaql/runtime/__init__.py b/src/teaql/runtime/__init__.py index bb731bb..e8146ed 100644 --- a/src/teaql/runtime/__init__.py +++ b/src/teaql/runtime/__init__.py @@ -13,6 +13,11 @@ from .audit import RawAuditEvent, SafeAuditEvent, MutationAuditKind from .business_clock import BusinessClock, FixedBusinessClock, SystemBusinessClock from .internal_id import InternalIdAllocator +from .extensions import ( + Checker, CheckerRegistry, DiagnosticTelemetryAdapter, RequestPolicy, + RequestPolicyInput, RuntimeTelemetryAdapter, RuntimeTelemetryInput, + SafeAuditEventSink, SchemaCapability, +) from .entity_reference import ( AeadEntityReferenceCodec, EntityReferenceClaims, @@ -61,6 +66,9 @@ __all__ = ["WireFieldMetadata", "WireEntityMetadata", "NormalizedWireInput", "WireInputError", "create_wire_entity_metadata", "normalize_wire_input", "encode_wire_output", "retain_submitted_paths", "EntityKey", "EntityChangeSet", "EntityRoot", "ContextEntityRef", "ContextRootError", "CheckException", "CheckResult", "I18nCatalog", "JsonFieldNamingProfile", "Locale", "ObjectLocation", "UnsupportedLocaleError", "UserContext", "TeaqlRuntime", "SqlLogEntry", "SqlLogOperation", "DiagnosticSqlLogSink", "TextDiagnosticSqlLogSink", "ServiceRuntimeFromEnv", "RuntimeModule", "DataStore", "RawAuditEvent", "SafeAuditEvent", "MutationAuditKind", "BusinessClock", "FixedBusinessClock", "SystemBusinessClock", "InternalIdAllocator", "AeadEntityReferenceCodec", "EntityReferenceClaims", "EntityReferenceCodec", "EntityReferenceTokenError", "ENTITY_REFERENCE_AAD", "UNSAFE_RAW_ENTITY_REFERENCES_ACKNOWLEDGEMENT", "UNSAFE_RAW_ENTITY_REFERENCES_ENVIRONMENT", "BUSINESS_ID_PERMUTATION_V1_ALPHABET", "BUSINESS_ID_PERMUTATION_V1_DOMAIN_SIZE", "BUSINESS_ID_PERMUTATION_V1_MAX_SEQUENCE", "BUSINESS_ID_PERMUTATION_V1_WIDTH", "DefaultBusinessIdProfileFactory", "DefaultBusinessIdService", "InMemoryBusinessIdAllocator", "PermutedDailyBusinessIdProfile", "StaticBusinessIdKeyProvider", "encode_business_id_permutation_v1", "ContextTools", "ExecutableHttpTool", "HTTP_TOOL", "HttpIntentPhase", "HttpTool", "HttpToolProvider", "ToolDeniedError", "ToolError", "ToolPolicy", "ToolRisk", "Tools", "ToolToken", "ToolUnavailableError"] __all__ += [ + "Checker", "CheckerRegistry", "DiagnosticTelemetryAdapter", "RequestPolicy", + "RequestPolicyInput", "RuntimeTelemetryAdapter", "RuntimeTelemetryInput", + "SafeAuditEventSink", "SchemaCapability", "GraphMutationSession", "GraphCommittedError", "MISSING_APPROVAL", "MISSING_POLICY", "DelegatingMutationGovernanceSink", diff --git a/src/teaql/runtime/context.py b/src/teaql/runtime/context.py index 73a43ec..f816b70 100644 --- a/src/teaql/runtime/context.py +++ b/src/teaql/runtime/context.py @@ -9,6 +9,13 @@ from .business_clock import BusinessClock, SystemBusinessClock from .internal_id import InternalIdAllocator, validate_internal_id_allocator +from .extensions import ( + Checker, CheckerRegistry, RequestPolicyInput, RuntimeTelemetryAdapter, + RuntimeTelemetryInput, + SafeAuditEventSink, SchemaCapability, validate_checker_registry, + validate_request_policy, validate_runtime_telemetry, + validate_safe_audit_sink, validate_schema_capability, +) from .entity_reference import ( EntityReferenceClaims, EntityReferenceCodec, @@ -125,6 +132,7 @@ def __init__(self): self._business_clock: BusinessClock = SystemBusinessClock() self._entity_reference_codec: Optional[EntityReferenceCodec] = None self._internal_id_allocator: Optional[InternalIdAllocator] = None + self._explicit_schema_provider = False from .mutation_policy import MutationPolicyRuntimeState self._mutation_policy = MutationPolicyRuntimeState() @@ -316,8 +324,8 @@ def initialize_entity(self, entity_name: str, entity: TEntity) -> TEntity: def managed_entities(self) -> List[Any]: return list(self._managed_entities) - def with_runtime_telemetry(self, telemetry: Any) -> "UserContext": - self._runtime_telemetry = telemetry + def with_runtime_telemetry(self, telemetry: RuntimeTelemetryInput) -> "UserContext": + self._runtime_telemetry = validate_runtime_telemetry(telemetry) return self def with_object_storage(self, provider) -> "UserContext": @@ -333,10 +341,10 @@ def object_storage(self, comment: str): from .object_storage import ObjectStorage return ObjectStorage(self, comment) - def set_runtime_telemetry(self, telemetry: Any) -> None: - self._runtime_telemetry = telemetry + def set_runtime_telemetry(self, telemetry: RuntimeTelemetryInput) -> None: + self.with_runtime_telemetry(telemetry) - def runtime_telemetry(self) -> Any: + def runtime_telemetry(self) -> RuntimeTelemetryAdapter: return self._runtime_telemetry def set_id_set_store(self, store: Any) -> None: @@ -378,9 +386,24 @@ def with_metadata(self, metadata: Any) -> 'UserContext': return self def insert_resource(self, resource_type: str, resource: Any): + inferred_schema = None + if resource_type == "request_policy": + resource = validate_request_policy(resource) + elif resource_type == "checker_registry": + resource = validate_checker_registry(resource) + elif resource_type == "schema_provider": + resource = validate_schema_capability(resource) + elif (resource_type == "dataService" and not self._explicit_schema_provider + and isinstance(resource, SchemaCapability)): + inferred_schema = validate_schema_capability(resource) self._resources[resource_type] = resource - if resource_type == "dataService" and hasattr(resource, "_ensure_schema"): - self._resources["schema_provider"] = resource + if resource_type == "schema_provider": + self._explicit_schema_provider = True + elif resource_type == "dataService" and not self._explicit_schema_provider: + if inferred_schema is None: + self._resources.pop("schema_provider", None) + else: + self._resources["schema_provider"] = inferred_schema return self def get_resource(self, resource_type: str) -> Optional[Any]: @@ -401,9 +424,7 @@ def prepare_query(self, query: Any) -> Any: if policy is None: return prepared_root - apply_policy = policy if callable(policy) else getattr(policy, "apply", None) - if apply_policy is None or not callable(apply_policy): - raise TypeError("request_policy must be callable or expose apply(query)") + apply_policy = policy if callable(policy) else policy.apply prepared_nodes: Dict[int, Any] = {} @@ -512,34 +533,34 @@ def with_entity_data_service_behavior_registry(self, registry: Any) -> 'UserCont def set_entity_data_service_behavior_registry(self, registry: Any): self.insert_resource("entity_data_service_behavior_registry", registry) - def with_request_policy(self, policy: Any) -> 'UserContext': - self.insert_resource("request_policy", policy) + def with_request_policy(self, policy: RequestPolicyInput) -> 'UserContext': + self.insert_resource("request_policy", validate_request_policy(policy)) return self - def set_request_policy(self, policy: Any): - self.insert_resource("request_policy", policy) + def set_request_policy(self, policy: RequestPolicyInput): + self.with_request_policy(policy) def clear_request_policy(self): self._resources.pop("request_policy", None) - def with_checker_registry(self, registry: Any) -> 'UserContext': - self.insert_resource("checker_registry", registry) + def with_checker_registry(self, registry: CheckerRegistry) -> 'UserContext': + self.insert_resource("checker_registry", validate_checker_registry(registry)) return self - def set_checker_registry(self, registry: Any): - self.insert_resource("checker_registry", registry) + def set_checker_registry(self, registry: CheckerRegistry): + self.with_checker_registry(registry) - def with_custom_event_sink(self, sink: Any) -> 'UserContext': - self._app_audit_sink = sink - return self + def with_custom_event_sink(self, sink: SafeAuditEventSink) -> 'UserContext': + return self.with_app_audit_event_sink(sink) - def set_custom_event_sink(self, sink: Any): - self._app_audit_sink = sink + def set_custom_event_sink(self, sink: SafeAuditEventSink): + self.with_app_audit_event_sink(sink) def initialize_audit(self, raw_sink: Any, app_sink: Any = None) -> 'UserContext': """Compatibility entry point for generated applications.""" + checked_app_sink = None if app_sink is None else validate_safe_audit_sink(app_sink) self._set_standard_audit_sink(raw_sink) - self._app_audit_sink = app_sink + self._app_audit_sink = checked_app_sink return self def configure_audit_policy(self, entity: str, mask_fields: List[str], @@ -588,12 +609,13 @@ async def ensure_business_id( self, definition, domain_root_key, aggregate_type, slot ) - def with_schema_provider(self, provider: Any) -> 'UserContext': - self.insert_resource("schema_provider", provider) + def with_schema_provider(self, provider: SchemaCapability) -> 'UserContext': + self._resources["schema_provider"] = validate_schema_capability(provider) + self._explicit_schema_provider = True return self - def set_schema_provider(self, provider: Any): - self.insert_resource("schema_provider", provider) + def set_schema_provider(self, provider: SchemaCapability): + self.with_schema_provider(provider) async def ensure_schema(self): provider = self.get_resource("schema_provider") @@ -688,14 +710,14 @@ def entity_data_service_behavior(self, entity: str) -> Optional[Any]: def has_checker(self, entity: str) -> bool: registry = self.get_resource("checker_registry") - return bool(registry and hasattr(registry, "checker") and registry.checker(entity)) + return bool(registry and registry.checker(entity)) def check_and_fix_record(self, entity: str, record: Any): self.check_and_fix_record_at(entity, record, None) def check_and_fix_record_at(self, entity: str, record: Any, location: Any): registry = self.get_resource("checker_registry") - if not registry or not hasattr(registry, "checker"): + if not registry: return checker = registry.checker(entity) @@ -703,8 +725,9 @@ def check_and_fix_record_at(self, entity: str, record: Any, location: Any): return results = [] - if hasattr(checker, "check_and_fix"): - checker.check_and_fix(self, record, location, results) + if not isinstance(checker, Checker): + raise TypeError("checker registry returned an object without check_and_fix") + checker.check_and_fix(self, record, location, results) if results: self.translate_check_results(results) @@ -804,8 +827,8 @@ async def _deliver_audit_event(self, event: Any, safe_event: Any): def _set_standard_audit_sink(self, sink: Any): self._standard_audit_sink = sink - def with_app_audit_event_sink(self, sink: Any) -> 'UserContext': - self._app_audit_sink = sink + def with_app_audit_event_sink(self, sink: SafeAuditEventSink) -> 'UserContext': + self._app_audit_sink = validate_safe_audit_sink(sink) return self def with_mutation_policy_registry(self, registry: Any) -> 'UserContext': diff --git a/src/teaql/runtime/extensions.py b/src/teaql/runtime/extensions.py new file mode 100644 index 0000000..8e9fa12 --- /dev/null +++ b/src/teaql/runtime/extensions.py @@ -0,0 +1,108 @@ +"""Typed, trusted Context extension contracts. + +Protocols describe the minimum runtime-consumed shape. Optional lifecycle +hooks are not required just to install an adapter. +""" + +from __future__ import annotations + +from inspect import iscoroutinefunction +from typing import Any, Awaitable, Callable, Protocol, TYPE_CHECKING, runtime_checkable + +from teaql.core.query import SelectQuery +from teaql.core.value import Value +from .audit import SafeAuditEvent +from .i18n import CheckResult, ObjectLocation +from .telemetry import ( + AttributeValue, NOOP_RUNTIME_TELEMETRY, RuntimeOperation, RuntimeTelemetryScope, +) + +if TYPE_CHECKING: + from .context import UserContext + + +@runtime_checkable +class RequestPolicy(Protocol): + def apply(self, query: SelectQuery) -> SelectQuery | None: ... + + +RequestPolicyInput = RequestPolicy | Callable[[SelectQuery], SelectQuery | None] + + +@runtime_checkable +class Checker(Protocol): + def check_and_fix(self, context: UserContext, record: dict[str, Value], + location: ObjectLocation | None, results: list[CheckResult]) -> None: ... + + +@runtime_checkable +class CheckerRegistry(Protocol): + def checker(self, entity: str) -> Checker | None: ... + + +@runtime_checkable +class SchemaCapability(Protocol): + async def _ensure_schema(self, context: UserContext, capability: object) -> None: ... + + +@runtime_checkable +class SafeAuditEventSink(Protocol): + def on_safe_event(self, context: UserContext, event: SafeAuditEvent) -> None | Awaitable[None]: ... + + +@runtime_checkable +class RuntimeTelemetryAdapter(Protocol): + def start(self, operation: RuntimeOperation) -> RuntimeTelemetryScope: ... + + +@runtime_checkable +class DiagnosticTelemetryAdapter(Protocol): + def record_diagnostic(self, attributes: dict[str, AttributeValue], correlation: Any) -> None: ... + + +RuntimeTelemetryInput = RuntimeTelemetryAdapter | DiagnosticTelemetryAdapter + + +class _DiagnosticOnlyTelemetry: + """Keep legacy diagnostic-only sinks while preserving a start-capable type.""" + + def __init__(self, delegate: DiagnosticTelemetryAdapter): + self._delegate = delegate + + def start(self, operation: RuntimeOperation) -> RuntimeTelemetryScope: + return NOOP_RUNTIME_TELEMETRY.start(operation) + + def record_diagnostic(self, attributes: dict[str, AttributeValue], correlation: Any) -> None: + self._delegate.record_diagnostic(attributes, correlation) + + +def validate_request_policy(policy: RequestPolicyInput) -> RequestPolicyInput: + if not callable(policy) and not isinstance(policy, RequestPolicy): + raise TypeError("request policy must be callable or expose apply(query)") + return policy + + +def validate_checker_registry(registry: CheckerRegistry) -> CheckerRegistry: + if not isinstance(registry, CheckerRegistry): + raise TypeError("checker registry must provide checker(entity)") + return registry + + +def validate_schema_capability(provider: SchemaCapability) -> SchemaCapability: + if not isinstance(provider, SchemaCapability) or not iscoroutinefunction(provider._ensure_schema): + raise TypeError("schema capability must provide async _ensure_schema(context, capability)") + return provider + + +def validate_safe_audit_sink(sink: SafeAuditEventSink) -> SafeAuditEventSink: + if not isinstance(sink, SafeAuditEventSink): + raise TypeError("application audit sink must provide on_safe_event(context, event)") + return sink + + +def validate_runtime_telemetry(telemetry: RuntimeTelemetryInput) -> RuntimeTelemetryAdapter: + if isinstance(telemetry, RuntimeTelemetryAdapter): + return telemetry + if isinstance(telemetry, DiagnosticTelemetryAdapter): + return _DiagnosticOnlyTelemetry(telemetry) + raise TypeError("runtime telemetry adapter must provide start(operation) or record_diagnostic") diff --git a/tests/provider/sqlite/test_sqlite.py b/tests/provider/sqlite/test_sqlite.py index 48fe160..ba4adb9 100644 --- a/tests/provider/sqlite/test_sqlite.py +++ b/tests/provider/sqlite/test_sqlite.py @@ -419,7 +419,9 @@ def invalidate(self, _key): raise RuntimeError("down") await service.query(context, QueryRequest(ttl, _comment='what: runtime regression fixture', _purpose='why: verify runtime behavior')) assert context.id_set_plan() == "ID_SET_BUILD" - source_one, source_two, policy_one, policy_two = object(), object(), object(), object() + source_one, source_two = object(), object() + policy_one = lambda query: query + policy_two = lambda query: query cases = [ ("tenant:user-a", 1, "one", source_one, policy_one), ("tenant:user-b", 1, "one", source_one, policy_one), diff --git a/tests/runtime/test_typed_extension_spi.py b/tests/runtime/test_typed_extension_spi.py new file mode 100644 index 0000000..2f8c726 --- /dev/null +++ b/tests/runtime/test_typed_extension_spi.py @@ -0,0 +1,228 @@ +"""Trusted Context extension registration rejects broken capabilities early.""" + +import pytest + +from teaql.core.meta import EntityDescriptor, PropertyDescriptor +from teaql.core.mutation import InsertCommand, MutationRequest +from teaql.core.query import SelectQuery +from teaql.core.value import DataType +from teaql.data_service import QueryRequest +from teaql.provider.sqlite import SimpleSchemaProvider, create_sqlite_service +from teaql.runtime import ( + CheckerRegistry, InternalIdAllocator, RequestPolicy, RuntimeModule, + RuntimeTelemetryAdapter, SafeAuditEventSink, SchemaCapability, UserContext, +) +from teaql.runtime.audit import MutationAuditKind, RawAuditEvent + + +def test_public_protocols_and_bad_installations_fail_before_business_execution(): + class Valid: + def apply(self, query): return query + def checker(self, entity): return None + async def _ensure_schema(self, context, capability): pass + def on_safe_event(self, context, event): pass + def start(self, operation): return None + async def next_id(self, entity): return 1 + async def ensure_id_floor(self, entity, floor): pass + + valid = Valid() + for protocol in (RequestPolicy, CheckerRegistry, SchemaCapability, + SafeAuditEventSink, RuntimeTelemetryAdapter, InternalIdAllocator): + assert isinstance(valid, protocol) + context = UserContext() + bad = object() + methods = ( + (context.with_request_policy, "request policy"), + (context.with_checker_registry, "checker registry"), + (context.with_schema_provider, "schema capability"), + (context.with_app_audit_event_sink, "application audit sink"), + (context.with_runtime_telemetry, "runtime telemetry"), + (context.with_internal_id_allocator, "internal ID allocator"), + ) + for install, message in methods: + with pytest.raises(TypeError, match=message): + install(bad) + with pytest.raises(TypeError, match="schema capability"): + context.with_schema_provider(type("SyncSchema", (), {"_ensure_schema": lambda *_: None})()) + with pytest.raises(TypeError, match="request policy"): + context.insert_resource("request_policy", bad) + with pytest.raises(TypeError, match="checker registry"): + context.insert_resource("checker_registry", bad) + with pytest.raises(TypeError, match="schema capability"): + context.insert_resource("schema_provider", bad) + with pytest.raises(TypeError, match="application audit sink"): + context.initialize_audit(valid, bad) + assert context._standard_audit_sink is None + + +@pytest.mark.asyncio +async def test_schema_only_capability_is_independent_and_preserves_bootstrap_order(): + events = [] + + class SchemaOnly: + async def _ensure_schema(self, context, capability): + events.append("schema") + + async def bootstrap(context): + events.append("bootstrap") + + schema = SchemaOnly() + assert isinstance(schema, SchemaCapability) + assert not hasattr(schema, "query") and not hasattr(schema, "mutate") + context = RuntimeModule.new().generated_bootstrap(bootstrap).into_context() + context.with_schema_provider(schema) + await context.ensure_schema() + await context.ensure_schema() + assert events == ["schema", "bootstrap", "schema", "bootstrap"] + + class CombinedService: + async def _ensure_schema(self, context, capability): + raise AssertionError("explicit schema provider must not be replaced") + + context.with_schema_provider(schema).insert_resource("dataService", CombinedService()) + await context.ensure_schema() + assert events[-2:] == ["schema", "bootstrap"] + + inferred = UserContext().insert_resource("dataService", CombinedService()) + assert isinstance(inferred.get_resource("schema_provider"), CombinedService) + inferred.insert_resource("dataService", object()) + assert inferred.get_resource("schema_provider") is None + + +async def school_fixture(path): + school = (EntityDescriptor("School").table_name("school_data") + .property(PropertyDescriptor("id", DataType.I64).is_id()) + .property(PropertyDescriptor("version", DataType.I64).is_version()) + .property(PropertyDescriptor("name", DataType.Text))) + provider = SimpleSchemaProvider() + provider.register_entity(school) + service = create_sqlite_service(str(path), provider) + context = (RuntimeModule.new().entity(school).into_context() + .with_schema_provider(service).insert_resource("dataService", service)) + context.disable_sql_log() + await context.ensure_schema() + return context, service + + +@pytest.mark.asyncio +async def test_object_policy_denies_query_before_sql_provider(tmp_path, monkeypatch): + context, service = await school_fixture(tmp_path / "policy.db") + called = [] + + async def forbidden_fetch(*args): + called.append("provider") + raise AssertionError("query provider must not be called") + + monkeypatch.setattr(service.transport, "fetch_all_sql", forbidden_fetch) + + class DenySchool: + def apply(self, query): + raise PermissionError("school query denied") + + context.with_request_policy(DenySchool()) + query = SelectQuery("School").project("id") + with pytest.raises(PermissionError, match="school query denied"): + prepared = context.prepare_query_request( + QueryRequest(query, _comment="load schools", _purpose="check policy")) + await service.query(context, prepared) + assert called == [] + await service.close() + + +@pytest.mark.asyncio +async def test_checker_failure_rejects_write_before_sql_provider(tmp_path, monkeypatch): + context, service = await school_fixture(tmp_path / "checker.db") + called = [] + + async def forbidden_execute(*args): + called.append("provider") + raise AssertionError("write provider must not be called") + + monkeypatch.setattr(service.transport, "execute_sql", forbidden_execute) + + class DenyChecker: + def check_and_fix(self, context, record, location, results): + raise ValueError("school name rejected") + + class Registry: + def checker(self, entity): + return DenyChecker() if entity == "School" else None + + context.with_checker_registry(Registry()) + request = MutationRequest(InsertCommand.new("School").value("name", "Rejected"), + comment="create rejected school") + with pytest.raises(ValueError, match="school name rejected"): + await service.mutate(context, request) + assert called == [] + await service.close() + + +@pytest.mark.asyncio +async def test_safe_audit_sink_and_telemetry_lifecycle_composed(): + events = [] + + class Provider: + async def begin(self, context): + events.append("begin") + return self + + async def commit(self, context): events.append("commit") + async def rollback(self, context): events.append("rollback") + + class Sink: + def on_safe_event(self, context, event): + events.append(("audit", event.entity)) + + class Telemetry: + def start(self, operation): + events.append(("start", operation.name)) + + class Scope: + def success(self, attributes=None): events.append(("success", operation.name)) + def failure(self, error): events.append(("failure", operation.name)) + + return Scope() + + context = (UserContext().insert_resource("dataService", Provider()) + .with_app_audit_event_sink(Sink()).with_runtime_telemetry(Telemetry())) + + async def save(graph): + await graph.context.send_audit_event( + RawAuditEvent(MutationAuditKind.CREATED, "School", 1, ())) + + await context.execute_graph_save(save, comment="create school") + assert events[0] == ("start", "graph.save") + assert events.index("begin") < events.index("commit") < events.index(("audit", "School")) + assert events[-1] == ("success", "graph.save") + + class BrokenTelemetry: + def start(self, operation): + raise RuntimeError("optional adapter is unavailable") + + context.with_runtime_telemetry(BrokenTelemetry()) + await context.execute_graph_save(save, comment="retry school") + assert events[-3:] == ["begin", "commit", ("audit", "School")] + + +def test_all_six_context_extensions_can_be_installed_together(): + class Extensions: + def apply(self, query): return query + def checker(self, entity): return None + async def _ensure_schema(self, context, capability): pass + def on_safe_event(self, context, event): pass + def start(self, operation): return None + async def next_id(self, entity): return 1 + async def ensure_id_floor(self, entity, floor): pass + + extension = Extensions() + context = (UserContext().with_request_policy(extension) + .with_checker_registry(extension) + .with_schema_provider(extension) + .with_app_audit_event_sink(extension) + .with_runtime_telemetry(extension) + .with_internal_id_allocator(extension)) + assert context.runtime_telemetry() is extension + assert context.internal_id_allocator() is extension + assert context.get_resource("schema_provider") is extension + assert context.get_resource("checker_registry") is extension + assert context.get_resource("request_policy") is extension