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
|
# -------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for
# license information.
# --------------------------------------------------------------------------
import time
import unittest
from unittest.mock import Mock
from azure.appconfiguration.provider import WatchKey
from devtools_testutils import recorded_by_proxy
from preparers import app_config_decorator_aad
from testcase import AppConfigTestCase, has_feature_flag
from test_constants import FEATURE_MANAGEMENT_KEY
class TestAppConfigurationProvider(AppConfigTestCase, unittest.TestCase):
# method: refresh
@recorded_by_proxy
@app_config_decorator_aad
def test_refresh(self, appconfiguration_endpoint_string, appconfiguration_keyvault_secret_url):
mock_callback = Mock()
client = self.create_aad_client(
appconfiguration_endpoint_string,
keyvault_secret_url=appconfiguration_keyvault_secret_url,
refresh_on=[WatchKey("refresh_message")],
refresh_interval=1,
on_refresh_success=mock_callback,
feature_flag_enabled=True,
feature_flag_refresh_enabled=True,
)
assert client["refresh_message"] == "original value"
assert client["my_json"]["key"] == "value"
assert FEATURE_MANAGEMENT_KEY in client
assert has_feature_flag(client, "Alpha")
appconfig_client = self.create_aad_sdk_client(appconfiguration_endpoint_string)
setting = appconfig_client.get_configuration_setting(key="refresh_message")
setting.value = "updated value"
feature_flag = appconfig_client.get_configuration_setting(key=".appconfig.featureflag/Alpha")
feature_flag.enabled = True
appconfig_client.set_configuration_setting(setting)
appconfig_client.set_configuration_setting(feature_flag)
# Waiting for the refresh interval to pass
time.sleep(2)
client.refresh()
assert client["refresh_message"] == "updated value"
assert has_feature_flag(client, "Alpha", True)
assert mock_callback.call_count == 1
setting.value = "original value"
feature_flag.enabled = False
appconfig_client.set_configuration_setting(setting)
appconfig_client.set_configuration_setting(feature_flag)
# Waiting for the refresh interval to pass
time.sleep(2)
client.refresh()
assert client["refresh_message"] == "original value"
assert has_feature_flag(client, "Alpha", False)
assert mock_callback.call_count == 2
setting.value = "updated value 2"
feature_flag.enabled = True
appconfig_client.set_configuration_setting(setting)
appconfig_client.set_configuration_setting(feature_flag)
# Not waiting for the refresh interval to pass
client.refresh()
assert client["refresh_message"] == "original value"
assert has_feature_flag(client, "Alpha", False)
assert mock_callback.call_count == 2
setting.value = "original value"
appconfig_client.set_configuration_setting(setting)
client.refresh()
assert client["refresh_message"] == "original value"
assert mock_callback.call_count == 2
# method: refresh
@recorded_by_proxy
@app_config_decorator_aad
def test_empty_refresh(self, appconfiguration_endpoint_string, appconfiguration_keyvault_secret_url):
mock_callback = Mock()
client = self.create_aad_client(
appconfiguration_endpoint_string,
keyvault_secret_url=appconfiguration_keyvault_secret_url,
on_refresh_success=mock_callback,
feature_flag_enabled=True,
)
assert client["refresh_message"] == "original value"
assert client["non_refreshed_message"] == "Static"
assert client["my_json"]["key"] == "value"
assert FEATURE_MANAGEMENT_KEY in client
assert has_feature_flag(client, "Alpha")
appconfig_client = self.create_aad_sdk_client(appconfiguration_endpoint_string)
setting = appconfig_client.get_configuration_setting(key="refresh_message")
setting.value = "updated value"
appconfig_client.set_configuration_setting(setting)
static_setting = appconfig_client.get_configuration_setting(key="non_refreshed_message")
static_setting.value = "updated static"
appconfig_client.set_configuration_setting(static_setting)
# Waiting for the refresh interval to pass
time.sleep(2)
client.refresh()
assert client["refresh_message"] == "original value"
assert client["non_refreshed_message"] == "Static"
assert mock_callback.call_count == 0
setting.value = "original value"
appconfig_client.set_configuration_setting(setting)
static_setting.value = "Static"
appconfig_client.set_configuration_setting(static_setting)
|