diff --git a/cosmo.example.yml b/cosmo.example.yml index e2b695d..15e85f8 100644 --- a/cosmo.example.yml +++ b/cosmo.example.yml @@ -1,5 +1,7 @@ fqdnSuffix: infra.example.com asn: 65542 +features: + interface-auto-descriptions: YES devices: router: - "router1" diff --git a/cosmo/__main__.py b/cosmo/__main__.py index e09121a..a98fd27 100644 --- a/cosmo/__main__.py +++ b/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") diff --git a/cosmo/clients/netbox.py b/cosmo/clients/netbox.py index efebda0..62e5819 100644 --- a/cosmo/clients/netbox.py +++ b/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( diff --git a/cosmo/features.py b/cosmo/features.py new file mode 100644 index 0000000..cb5fc18 --- /dev/null +++ b/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}) diff --git a/cosmo/serializer.py b/cosmo/serializer.py index d85a531..c4b644e 100644 --- a/cosmo/serializer.py +++ b/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( diff --git a/cosmo/tests/cosmo-test-features-toggles.yaml b/cosmo/tests/cosmo-test-features-toggles.yaml new file mode 100644 index 0000000..6afae3c --- /dev/null +++ b/cosmo/tests/cosmo-test-features-toggles.yaml @@ -0,0 +1,3 @@ +features: + feature_a: YES + feature_b: NO diff --git a/cosmo/tests/test_features.py b/cosmo/tests/test_features.py new file mode 100644 index 0000000..f47722c --- /dev/null +++ b/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")