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
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
|
# coding: utf-8
# -------------------------------------------------------------------------
# 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 AzureRecordedTestCase
from azure.appconfiguration import (
AzureAppConfigurationClient,
ConfigurationSetting,
FeatureFlagConfigurationSetting,
SecretReferenceConfigurationSetting,
)
from azure.appconfiguration.provider import load, AzureAppConfigurationKeyVaultOptions
from azure.appconfiguration.provider._constants import NULL_CHAR
from test_constants import FEATURE_MANAGEMENT_KEY, FEATURE_FLAG_KEY
class AppConfigTestCase(AzureRecordedTestCase):
def create_client(self, **kwargs):
credential = self.get_credential(AzureAppConfigurationClient)
client = None
if "connection_string" in kwargs:
client = AzureAppConfigurationClient.from_connection_string(kwargs["connection_string"])
else:
client = AzureAppConfigurationClient(kwargs["endpoint"], credential)
setup_configs(client, kwargs.get("keyvault_secret_url"), kwargs.get("keyvault_secret_url2"))
kwargs["user_agent"] = "SDK/Integration"
if "endpoint" in kwargs:
kwargs["credential"] = credential
if "secret_resolver" not in kwargs and kwargs.get("keyvault_secret_url") and "key_vault_options" not in kwargs:
kwargs["keyvault_credential"] = credential
if "key_vault_options" in kwargs:
key_vault_options = kwargs.pop("key_vault_options")
if not key_vault_options.secret_resolver:
key_vault_options = AzureAppConfigurationKeyVaultOptions(credential=credential)
kwargs["key_vault_options"] = key_vault_options
return load(**kwargs)
@staticmethod
def create_sdk_client(appconfiguration_connection_string):
return AzureAppConfigurationClient.from_connection_string(
appconfiguration_connection_string, user_agent="SDK/Integration"
)
def create_aad_sdk_client(self, appconfiguration_endpoint_string):
cred = self.get_credential(AzureAppConfigurationClient)
return AzureAppConfigurationClient(appconfiguration_endpoint_string, cred, user_agent="SDK/Integration")
def setup_configs(client, keyvault_secret_url, keyvault_secret_url2):
for config in get_configs(keyvault_secret_url, keyvault_secret_url2):
client.set_configuration_setting(config)
def get_configs(keyvault_secret_url, keyvault_secret_url2):
configs = []
configs.append(create_config_setting("message", NULL_CHAR, "hi"))
configs.append(create_config_setting("message", "dev", "test"))
configs.append(create_config_setting("my_json", NULL_CHAR, '{"key": "value"}', "application/json"))
configs.append(create_config_setting("test.trimmed", NULL_CHAR, "key"))
configs.append(create_config_setting("refresh_message", NULL_CHAR, "original value"))
configs.append(create_config_setting("non_refreshed_message", NULL_CHAR, "Static"))
configs.append(create_config_setting("tagged_config", NULL_CHAR, None, tags={"a": "b"}))
configs.append(create_config_setting("two_tagged", NULL_CHAR, None, tags={"a": "b", "second": "tag"}))
configs.append(create_config_setting("only_second_tag", NULL_CHAR, None, tags={"second": "tag"}))
configs.append(
create_config_setting(
"complex_tag", NULL_CHAR, None, tags={"Special:Tag": "Value:With:Colons", "Tag@With@At": "Value@With@At"}
)
)
configs.append(
create_config_setting(
".appconfig.featureflag/Alpha",
NULL_CHAR,
'{ "id": "Alpha", "description": "", "enabled": false, "conditions": { "client_filters": [] }}',
"application/vnd.microsoft.appconfig.ff+json;charset=utf-8",
)
)
configs.append(
create_config_setting(
".appconfig.featureflag/TaggedFeatureFlag",
NULL_CHAR,
'{ "id": "TaggedFeatureFlag", "description": "", "enabled": false, "conditions": { "client_filters": [] }}',
"application/vnd.microsoft.appconfig.ff+json;charset=utf-8",
tags={"a": "b"},
)
)
# Configuration with multiple tags
configs.append(create_config_setting("multi_tagged", NULL_CHAR, "multi tagged value", tags={"a": "b", "c": "d"}))
# Configuration with tag that has special characters
configs.append(create_config_setting("special_chars", NULL_CHAR, "special", tags={"special@tag": "special:value"}))
# Configuration with no tags
configs.append(create_config_setting("no_tags", NULL_CHAR, "no tags"))
# Configuration with tag that has no value
configs.append(create_config_setting("tag_no_value", NULL_CHAR, "no value", tags={"a": ""}))
# Configuration with different tag
configs.append(create_config_setting("different_tag", NULL_CHAR, "different", tags={"different": "tag"}))
# Configuration with null tag
configs.append(create_config_setting("null_tag", NULL_CHAR, "null tag", tags={"tag": None}))
if keyvault_secret_url:
configs.append(
create_secret_config_setting(
"secret",
"prod",
keyvault_secret_url,
)
)
if keyvault_secret_url2:
configs.append(
create_secret_config_setting(
"secret2",
"prod",
keyvault_secret_url2,
)
)
return configs
def create_config_setting(key, label, value, content_type="text/plain", tags=None):
return ConfigurationSetting(key=key, label=label, value=value, content_type=content_type, tags=tags)
def create_secret_config_setting(key, label, value):
return SecretReferenceConfigurationSetting(
key=key,
label=label,
secret_id=value,
)
def create_feature_flag_config_setting(key, label, enabled, tags=None):
return FeatureFlagConfigurationSetting(feature_id=key, label=label, enabled=enabled, tags=tags)
def get_feature_flag(client, feature_id):
for feature_flag in client[FEATURE_MANAGEMENT_KEY][FEATURE_FLAG_KEY]:
if feature_flag["id"] == feature_id:
return feature_flag
return None
def has_feature_flag(client, feature_id, enabled=False):
feature_flag = get_feature_flag(client, feature_id)
if feature_flag:
return feature_flag["enabled"] == enabled
return False
|
