diff --git a/CHANGELOG.md b/CHANGELOG.md index 54653c1..6fde7b4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,10 +22,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Security --> -## [Unreleased] - YYYY-MM-DD +## [0.1.0a7] - 2026-09-07 ### Added - `SQLStorage` - SQL based storage module. +- `MPCompute` - Multiprocess compute module that offloads compute operations to a worker process pool. - `StorageModule` and `ComputeModule` now automatically translate exceptions raised in their methods into the method's expected result body. - Full class and method docstrings for `StorageModule` and `ComputeModule`, including success and error return examples, call examples with the full config body, and notes on the automatic exception translation. - `ComputeModule` docstring notes that a compute module must handle all errors returned from storage. @@ -35,8 +36,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `ComputeModule` and `StorageModule` base classes now inherit from ABC. - `DictStorage` now stores storage latch `created_at` as an ISO 8601 string instead of a `datetime` object. -### Deprecated - ### Removed - `NotImplementedError` since base classes now use auto checks from ABC. @@ -46,8 +45,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - The non-list (`get_*`) identity lookup in `validate_request` now populates the identity lookup and returns the correct identity error message. - `validate_batch_request` no longer raises `KeyError` on the non-list identity lookup path and no longer silently succeeds for an unregistered root definition in the list path. -### Security - ## [0.1.0a6] - 2026-08-27 diff --git a/src/authzee/compute/compute_module.py b/src/authzee/compute/compute_module.py index 1cdb029..78fc265 100644 --- a/src/authzee/compute/compute_module.py +++ b/src/authzee/compute/compute_module.py @@ -90,15 +90,15 @@ class ComputeModule(metaclass=_ComputeMeta): than implemented on compute: - The definition and grant persistence and retrieval configs: - `GetContextDefConfig`, `PutContextDefConfig`, `DeleteContextDefConfig`, - `GetIdentityDefConfig`, `PutIdentityDefConfig`, `DeleteIdentityDefConfig`, - `GetResourceDefConfig`, `PutResourceDefConfig`, `DeleteResourceDefConfig`, - `GetGrantConfig`, `EnactConfig`, and `RepealConfig`, along with the - standalone `ListContextDefsConfig`, `ListIdentityDefsConfig`, - `ListResourceDefsConfig`, `ListGrantsConfig`, and `ListGrantRefsConfig` as - top-level (non-embedded) configs. + `GetContextDefConfig`, `PutContextDefConfig`, `DeleteContextDefConfig`, + `GetIdentityDefConfig`, `PutIdentityDefConfig`, `DeleteIdentityDefConfig`, + `GetResourceDefConfig`, `PutResourceDefConfig`, `DeleteResourceDefConfig`, + `GetGrantConfig`, `EnactConfig`, and `RepealConfig`, along with the + standalone `ListContextDefsConfig`, `ListIdentityDefsConfig`, + `ListResourceDefsConfig`, `ListGrantsConfig`, and `ListGrantRefsConfig` as + top-level (non-embedded) configs. - The storage latch configs: `CreateLatchConfig`, `GetLatchConfig`, - `SetLatchConfig`, `DeleteLatchConfig`, and `CleanupLatchesConfig`. + `SetLatchConfig`, `DeleteLatchConfig`, and `CleanupLatchesConfig`. A compute module does still cause several of these storage calls to run (for example listing grants during an audit or authorize); when it does, it uses the diff --git a/src/authzee/compute/mp_compute.py b/src/authzee/compute/mp_compute.py new file mode 100644 index 0000000..60d2ecb --- /dev/null +++ b/src/authzee/compute/mp_compute.py @@ -0,0 +1,343 @@ +"""Multiprocess compute module for Authzee. + +All requests are offloaded to a worker process pool. +""" + +__all__ = [ + "MPCompute" +] + +import asyncio +from concurrent.futures import ProcessPoolExecutor +import multiprocessing +from typing import Any, Callable, Type + +from authzee.compute.compute_module import ComputeModule +from authzee.compute.in_process_compute import InProcessCompute +from authzee.module_locality import ModuleLocality +from authzee.storage.storage_module import StorageModule +from authzee.types.authzee import * +from authzee.types.config import ( + AuditConfig, + AuthorizeConfig, + BatchAuditConfig, + BatchAuthorizeConfig, + ComputeConstructConfig, + ComputeDestroyConfig, + ComputeShutdownConfig, + ComputeStartConfig, + ValidateBatchRequestConfig, + ValidateContextDefConfig, + ValidateGrantConfig, + ValidateIdentityDefConfig, + ValidateRequestConfig, + ValidateResourceDefConfig +) + + +class MPCompute(ComputeModule): + """Multiprocess Compute Module. + + Parameters + ---------- + max_workers : int | None + Maximum number of worker processes. If None, defaults to number of machine processors. + worker_compute : Type[ComputeModule] + The type of the compute module for each worker process to use. + worker_kwargs : dict[str, Any] + KWArgs to pass when creating compute modules for the worker processes. + + Examples + -------- + ```python + from authzee import ( + Authzee, + DictStorage, + InProcessCompute, + jmespath_execute, + MPCompute + ) + + + storage_dict = {} + authz = Authzee( + execute=jmespath_execute, + compute_type=MPCompute, + compute_kwargs={ + "max_workers": None, + "worker_compute": InProcessCompute, + "worker_kwargs": {} + }, + storage_type=DictStorage, + storage_kwargs={ + "storage_dict": storage_dict + }, + config={ # optional - AuthzeeConfigOverride | None - All keys are optional + "authzee": { + "raise_errors": True + } + # "method_name": {} + } + ) + """ + + + def __init__( + self, + max_workers: int | None, + worker_compute: Type[ComputeModule], + worker_kwargs: dict[str, Any] + ): + self._max_workers = max_workers + self._worker_compute = worker_compute + self._worker_kwargs = worker_kwargs + self._executor = None + + + async def start( + self, + execute: Callable[[str, Any], Any], + storage_type: Type[StorageModule], + storage_kwargs: dict[str, Any], + config: ComputeStartConfig + ) -> GenericResult: + await super().start( + execute=execute, + storage_type=storage_type, + storage_kwargs=storage_kwargs, + config=config + ) + self.locality = ModuleLocality.SYSTEM + self.has_parallel_paging = False + self._executor = ProcessPoolExecutor( + max_workers=self._max_workers, + mp_context=multiprocessing.get_context("spawn"), + initializer=_executor_start, + initargs=( + self._worker_compute, + self._worker_kwargs, + execute, + storage_type, + storage_kwargs, + config + ) + ) + + return { + "error": None + } + + + async def shutdown(self, config: ComputeShutdownConfig) -> GenericResult: + if self._executor is not None: + self._executor.shutdown(wait=True) + self._executor = None + + return { + "error": None + } + + + async def construct(self, config: ComputeConstructConfig) -> GenericResult: + return { + "error": None + } + + + async def destroy(self, config: ComputeDestroyConfig) -> GenericResult: + return { + "error": None + } + + + async def validate_context_def( + self, + context_def: ContextDef, + config: ValidateContextDefConfig + ) -> GenericResult: + return await asyncio.get_running_loop().run_in_executor( + self._executor, + _executor_run, + "validate_context_def", + { + "context_def": context_def, + "config": config + } + ) + + + async def validate_identity_def( + self, + identity_def: IdentityDef, + config: ValidateIdentityDefConfig + ) -> GenericResult: + return await asyncio.get_running_loop().run_in_executor( + self._executor, + _executor_run, + "validate_identity_def", + { + "identity_def": identity_def, + "config": config + } + ) + + + async def validate_resource_def( + self, + resource_def: ResourceDef, + config: ValidateResourceDefConfig + ) -> GenericResult: + return await asyncio.get_running_loop().run_in_executor( + self._executor, + _executor_run, + "validate_resource_def", + { + "resource_def": resource_def, + "config": config + } + ) + + + async def validate_grant( + self, + grant: Grant, + config: ValidateGrantConfig + ) -> GenericResult: + return await asyncio.get_running_loop().run_in_executor( + self._executor, + _executor_run, + "validate_grant", + { + "grant": grant, + "config": config + } + ) + + + async def validate_request( + self, + request: AuthzeeRequest, + config: ValidateRequestConfig + ) -> GenericResult: + return await asyncio.get_running_loop().run_in_executor( + self._executor, + _executor_run, + "validate_request", + { + "request": request, + "config": config + } + ) + + + async def validate_batch_request( + self, + batch_request: AuthzeeBatchRequest, + config: ValidateBatchRequestConfig + ) -> ValidateBatchRequestResult: + return await asyncio.get_running_loop().run_in_executor( + self._executor, + _executor_run, + "validate_batch_request", + { + "batch_request": batch_request, + "config": config + } + ) + + + async def audit( + self, + request: AuthzeeRequest, + page_ref: str | None, + config: AuditConfig + ) -> AuditResultPage: + return await asyncio.get_running_loop().run_in_executor( + self._executor, + _executor_run, + "audit", + { + "request": request, + "page_ref": page_ref, + "config": config + } + ) + + + async def authorize( + self, + request: AuthzeeRequest, + config: AuthorizeConfig + ) -> AuthorizeResult: + return await asyncio.get_running_loop().run_in_executor( + self._executor, + _executor_run, + "authorize", + { + "request": request, + "config": config + } + ) + + + async def batch_audit( + self, + batch_request: AuthzeeBatchRequest, + page_ref: str | None, + config: BatchAuditConfig + ) -> BatchAuditResultPage: + return await asyncio.get_running_loop().run_in_executor( + self._executor, + _executor_run, + "batch_audit", + { + "batch_request": batch_request, + "page_ref": page_ref, + "config": config + } + ) + + + async def batch_authorize( + self, + batch_request: AuthzeeBatchRequest, + config: BatchAuthorizeConfig + ) -> BatchAuthorizeResult: + return await asyncio.get_running_loop().run_in_executor( + self._executor, + _executor_run, + "batch_authorize", + { + "batch_request": batch_request, + "config": config + } + ) + + +def _executor_start( + worker_compute: Type[ComputeModule], + worker_kwargs: dict[str, Any], + execute: Callable[[str, Any], Any], + storage_type: Type[StorageModule], + storage_kwargs: dict[str, Any], + config: ComputeStartConfig +) -> None: + global _authzee_compute + _authzee_compute = worker_compute(**worker_kwargs) + + return asyncio.run( + _authzee_compute.start( + execute=execute, + storage_type=storage_type, + storage_kwargs=storage_kwargs, + config=config + ) + ) + + +def _executor_run(method: str, method_kwargs: dict[str, Any]) -> Any: + global _authzee_compute + + return asyncio.run( + getattr(_authzee_compute, method)(**method_kwargs) + ) diff --git a/src/authzee/compute/multiprocess_compute.py.bak b/src/authzee/compute/multiprocess_compute.py.bak deleted file mode 100644 index 084cb83..0000000 --- a/src/authzee/compute/multiprocess_compute.py.bak +++ /dev/null @@ -1,169 +0,0 @@ -__all__ = [ - "MultiprocessCompute" -] - -import asyncio -import multiprocessing -from concurrent.futures import ProcessPoolExecutor -from typing import Any, Callable, Dict, List, Type - -from authzee.compute.compute_module import ComputeModule -from authzee.module_locality import ModuleLocality -from authzee.storage.storage_module import StorageModule - - -class MultiprocessCompute(ComputeModule): - """Compute using multiple processes with system locality. - - Requests for audit page or authorize are forwarded to a worker to handle the request. - Each worker will create it's own instance of ``compute_type`` and use that to process the requests. - - Acts as a base class to offload to worker processes. - - Parameters - ---------- - max_workers : int | None - Maximum number of worker processes. If None, defaults to number of machine processors. - compute_type : Type[ComputeModule] - The type of the compute module for each worker process to use to process requests. - compute_kwargs : Dict[str, Any] - KWArgs to pass when creating compute modules for the worker processes. - - Examples - -------- - .. code-block:: python - - from Authzee import Authzee, InProcessCompute, MultiprocessCompute, FanOutMPCompute - - azee_app = Authzee( - - ) - """ - - def __init__( - self, - max_workers: int | None, - compute_type: Type[ComputeModule], - compute_kwargs: Dict[str, Any] - ): - self.max_workers = max_workers - self.compute_type = compute_type - self.compute_kwargs = compute_kwargs - self._executor = None - - - async def start( - self, - identity_defs: List[Dict[str, Any]], - resource_defs: List[Dict[str, Any]], - search: Callable[[str, Any], Any], - storage_type: Type[StorageModule], - storage_kwargs: Dict[str, Any] - ) -> None: - """Create runtime resources for the compute module.""" - await super().start( - identity_defs=identity_defs, - resource_defs=resource_defs, - search=search, - storage_type=storage_type, - storage_kwargs=storage_kwargs - ) - self.locality = ModuleLocality.SYSTEM - self._executor = ProcessPoolExecutor( - max_workers=self.max_workers, - mp_context=multiprocessing.get_context("spawn"), - initializer=_executor_start, - initargs=( - self.compute_type, - self.compute_kwargs, - { - "identity_defs": identity_defs, - "resource_defs": resource_defs, - "search": search, - "storage_type": storage_type, - "storage_kwargs": storage_kwargs - }, - ) - ) - - - async def shutdown(self) -> None: - if self._executor: - self._executor.shutdown(wait=True) - self._executor = None - - - async def audit( - self, - request: dict, - page_ref: str | None, - grants_page_size: int, - parallel_paging: bool, - refs_page_size: int - ) -> dict: - return await asyncio.get_running_loop().run_in_executor( - self._executor, - _executor_audit, - ( - { - "request": request, - "page_ref": page_ref, - "grants_page_size": grants_page_size, - "parallel_paging": parallel_paging, - "refs_page_size": refs_page_size - }, - ) - ) - - - async def authorize( - self, - request: dict, - grants_page_size: int, - parallel_paging: bool, - refs_page_size: int - ) -> dict: - return await asyncio.get_running_loop().run_in_executor( - self._executor, - _executor_authorize, - ( - { - "request": request, - "grants_page_size": grants_page_size, - "parallel_paging": parallel_paging, - "refs_page_size": refs_page_size - }, - ) - ) - - -def _executor_start( - compute_type: Type[ComputeModule], - compute_kwargs: Dict[str, Any], - start_kwargs: Dict[str, Any] -) -> None: - global authzee_compute - authzee_compute = compute_type(**compute_kwargs) - return asyncio.run(authzee_compute.start(**start_kwargs)) - - -def _executor_audit( - audit_kwargs: Dict[str, Any] -) -> dict: - global authzee_compute - return asyncio.run( - authzee_compute.audit( - **audit_kwargs - ) - ) - - -def _executor_authorize( - authorize_kwargs: Dict[str, Any] -) -> dict: - global authzee_compute - return asyncio.run( - authzee_compute.authorize( - **authorize_kwargs - ) - ) diff --git a/src/authzee/storage/storage_module.py b/src/authzee/storage/storage_module.py index 7bab01e..7239693 100644 --- a/src/authzee/storage/storage_module.py +++ b/src/authzee/storage/storage_module.py @@ -87,13 +87,13 @@ class StorageModule(metaclass=_StorageMeta): storage layer: - The nested `get_*` / `use_list_*` / `list_*` sub-configs inside the put and - delete definition configs and the repeal config - specifically - `PutContextDefConfig`, `DeleteContextDefConfig`, `PutIdentityDefConfig`, - `DeleteIdentityDefConfig`, `PutResourceDefConfig`, `DeleteResourceDefConfig`, - and `RepealConfig`. A storage module puts, deletes, or repeals the target - directly by its type or UUID; those nested sub-configs describe an optional - "look the target up first via a get or a list" step that belongs to the - orchestration layer, not to storage. Subclasses should ignore them. + delete definition configs and the repeal config - specifically + `PutContextDefConfig`, `DeleteContextDefConfig`, `PutIdentityDefConfig`, + `DeleteIdentityDefConfig`, `PutResourceDefConfig`, `DeleteResourceDefConfig`, + and `RepealConfig`. A storage module puts, deletes, or repeals the target + directly by its type or UUID; those nested sub-configs describe an optional + "look the target up first via a get or a list" step that belongs to the + orchestration layer, not to storage. Subclasses should ignore them. Any config type not listed above is used by the corresponding storage method where its keys map to that method's behavior. diff --git a/tests/unit/test_mp_compute.py b/tests/unit/test_mp_compute.py new file mode 100644 index 0000000..79de904 --- /dev/null +++ b/tests/unit/test_mp_compute.py @@ -0,0 +1,520 @@ +"""Unit tests for authzee.compute MPCompute. + +`MPCompute` offloads every compute operation to a worker process pool where a +per-worker `InProcessCompute` does the real work. To exercise `MPCompute`'s own +code deterministically (without the cost and `__main__` requirements of spawning +real processes), these tests: + +- Patch the running loop's `run_in_executor` so the executor task runs inline in +the test process, driving each delegation method through its real body. +- Seed a real in-process worker `InProcessCompute` via the module-level +`_executor_start` so `_executor_run` resolves against it. +- Patch `ProcessPoolExecutor` in `start` so no real pool is spawned. + +The module-level `_executor_start` / `_executor_run` helpers are also tested +directly. +""" + +import asyncio +import os +import sys +from unittest.mock import MagicMock, patch +from uuid import uuid4 + +import pytest + + +sys.path.insert(0, os.path.dirname(__file__)) + +from authzee.compute.in_process_compute import InProcessCompute +import authzee.compute.mp_compute as mp_compute_module +from authzee.compute.mp_compute import _executor_run, _executor_start, MPCompute +from authzee.jmespath import jmespath_execute +from authzee.module_locality import ModuleLocality +from authzee.storage.dict_storage import DictStorage + + +CONTEXT_DEF = { + "context_type": "NONE", + "schema": { + "type": "object", + "additionalProperties": False + } +} +IDENTITY_DEF = { + "identity_type": "user", + "schema": { + "type": "object", + "required": [ + "username", + "department" + ], + "additionalProperties": False, + "properties": { + "username": { + "type": "string" + }, + "department": { + "type": "string" + } + } + } +} +RESOURCE_DEF = { + "resource_type": "balloon", + "actions": [ + "balloon:read", + "balloon:inflate" + ], + "schema": { + "type": "object", + "required": [ + "color", + "is_inflated" + ], + "additionalProperties": False, + "properties": { + "color": { + "type": "string" + }, + "is_inflated": { + "type": "boolean" + } + } + } +} +GRANT = { + "grant_uuid": "0da5dfc6-c919-4bd6-b80f-a351a9ac8d27", + "name": "Allow inflate", + "description": "", + "tags": {}, + "effect": "allow", + "actions": [ + "balloon:read", + "balloon:inflate" + ], + "query": "length(request.identities.user[?department == 'Balloon Dept']) > `0`", + "equality": True, + "applicable_on_failure": False, + "data": {} +} +REQUEST = { + "identities": { + "user": [ + { + "username": "balloon_person", + "department": "Balloon Dept" + } + ] + }, + "action": "balloon:inflate", + "resource_type": "balloon", + "resource": { + "color": "red", + "is_inflated": False + }, + "context_type": "NONE", + "context": {} +} +BATCH_REQUEST = { + "identities": { + "user": [ + { + "username": "balloon_person", + "department": "Balloon Dept" + } + ] + }, + "action": "balloon:inflate", + "resource_type": "balloon", + "resource": { + "color": "red", + "is_inflated": False + }, + "context_type": "NONE", + "context": {}, + "batch": [ + {} + ] +} +VALIDATE_REQUEST_CONFIG = { + "get_context_def": { + "use_cache": False + }, + "use_list_context_defs": False, + "list_context_defs": { + "page_size": 1000, + "use_cache": False + }, + "get_identity_def": { + "use_cache": False + }, + "use_list_identity_defs": False, + "list_identity_defs": { + "page_size": 1000, + "use_cache": False + }, + "get_resource_def": { + "use_cache": False + }, + "use_list_resource_defs": False, + "list_resource_defs": { + "page_size": 1000, + "use_cache": False + } +} +LIST_GRANTS_CONFIG = { + "page_size": 1000, + "use_cache": False +} + + +class _InlineFuture: + """Awaitable wrapper that resolves to a precomputed value.""" + + + def __init__(self, value): + self._value = value + + + def __await__(self): + if False: + yield + + return self._value + + +@pytest.fixture +def storage_dict(): + return {} + + +@pytest.fixture +def worker(storage_dict): + """Seed the module-level worker `InProcessCompute` used by `_executor_run`. + + Also seeds a `DictStorage` (sharing `storage_dict`) with defs and a grant so + the request/audit/authorize delegations have data to work against. + """ + + async def setup(): + storage = DictStorage(storage_dict=storage_dict) + await storage.construct(config={}) + await storage.start(config={}) + await storage.put_context_def(CONTEXT_DEF, config={}) + await storage.put_identity_def(IDENTITY_DEF, config={}) + await storage.put_resource_def(RESOURCE_DEF, config={}) + await storage.enact(grant=GRANT, config={}) + + asyncio.run(setup()) + _executor_start( + InProcessCompute, + {}, + jmespath_execute, + DictStorage, + { + "storage_dict": storage_dict + }, + { + "storage": {} + } + ) + + yield mp_compute_module._authzee_compute + + asyncio.run( + mp_compute_module._authzee_compute.shutdown(config={}) + ) + + +@pytest.fixture +def compute(worker): + """An `MPCompute` whose `run_in_executor` runs inline against the worker.""" + c = MPCompute( + max_workers=2, + worker_compute=InProcessCompute, + worker_kwargs={} + ) + c._executor = MagicMock() + + return c + + +class _InlineLoop: + """Fake event loop whose ``run_in_executor`` runs the target inline.""" + + + def run_in_executor(self, executor, func, *args): + return _InlineFuture(func(*args)) + + +def _run_inline(coro_func): + """Drive a single delegation coroutine to completion with no running loop. + + The delegation methods are ``return await asyncio.get_running_loop().run_in_executor(...)``. + We patch `get_running_loop` to a fake loop that runs the executor target + inline and stepping the coroutine manually via ``send`` means there is no + real running event loop when the target `_executor_run` calls `asyncio.run`, + faithfully mirroring a real worker process. + """ + with patch.object( + asyncio, + "get_running_loop", + return_value=_InlineLoop() + ): + coro = coro_func() + try: + coro.send(None) + except StopIteration as stop: + return stop.value + + raise AssertionError("coroutine did not complete synchronously") + + +def test_mp_start_sets_locality_and_creates_executor(storage_dict): + c = MPCompute( + max_workers=3, + worker_compute=InProcessCompute, + worker_kwargs={} + ) + fake_executor = MagicMock() + with patch.object( + mp_compute_module, + "ProcessPoolExecutor", + return_value=fake_executor + ) as ppe: + async def run(): + return await c.start( + execute=jmespath_execute, + storage_type=DictStorage, + storage_kwargs={ + "storage_dict": storage_dict + }, + config={ + "storage": {} + } + ) + + result = asyncio.run(run()) + + assert result['error'] is None + assert c.locality == ModuleLocality.SYSTEM + assert c.has_parallel_paging is False + assert c._executor is fake_executor + assert ppe.call_args.kwargs['max_workers'] == 3 + assert ppe.call_args.kwargs['initargs'][0] is InProcessCompute + assert ppe.call_args.kwargs['initargs'][1] == {} + + +def test_mp_shutdown_shuts_down_executor(): + c = MPCompute( + max_workers=None, + worker_compute=InProcessCompute, + worker_kwargs={} + ) + fake_executor = MagicMock() + c._executor = fake_executor + result = asyncio.run(c.shutdown(config={})) + assert result['error'] is None + fake_executor.shutdown.assert_called_once_with(wait=True) + assert c._executor is None + + +def test_mp_shutdown_with_no_executor(): + c = MPCompute( + max_workers=None, + worker_compute=InProcessCompute, + worker_kwargs={} + ) + result = asyncio.run(c.shutdown(config={})) + assert result['error'] is None + assert c._executor is None + + +def test_mp_construct(): + c = MPCompute( + max_workers=None, + worker_compute=InProcessCompute, + worker_kwargs={} + ) + result = asyncio.run(c.construct(config={})) + assert result['error'] is None + + +def test_mp_destroy(): + c = MPCompute( + max_workers=None, + worker_compute=InProcessCompute, + worker_kwargs={} + ) + result = asyncio.run(c.destroy(config={})) + assert result['error'] is None + + +def test_mp_validate_context_def(compute): + result = _run_inline( + lambda: compute.validate_context_def( + context_def=CONTEXT_DEF, + config={} + ) + ) + assert result['error'] is None + + +def test_mp_validate_identity_def(compute): + result = _run_inline( + lambda: compute.validate_identity_def( + identity_def=IDENTITY_DEF, + config={} + ) + ) + assert result['error'] is None + + +def test_mp_validate_resource_def(compute): + result = _run_inline( + lambda: compute.validate_resource_def( + resource_def=RESOURCE_DEF, + config={} + ) + ) + assert result['error'] is None + + +def test_mp_validate_grant(compute): + result = _run_inline( + lambda: compute.validate_grant(grant=GRANT, config={}) + ) + assert result['error'] is None + + +def test_mp_validate_request(compute): + result = _run_inline( + lambda: compute.validate_request( + request=REQUEST, + config=VALIDATE_REQUEST_CONFIG + ) + ) + assert result['error'] is None + + +def test_mp_validate_batch_request(compute): + result = _run_inline( + lambda: compute.validate_batch_request( + batch_request=BATCH_REQUEST, + config=VALIDATE_REQUEST_CONFIG + ) + ) + assert result['error'] is None + assert result['batch'] == [ + { + "error": None + } + ] + + +def test_mp_audit(compute): + result = _run_inline( + lambda: compute.audit( + request=REQUEST, + page_ref=None, + config={ + "validate_request": VALIDATE_REQUEST_CONFIG, + "list_grants": LIST_GRANTS_CONFIG + } + ) + ) + assert result['error'] is None + assert len(result['results']) == 1 + + +def test_mp_authorize(compute): + result = _run_inline( + lambda: compute.authorize( + request=REQUEST, + config={ + "validate_request": VALIDATE_REQUEST_CONFIG, + "list_grants": LIST_GRANTS_CONFIG, + "parallel_paging": False, + "list_grant_refs": { + "page_size": 10, + "use_cache": False + } + } + ) + ) + assert result['error'] is None + assert result['is_authorized'] is True + + +def test_mp_batch_audit(compute): + result = _run_inline( + lambda: compute.batch_audit( + batch_request=BATCH_REQUEST, + page_ref=None, + config={ + "validate_batch_request": VALIDATE_REQUEST_CONFIG, + "list_grants": LIST_GRANTS_CONFIG + } + ) + ) + assert result['error'] is None + assert len(result['batch']) == 1 + + +def test_mp_batch_authorize(compute): + result = _run_inline( + lambda: compute.batch_authorize( + batch_request=BATCH_REQUEST, + config={ + "validate_batch_request": VALIDATE_REQUEST_CONFIG, + "list_grants": LIST_GRANTS_CONFIG, + "parallel_paging": False, + "list_grant_refs": { + "page_size": 10, + "use_cache": False + } + } + ) + ) + assert result['error'] is None + assert len(result['batch']) == 1 + assert result['batch'][0]['is_authorized'] is True + + +def test_executor_start_and_run(storage_dict): + async def seed(): + storage = DictStorage(storage_dict=storage_dict) + await storage.construct(config={}) + + asyncio.run(seed()) + start_result = _executor_start( + InProcessCompute, + {}, + jmespath_execute, + DictStorage, + { + "storage_dict": storage_dict + }, + { + "storage": {} + } + ) + assert start_result['error'] is None + assert isinstance( + mp_compute_module._authzee_compute, + InProcessCompute + ) + + run_result = _executor_run( + "validate_context_def", + { + "context_def": CONTEXT_DEF, + "config": {} + } + ) + assert run_result['error'] is None + + asyncio.run( + mp_compute_module._authzee_compute.shutdown(config={}) + )