Feature flags / config with flag for interface auto descriptions

cosmo.example.yml

@@ -1,5 +1,7 @@
 fqdnSuffix: infra.example.com
 asn: 65542
+features:
+  interface-auto-descriptions: YES
 devices:
   router:
     - "router1"

cosmo/__main__.py

@@ -7,6 +7,7 @@
 import argparse
 
 from cosmo.clients.netbox import NetboxClient
+from cosmo.features import features
 from cosmo.log import (
     info,
     logger,
@@ -15,7 +16,7 @@
     HumanReadableLoggingStrategy,
 )
 from cosmo.serializer import RouterSerializer, SwitchSerializer
-from cosmo.common import DeviceSerializationError
+from cosmo.common import DeviceSerializationError, APP_NAME
 
 
 def main() -> int:
@@ -41,6 +42,24 @@ def main() -> int:
     parser.add_argument(
         "--json", "-j", action="store_true", help="Toggle machine readable output on"
     )
+    parser.add_argument(
+        "--disable-feature",
+        default=[],
+        metavar="DISABLED_FEATURE",
+        action=features.toggleFeatureActionFactory(False),
+        choices=features.getAllFeatureNames(),
+        dest="nofeatures",
+        help="selectively disable cosmo feature. can be repeated.",
+    )
+    parser.add_argument(
+        "--enable-feature",
+        default=[],
+        metavar="ENABLED_FEATURE",
+        action=features.toggleFeatureActionFactory(True),
+        choices=features.getAllFeatureNames(),
+        dest="yesfeatures",
+        help="selectively enable cosmo feature. can be repeated.",
+    )
 
     args = parser.parse_args()
 
@@ -70,11 +89,13 @@ def main() -> int:
     cosmo_configuration = {}
     with open(args.config, "r") as cfg_file:
         cosmo_configuration = yaml.safe_load(cfg_file)
+        features.setFeaturesFromConfig(cosmo_configuration)
 
     if not "asn" in cosmo_configuration:
         error(f"Field 'asn' not defined in configuration file", None)
         return 1
 
+    info(f"Feature toggles for {APP_NAME}: {features}")
     info(f"Fetching information from Netbox, make sure VPN is enabled on your system.")
 
     netbox_url = os.environ.get("NETBOX_URL")

cosmo/clients/netbox.py

@@ -32,7 +32,7 @@ def __init__(self, url, token):
             raise Exception("Unknown Version")
 
         for f, e in feature_flags.items():
-            log.info(f"Feature {f}: {e}")
+            log.info(f"Netbox feature {f}: {e}")
 
     def query_version(self):
         r = requests.get(

cosmo/features.py

@@ -0,0 +1,86 @@
+# implementation guide
+# https://martinfowler.com/articles/feature-toggles.html
+from argparse import Action, ArgumentParser
+from typing import Never, Self, Optional, TextIO, Sequence, Any
+
+import yaml
+
+
+class NonExistingFeatureToggleException(Exception):
+    pass
+
+
+class FeatureToggle:
+    CFG_KEY = "features"
+
+    def __init__(self, features_default_config: dict[str, bool]):
+        self._store: dict[str, bool] = features_default_config
+        self._authorized_keys = list(features_default_config.keys())
+
+    def checkFeatureExistsOrRaise(self, key: str) -> bool | Never:
+        if key in self._authorized_keys:
+            return True
+        raise NonExistingFeatureToggleException(
+            f"feature toggle {key} is unknown, please check your code"
+        )
+
+    def getAllFeatureNames(self) -> list[str]:
+        return self._authorized_keys
+
+    def featureIsEnabled(self, key: str) -> bool:
+        self.checkFeatureExistsOrRaise(key)
+        return bool(self._store.get(key))
+
+    def setFeature(self, key: str, toggle: bool) -> Self:
+        self.checkFeatureExistsOrRaise(key)
+        self._store[key] = toggle
+        return self  # chain
+
+    def setFeatures(self, config: dict[str, bool]) -> Self:
+        for feature_key, feature_toggle in config.items():
+            self.setFeature(feature_key, feature_toggle)
+        return self  # chain
+
+    def setFeaturesFromConfig(self, config: dict) -> Self:
+        config_dict = dict(config.get(self.CFG_KEY, dict()))
+        self.setFeatures(config_dict)
+        return self
+
+    def setFeaturesFromYAML(self, yaml_stream_or_str: TextIO | str) -> Self:
+        config = yaml.safe_load(yaml_stream_or_str)
+        self.setFeaturesFromConfig(config)
+        return self
+
+    def setFeaturesFromYAMLFile(self, path: str) -> Self:
+        with open(path, "r") as cfg_file:
+            self.setFeaturesFromYAML(cfg_file)
+        return self
+
+    def toggleFeatureActionFactory(self, toggle_to: bool) -> type[Action]:
+        feature_toggle_instance = self
+
+        class ToggleFeatureAction(Action):
+            def __call__(
+                self,
+                parser: ArgumentParser,
+                namespace: object,
+                values: str | Sequence[Any] | None,
+                option_string: Optional[str] = None,
+            ):
+                if type(values) is list:
+                    [feature_toggle_instance.setFeature(v, toggle_to) for v in values]
+                elif type(values) is str:
+                    feature_toggle_instance.setFeature(values, toggle_to)
+                setattr(namespace, self.dest, values)
+
+        return ToggleFeatureAction
+
+    def __str__(self):
+        features_desc = []
+        conv = {True: "ENABLED", False: "DISABLED"}
+        for feature, state in self._store.items():
+            features_desc.append(f"{feature}: {conv.get(state)}")
+        return ", ".join(features_desc)
+
+
+features = FeatureToggle({"interface-auto-descriptions": True})

cosmo/serializer.py

@@ -11,6 +11,7 @@
     head,
     CosmoOutputType,
 )
+from cosmo.features import features
 from cosmo.log import error
 from cosmo.netbox_types import DeviceType, CosmoLoopbackType, AbstractNetboxType
 from cosmo.loopbacks import LoopbackHelper
@@ -40,6 +41,8 @@ def getMerger():
 
     @staticmethod
     def autoDescPreprocess(_: CosmoOutputType, value: AbstractNetboxType):
+        if not features.featureIsEnabled("interface-auto-descriptions"):
+            return  # early return / skip
         MutatingAutoDescVisitor().accept(value)
 
     def walk(

cosmo/tests/cosmo-test-features-toggles.yaml

@@ -0,0 +1,3 @@
+features:
+  feature_a: YES
+  feature_b: NO

cosmo/tests/test_features.py

@@ -0,0 +1,96 @@
+import argparse
+import os.path
+
+import pytest
+
+from cosmo.features import NonExistingFeatureToggleException, FeatureToggle
+
+
+def test_set_get():
+    ft = FeatureToggle({"feature_a": False, "feature_b": False})
+
+    ft.setFeature("feature_a", True)
+    assert ft.featureIsEnabled("feature_a")
+
+    ft.setFeature("feature_a", False)
+    assert not ft.featureIsEnabled("feature_a")
+
+    ft.setFeatures({"feature_a": True, "feature_b": True})
+    assert ft.featureIsEnabled("feature_a")
+    assert ft.featureIsEnabled("feature_b")
+
+
+def test_config_from_str():
+    ft = FeatureToggle({"feature_a": False, "feature_b": False})
+    yaml_config = """
+    features:
+      feature_a: NO
+      feature_b: YES
+    """
+
+    ft.setFeaturesFromYAML(yaml_config)
+    assert not ft.featureIsEnabled("feature_a")
+    assert ft.featureIsEnabled("feature_b")
+
+    ft.setFeaturesFromYAMLFile(
+        os.path.join(os.path.dirname(__file__), "cosmo-test-features-toggles.yaml")
+    )
+    assert ft.featureIsEnabled("feature_a")
+    assert not ft.featureIsEnabled("feature_b")
+
+
+def test_get_feature_names():
+    features_dict = {
+        "feature_a": True,
+        "feature_b": False,
+        "feature_c": False,
+        "feature_d": True,
+    }
+    ft = FeatureToggle(features_dict)
+    assert list(features_dict.keys()) == ft.getAllFeatureNames()
+
+
+def test_non_existing_features():
+    ft = FeatureToggle({"feature_a": True})
+
+    with pytest.raises(NonExistingFeatureToggleException):
+        ft.setFeature("i-do-not-exist", True)
+
+
+def test_argparse_integration():
+    ft = FeatureToggle({"feature_a": False, "feature_b": False, "feature_c": True})
+
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--enable-feature",
+        default=[],
+        metavar="ENABLED_FEATURE",
+        action=ft.toggleFeatureActionFactory(True),
+        choices=ft.getAllFeatureNames(),
+        dest="yesfeatures",
+        help="selectively enable features",
+    )
+    parser.add_argument(
+        "--disable-feature",
+        default=[],
+        metavar="DISABLED_FEATURE",
+        action=ft.toggleFeatureActionFactory(False),
+        choices=ft.getAllFeatureNames(),
+        dest="nofeatures",
+        help="selectively disable features",
+    )
+
+    parser.parse_args(
+        [
+            "--enable-feature",
+            "feature_a",
+            "--enable-feature",
+            "feature_b",
+            "--disable-feature",
+            "feature_c",
+        ]
+    )
+
+    assert ft.featureIsEnabled("feature_a")
+    assert ft.featureIsEnabled("feature_b")
+    assert not ft.featureIsEnabled("feature_c")