Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_value_for_CNAME(self):
provider = DynProvider('test', 'cust', 'user', 'pass')
class DummyRecord(object):
def __init__(self, cname, weight):
self.cname = cname
self.weight = weight
record = DummyRecord('foo.unit.tests.', 32)
self.assertEquals({
'value': record.cname,
'weight': record.weight,
}, provider._value_for_CNAME('CNAME', record))
def test_populate_dynamic_traffic_director(self):
provider = DynProvider('test', 'cust', 'user', 'pass')
fqdn = 'dynamic.unit.tests.'
multi_a = [
DummyRecord('1.2.3.5', 1, 90),
DummyRecord('1.2.3.6', 1, 90),
DummyRecord('1.2.3.7', 1, 90),
]
default_response_pool = DummyResponsePool('default', multi_a)
pool1_response_pool = DummyResponsePool('pool1', multi_a)
rulesets = [
DummyRuleset('default', [default_response_pool]),
DummyRuleset('0:abcdef', [pool1_response_pool], 'geoip', {
'geoip': {
'country': ['US'],
'province': ['or'],
'region': [14],
def test_sync(self, execute_mock):
provider = DynProvider('test', 'cust', 'user', 'pass')
# Test Zone create
execute_mock.side_effect = [
# No such zone, during populate
DynectGetError('foo'),
# No such zone, during sync
DynectGetError('foo'),
# get empty Zone
{'data': {}},
# get zone we can modify & delete with
{'data': {
# A top-level to delete
'a_records': [{
'fqdn': 'unit.tests',
'rdata': {'address': '1.2.3.4'},
'record_id': 1,
def test_mod_geo_create(self, mock):
provider = DynProvider('test', 'cust', 'user', 'pass',
traffic_directors_enabled=True)
# will be tested separately
provider._mod_geo_rulesets = MagicMock()
mock.side_effect = [
# create traffic director
self.traffic_director_response,
# get traffic directors
self.traffic_directors_response
]
provider._mod_geo_Create(None, Create(self.geo_record))
# td now lives in cache
self.assertTrue('A' in provider.traffic_directors['unit.tests.'])
# should have seen 1 gen call
provider._mod_geo_rulesets.assert_called_once()
def test_mod_geo_update_geo_geo(self):
provider = DynProvider('test', 'cust', 'user', 'pass',
traffic_directors_enabled=True)
# update of an existing td
# pre-populate the cache with our mock td
provider._traffic_directors = {
'unit.tests.': {
'A': 42,
}
}
# mock _mod_geo_rulesets
provider._mod_geo_rulesets = MagicMock()
geo = self.geo_record
change = Update(geo, geo)
provider._mod_geo_Update(None, change)
def test_populate_traffic_director_busted(self, mock):
provider = DynProvider('test', 'cust', 'user', 'pass',
traffic_directors_enabled=True)
busted_traffic_director_response = {
"status": "success",
"data": {
"notifiers": [],
"rulesets": [],
"ttl": "300",
"active": "Y",
"service_id": "oIRZ4lM-W64NUelJGuzuVziZ4MI",
"nodes": [{
"fqdn": "unit.tests",
"zone": "unit.tests"
}],
"pending_change": "",
"label": "unit.tests.:A"
def test_find_or_create_geo_pool(self, mock):
provider = DynProvider('test', 'cust', 'user', 'pass',
traffic_directors_enabled=True)
td = 42
# no candidates cache miss, so create
values = ['1.2.3.4', '1.2.3.5']
pool = provider._find_or_create_geo_pool(td, [], 'default', 'A',
values)
self.assertIsInstance(pool, DSFResponsePool)
self.assertEquals(1, len(pool.rs_chains))
records = pool.rs_chains[0].record_sets[0].records
self.assertEquals(values, [r.address for r in records])
mock.assert_called_once_with(td)
# cache hit, use the one we just created
mock.reset_mock()
def test_populate_dynamic_rules(self):
provider = DynProvider('test', 'cust', 'user', 'pass')
# Empty
rulesets = []
pools = {}
rules = provider._populate_dynamic_rules(rulesets, pools)
self.assertEquals([], rules)
# default: is ignored
rulesets = [DummyRuleset('default:')]
pools = {}
rules = provider._populate_dynamic_rules(rulesets, pools)
self.assertEquals([], rules)
# No ResponsePools in RuleSet, ignored
rulesets = [DummyRuleset('0:abcdefg')]
pools = {}
def test_mod_dynamic_update_regular_dynamic(self, _):
provider = DynProvider('test', 'cust', 'user', 'pass',
traffic_directors_enabled=True)
# convert a regular record to a dynamic td
provider._mod_dynamic_Create = MagicMock()
provider._mod_Delete = MagicMock()
change = Update(self.regular_a_record, self.dynamic_a_record)
provider._mod_dynamic_Update(42, change)
# should have seen a call to create the new geo record
provider._mod_dynamic_Create.assert_called_once_with(42, change)
# should have seen a call to delete the old regular record
provider._mod_Delete.assert_called_once_with(42, change)
def __init__(self, id, customer, username, password,
traffic_directors_enabled=False, *args, **kwargs):
self.log = getLogger('DynProvider[{}]'.format(id))
self.log.debug('__init__: id=%s, customer=%s, username=%s, '
'password=***, traffic_directors_enabled=%s', id,
customer, username, traffic_directors_enabled)
# we have to set this before calling super b/c SUPPORTS_GEO requires it
self.traffic_directors_enabled = traffic_directors_enabled
super(DynProvider, self).__init__(id, *args, **kwargs)
self.customer = customer
self.username = username
self.password = password
self._cache = {}
self._traffic_directors = None
self._traffic_director_monitors = None