Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
'HealthCheckConfig': {
'Type': 'HTTPS',
'FullyQualifiedDomainName': 'unit.tests',
'IPAddress': '2.2.3.4',
},
'HealthCheckVersion': 2,
}],
'IsTruncated': False,
'MaxItems': '100',
'Marker': '',
})
extra = provider._extra_changes(existing, [])
self.assertEquals(1, len(extra))
stubber.assert_no_pending_responses()
for change in (Create(record), Update(record, record), Delete(record)):
extra = provider._extra_changes(existing, [change])
self.assertEquals(0, len(extra))
stubber.assert_no_pending_responses()
'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')
def test_empty(self):
out = StringIO()
PlanHtml('html').run([], fh=out)
self.assertEquals('<b>No changes were planned</b>', out.getvalue())
def test_simple(self):
'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': 3600,
'type': 'CNAME',
'value': 'foo.unit.tests.'
}, simple))
update = Update(existing, new)
delete = Delete(new)
changes = [create, delete, update]
plan = Plan(zone, zone, changes)
self.assertEquals(3600, create.new.ttl)
self.assertEquals(300, update.new.ttl)
plan.make_cautious()
self.assertEquals(60, create.new.ttl)
self.assertEquals(60, update.new.ttl)
def test_mod_geo_delete(self, mock):
provider = DynProvider('test', 'cust', 'user', 'pass',
traffic_directors_enabled=True)
td_mock = MagicMock()
provider._traffic_directors = {
'unit.tests.': {
'A': td_mock,
}
}
provider._mod_geo_Delete(None, Delete(self.geo_record))
# delete called
td_mock.delete.assert_called_once()
# removed from cache
self.assertFalse('A' in provider.traffic_directors['unit.tests.'])
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()
provider._mod_Create.assert_called_once()
provider._mod_Update.assert_called_once()
provider._mod_Delete.assert_called_once()
{'answer': ['3.4.5.6'], 'meta': {'country': ['US']}},
{'answer': ['4.5.6.7'],
'meta': {'iso_region_code': ['NA-US-WA']}},
],
'ttl': 34,
},
]
nsone_zone.search = zone_search
load_mock.side_effect = [nsone_zone, nsone_zone]
plan = provider.plan(desired)
self.assertEquals(3, len(plan.changes))
# Shouldn't rely on order so just count classes
classes = defaultdict(lambda: 0)
for change in plan.changes:
classes[change.__class__] += 1
self.assertEquals(1, classes[Delete])
self.assertEquals(2, classes[Update])
# ugh, we need a mock record that can be returned from loadRecord for
# the update and delete targets, we can add our side effects to that to
# trigger rate limit handling
mock_record = Mock()
mock_record.update.side_effect = [
RateLimitException('one', period=0),
None,
None,
]
mock_record.delete.side_effect = [
RateLimitException('two', period=0),
None,
None,
]
nsone_zone.loadRecord.side_effect = [mock_record, mock_record,
# than MIN_EXISTING_RECORDS exist
zone = Zone('unit.tests.', [])
record = Record.new(zone, 'a', {
'ttl': 30,
'type': 'A',
'value': '1.2.3.4',
})
for i in range(int(Plan.MIN_EXISTING_RECORDS)):
zone.add_record(Record.new(zone, str(i), {
'ttl': 60,
'type': 'A',
'value': '2.3.4.5'
}))
changes = [Delete(record)
for i in range(int(Plan.MIN_EXISTING_RECORDS *
Plan.MAX_SAFE_DELETE_PCENT) + 1)]
with self.assertRaises(UnsafePlan) as ctx:
Plan(zone, zone, changes).raise_if_unsafe()
self.assertTrue('Too many deletes' in ctx.exception.message)
target = SimpleProvider()
# before == after -> no changes
self.assertFalse(before.changes(after, target))
# add a record, delete a record -> [Delete, Create]
c = ARecord(before, 'c', {'ttl': 42, 'value': '1.1.1.1'})
after.add_record(c)
after._remove_record(b)
self.assertEquals(after.records, set([a, c]))
changes = before.changes(after, target)
self.assertEquals(2, len(changes))
for change in changes:
if isinstance(change, Create):
create = change
elif isinstance(change, Delete):
delete = change
self.assertEquals(b, delete.existing)
self.assertFalse(delete.new)
self.assertEquals(c, create.new)
self.assertFalse(create.existing)
delete.__repr__()
create.__repr__()
after = Zone('unit.tests.', [])
changed = ARecord(before, 'a', {'ttl': 42, 'value': '2.2.2.2'})
after.add_record(changed)
after.add_record(b)
changes = before.changes(after, target)
self.assertEquals(1, len(changes))
update = changes[0]
self.assertIsInstance(update, Update)
# MAX_SAFE_DELETE_PCENT is safe when more
# than MIN_EXISTING_RECORDS exist
zone = Zone('unit.tests.', [])
record = Record.new(zone, 'a', {
'ttl': 30,
'type': 'A',
'value': '1.2.3.4',
})
for i in range(int(Plan.MIN_EXISTING_RECORDS)):
zone.add_record(Record.new(zone, str(i), {
'ttl': 60,
'type': 'A',
'value': '2.3.4.5'
}))
changes = [Delete(record)
for i in range(int(Plan.MIN_EXISTING_RECORDS *
Plan.MAX_SAFE_DELETE_PCENT))]
Plan(zone, zone, changes).raise_if_unsafe()
self.assertEquals(1, len(filter(lambda d: isinstance(d, Delete),
plan.changes)))