From be7594daa06ac30ccec76f2be3f8209912240f64 Mon Sep 17 00:00:00 2001 From: Abram Booth Date: Fri, 9 Feb 2024 07:23:26 -0500 Subject: [PATCH] addon_interface decorator --- addon_toolkit/__init__.py | 12 +- addon_toolkit/categories/__init__.py | 4 +- addon_toolkit/categories/storage.py | 16 +- addon_toolkit/category.py | 32 ---- addon_toolkit/interface.py | 154 ++++++++++++++--- addon_toolkit/network/__init__.py | 25 +++ addon_toolkit/operation.py | 157 ++++-------------- addon_toolkit/tests/test_addon_category.py | 132 --------------- .../tests/test_addon_interface_declaration.py | 148 +++++++++++++++++ 9 files changed, 349 insertions(+), 331 deletions(-) delete mode 100644 addon_toolkit/category.py create mode 100644 addon_toolkit/network/__init__.py delete mode 100644 addon_toolkit/tests/test_addon_category.py create mode 100644 addon_toolkit/tests/test_addon_interface_declaration.py diff --git a/addon_toolkit/__init__.py b/addon_toolkit/__init__.py index 4e4c5278..5d1ee9b9 100644 --- a/addon_toolkit/__init__.py +++ b/addon_toolkit/__init__.py @@ -1,10 +1,10 @@ -from .category import AddonCategory from .interface import ( - AddonInterface, + AddonInterfaceDeclaration, PagedResult, + addon_interface, ) from .operation import ( - AddonOperation, + AddonOperationDeclaration, AddonOperationType, proxy_operation, redirect_operation, @@ -12,11 +12,11 @@ __all__ = ( - "AddonCategory", - "AddonInterface", - "AddonOperation", + "AddonInterfaceDeclaration", + "AddonOperationDeclaration", "AddonOperationType", "PagedResult", + "addon_interface", "proxy_operation", "redirect_operation", ) diff --git a/addon_toolkit/categories/__init__.py b/addon_toolkit/categories/__init__.py index f71eb0fc..65ad1cd9 100644 --- a/addon_toolkit/categories/__init__.py +++ b/addon_toolkit/categories/__init__.py @@ -1,4 +1,4 @@ -from .storage import StorageAddonCategory +from .storage import StorageAddon -__all__ = ("StorageAddonCategory",) +__all__ = ("StorageAddon",) diff --git a/addon_toolkit/categories/storage.py b/addon_toolkit/categories/storage.py index f0ae1192..122c211d 100644 --- a/addon_toolkit/categories/storage.py +++ b/addon_toolkit/categories/storage.py @@ -1,15 +1,14 @@ import enum from addon_toolkit import ( - AddonCategory, - AddonInterface, PagedResult, + addon_interface, proxy_operation, redirect_operation, ) -__all__ = ("StorageAddonCategory",) +__all__ = ("StorageAddon",) class StorageCapability(enum.Enum): @@ -17,8 +16,9 @@ class StorageCapability(enum.Enum): UPDATE = "update" -# what a base StorageInterface could be like (incomplete) -class StorageInterface(AddonInterface): +# what a base StorageAddon could be like (incomplete) +@addon_interface(capabilities=StorageCapability) +class StorageAddon: ## # "item-read" operations: @@ -88,9 +88,3 @@ async def get_version_ids(self, item_id: str) -> PagedResult: @proxy_operation(capability=StorageCapability.UPDATE) async def pls_restore_version(self, item_id: str, version_id: str): raise NotImplementedError - - -StorageAddonCategory = AddonCategory( - capabilities=StorageCapability, - base_interface=StorageInterface, -) diff --git a/addon_toolkit/category.py b/addon_toolkit/category.py deleted file mode 100644 index eca9b4aa..00000000 --- a/addon_toolkit/category.py +++ /dev/null @@ -1,32 +0,0 @@ -import dataclasses -import enum -from typing import Iterator - -from .interface import AddonInterface -from .operation import AddonOperation - - -@dataclasses.dataclass(frozen=True) -class AddonCategory: - capabilities: type[enum.Enum] - base_interface: type[AddonInterface] - - def operations_declared( - self, - *, - capability: enum.Enum | None = None, - ) -> Iterator[AddonOperation]: - for _interface_cls in self.base_interface.__mro__: - yield from AddonOperation.operations_declared_on_interface( - _interface_cls, capability=capability - ) - - def operations_implemented( - self, - interface_cls: type[AddonInterface], - capability: enum.Enum | None = None, - ) -> Iterator[AddonOperation]: - assert issubclass(interface_cls, self.base_interface) - for _op in self.operations_declared(capability=capability): - if _op.is_implemented_on(interface_cls): - yield _op diff --git a/addon_toolkit/interface.py b/addon_toolkit/interface.py index 2b46035d..bc7884b5 100644 --- a/addon_toolkit/interface.py +++ b/addon_toolkit/interface.py @@ -1,18 +1,23 @@ import dataclasses +import enum +import inspect import logging -from http import HTTPMethod +import weakref +from typing import ( + Callable, + ClassVar, + Iterator, + Optional, +) +from .operation import AddonOperationDeclaration -__all__ = ("AddonInterface", "PagedResult") +__all__ = ("addon_interface", "PagedResult") _logger = logging.getLogger(__name__) -### -# addon interface - - @dataclasses.dataclass class PagedResult: # TODO: consistent handling of paged results page: list @@ -20,25 +25,120 @@ class PagedResult: # TODO: consistent handling of paged results @dataclasses.dataclass -class AddonInterface: +class AddonInterfaceDeclaration: + """dataclass for the operations declared on a class decorated with `addon_interface`""" + + interface_cls: type + capabilities: type[enum.Enum] + method_name_by_op: dict[AddonOperationDeclaration, str] = dataclasses.field( + default_factory=dict, + ) + ops_by_capability: dict[ + enum.Enum, set[AddonOperationDeclaration] + ] = dataclasses.field( + default_factory=dict, + ) + + ### + # AddonInterface stores references to declared interface classes + + # private storage linking a class to data gleaned from its decorator + __declarations_by_cls: ClassVar[ + weakref.WeakKeyDictionary[type, "AddonInterfaceDeclaration"] + ] = weakref.WeakKeyDictionary() + + @staticmethod + def declare(capabilities: type[enum.Enum]): + def _cls_decorator(interface_cls: type) -> type: + AddonInterfaceDeclaration.__declarations_by_cls[ + interface_cls + ] = AddonInterfaceDeclaration(interface_cls, capabilities) + return interface_cls + + return _cls_decorator + + @staticmethod + def for_class(interface_cls: type) -> Optional["AddonInterfaceDeclaration"]: + return AddonInterfaceDeclaration.__declarations_by_cls.get(interface_cls) + ### - # public api for use on `self` when implementing operations - - # TODO: consider intermediate dataclasses to limit direct use of data models - # authorized_account: AuthorizedStorageAccount - # configured_addon: ConfiguredStorageAddon | None - - async def send_request(self, http_method: HTTPMethod, url: str, **kwargs): - """helper for external requests in addon implementations - - subclasses SHOULD use this instead of sending requests by hand - """ - _logger.info("sending %s to %s", http_method, url) - # TODO: common http handling (retry, backoff, etc) to ease implementer load - # async with httpx.AsyncClient() as _client: # TODO: shared client? - # _response = await _client.request( - # http_method, - # url, - # **kwargs, - # ) - # return _response + # AddonInterfaceDeclaration instance methods + + def __post_init__(self): + self._gather_operations() + + def _gather_operations(self): + for _name, _fn in inspect.getmembers(self.interface_cls, inspect.isfunction): + _maybe_op = AddonOperationDeclaration.for_function(_fn) + if _maybe_op is not None: + self._add_operation(_name, _maybe_op) + + def _add_operation(self, method_name: str, operation: AddonOperationDeclaration): + assert operation not in self.method_name_by_op, ( + f"duplicate operation '{operation}'" f" on {self.interface_cls}" + ) + self.method_name_by_op[operation] = method_name + self.ops_by_capability.setdefault( + operation.capability, + set(), + ).add(operation) + + def get_operation_method( + self, cls_or_instance: type | object, operation: AddonOperationDeclaration + ) -> Callable: + _cls = ( + cls_or_instance + if isinstance(cls_or_instance, type) + else type(cls_or_instance) + ) + assert self.interface_cls is not _cls + assert issubclass(_cls, self.interface_cls) + try: + _method_name = self.method_name_by_op[operation] + except KeyError: + raise ValueError # TODO: helpful exception type + _declared_fn = getattr(self.interface_cls, _method_name) + _implemented_fn = getattr(_cls, _method_name) + if _implemented_fn is _declared_fn: + raise NotImplementedError( # TODO: helpful exception type + f"operation '{_method_name}' not implemented by {_cls}" + ) + # now get the method directly on what was passed in + # to ensure a bound method when that arg is an interface instance + return getattr(cls_or_instance, _method_name) + + def operation_is_implemented( + self, implementation_cls: type, operation: AddonOperationDeclaration + ): + try: + return bool(self.get_operation_method(implementation_cls, operation)) + except NotImplementedError: # TODO: more specific error + return False + + def invoke(self, operation, interface_instance, /, args=None, kwargs=None): + # TODO: reconsider + _op_method = self.get_operation_method(interface_instance, operation) + return _op_method(*(args or ()), **(kwargs or {})) + + def get_declared_operations( + self, + *, + capability: enum.Enum | None = None, + ) -> Iterator[AddonOperationDeclaration]: + if capability is None: + yield from self.method_name_by_op.keys() + else: + yield from self.ops_by_capability.get(capability, ()) + + def get_implemented_operations( + self, + implementation_cls: type, + capability: enum.Enum | None = None, + ) -> Iterator[AddonOperationDeclaration]: + for _op in self.get_declared_operations(capability=capability): + if self.operation_is_implemented(implementation_cls, _op): + yield _op + + +# meant for use as decorator on a class, `@addon_interface(MyCapabilitiesEnum)` +addon_interface = AddonInterfaceDeclaration.declare diff --git a/addon_toolkit/network/__init__.py b/addon_toolkit/network/__init__.py new file mode 100644 index 00000000..18efda13 --- /dev/null +++ b/addon_toolkit/network/__init__.py @@ -0,0 +1,25 @@ +"""TODO: give addon implementers an easy way to declare the network requests +their addon needs while allowing consistent handling in any given addon_service +implementation +""" +import http +import logging + + +_logger = logging.getLogger(__name__) + + +async def send_request(self, http_method: http.HTTPMethod, url: str, **kwargs): + """helper for external requests in addon implementations + + subclasses SHOULD use this instead of sending requests by hand + """ + _logger.info("sending %s to %s", http_method, url) + # TODO: common http handling (retry, backoff, etc) to ease implementer load + # async with httpx.AsyncClient() as _client: # TODO: shared client? + # _response = await _client.request( + # http_method, + # url, + # **kwargs, + # ) + # return _response diff --git a/addon_toolkit/operation.py b/addon_toolkit/operation.py index 780b0151..d9a0f52d 100644 --- a/addon_toolkit/operation.py +++ b/addon_toolkit/operation.py @@ -5,14 +5,12 @@ from typing import ( Callable, ClassVar, - Iterator, + Optional, ) -from .interface import AddonInterface - __all__ = ( - "AddonOperation", + "AddonOperationDeclaration", "AddonOperationType", "proxy_operation", "redirect_operation", @@ -20,153 +18,70 @@ class AddonOperationType(enum.Enum): - REDIRECT = enum.auto() - PROXY = enum.auto() - - -@dataclasses.dataclass -class _InterfaceOperations: - by_capability: dict[enum.Enum, set["AddonOperation"]] = dataclasses.field( - default_factory=dict, - ) - by_method_name: dict[str, "AddonOperation"] = dataclasses.field( - default_factory=dict, - ) - - def add_operation(self, operation: "AddonOperation"): - assert operation.method_name not in self.by_method_name, ( - f"duplicate operation '{operation.method_name}'" - f" on {operation.declaration_cls}" - ) - self.by_method_name[operation.method_name] = operation - self.by_capability.setdefault( - operation.capability, - set(), - ).add(operation) + REDIRECT = "redirect" + PROXY = "redirect" @dataclasses.dataclass(frozen=True) -class _UnboundAddonOperation: - operation_fn: Callable - operation_type: AddonOperationType - capability: enum.Enum - - def __set_name__(self, cls, name): - AddonOperation._register_operation(self, cls, name) - # replace this _UnboundAddonOperation with the original function - setattr(cls, name, self.operation_fn) +class AddonOperationDeclaration: + """dataclass for a declared operation method on a subclass of AddonInterface + created by the decorators "proxy_operation" and "redirect_operation" + """ -@dataclasses.dataclass(frozen=True) -class AddonOperation: operation_type: AddonOperationType capability: enum.Enum - declaration_cls: type[AddonInterface] - method_name: str - - # private registry for all decorated operations - _registry: ClassVar[ - weakref.WeakKeyDictionary[ - type[AddonInterface], - _InterfaceOperations, - ] - ] = weakref.WeakKeyDictionary() + operation_fn: Callable ### - # methods for _registry interaction - - @classmethod - def _register_operation( - cls, - unbound_operation: _UnboundAddonOperation, - declaration_cls: type, - method_name: str, + # AddonOperationDeclaration stores references to declared operations + + # private storage linking a function to data gleaned from its decorator + __operations_by_fn: ClassVar[ + weakref.WeakKeyDictionary[Callable, "AddonOperationDeclaration"] + ] = weakref.WeakKeyDictionary() + + @staticmethod + def declared( + operation_fn: Callable, + capability: enum.Enum, + operation_type: AddonOperationType, ): - _operation = cls( - operation_type=unbound_operation.operation_type, - capability=unbound_operation.capability, - declaration_cls=declaration_cls, - method_name=method_name, + AddonOperationDeclaration.__operations_by_fn[ + operation_fn + ] = AddonOperationDeclaration( + operation_type=operation_type, + capability=capability, + operation_fn=operation_fn, ) - try: - _interface_ops = cls._registry[_operation.declaration_cls] - except KeyError: - _interface_ops = cls._registry[ - _operation.declaration_cls - ] = _InterfaceOperations() - _interface_ops.add_operation(_operation) - - @classmethod - def operations_declared_on_interface( - cls, - interface_cls: type[AddonInterface], - capability: enum.Enum | None = None, - ) -> Iterator["AddonOperation"]: - try: - _interface_ops = cls._registry[interface_cls] - except KeyError: - return # zero - else: - if capability is None: - yield from _interface_ops.by_method_name.values() - else: - yield from _interface_ops.by_capability.get(capability, ()) - ### - # instance methods - - def __post_init__(self): - assert self.declaration_cls is not AddonInterface - assert issubclass(self.declaration_cls, AddonInterface) - - def get_operation_method( - self, interface: AddonInterface | type[AddonInterface] - ) -> Callable: - _interface_cls = interface if isinstance(interface, type) else type(interface) - assert _interface_cls is not self.declaration_cls - assert issubclass(_interface_cls, self.declaration_cls) - _declared_fn = getattr(self.declaration_cls, self.method_name) - _implemented_fn = getattr(_interface_cls, self.method_name) - if _implemented_fn is _declared_fn: - raise NotImplementedError( # TODO: specific exception class - f"operation '{self.method_name}' not implemented by {type(interface)}" - ) - # now get the method directly on the arg (instead of _interface_cls) - # to ensure a bound method when that arg is an interface instance - return getattr(interface, self.method_name) - - def is_implemented_on(self, interface): - try: - return bool(self.get_operation_method(interface)) - except NotImplementedError: - return False - - def invoke(self, interface, /, args=None, kwargs=None): - # TODO: reconsider - _op_method = self.get_operation_method(interface) - return _op_method(*(args or ()), **(kwargs or {})) + @staticmethod + def for_function(fn: Callable) -> Optional["AddonOperationDeclaration"]: + return AddonOperationDeclaration.__operations_by_fn.get(fn) def redirect_operation(capability: enum.Enum): - def _redirect_operation_decorator(fn: Callable) -> _UnboundAddonOperation: + def _redirect_operation_decorator(fn: Callable) -> Callable: # decorator for operations that may be performed by a client request # (e.g. redirect to waterbutler) assert inspect.isfunction(fn) # TODO: inspect function params assert not inspect.isawaitable(fn) # TODO: helpful error messaging for implementers - return _UnboundAddonOperation(fn, AddonOperationType.REDIRECT, capability) + AddonOperationDeclaration.declared(fn, capability, AddonOperationType.REDIRECT) + return fn return _redirect_operation_decorator def proxy_operation(capability: enum.Enum): - def _proxy_operation_decorator(fn: Callable) -> _UnboundAddonOperation: + def _proxy_operation_decorator(fn: Callable) -> Callable: # decorator for operations that require fetching data from elsewhere, # but make no changes (e.g. get a metadata description of an item, # list items in a given folder) # TODO: assert inspect.isasyncgenfunction(fn) # generate rdf triples? # TODO: assert based on `inspect.signature(fn).parameters` # TODO: assert based on return value? - return _UnboundAddonOperation(fn, AddonOperationType.PROXY, capability) + AddonOperationDeclaration.declared(fn, capability, AddonOperationType.PROXY) + return fn return _proxy_operation_decorator diff --git a/addon_toolkit/tests/test_addon_category.py b/addon_toolkit/tests/test_addon_category.py deleted file mode 100644 index 3aa51148..00000000 --- a/addon_toolkit/tests/test_addon_category.py +++ /dev/null @@ -1,132 +0,0 @@ -import enum -import unittest - -from addon_toolkit import ( - AddonCategory, - AddonInterface, - AddonOperation, - AddonOperationType, - proxy_operation, - redirect_operation, -) - - -class TestAddonCategory(unittest.TestCase): - def setUp(self): - class _MyChecksumArchiveCapability(enum.Enum): - GET_IT = "get-it" - PUT_IT = "put-it" - UNUSED = "unused" # for testing when a capability has no operations - - class _MyChecksumArchiveInterface(AddonInterface): - """this is a docstring for _MyChecksumArchiveInterface - - it should find its way to browsable docs somewhere - """ - - @redirect_operation(capability=_MyChecksumArchiveCapability.GET_IT) - def url_for_get(self, checksum_iri) -> str: - """this url_for_get docstring should find its way to docs""" - raise NotImplementedError - - @proxy_operation(capability=_MyChecksumArchiveCapability.GET_IT) - async def query_relations(self, checksum_iri, query=None): - """this query_relations docstring should find its way to docs""" - raise NotImplementedError - - @redirect_operation(capability=_MyChecksumArchiveCapability.PUT_IT) - def url_for_put(self, checksum_iri): - """this url_for_put docstring should find its way to docs""" - raise NotImplementedError - - self.my_addon_category = AddonCategory( - capabilities=_MyChecksumArchiveCapability, - base_interface=_MyChecksumArchiveInterface, - ) - - class _MyChecksumArchiveImplementation(_MyChecksumArchiveInterface): - def url_for_get(self, checksum_iri) -> str: - return f"https://myarchive.example///{checksum_iri}" - - def url_for_put(self, checksum_iri): - # TODO: how to represent "send a PUT request here"? - # return RedirectLadle( - # HTTPMethod.PUT, - # f'https://myarchive.example///{checksum_iri}', - # )? - return f"https://myarchive.example///{checksum_iri}" - - self._MyChecksumArchiveImplementation = _MyChecksumArchiveImplementation - - def test_operation_list(self): - _my_cap = self.my_addon_category.capabilities - _my_interface_cls = self.my_addon_category.base_interface - _expected_get_op = AddonOperation( - operation_type=AddonOperationType.REDIRECT, - capability=_my_cap.GET_IT, - declaration_cls=_my_interface_cls, - method_name="url_for_get", - ) - _expected_put_op = AddonOperation( - operation_type=AddonOperationType.REDIRECT, - capability=_my_cap.PUT_IT, - declaration_cls=_my_interface_cls, - method_name="url_for_put", - ) - _query_operation = AddonOperation( - operation_type=AddonOperationType.PROXY, - capability=_my_cap.GET_IT, - declaration_cls=_my_interface_cls, - method_name="query_relations", - ) - self.assertEqual( - set(self.my_addon_category.operations_declared()), - {_expected_get_op, _expected_put_op, _query_operation}, - ) - self.assertEqual( - set(self.my_addon_category.operations_declared(capability=_my_cap.GET_IT)), - {_expected_get_op, _query_operation}, - ) - self.assertEqual( - set(self.my_addon_category.operations_declared(capability=_my_cap.PUT_IT)), - {_expected_put_op}, - ) - self.assertEqual( - set(self.my_addon_category.operations_declared(capability=_my_cap.UNUSED)), - set(), - ) - self.assertEqual( - set( - self.my_addon_category.operations_implemented( - self._MyChecksumArchiveImplementation, - ) - ), - {_expected_get_op, _expected_put_op}, - ) - self.assertEqual( - set( - self.my_addon_category.operations_implemented( - self._MyChecksumArchiveImplementation, - capability=_my_cap.GET_IT, - ) - ), - {_expected_get_op}, - ) - self.assertEqual( - set( - self.my_addon_category.operations_implemented( - self._MyChecksumArchiveImplementation, - capability=_my_cap.PUT_IT, - ) - ), - {_expected_put_op}, - ) - self.assertEqual( - set( - self.my_addon_category.operations_implemented( - self._MyChecksumArchiveImplementation, - capability=_my_cap.UNUSED, - ) - ), - set(), - ) diff --git a/addon_toolkit/tests/test_addon_interface_declaration.py b/addon_toolkit/tests/test_addon_interface_declaration.py new file mode 100644 index 00000000..cbe8c4aa --- /dev/null +++ b/addon_toolkit/tests/test_addon_interface_declaration.py @@ -0,0 +1,148 @@ +import enum +import unittest + +from addon_toolkit import ( + AddonInterfaceDeclaration, + AddonOperationDeclaration, + AddonOperationType, + addon_interface, + proxy_operation, + redirect_operation, +) + + +class TestAddonInterfaceDeclaration(unittest.TestCase): + @classmethod + def setUpClass(cls): + ### + # declare the capabilities and interface for a category of addons + + class _MyCapability(enum.Enum): + GET_IT = "get-it" + PUT_IT = "put-it" + UNUSED = "unused" # for testing when a capability has no operations + + @addon_interface(capabilities=_MyCapability) + class _MyInterface: + """this is a docstring for _MyInterface + + it should find its way to browsable docs somewhere + """ + + @redirect_operation(capability=_MyCapability.GET_IT) + def url_for_get(self, checksum_iri) -> str: + """this url_for_get docstring should find its way to docs""" + raise NotImplementedError + + @proxy_operation(capability=_MyCapability.GET_IT) + async def query_relations(self, checksum_iri, query=None): + """this query_relations docstring should find its way to docs""" + raise NotImplementedError + + @redirect_operation(capability=_MyCapability.PUT_IT) + def url_for_put(self, checksum_iri): + """this url_for_put docstring should find its way to docs""" + raise NotImplementedError + + ### + # implement (some of) the interface's declared operations + + class _MyImplementation(_MyInterface): + def url_for_get(self, checksum_iri) -> str: + return f"https://myarchive.example///{checksum_iri}" + + def url_for_put(self, checksum_iri): + # TODO: how to represent "send a PUT request here"? + # return RedirectLadle( + # HTTPMethod.PUT, + # f'https://myarchive.example///{checksum_iri}', + # )? + return f"https://myarchive.example///{checksum_iri}" + + cls._MyCapability = _MyCapability + cls._MyInterface = _MyInterface + cls._MyImplementation = _MyImplementation + cls._declaration = AddonInterfaceDeclaration.for_class(_MyInterface) + + cls._expected_get_op = AddonOperationDeclaration( + operation_type=AddonOperationType.REDIRECT, + capability=_MyCapability.GET_IT, + operation_fn=_MyInterface.url_for_get, + ) + cls._expected_put_op = AddonOperationDeclaration( + operation_type=AddonOperationType.REDIRECT, + capability=_MyCapability.PUT_IT, + operation_fn=_MyInterface.url_for_put, + ) + cls._expected_query_op = AddonOperationDeclaration( + operation_type=AddonOperationType.PROXY, + capability=_MyCapability.GET_IT, + operation_fn=_MyInterface.query_relations, + ) + + def test_get_declared_operations(self): + self.assertEqual( + set(self._declaration.get_declared_operations()), + {self._expected_get_op, self._expected_put_op, self._expected_query_op}, + ) + self.assertEqual( + set( + self._declaration.get_declared_operations( + capability=self._MyCapability.GET_IT + ) + ), + {self._expected_get_op, self._expected_query_op}, + ) + self.assertEqual( + set( + self._declaration.get_declared_operations( + capability=self._MyCapability.PUT_IT + ) + ), + {self._expected_put_op}, + ) + self.assertEqual( + set( + self._declaration.get_declared_operations( + capability=self._MyCapability.UNUSED + ) + ), + set(), + ) + + def test_get_implemented_operations(self): + self.assertEqual( + set( + self._declaration.get_implemented_operations( + self._MyImplementation, + ) + ), + {self._expected_get_op, self._expected_put_op}, + ) + self.assertEqual( + set( + self._declaration.get_implemented_operations( + self._MyImplementation, + capability=self._MyCapability.GET_IT, + ) + ), + {self._expected_get_op}, + ) + self.assertEqual( + set( + self._declaration.get_implemented_operations( + self._MyImplementation, + capability=self._MyCapability.PUT_IT, + ) + ), + {self._expected_put_op}, + ) + self.assertEqual( + set( + self._declaration.get_implemented_operations( + self._MyImplementation, + capability=self._MyCapability.UNUSED, + ) + ), + set(), + )