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_safe_creates(self):
# Creates are safe when existing records is under MIN_EXISTING_RECORDS
zone = Zone('unit.tests.', [])
record = Record.new(zone, 'a', {
'ttl': 30,
'type': 'A',
'value': '1.2.3.4',
})
Plan(zone, zone, [Create(record) for i in range(10)]).raise_if_unsafe()
td_mock = MagicMock()
td_mock._rulesets = []
provider._traffic_director_monitor = MagicMock()
provider._find_or_create_dynamic_pool = MagicMock()
td_mock.all_response_pools = []
provider._find_or_create_dynamic_pool.side_effect = [
_DummyPool('default'),
_DummyPool('one'),
_DummyPool('two'),
_DummyPool('three'),
]
change = Create(self.dynamic_cname_record)
provider._mod_dynamic_rulesets(td_mock, change)
add_response_pool_mock.assert_has_calls((
# default
call('default'),
# first dynamic and it's fallback
call('one'),
call('default', index=999),
# 2nd dynamic and it's fallback
call('three'),
call('default', index=999),
# 3nd dynamic and it's fallback
call('two'),
call('default', index=999),
))
ruleset_create_mock.assert_has_calls((
call(td_mock, index=0),
}
stubber.add_response('list_hosted_zones', list_hosted_zones_resp,
{})
list_resource_record_sets_resp = {
'ResourceRecordSets': [],
'IsTruncated': False,
'MaxItems': '100',
}
stubber.add_response('list_resource_record_sets',
list_resource_record_sets_resp,
{'HostedZoneId': 'z42'})
plan = provider.plan(self.expected)
self.assertEquals(8, len(plan.changes))
for change in plan.changes:
self.assertIsInstance(change, Create)
stubber.assert_no_pending_responses()
stubber.add_response('list_health_checks',
{
'HealthChecks': self.health_checks,
'IsTruncated': False,
'MaxItems': '100',
'Marker': '',
})
stubber.add_response('change_resource_record_sets',
{'ChangeInfo': {
'Id': 'id',
'Status': 'PENDING',
'SubmittedAt': '2017-01-29T01:02:03Z',
}}, {'HostedZoneId': 'z42', 'ChangeBatch': ANY})
self.assertEquals(15, len(filter(lambda c: isinstance(c, Create),
plan.changes)))
self.assertEquals(14, len(filter(lambda c: isinstance(c, Create),
plan.changes)))
'ttl': 300,
'type': 'A',
# This matches the zone data above, one to swap, one to leave
'values': ['1.1.1.1', '2.2.2.2'],
})
new = Record.new(zone, 'a', {
'geo': {
'AF': ['5.5.5.5'],
'NA-US': ['6.6.6.6']
},
'ttl': 300,
'type': 'A',
# This leaves one, swaps ones, and adds one
'values': ['2.2.2.2', '3.3.3.3', '4.4.4.4'],
}, simple)
create = Create(Record.new(zone, 'b', {
'ttl': 60,
'type': 'CNAME',
'value': 'foo.unit.tests.'
}, simple))
update = Update(existing, new)
delete = Delete(new)
changes = [create, delete, update]
plans = [
(simple, Plan(zone, zone, changes)),
(simple, Plan(zone, zone, changes)),
]
class TestPlanHtml(TestCase):
log = getLogger('TestPlanHtml')
provider._mod_Update = MagicMock()
provider._mod_Delete = MagicMock()
# busted traffic director
mock.side_effect = [
# get zone
{'data': {}},
# accept publish
{'data': {}},
]
desired = Zone('unit.tests.', [])
geo = self.geo_record
regular = self.regular_record
changes = [
Create(geo),
Create(regular),
Update(geo, geo),
Update(regular, regular),
Delete(geo),
Delete(regular),
]
plan = Plan(None, desired, changes, True)
provider._apply(plan)
mock.assert_has_calls([
call('/Zone/unit.tests/', 'GET', {}),
call('/Zone/unit.tests/', 'PUT', {'publish': True})
])
# should have seen 1 call to each
provider._mod_geo_Create.assert_called_once()
provider._mod_geo_Update.assert_called_once()
provider._mod_geo_Delete.assert_called_once()
# sorting mostly to make things deterministic for testing, but in
# theory it let us find what we're after quicker (though sorting would
# be more expensive.)
for record in sorted(existing.records):
if record == ns:
# We've found the top-level NS record, return any changes
change = record.changes(ns, self)
self.log.debug('_extra_changes: change=%s', change)
if change:
# We need to modify an existing record
return [change]
# No change is necessary
return []
# No existing top-level NS
self.log.debug('_extra_changes: create')
return [Create(ns)]