service: Add feature toggle support (#421)

* service: Add dependency on python-decouple

* Add python-decouple to poetry.lock

* service: Add feature toggle implementation

* tests: Add feature toggle tests

* service: Change feature decorator into a free function

* tests: Add use_code_readiness

* service: Remove runtime typing_extensions import

* service: Rename a constant and remove an extra import

* service: Rename READINESS_LEVEL and hide it behind an accessor function

* service: Make CodeReadiness an ordered enum

* tests: Fix incorrect feature toggle name

* service: Remove MEASUREMENTLINK_CODE_READINESS_LEVEL override

* tests: Change default code readiness level to PROTOTYPE

* tests: Add one more test

ni_measurementlink_service/_featuretoggles.py

@@ -0,0 +1,143 @@
+from __future__ import annotations
+
+import functools
+import sys
+from enum import Enum
+from typing import TYPE_CHECKING, Callable, TypeVar
+
+from decouple import config
+
+if TYPE_CHECKING:
+    if sys.version_info >= (3, 10):
+        from typing import ParamSpec
+    else:
+        from typing_extensions import ParamSpec
+
+    if sys.version_info >= (3, 11):
+        from typing import Self
+    else:
+        from typing_extensions import Self
+
+    _P = ParamSpec("_P")
+    _T = TypeVar("_T")
+
+_PREFIX = "MEASUREMENTLINK"
+
+
+# Based on the recipe at https://docs.python.org/3/howto/enum.html
+class _OrderedEnum(Enum):
+    def __ge__(self, other: Self) -> bool:
+        if self.__class__ is other.__class__:
+            return self.value >= other.value
+        return NotImplemented
+
+    def __gt__(self, other: Self) -> bool:
+        if self.__class__ is other.__class__:
+            return self.value > other.value
+        return NotImplemented
+
+    def __le__(self, other: Self) -> bool:
+        if self.__class__ is other.__class__:
+            return self.value <= other.value
+        return NotImplemented
+
+    def __lt__(self, other: Self) -> bool:
+        if self.__class__ is other.__class__:
+            return self.value < other.value
+        return NotImplemented
+
+
+class CodeReadiness(_OrderedEnum):
+    """Indicates whether code is ready to be supported."""
+
+    RELEASE = 0
+    NEXT_RELEASE = 1
+    INCOMPLETE = 2
+    PROTOTYPE = 3
+
+
+def _init_code_readiness_level() -> CodeReadiness:
+    if config(f"{_PREFIX}_ALLOW_INCOMPLETE", default=False, cast=bool):
+        return CodeReadiness.INCOMPLETE
+    elif config(f"{_PREFIX}_ALLOW_NEXT_RELEASE", default=False, cast=bool):
+        return CodeReadiness.NEXT_RELEASE
+    else:
+        return CodeReadiness.RELEASE
+
+
+# This is not public because `from _featuretoggles import CODE_READINESS_LEVEL`
+# is incompatible with the patching performed by the use_code_readiness mark.
+_CODE_READINESS_LEVEL = _init_code_readiness_level()
+
+
+def get_code_readiness_level() -> CodeReadiness:
+    """Get the current code readiness level.
+
+    You can override this in tests by specifying the ``use_code_readiness``
+    mark.
+    """
+    return _CODE_READINESS_LEVEL
+
+
+class FeatureNotSupportedError(Exception):
+    """The feature is not supported at the current code readiness level."""
+
+
+class FeatureToggle:
+    """A run-time feature toggle."""
+
+    name: str
+    """The name of the feature."""
+
+    readiness: CodeReadiness
+    """The code readiness at which this feature is enabled."""
+
+    def __init__(self, name: str, readiness: CodeReadiness) -> None:
+        """Initialize the feature toggle."""
+        assert name == name.upper()
+        self.name = name
+        self.readiness = readiness
+        self._is_enabled_override = None
+        # Only read the env var at initialization time.
+        if config(f"{_PREFIX}_ENABLE_{name}", default=False, cast=bool):
+            self._is_enabled_override = True
+
+    @property
+    def is_enabled(self) -> bool:
+        """Indicates whether the feature is currently enabled.
+
+        You can enable/disable features in tests by specifying the
+        ``enable_feature_toggle`` or ``disable_feature_toggle`` marks.
+        """
+        if self._is_enabled_override is not None:
+            return self._is_enabled_override
+        return self.readiness <= get_code_readiness_level()
+
+    def _raise_if_disabled(self) -> None:
+        if self.is_enabled:
+            return
+
+        env_vars = f"{_PREFIX}_ENABLE_{self.name}"
+        if self.readiness in [CodeReadiness.NEXT_RELEASE, CodeReadiness.INCOMPLETE]:
+            env_vars += f" or {_PREFIX}_ALLOW_{self.readiness.name}"
+        message = (
+            f"The {self.name} feature is not supported at the current code readiness level. "
+            f" To enable it, set {env_vars}."
+        )
+        raise FeatureNotSupportedError(message)
+
+
+def requires_feature(
+    feature_toggle: FeatureToggle,
+) -> Callable[[Callable[_P, _T]], Callable[_P, _T]]:
+    """Decorator specifying that the function requires the specified feature toggle."""
+
+    def decorator(func: Callable[_P, _T]) -> Callable[_P, _T]:
+        @functools.wraps(func)
+        def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> _T:
+            feature_toggle._raise_if_disabled()
+            return func(*args, **kwargs)
+
+        return wrapper
+
+    return decorator

pyproject.toml

@@ -36,6 +36,7 @@ pywin32 = {version = ">=303", platform = "win32"}
 deprecation = ">=2.1"
 # https://github.com/microsoft/tracelogging/issues/58 - traceloggingdynamic raises TypeError with Python 3.8
 traceloggingdynamic = {version = ">=1.0", platform = "win32", python = "^3.9"}
+python-decouple = ">=3.8"
 
 [tool.poetry.group.dev.dependencies]
 pytest = ">=7.2.0"
@@ -73,7 +74,10 @@ addopts = "--doctest-modules --strict-markers"
 filterwarnings = ["always::ImportWarning", "always::ResourceWarning"]
 testpaths = ["tests"]
 markers = [
+  "disable_feature_toggle: specifies a feature toggle to disable for the test function/module.",
+  "enable_feature_toggle: specifies a feature toggle to enable for the test function/module.",
   "service_class: specifies which test service to use.",
+  "use_code_readiness: specifies a code readiness level to use for the test function/module.",
 ]
 
 [tool.mypy]
@@ -85,6 +89,8 @@ warn_unused_ignores = true
 
 [[tool.mypy.overrides]]
 module = [
+  # https://github.com/HBNetwork/python-decouple/issues/122 - Add support for type stubs
+  "decouple.*",
   # https://github.com/briancurtin/deprecation/issues/56 - Add type information (PEP 561)
   "deprecation.*",
   "grpc.framework.foundation.*",

tests/conftest.py

@@ -1,11 +1,13 @@
 """Pytest configuration file."""
 import pathlib
 import sys
-from typing import Generator
+from typing import Generator, List
 
 import grpc
 import pytest
 
+from ni_measurementlink_service import _featuretoggles
+from ni_measurementlink_service._featuretoggles import CodeReadiness, FeatureToggle
 from ni_measurementlink_service._internal.discovery_client import (
     DiscoveryClient,
     _get_registration_json_file_path,
@@ -95,3 +97,34 @@ def session_management_client(
     return SessionManagementClient(
         discovery_client=discovery_client, grpc_channel_pool=grpc_channel_pool
     )
+
+
+@pytest.fixture
+def feature_toggles(monkeypatch: pytest.MonkeyPatch, request: pytest.FixtureRequest) -> None:
+    """Test fixture that disables or enables feature toggles."""
+    for mark in request.node.iter_markers():
+        if mark.name in ["disable_feature_toggle", "enable_feature_toggle"]:
+            feature_toggle = mark.args[0]
+            assert isinstance(feature_toggle, FeatureToggle)
+            monkeypatch.setattr(
+                feature_toggle, "_is_enabled_override", mark.name == "enable_feature_toggle"
+            )
+        elif mark.name == "use_code_readiness":
+            code_readiness = mark.args[0]
+            assert isinstance(code_readiness, CodeReadiness)
+            monkeypatch.setattr(_featuretoggles, "_CODE_READINESS_LEVEL", code_readiness)
+
+
+def pytest_collection_modifyitems(items: List[pytest.Item]) -> None:
+    """Hook to inject fixtures based on marks."""
+    # By default, all features are enabled when running tests.
+    _featuretoggles._CODE_READINESS_LEVEL = CodeReadiness.PROTOTYPE
+
+    for item in items:
+        if (
+            item.get_closest_marker("disable_feature_toggle")
+            or item.get_closest_marker("enable_feature_toggle")
+            or item.get_closest_marker("use_code_readiness")
+        ):
+            assert hasattr(item, "fixturenames")
+            item.fixturenames.append("feature_toggles")

tests/unit/__init__.py

@@ -0,0 +1 @@
+"""Unit tests."""

tests/unit/test_featuretoggles.py

@@ -0,0 +1,108 @@
+from typing import List
+
+import pytest
+from pytest_mock import MockerFixture
+
+from ni_measurementlink_service._featuretoggles import (
+    CodeReadiness,
+    FeatureNotSupportedError,
+    FeatureToggle,
+    get_code_readiness_level,
+    requires_feature,
+)
+
+RELEASE_FEATURE = FeatureToggle("RELEASE_FEATURE", CodeReadiness.RELEASE)
+NEXT_RELEASE_FEATURE = FeatureToggle("NEXT_RELEASE_FEATURE", CodeReadiness.NEXT_RELEASE)
+INCOMPLETE_FEATURE = FeatureToggle("INCOMPLETE_FEATURE", CodeReadiness.INCOMPLETE)
+PROTOTYPE_FEATURE = FeatureToggle("PROTOTYPE_FEATURE", CodeReadiness.PROTOTYPE)
+
+
+@requires_feature(PROTOTYPE_FEATURE)
+def _prototype_function(x: int, y: str, z: List[int]) -> str:
+    return _prototype_function_impl(x, y, z)
+
+
+def _prototype_function_impl(x: int, y: str, z: List[int]) -> str:
+    return ""
+
+
+def test___default_code_readiness_level___get_code_readiness_level___equals_prototype() -> None:
+    assert get_code_readiness_level() == CodeReadiness.PROTOTYPE
+
+
+@pytest.mark.use_code_readiness(CodeReadiness.RELEASE)
+def test___use_release_readiness___get_code_readiness_level___equals_release() -> None:
+    assert get_code_readiness_level() == CodeReadiness.RELEASE
+
+
+@pytest.mark.use_code_readiness(CodeReadiness.NEXT_RELEASE)
+def test___use_next_release_readiness___get_code_readiness_level___equals_next_release() -> None:
+    assert get_code_readiness_level() == CodeReadiness.NEXT_RELEASE
+
+
+def test___default_code_readiness_level___is_enabled___returns_true() -> None:
+    assert RELEASE_FEATURE.is_enabled
+    assert NEXT_RELEASE_FEATURE.is_enabled
+    assert INCOMPLETE_FEATURE.is_enabled
+    assert PROTOTYPE_FEATURE.is_enabled
+
+
+@pytest.mark.use_code_readiness(CodeReadiness.INCOMPLETE)
+def test___use_incomplete_readiness___is_enabled___reflects_code_readiness_level() -> None:
+    assert RELEASE_FEATURE.is_enabled
+    assert NEXT_RELEASE_FEATURE.is_enabled
+    assert INCOMPLETE_FEATURE.is_enabled
+    assert not PROTOTYPE_FEATURE.is_enabled
+
+
+@pytest.mark.use_code_readiness(CodeReadiness.NEXT_RELEASE)
+def test___use_next_release_readiness___is_enabled___reflects_code_readiness_level() -> None:
+    assert RELEASE_FEATURE.is_enabled
+    assert NEXT_RELEASE_FEATURE.is_enabled
+    assert not INCOMPLETE_FEATURE.is_enabled
+    assert not PROTOTYPE_FEATURE.is_enabled
+
+
+@pytest.mark.use_code_readiness(CodeReadiness.RELEASE)
+def test___release_readiness_level___is_enabled___reflects_code_readiness_level() -> None:
+    assert RELEASE_FEATURE.is_enabled
+    assert not NEXT_RELEASE_FEATURE.is_enabled
+    assert not INCOMPLETE_FEATURE.is_enabled
+    assert not PROTOTYPE_FEATURE.is_enabled
+
+
+@pytest.mark.enable_feature_toggle(PROTOTYPE_FEATURE)
+def test___feature_toggle_enabled___is_enabled___returns_true() -> None:
+    assert PROTOTYPE_FEATURE.is_enabled
+
+
+@pytest.mark.disable_feature_toggle(PROTOTYPE_FEATURE)
+def test___feature_toggle_disabled___is_enabled___returns_false() -> None:
+    assert not PROTOTYPE_FEATURE.is_enabled
+
+
+@pytest.mark.enable_feature_toggle(PROTOTYPE_FEATURE)
+def test___feature_toggle_enabled___call_decorated_function___impl_called(
+    mocker: MockerFixture,
+) -> None:
+    impl = mocker.patch("tests.unit.test_featuretoggles._prototype_function_impl")
+    impl.return_value = "def"
+
+    result = _prototype_function(123, "abc", [4, 5, 6])
+
+    impl.assert_called_once_with(123, "abc", [4, 5, 6])
+    assert result == "def"
+
+
+@pytest.mark.disable_feature_toggle(PROTOTYPE_FEATURE)
+def test___feature_toggle_disabled___call_decorated_function___error_raised(
+    mocker: MockerFixture,
+) -> None:
+    impl = mocker.patch("tests.unit.test_featuretoggles._prototype_function_impl")
+    impl.return_value = "def"
+
+    with pytest.raises(FeatureNotSupportedError) as exc_info:
+        _ = _prototype_function(123, "abc", [4, 5, 6])
+
+    impl.assert_not_called()
+    assert "set MEASUREMENTLINK_ENABLE_PROTOTYPE_FEATURE" in exc_info.value.args[0]