File: async_wrapper.py

package info (click to toggle)
python-azure 20201208%2Bgit-6
  • links: PTS, VCS
  • area: main
  • in suites: bullseye
  • size: 1,437,920 kB
  • sloc: python: 4,287,452; javascript: 269; makefile: 198; sh: 187; xml: 106
file content (119 lines) | stat: -rw-r--r-- 4,054 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119

# -------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for
# license information.
# --------------------------------------------------------------------------
from devtools_testutils import PowerShellPreparer

from azure.appconfiguration import (
    ResourceReadOnlyError,
    ConfigurationSetting,
)
from azure.appconfiguration.aio import AzureAppConfigurationClient
from azure.core.exceptions import (
    AzureError,
)
from consts import (
    KEY,
    LABEL,
    TEST_VALUE,
    TEST_CONTENT_TYPE,
    LABEL_RESERVED_CHARS,
    PAGE_SIZE,
    KEY_UUID,
)
from async_proxy import AzureAppConfigurationClientProxy
import functools
import inspect

AppConfigPreparer = functools.partial(
    PowerShellPreparer,
    'appconfiguration',
    appconfiguration_connection_string="Endpoint=https://fake_app_config.azconfig-test.io;Id=0-l4-s0:h5htBaY5Z1LwFz50bIQv;Secret=lamefakesecretlamefakesecretlamefakesecrett=",
    appconfiguration_endpoint_string="https://fake_app_config.azconfig-test.io")

def _add_for_test(client, kv):
    exist = bool(
        list(
            client.list_configuration_settings(
                key_filter=kv.key, label_filter=kv.label
            )
        )
    )
    if exist:
        _delete_from_test(client, kv.key, kv.label)
    return client.add_configuration_setting(kv)

def _delete_from_test(client, key, label):
    try:
        client.delete_configuration_setting(key=key, label=label)
    except AzureError:
        logging.debug(
            "Error occurred removing configuration setting %s %s during unit test"
            % (key, label)
        )

def trim_kwargs_from_test_function(fn, kwargs):
    # the next function is the actual test function. the kwargs need to be trimmed so
    # that parameters which are not required will not be passed to it.
    if not getattr(fn, '__is_preparer', False):
        try:
            args, _, kw, _, _, _, _ = inspect.getfullargspec(fn)
        except AttributeError:
            args, _, kw, _ = inspect.getargspec(fn) # pylint: disable=deprecated-method
        if kw is None:
            args = set(args)
            for key in [k for k in kwargs if k not in args]:
                del kwargs[key]


def app_config_decorator(func, **kwargs):

    @AppConfigPreparer()
    def wrapper(*args, **kwargs):
        appconfiguration_connection_string = kwargs.pop("appconfiguration_connection_string")
        client = AzureAppConfigurationClient.from_connection_string(appconfiguration_connection_string)
        client = AzureAppConfigurationClientProxy(client)

        kwargs['client'] = client
        kwargs['appconfiguration_connection_string'] = appconfiguration_connection_string

        # Do setUp on client
        test_config_setting = _add_for_test(
            client,
            ConfigurationSetting(
                key=KEY,
                label=LABEL,
                value=TEST_VALUE,
                content_type=TEST_CONTENT_TYPE,
                tags={"tag1": "tag1", "tag2": "tag2"},
            )
        )
        test_config_setting_no_label = _add_for_test(
            client,
            ConfigurationSetting(
                key=KEY,
                label=None,
                value=TEST_VALUE,
                content_type=TEST_CONTENT_TYPE,
                tags={"tag1": "tag1", "tag2": "tag2"},
            )
        )
        to_delete = [test_config_setting, test_config_setting_no_label]

        kwargs['test_config_setting'] = test_config_setting
        kwargs['test_config_setting_no_label'] = test_config_setting_no_label

        trimmed_kwargs = {k:v for k, v in kwargs.items()}
        trim_kwargs_from_test_function(func, trimmed_kwargs)

        func(*args, **trimmed_kwargs)

        # do tearDown on client
        for item in to_delete:
            client.delete_configuration_setting(
                key=item.key, label=item.label
            )

    return wrapper