python-server-sdk/testing/test_ldclient.py at main · bkodes/python-server-sdk · GitHub

History

190 lines (145 loc) · 7.03 KB

Raw

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

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

from ldclient.client import LDClient, Config, Context

from ldclient.feature_store import InMemoryFeatureStore

from ldclient.impl.datasource.polling import PollingUpdateProcessor

from ldclient.impl.datasource.streaming import StreamingUpdateProcessor

from ldclient.impl.stubs import NullUpdateProcessor

from ldclient.interfaces import UpdateProcessor

from ldclient.versioned_data_kind import FEATURES, SEGMENTS

import pytest

from testing.builders import *

from testing.stub_util import CapturingFeatureStore, MockEventProcessor, MockUpdateProcessor

unreachable_uri="http://fake"

context = Context.builder('xyz').set('bizzle', 'def').build()

user = {

u'key': u'xyz',

u'custom': {

u'bizzle': u'def'

}

anonymous_user = {

u'key': u'abc',

u'anonymous': True

}

def make_client(store = InMemoryFeatureStore()):

return LDClient(config=Config(sdk_key = 'SDK_KEY',

base_uri=unreachable_uri,

events_uri=unreachable_uri,

stream_uri=unreachable_uri,

event_processor_class=MockEventProcessor,

update_processor_class=MockUpdateProcessor,

feature_store=store))

def make_offline_client():

return LDClient(config=Config(sdk_key="secret",

offline=True,

base_uri=unreachable_uri,

events_uri=unreachable_uri,

stream_uri=unreachable_uri))

def make_ldd_client():

return LDClient(config=Config(sdk_key="secret",

use_ldd=True,

base_uri=unreachable_uri,

events_uri=unreachable_uri,

stream_uri=unreachable_uri))

def get_first_event(c):

e = c._event_processor._events.pop(0)

c._event_processor._events = []

return e

def count_events(c):

n = len(c._event_processor._events)

c._event_processor._events = []

return n

def test_client_has_null_update_processor_in_offline_mode():

with make_offline_client() as client:

assert isinstance(client._update_processor, NullUpdateProcessor)

def test_client_has_null_update_processor_in_ldd_mode():

with make_ldd_client() as client:

assert isinstance(client._update_processor, NullUpdateProcessor)

@pytest.mark.skip("Can't currently use a live stream processor in tests because its error logging will disrupt other tests.")

def test_client_has_streaming_processor_by_default():

config = Config(sdk_key="secret", base_uri=unreachable_uri, stream_uri=unreachable_uri, send_events=False)

with LDClient(config=config, start_wait=0) as client:

assert isinstance(client._update_processor, StreamingUpdateProcessor)

@pytest.mark.skip("Can't currently use a live polling processor in tests because its error logging will disrupt other tests.")

def test_client_has_polling_processor_if_streaming_is_disabled():

config = Config(sdk_key="secret", stream=False, base_uri=unreachable_uri, stream_uri=unreachable_uri, send_events=False)

with LDClient(config=config, start_wait=0) as client:

assert isinstance(client._update_processor, PollingUpdateProcessor)

def test_toggle_offline():

with make_offline_client() as client:

assert client.variation('feature.key', user, default=None) is None

def test_defaults():

config=Config("SDK_KEY", base_uri="http://localhost:3000", defaults={"foo": "bar"}, offline=True)

with LDClient(config=config) as client:

assert "bar" == client.variation('foo', user, default=None)

def test_defaults_and_online():

expected = "bar"

my_client = LDClient(config=Config("SDK_KEY",

base_uri="http://localhost:3000",

defaults={"foo": expected},

event_processor_class=MockEventProcessor,

update_processor_class=MockUpdateProcessor,

feature_store=InMemoryFeatureStore()))

actual = my_client.variation('foo', user, default="originalDefault")

assert actual == expected

def test_defaults_and_online_no_default():

my_client = LDClient(config=Config("SDK_KEY",

base_uri="http://localhost:3000",

defaults={"foo": "bar"},

event_processor_class=MockEventProcessor,

update_processor_class=MockUpdateProcessor))

assert "jim" == my_client.variation('baz', user, default="jim")

def test_no_defaults():

with make_offline_client() as client:

assert "bar" == client.variation('foo', user, default="bar")

def test_secure_mode_hash():

context_to_hash = Context.create('Message')

equivalent_user_to_hash = {'key': 'Message'}

expected_hash = "aa747c502a898200f9e4fa21bac68136f886a0e27aec70ba06daf2e2a5cb5597"

with make_offline_client() as client:

assert client.secure_mode_hash(context_to_hash) == expected_hash

assert client.secure_mode_hash(equivalent_user_to_hash) == expected_hash

dependency_ordering_test_data = {

FEATURES: {

"a": { "key": "a", "prerequisites": [ { "key": "b" }, { "key": "c" } ] },

"b": { "key": "b", "prerequisites": [ { "key": "c" }, { "key": "e" } ] },

"c": { "key": "c" },

"d": { "key": "d" },

"e": { "key": "e" },

"f": { "key": "f" }

SEGMENTS: {

"o": { "key": "o" }

}

class DependencyOrderingDataUpdateProcessor(UpdateProcessor):

def __init__(self, config, store, ready):

store.init(dependency_ordering_test_data)

ready.set()

def start(self):

pass

def initialized(self):

return True

def test_store_data_set_ordering():

store = CapturingFeatureStore()

config = Config(sdk_key = 'SDK_KEY', send_events=False, feature_store=store,

update_processor_class=DependencyOrderingDataUpdateProcessor)

LDClient(config=config)

data = store.received_data

assert data is not None

assert len(data) == 2

keys = list(data.keys())

values = list(data.values())

assert keys[0] == SEGMENTS

assert len(values[0]) == len(dependency_ordering_test_data[SEGMENTS])

assert keys[1] == FEATURES

flags_map = values[1]

flags_list = list(flags_map.values())

assert len(flags_list) == len(dependency_ordering_test_data[FEATURES])

for item_index, item in enumerate(flags_list):

for prereq in item.get("prerequisites", []):

prereq_item = flags_map[prereq["key"]]

prereq_index = flags_list.index(prereq_item)

if prereq_index > item_index:

all_keys = (f["key"] for f in flags_list)

raise Exception("%s depends on %s, but %s was listed first; keys in order are [%s]" %

(item["key"], prereq["key"], item["key"], ", ".join(all_keys)))

python-server-sdk/testing/test_ldclient.py at main · bkodes/python-server-sdk · GitHub

Expand file tree

Breadcrumbs

test_ldclient.py

Latest commit

History

Breadcrumbs

test_ldclient.py

File metadata and controls