mirror/github-octodns
1
0
Fork 0
mirror of https://github.com/github/octodns.git synced 2024-05-11 05:55:00 +00:00
Code Issues Releases Activity

Merge branch 'master' into googledns-semicolon-escapes

Browse Source
This commit is contained in:
Ross McFarland
2018-03-03 10:43:26 -08:00
committed by GitHub
parent 7215d80230 b153d56289
commit fc4bf6b8ce
38 changed files with 328 additions and 203 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_octodns_plan.py
+34 -13
View File
@@ -16,15 +16,6 @@ from octodns.zone import Zone
from helpers import SimpleProvider
class TestPlanLogger(TestCase):
def test_invalid_level(self):
with self.assertRaises(Exception) as ctx:
PlanLogger('invalid', 'not-a-level')
self.assertEquals('Unsupported level: not-a-level',
ctx.exception.message)
simple = SimpleProvider()
zone = Zone('unit.tests.', [])
existing = Record.new(zone, 'a', {
@@ -48,15 +39,45 @@ create = Create(Record.new(zone, 'b', {
'type': 'CNAME',
'value': 'foo.unit.tests.'
}, simple))
create2 = Create(Record.new(zone, 'c', {
'ttl': 60,
'type': 'CNAME',
'value': 'foo.unit.tests.'
}))
update = Update(existing, new)
delete = Delete(new)
changes = [create, delete, update]
changes = [create, create2, delete, update]
plans = [
(simple, Plan(zone, zone, changes)),
(simple, Plan(zone, zone, changes)),
(simple, Plan(zone, zone, changes, True)),
(simple, Plan(zone, zone, changes, False)),
]
class TestPlanLogger(TestCase):
def test_invalid_level(self):
with self.assertRaises(Exception) as ctx:
PlanLogger('invalid', 'not-a-level')
self.assertEquals('Unsupported level: not-a-level',
ctx.exception.message)
def test_create(self):
class MockLogger(object):
def __init__(self):
self.out = StringIO()
def log(self, level, msg):
self.out.write(msg)
log = MockLogger()
PlanLogger('logger').run(log, plans)
out = log.out.getvalue()
self.assertTrue('Summary: Creates=2, Updates=1, '
'Deletes=1, Existing Records=0' in out)
class TestPlanHtml(TestCase):
log = getLogger('TestPlanHtml')
@@ -69,7 +90,7 @@ class TestPlanHtml(TestCase):
out = StringIO()
PlanHtml('html').run(plans, fh=out)
out = out.getvalue()
self.assertTrue(' <td colspan=6>Summary: Creates=1, Updates=1, '
self.assertTrue(' <td colspan=6>Summary: Creates=2, Updates=1, '
'Deletes=1, Existing Records=0</td>' in out)
tests/test_octodns_provider_azuredns.py
+10 -5
View File
@@ -302,7 +302,8 @@ class TestAzureDnsProvider(TestCase):
record_list = provider._dns_client.record_sets.list_by_dns_zone
record_list.return_value = rs
provider.populate(zone)
exists = provider.populate(zone)
self.assertTrue(exists)
self.assertEquals(len(zone.records), 16)
@@ -338,8 +339,10 @@ class TestAzureDnsProvider(TestCase):
changes.append(Create(i))
deletes.append(Delete(i))
self.assertEquals(13, provider.apply(Plan(None, zone, changes)))
self.assertEquals(13, provider.apply(Plan(zone, zone, deletes)))
self.assertEquals(13, provider.apply(Plan(None, zone,
changes, True)))
self.assertEquals(13, provider.apply(Plan(zone, zone,
deletes, True)))
def test_create_zone(self):
provider = self._get_provider()
@@ -354,7 +357,8 @@ class TestAzureDnsProvider(TestCase):
_get = provider._dns_client.zones.get
_get.side_effect = CloudError(Mock(status=404), err_msg)
self.assertEquals(13, provider.apply(Plan(None, desired, changes)))
self.assertEquals(13, provider.apply(Plan(None, desired, changes,
True)))
def test_check_zone_no_create(self):
provider = self._get_provider()
@@ -374,6 +378,7 @@ class TestAzureDnsProvider(TestCase):
_get = provider._dns_client.zones.get
_get.side_effect = CloudError(Mock(status=404), err_msg)
provider.populate(Zone('unit3.test.', []))
exists = provider.populate(Zone('unit3.test.', []))
self.assertFalse(exists)
self.assertEquals(len(zone.records), 0)
tests/test_octodns_provider_base.py
+22 -20
View File
@@ -63,14 +63,14 @@ class TestBaseProvider(TestCase):
zone = Zone('unit.tests.', [])
with self.assertRaises(NotImplementedError) as ctx:
HasSupportsGeo('hassupportesgeo').populate(zone)
HasSupportsGeo('hassupportsgeo').populate(zone)
self.assertEquals('Abstract base class, SUPPORTS property missing',
ctx.exception.message)
class HasSupports(HasSupportsGeo):
SUPPORTS = set(('A',))
with self.assertRaises(NotImplementedError) as ctx:
HasSupports('hassupportes').populate(zone)
HasSupports('hassupports').populate(zone)
self.assertEquals('Abstract base class, populate method missing',
ctx.exception.message)
@@ -94,7 +94,7 @@ class TestBaseProvider(TestCase):
'value': '1.2.3.4'
}))
self.assertTrue(HasSupports('hassupportesgeo')
self.assertTrue(HasSupports('hassupportsgeo')
.supports(list(zone.records)[0]))
plan = HasPopulate('haspopulate').plan(zone)
@@ -153,7 +153,7 @@ class TestBaseProvider(TestCase):
def test_safe_none(self):
# No changes is safe
Plan(None, None, []).raise_if_unsafe()
Plan(None, None, [], True).raise_if_unsafe()
def test_safe_creates(self):
# Creates are safe when existing records is under MIN_EXISTING_RECORDS
@@ -164,7 +164,8 @@ class TestBaseProvider(TestCase):
'type': 'A',
'value': '1.2.3.4',
})
Plan(zone, zone, [Create(record) for i in range(10)]).raise_if_unsafe()
Plan(zone, zone, [Create(record) for i in range(10)], True) \
.raise_if_unsafe()
def test_safe_min_existing_creates(self):
# Creates are safe when existing records is over MIN_EXISTING_RECORDS
@@ -177,13 +178,14 @@ class TestBaseProvider(TestCase):
})
for i in range(int(Plan.MIN_EXISTING_RECORDS)):
zone.add_record(Record.new(zone, str(i), {
zone.add_record(Record.new(zone, unicode(i), {
'ttl': 60,
'type': 'A',
'value': '2.3.4.5'
}))
Plan(zone, zone, [Create(record) for i in range(10)]).raise_if_unsafe()
Plan(zone, zone, [Create(record) for i in range(10)], True) \
.raise_if_unsafe()
def test_safe_no_existing(self):
# existing records fewer than MIN_EXISTING_RECORDS is safe
@@ -195,7 +197,7 @@ class TestBaseProvider(TestCase):
})
updates = [Update(record, record), Update(record, record)]
Plan(zone, zone, updates).raise_if_unsafe()
Plan(zone, zone, updates, True).raise_if_unsafe()
def test_safe_updates_min_existing(self):
# MAX_SAFE_UPDATE_PCENT+1 fails when more
@@ -208,7 +210,7 @@ class TestBaseProvider(TestCase):
})
for i in range(int(Plan.MIN_EXISTING_RECORDS)):
zone.add_record(Record.new(zone, str(i), {
zone.add_record(Record.new(zone, unicode(i), {
'ttl': 60,
'type': 'A',
'value': '2.3.4.5'
@@ -219,7 +221,7 @@ class TestBaseProvider(TestCase):
Plan.MAX_SAFE_UPDATE_PCENT) + 1)]
with self.assertRaises(UnsafePlan) as ctx:
Plan(zone, zone, changes).raise_if_unsafe()
Plan(zone, zone, changes, True).raise_if_unsafe()
self.assertTrue('Too many updates' in ctx.exception.message)
@@ -234,7 +236,7 @@ class TestBaseProvider(TestCase):
})
for i in range(int(Plan.MIN_EXISTING_RECORDS)):
zone.add_record(Record.new(zone, str(i), {
zone.add_record(Record.new(zone, unicode(i), {
'ttl': 60,
'type': 'A',
'value': '2.3.4.5'
@@ -243,7 +245,7 @@ class TestBaseProvider(TestCase):
for i in range(int(Plan.MIN_EXISTING_RECORDS *
Plan.MAX_SAFE_UPDATE_PCENT))]
Plan(zone, zone, changes).raise_if_unsafe()
Plan(zone, zone, changes, True).raise_if_unsafe()
def test_safe_deletes_min_existing(self):
# MAX_SAFE_DELETE_PCENT+1 fails when more
@@ -256,7 +258,7 @@ class TestBaseProvider(TestCase):
})
for i in range(int(Plan.MIN_EXISTING_RECORDS)):
zone.add_record(Record.new(zone, str(i), {
zone.add_record(Record.new(zone, unicode(i), {
'ttl': 60,
'type': 'A',
'value': '2.3.4.5'
@@ -267,7 +269,7 @@ class TestBaseProvider(TestCase):
Plan.MAX_SAFE_DELETE_PCENT) + 1)]
with self.assertRaises(UnsafePlan) as ctx:
Plan(zone, zone, changes).raise_if_unsafe()
Plan(zone, zone, changes, True).raise_if_unsafe()
self.assertTrue('Too many deletes' in ctx.exception.message)
@@ -282,7 +284,7 @@ class TestBaseProvider(TestCase):
})
for i in range(int(Plan.MIN_EXISTING_RECORDS)):
zone.add_record(Record.new(zone, str(i), {
zone.add_record(Record.new(zone, unicode(i), {
'ttl': 60,
'type': 'A',
'value': '2.3.4.5'
@@ -291,7 +293,7 @@ class TestBaseProvider(TestCase):
for i in range(int(Plan.MIN_EXISTING_RECORDS *
Plan.MAX_SAFE_DELETE_PCENT))]
Plan(zone, zone, changes).raise_if_unsafe()
Plan(zone, zone, changes, True).raise_if_unsafe()
def test_safe_updates_min_existing_override(self):
safe_pcent = .4
@@ -305,7 +307,7 @@ class TestBaseProvider(TestCase):
})
for i in range(int(Plan.MIN_EXISTING_RECORDS)):
zone.add_record(Record.new(zone, str(i), {
zone.add_record(Record.new(zone, unicode(i), {
'ttl': 60,
'type': 'A',
'value': '2.3.4.5'
@@ -316,7 +318,7 @@ class TestBaseProvider(TestCase):
safe_pcent) + 1)]
with self.assertRaises(UnsafePlan) as ctx:
Plan(zone, zone, changes,
Plan(zone, zone, changes, True,
update_pcent_threshold=safe_pcent).raise_if_unsafe()
self.assertTrue('Too many updates' in ctx.exception.message)
@@ -333,7 +335,7 @@ class TestBaseProvider(TestCase):
})
for i in range(int(Plan.MIN_EXISTING_RECORDS)):
zone.add_record(Record.new(zone, str(i), {
zone.add_record(Record.new(zone, unicode(i), {
'ttl': 60,
'type': 'A',
'value': '2.3.4.5'
@@ -344,7 +346,7 @@ class TestBaseProvider(TestCase):
safe_pcent) + 1)]
with self.assertRaises(UnsafePlan) as ctx:
Plan(zone, zone, changes,
Plan(zone, zone, changes, True,
delete_pcent_threshold=safe_pcent).raise_if_unsafe()
self.assertTrue('Too many deletes' in ctx.exception.message)
tests/test_octodns_provider_cloudflare.py
+8 -5
View File
@@ -166,6 +166,7 @@ class TestCloudflareProvider(TestCase):
plan = provider.plan(self.expected)
self.assertEquals(12, len(plan.changes))
self.assertEquals(12, provider.apply(plan))
self.assertFalse(plan.exists)
provider._request.assert_has_calls([
# created the domain
@@ -285,6 +286,7 @@ class TestCloudflareProvider(TestCase):
# only see the delete & ttl update, below min-ttl is filtered out
self.assertEquals(2, len(plan.changes))
self.assertEquals(2, provider.apply(plan))
self.assertTrue(plan.exists)
# recreate for update, and deletes for the 2 parts of the other
provider._request.assert_has_calls([
call('PUT', '/zones/ff12ab34cd5611334422ab3322997650/dns_records/'
@@ -366,7 +368,7 @@ class TestCloudflareProvider(TestCase):
'values': ['2.2.2.2', '3.3.3.3', '4.4.4.4'],
})
change = Update(existing, new)
plan = Plan(zone, zone, [change])
plan = Plan(zone, zone, [change], True)
provider._apply(plan)
provider._request.assert_has_calls([
@@ -451,7 +453,7 @@ class TestCloudflareProvider(TestCase):
'value': 'ns2.foo.bar.',
})
change = Update(existing, new)
plan = Plan(zone, zone, [change])
plan = Plan(zone, zone, [change], True)
provider._apply(plan)
provider._request.assert_has_calls([
@@ -599,7 +601,8 @@ class TestCloudflareProvider(TestCase):
zone = Zone('unit.tests.', [])
provider.populate(zone)
# the two A records get merged into one CNAME record poining to the CDN
# the two A records get merged into one CNAME record pointing to
# the CDN.
self.assertEquals(3, len(zone.records))
record = list(zone.records)[0]
@@ -621,7 +624,7 @@ class TestCloudflareProvider(TestCase):
self.assertEquals('a.unit.tests.cdn.cloudflare.net.', record.value)
# CDN enabled records can't be updated, we don't know the real values
# never point a Cloudflare record to itsself.
# never point a Cloudflare record to itself.
wanted = Zone('unit.tests.', [])
wanted.add_record(Record.new(wanted, 'cname', {
'ttl': 300,
@@ -676,7 +679,7 @@ class TestCloudflareProvider(TestCase):
self.assertEquals('unit.tests.cdn.cloudflare.net.', record.value)
# CDN enabled records can't be updated, we don't know the real values
# never point a Cloudflare record to itsself.
# never point a Cloudflare record to itself.
wanted = Zone('unit.tests.', [])
wanted.add_record(Record.new(wanted, '', {
'ttl': 300,
tests/test_octodns_provider_digitalocean.py
+2
View File
@@ -165,6 +165,7 @@ class TestDigitalOceanProvider(TestCase):
n = len(self.expected.records) - 7
self.assertEquals(n, len(plan.changes))
self.assertEquals(n, provider.apply(plan))
self.assertFalse(plan.exists)
provider._client._request.assert_has_calls([
# created the domain
@@ -225,6 +226,7 @@ class TestDigitalOceanProvider(TestCase):
}))
plan = provider.plan(wanted)
self.assertTrue(plan.exists)
self.assertEquals(2, len(plan.changes))
self.assertEquals(2, provider.apply(plan))
# recreate for update, and delete for the 2 parts of the other
tests/test_octodns_provider_dnsimple.py
+2
View File
@@ -133,6 +133,7 @@ class TestDnsimpleProvider(TestCase):
n = len(self.expected.records) - 3
self.assertEquals(n, len(plan.changes))
self.assertEquals(n, provider.apply(plan))
self.assertFalse(plan.exists)
provider._client._request.assert_has_calls([
# created the domain
@@ -186,6 +187,7 @@ class TestDnsimpleProvider(TestCase):
}))
plan = provider.plan(wanted)
self.assertTrue(plan.exists)
self.assertEquals(2, len(plan.changes))
self.assertEquals(2, provider.apply(plan))
# recreate for update, and deletes for the 2 parts of the other
tests/test_octodns_provider_dyn.py
+11 -9
View File
@@ -430,6 +430,7 @@ class TestDynProvider(TestCase):
update_mock.assert_not_called()
provider.apply(plan)
update_mock.assert_called()
self.assertFalse(plan.exists)
add_mock.assert_called()
# Once for each dyn record (8 Records, 2 of which have dual values)
self.assertEquals(15, len(add_mock.call_args_list))
@@ -474,6 +475,7 @@ class TestDynProvider(TestCase):
plan = provider.plan(new)
provider.apply(plan)
update_mock.assert_called()
self.assertTrue(plan.exists)
# we expect 4 deletes, 2 from actual deletes and 2 from
# updates which delete and recreate
self.assertEquals(4, len(delete_mock.call_args_list))
@@ -491,7 +493,7 @@ class TestDynProviderGeo(TestCase):
traffic_director_response = loads(fh.read())
@property
def traffic_directors_reponse(self):
def traffic_directors_response(self):
return {
'data': [{
'active': 'Y',
@@ -607,7 +609,7 @@ class TestDynProviderGeo(TestCase):
mock.side_effect = [{'data': []}]
self.assertEquals({}, provider.traffic_directors)
# a supported td and an ingored one
# a supported td and an ignored one
response = {
'data': [{
'active': 'Y',
@@ -650,7 +652,7 @@ class TestDynProviderGeo(TestCase):
set(tds.keys()))
self.assertEquals(['A'], tds['unit.tests.'].keys())
self.assertEquals(['A'], tds['geo.unit.tests.'].keys())
provider.log.warn.assert_called_with("Failed to load TraficDirector "
provider.log.warn.assert_called_with("Failed to load TrafficDirector "
"'%s': %s", 'something else',
'need more than 1 value to '
'unpack')
@@ -758,7 +760,7 @@ class TestDynProviderGeo(TestCase):
# only traffic director
mock.side_effect = [
# get traffic directors
self.traffic_directors_reponse,
self.traffic_directors_response,
# get traffic director
self.traffic_director_response,
# get zone
@@ -809,7 +811,7 @@ class TestDynProviderGeo(TestCase):
# both traffic director and regular, regular is ignored
mock.side_effect = [
# get traffic directors
self.traffic_directors_reponse,
self.traffic_directors_response,
# get traffic director
self.traffic_director_response,
# get zone
@@ -859,7 +861,7 @@ class TestDynProviderGeo(TestCase):
# busted traffic director
mock.side_effect = [
# get traffic directors
self.traffic_directors_reponse,
self.traffic_directors_response,
# get traffic director
busted_traffic_director_response,
# get zone
@@ -913,7 +915,7 @@ class TestDynProviderGeo(TestCase):
Delete(geo),
Delete(regular),
]
plan = Plan(None, desired, changes)
plan = Plan(None, desired, changes, True)
provider._apply(plan)
mock.assert_has_calls([
call('/Zone/unit.tests/', 'GET', {}),
@@ -932,14 +934,14 @@ class TestDynProviderGeo(TestCase):
provider = DynProvider('test', 'cust', 'user', 'pass',
traffic_directors_enabled=True)
# will be tested seperately
# will be tested separately
provider._mod_rulesets = MagicMock()
mock.side_effect = [
# create traffic director
self.traffic_director_response,
# get traffic directors
self.traffic_directors_reponse
self.traffic_directors_response
]
provider._mod_geo_Create(None, Create(self.geo_record))
# td now lives in cache
tests/test_octodns_provider_googlecloud.py
+13 -9
View File
@@ -263,7 +263,8 @@ class TestGoogleCloudProvider(TestCase):
provider.apply(Plan(
existing=[update_existing_r, delete_r],
desired=desired,
changes=changes
changes=changes,
exists=True
))
calls_mock = gcloud_zone_mock.changes.return_value
@@ -295,7 +296,8 @@ class TestGoogleCloudProvider(TestCase):
provider.apply(Plan(
existing=[update_existing_r, delete_r],
desired=desired,
changes=changes
changes=changes,
exists=True
))
unsupported_change = Mock()
@@ -357,15 +359,17 @@ class TestGoogleCloudProvider(TestCase):
"unit.tests.")
test_zone = Zone('unit.tests.', [])
provider.populate(test_zone)
exists = provider.populate(test_zone)
self.assertTrue(exists)
# test_zone gets fed the same records as zone does, except it's in
# the format returned by google API, so after populate they should look
# excactly the same.
# exactly the same.
self.assertEqual(test_zone.records, zone.records)
test_zone2 = Zone('nonexistant.zone.', [])
provider.populate(test_zone2, False, False)
test_zone2 = Zone('nonexistent.zone.', [])
exists = provider.populate(test_zone2, False, False)
self.assertFalse(exists)
self.assertEqual(len(test_zone2.records), 0,
msg="Zone should not get records from wrong domain")
@@ -401,8 +405,8 @@ class TestGoogleCloudProvider(TestCase):
provider.gcloud_client.list_zones = Mock(
return_value=DummyIterator([]))
self.assertIsNone(provider.gcloud_zones.get("nonexistant.xone"),
msg="Check that nonexistant zones return None when"
self.assertIsNone(provider.gcloud_zones.get("nonexistent.zone"),
msg="Check that nonexistent zones return None when"
"there's no create=True flag")
def test__get_rrsets(self):
@@ -423,7 +427,7 @@ class TestGoogleCloudProvider(TestCase):
provider.gcloud_client.list_zones = Mock(
return_value=DummyIterator([]))
mock_zone = provider._create_gcloud_zone("nonexistant.zone.mock")
mock_zone = provider._create_gcloud_zone("nonexistent.zone.mock")
mock_zone.create.assert_called()
provider.gcloud_client.zone.assert_called()
tests/test_octodns_provider_ns1.py
+3 -1
View File
@@ -196,9 +196,10 @@ class TestNs1Provider(TestCase):
load_mock.side_effect = \
ResourceException('server error: zone not found')
zone = Zone('unit.tests.', [])
provider.populate(zone)
exists = provider.populate(zone)
self.assertEquals(set(), zone.records)
self.assertEquals(('unit.tests',), load_mock.call_args[0])
self.assertFalse(exists)
# Existing zone w/o records
load_mock.reset_mock()
@@ -269,6 +270,7 @@ class TestNs1Provider(TestCase):
# everything except the root NS
expected_n = len(self.expected) - 1
self.assertEquals(expected_n, len(plan.changes))
self.assertTrue(plan.exists)
# Fails, general error
load_mock.reset_mock()
tests/test_octodns_provider_ovh.py
+19 -7
View File
@@ -8,7 +8,7 @@ from __future__ import absolute_import, division, print_function, \
from unittest import TestCase
from mock import patch, call
from ovh import APIError
from ovh import APIError, ResourceNotFoundError, InvalidCredential
from octodns.provider.ovh import OvhProvider
from octodns.record import Record
@@ -199,14 +199,14 @@ class TestOvhProvider(TestCase):
api_record.append({
'fieldType': 'SPF',
'ttl': 1000,
'target': 'v=spf1 include:unit.texts.rerirect ~all',
'target': 'v=spf1 include:unit.texts.redirect ~all',
'subDomain': '',
'id': 13
})
expected.add(Record.new(zone, '', {
'ttl': 1000,
'type': 'SPF',
'value': 'v=spf1 include:unit.texts.rerirect ~all'
'value': 'v=spf1 include:unit.texts.redirect ~all'
}))
# SSHFP
@@ -307,18 +307,30 @@ class TestOvhProvider(TestCase):
with patch.object(provider._client, 'get') as get_mock:
zone = Zone('unit.tests.', [])
get_mock.side_effect = APIError('boom')
get_mock.side_effect = ResourceNotFoundError('boom')
with self.assertRaises(APIError) as ctx:
provider.populate(zone)
self.assertEquals(get_mock.side_effect, ctx.exception)
with patch.object(provider._client, 'get') as get_mock:
get_mock.side_effect = InvalidCredential('boom')
with self.assertRaises(APIError) as ctx:
provider.populate(zone)
self.assertEquals(get_mock.side_effect, ctx.exception)
zone = Zone('unit.tests.', [])
get_mock.side_effect = ResourceNotFoundError('This service does '
'not exist')
exists = provider.populate(zone)
self.assertEquals(set(), zone.records)
self.assertFalse(exists)
zone = Zone('unit.tests.', [])
get_returns = [[record['id'] for record in self.api_record]]
get_returns += self.api_record
get_mock.side_effect = get_returns
provider.populate(zone)
exists = provider.populate(zone)
self.assertEquals(self.expected, zone.records)
self.assertTrue(exists)
@patch('ovh.Client')
def test_is_valid_dkim(self, client_mock):
@@ -404,7 +416,7 @@ class TestOvhProvider(TestCase):
call(u'/domain/zone/unit.tests/record', fieldType=u'SPF',
subDomain=u'', ttl=1000,
target=u'v=spf1 include:unit.texts.'
u'rerirect ~all',
u'redirect ~all',
),
call(u'/domain/zone/unit.tests/record', fieldType=u'A',
subDomain='sub', target=u'1.2.3.4', ttl=200),
tests/test_octodns_provider_powerdns.py
+4 -2
View File
@@ -100,12 +100,13 @@ class TestPowerDnsProvider(TestCase):
# No existing records -> creates for every record in expected
with requests_mock() as mock:
mock.get(ANY, status_code=200, text=EMPTY_TEXT)
# post 201, is reponse to the create with data
# post 201, is response to the create with data
mock.patch(ANY, status_code=201, text=assert_rrsets_callback)
plan = provider.plan(expected)
self.assertEquals(expected_n, len(plan.changes))
self.assertEquals(expected_n, provider.apply(plan))
self.assertTrue(plan.exists)
# Non-existent zone -> creates for every record in expected
# OMG this is fucking ugly, probably better to ditch requests_mocks and
@@ -118,12 +119,13 @@ class TestPowerDnsProvider(TestCase):
mock.get(ANY, status_code=422, text='')
# patch 422's, unknown zone
mock.patch(ANY, status_code=422, text=dumps(not_found))
# post 201, is reponse to the create with data
# post 201, is response to the create with data
mock.post(ANY, status_code=201, text=assert_rrsets_callback)
plan = provider.plan(expected)
self.assertEquals(expected_n, len(plan.changes))
self.assertEquals(expected_n, provider.apply(plan))
self.assertFalse(plan.exists)
with requests_mock() as mock:
# get 422's, unknown zone
tests/test_octodns_provider_rackspace.py
+3 -1
View File
@@ -73,9 +73,10 @@ class TestRackspaceProvider(TestCase):
json={'error': "Could not find domain 'unit.tests.'"})
zone = Zone('unit.tests.', [])
self.provider.populate(zone)
exists = self.provider.populate(zone)
self.assertEquals(set(), zone.records)
self.assertTrue(mock.called_once)
self.assertFalse(exists)
def test_multipage_populate(self):
with requests_mock() as mock:
@@ -109,6 +110,7 @@ class TestRackspaceProvider(TestCase):
plan = self.provider.plan(expected)
self.assertTrue(mock.called)
self.assertTrue(plan.exists)
# OctoDNS does not propagate top-level NS records.
self.assertEquals(1, len(plan.changes))
tests/test_octodns_provider_route53.py
+5 -3
View File
@@ -331,9 +331,9 @@ class TestRoute53Provider(TestCase):
stubber.assert_no_pending_responses()
# Populate a zone that doesn't exist
noexist = Zone('does.not.exist.', [])
provider.populate(noexist)
self.assertEquals(set(), noexist.records)
nonexistent = Zone('does.not.exist.', [])
provider.populate(nonexistent)
self.assertEquals(set(), nonexistent.records)
def test_sync(self):
provider, stubber = self._get_stubbed_provider()
@@ -361,6 +361,7 @@ class TestRoute53Provider(TestCase):
plan = provider.plan(self.expected)
self.assertEquals(9, len(plan.changes))
self.assertTrue(plan.exists)
for change in plan.changes:
self.assertIsInstance(change, Create)
stubber.assert_no_pending_responses()
@@ -593,6 +594,7 @@ class TestRoute53Provider(TestCase):
plan = provider.plan(self.expected)
self.assertEquals(9, len(plan.changes))
self.assertFalse(plan.exists)
for change in plan.changes:
self.assertIsInstance(change, Create)
stubber.assert_no_pending_responses()
tests/test_octodns_record.py
+2 -2
View File
@@ -430,7 +430,7 @@ class TestRecord(TestCase):
self.assertEqual(change.new, other)
# full sorting
# equivilent
# equivalent
b_naptr_value = b.values[0]
self.assertEquals(0, b_naptr_value.__cmp__(b_naptr_value))
# by order
@@ -710,7 +710,7 @@ class TestRecord(TestCase):
Record.new(self.zone, 'unknown', {})
self.assertTrue('missing type' in ctx.exception.message)
# Unkown type
# Unknown type
with self.assertRaises(Exception) as ctx:
Record.new(self.zone, 'unknown', {
'type': 'XXX',