From eac5f64b8432b550ad0bfdf8608f2cc5447f98da Mon Sep 17 00:00:00 2001
From: Sylvain Hellegouarch <sh@defuze.org>
Date: Fri, 19 Jun 2020 15:16:57 +0200
Subject: [PATCH] Add locate_settings_entry function

Contributes to https://github.com/chaostoolkit/chaostoolkit/issues/65

Signed-off-by: Sylvain Hellegouarch <sh@defuze.org>
---
 CHANGELOG.md           |  7 +++
 chaoslib/settings.py   | 75 +++++++++++++++++++++++++++++++-
 tests/test_settings.py | 97 +++++++++++++++++++++++++++++++++++++++++-
 3 files changed, 177 insertions(+), 2 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index a181ff2..d0f02eb 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,6 +4,13 @@
 
 [Unreleased]: https://github.com/chaostoolkit/chaostoolkit-lib/compare/1.9.0...HEAD
 
+### Added
+
+- Added function to lookup a value in the settings from a dotted path
+  key [chaostoolkit#65][65]
+
+[65]: https://github.com/chaostoolkit/chaostoolkit/issues/65
+
 ## [1.9.0][] - 2020-04-29
 
 [1.9.0]: https://github.com/chaostoolkit/chaostoolkit-lib/compare/1.8.1...1.9.0
diff --git a/chaoslib/settings.py b/chaoslib/settings.py
index acddcda..7970ee6 100644
--- a/chaoslib/settings.py
+++ b/chaoslib/settings.py
@@ -1,6 +1,8 @@
 # -*- coding: utf-8 -*-
 import os
 import os.path
+import re
+from typing import Any, Dict, List, Optional, Tuple, Union
 
 import contextvars
 from logzero import logger
@@ -8,7 +10,8 @@
 
 from chaoslib.types import Settings
 
-__all__ = ["get_loaded_settings", "load_settings", "save_settings"]
+__all__ = ["get_loaded_settings", "load_settings", "save_settings",
+           "locate_settings_entry"]
 CHAOSTOOLKIT_CONFIG_PATH = os.path.abspath(
     os.path.expanduser("~/.chaostoolkit/settings.yaml"))
 loaded_settings = contextvars.ContextVar('loaded_settings', default={})
@@ -53,3 +56,73 @@ def get_loaded_settings() -> Settings:
     Settings that have been loaded in the current context.
     """
     return loaded_settings.get()
+
+
+def locate_settings_entry(settings: Settings, key: str) \
+        -> Optional[
+            Tuple[
+                Union[Dict[str, Any], List],
+                Any,
+                Optional[str],
+                Optional[int]
+            ]
+           ]:
+    """
+    Lookup the entry at the given dotted key in the provided settings and
+    return a a tuple as follows:
+
+    * the parent of the found entry (can be a list or a dict)
+    * the entry (can eb anything: string, int, list, dict)
+    * the key on the parent that has the entry (in case parent is a dict)
+    * the index in the parent that has the entry (in case parent is a list)
+
+    Otherwise, returns `None`.
+
+    When the key in the settings has at least one dot, it must be escaped
+    with two backslahes.
+
+    Examples of valid keys:
+
+    * auths
+    * auths.example\\.com
+    * auths.example\\.com:8443
+    * auths.example\\.com.type
+    * controls[0].name
+    """
+    array_index = re.compile(r"\[([0-9]*)\]$")
+    # borrowed from https://github.com/carlosescri/DottedDict (MIT)
+    # this kindly preserves escaped dots
+    parts = [x for x in re.split(r"(?<!\\)(\.)", key) if x != "."]
+
+    current = settings
+    parent = settings
+    last_key = None
+    last_index = None
+    for part in parts:
+        # we don't know to escape now that we have our part
+        part = part.replace('\\', '')
+        m = array_index.search(part)
+        if m:
+            # this is part with an index
+            part = part[:m.start()]
+            if part not in current:
+                return
+            current = current.get(part)
+            parent = current
+            index = int(m.groups()[0])
+            last_key = None
+            last_index = index
+            try:
+                current = current[index]
+            except (KeyError, IndexError):
+                return
+        else:
+            # this is just a regular key
+            if part not in current:
+                return
+            parent = current
+            last_key = part
+            last_index = None
+            current = current.get(part)
+
+    return (parent, current, last_key, last_index)
diff --git a/tests/test_settings.py b/tests/test_settings.py
index ed5e6c0..a6fb275 100644
--- a/tests/test_settings.py
+++ b/tests/test_settings.py
@@ -1,6 +1,7 @@
 # -*- coding: utf-8 -*-
 import os.path
-from chaoslib.settings import get_loaded_settings, load_settings, save_settings
+from chaoslib.settings import get_loaded_settings, load_settings, \
+    locate_settings_entry, save_settings
 
 
 settings_dir = os.path.join(os.path.dirname(__file__), "fixtures")
@@ -51,3 +52,97 @@ def test_create_settings_file_on_save():
 def test_get_loaded_settings():
     settings = load_settings(os.path.join(settings_dir, "settings.yaml"))
     assert get_loaded_settings() is settings
+
+
+def test_locate_root_level_entry():
+    settings = {
+        "auths": {
+            "chaos.example.com": {
+                "type": "bearer"
+            }
+        }
+    }
+    parent, entry, k, i = locate_settings_entry(settings, 'auths')
+    assert parent == settings
+    assert entry == settings['auths']
+    assert k == 'auths'
+    assert i == None
+
+
+def test_locate_dotted_entry():
+    settings = {
+        "auths": {
+            "chaos.example.com": {
+                "type": "bearer"
+            }
+        }
+    }
+    parent, entry, k, i = locate_settings_entry(
+        settings, 'auths.chaos\\.example\\.com')
+    assert parent == settings['auths']
+    assert entry == {"type": "bearer"}
+    assert k == 'chaos.example.com'
+    assert i == None
+
+
+def test_locate_indexed_entry():
+    settings = {
+        "auths": {
+            "chaos.example.com": {
+                "type": "bearer",
+                "headers": [
+                    {
+                        "name": "X-Client",
+                        "value": "blah"
+                    },
+                    {
+                        "name": "X-For",
+                        "value": "other"
+                    }
+                ]
+            }
+        }
+    }
+    parent, entry, k, i = locate_settings_entry(
+        settings, 'auths.chaos\\.example\\.com.headers[1]')
+    assert parent == settings['auths']["chaos.example.com"]["headers"]
+    assert entry == {"name": "X-For", "value": "other"}
+    assert k == None
+    assert i == 1
+
+
+def test_locate_dotted_key_from_indexed_entry():
+    settings = {
+        "auths": {
+            "chaos.example.com": {
+                "type": "bearer",
+                "headers": [
+                    {
+                        "name": "X-Client",
+                        "value": "blah"
+                    },
+                    {
+                        "name": "X-For",
+                        "value": "other"
+                    }
+                ]
+            }
+        }
+    }
+    parent, entry, k, i = locate_settings_entry(
+        settings, 'auths.chaos\\.example\\.com.headers[1].name')
+    assert parent == settings['auths']["chaos.example.com"]["headers"][1]
+    assert entry == "X-For"
+    assert k == "name"
+    assert i == None
+
+
+def test_cannot_locate_dotted_entry():
+    settings = {
+        "auths": {
+            "chaos.example.com": {
+                "type": "bearer"
+            }
+        }
+    }
+    assert locate_settings_entry(settings, 'auths.chaos.example.com') == None