Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
75 changes: 74 additions & 1 deletion chaoslib/settings.py
Original file line number Diff line number Diff line change
@@ -1,14 +1,17 @@
# -*- 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
import yaml

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={})
Expand Down Expand Up @@ -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)
97 changes: 96 additions & 1 deletion tests/test_settings.py
Original file line number Diff line number Diff line change
@@ -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")
Expand Down Expand Up @@ -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