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 pull request #850 from octodns/address-warnings

Browse Source 
Quell all python warnings, mostly *Equals -> Equal and warn -> warning
This commit is contained in:
Ross McFarland
2022-01-14 10:09:52 -08:00
committed by GitHub
co-authored by GitHub
parent db2f06abdd 1498ced567
commit 69e7632df3
44 changed files with 779 additions and 777 deletions
Download Patch File Download Diff File Expand all files Collapse all files
octodns/manager.py
+4 -4
View File
@@ -260,8 +260,8 @@ class Manager(object):
if ("unexpected keyword argument 'lenient'"
not in str(e)):
raise
self.log.warn('provider %s does not accept lenient '
'param', source.__class__.__name__)
self.log.warning('provider %s does not accept lenient '
'param', source.__class__.__name__)
source.populate(zone)
for processor in processors:
@@ -283,8 +283,8 @@ class Manager(object):
except TypeError as e:
if "keyword argument 'processors'" not in str(e):
raise
self.log.warn('provider.plan %s does not accept processors '
'param', target.__class__.__name__)
self.log.warning('provider.plan %s does not accept processors '
'param', target.__class__.__name__)
plan = target.plan(zone)
for processor in processors:
octodns/processor/awsacm.py
+3 -3
View File
@@ -9,9 +9,9 @@ from logging import getLogger
logger = getLogger('Route53')
try:
logger.warn('octodns_route53 shimmed. Update your processor class to '
'octodns_route53.processor.AwsAcmMangingProcessor. '
'Shim will be removed in 1.0')
logger.warning('octodns_route53 shimmed. Update your processor class to '
'octodns_route53.processor.AwsAcmMangingProcessor. '
'Shim will be removed in 1.0')
from octodns_route53.processor import AwsAcmMangingProcessor
AwsAcmMangingProcessor # pragma: no cover
except ModuleNotFoundError:
octodns/provider/azuredns.py
+3 -3
View File
@@ -9,9 +9,9 @@ from logging import getLogger
logger = getLogger('Azure')
try:
logger.warn('octodns_azure shimmed. Update your provider class to '
'octodns_azure.AzureProvider. '
'Shim will be removed in 1.0')
logger.warning('octodns_azure shimmed. Update your provider class to '
'octodns_azure.AzureProvider. '
'Shim will be removed in 1.0')
from octodns_azure import AzureProvider
AzureProvider # pragma: no cover
except ModuleNotFoundError:
octodns/provider/base.py
+2 -2
View File
@@ -132,8 +132,8 @@ class BaseProvider(BaseSource):
if exists is None:
# If your code gets this warning see Source.populate for more
# information
self.log.warn('Provider %s used in target mode did not return '
'exists', self.id)
self.log.warning('Provider %s used in target mode did not return '
'exists', self.id)
for processor in processors:
existing = processor.process_target_zone(existing, target=self)
octodns/provider/cloudflare.py
+3 -3
View File
@@ -9,9 +9,9 @@ from logging import getLogger
logger = getLogger('Cloudflare')
try:
logger.warn('octodns_cloudflare shimmed. Update your provider class to '
'octodns_cloudflare.CloudflareProvider. '
'Shim will be removed in 1.0')
logger.warning('octodns_cloudflare shimmed. Update your provider class to '
'octodns_cloudflare.CloudflareProvider. '
'Shim will be removed in 1.0')
from octodns_cloudflare import CloudflareProvider
CloudflareProvider # pragma: no cover
except ModuleNotFoundError:
octodns/provider/constellix.py
+3 -3
View File
@@ -9,9 +9,9 @@ from logging import getLogger
logger = getLogger('Constellix')
try:
logger.warn('octodns_constellix shimmed. Update your provider class to '
'octodns_constellix.ConstellixProvider. '
'Shim will be removed in 1.0')
logger.warning('octodns_constellix shimmed. Update your provider class to '
'octodns_constellix.ConstellixProvider. '
'Shim will be removed in 1.0')
from octodns_constellix import ConstellixProvider
ConstellixProvider # pragma: no cover
except ModuleNotFoundError:
octodns/provider/digitalocean.py
+3 -3
View File
@@ -9,9 +9,9 @@ from logging import getLogger
logger = getLogger('DigitalOcean')
try:
logger.warn('octodns_digitalocean shimmed. Update your provider class to '
'octodns_digitalocean.DigitalOceanProvider. '
'Shim will be removed in 1.0')
logger.warning('octodns_digitalocean shimmed. Update your provider class '
'to octodns_digitalocean.DigitalOceanProvider. Shim will '
'be removed in 1.0')
from octodns_digitalocean import DigitalOceanProvider
DigitalOceanProvider # pragma: no cover
except ModuleNotFoundError:
octodns/provider/dnsimple.py
+3 -3
View File
@@ -9,9 +9,9 @@ from logging import getLogger
logger = getLogger('Dnsimple')
try:
logger.warn('octodns_dnsimple shimmed. Update your provider class to '
'octodns_dnsimple.DnsimpleProvider. '
'Shim will be removed in 1.0')
logger.warning('octodns_dnsimple shimmed. Update your provider class to '
'octodns_dnsimple.DnsimpleProvider. '
'Shim will be removed in 1.0')
from octodns_dnsimple import DnsimpleProvider
DnsimpleProvider # pragma: no cover
except ModuleNotFoundError:
octodns/provider/dnsmadeeasy.py
+3 -3
View File
@@ -9,9 +9,9 @@ from logging import getLogger
logger = getLogger('DnsMadeEasy')
try:
logger.warn('octodns_dnsmadeeasy shimmed. Update your provider class to '
'octodns_dnsmadeeasy.DnsMadeEasyProvider. '
'Shim will be removed in 1.0')
logger.warning('octodns_dnsmadeeasy shimmed. Update your provider class '
'to octodns_dnsmadeeasy.DnsMadeEasyProvider. '
'Shim will be removed in 1.0')
from octodns_dnsmadeeasy import DnsMadeEasyProvider
DnsMadeEasyProvider # pragma: no cover
except ModuleNotFoundError:
octodns/provider/dyn.py
+3 -3
View File
@@ -9,9 +9,9 @@ from logging import getLogger
logger = getLogger('Dyn')
try:
logger.warn('octodns_dyn shimmed. Update your provider class to '
'octodns_dyn.DynProvider. '
'Shim will be removed in 1.0')
logger.warning('octodns_dyn shimmed. Update your provider class to '
'octodns_dyn.DynProvider. '
'Shim will be removed in 1.0')
from octodns_dyn import DynProvider
DynProvider # pragma: no cover
except ModuleNotFoundError:
octodns/provider/easydns.py
+3 -3
View File
@@ -9,9 +9,9 @@ from logging import getLogger
logger = getLogger('EasyDns')
try:
logger.warn('octodns_easydns shimmed. Update your provider class to '
'octodns_easydns.EasyDnsProvider. '
'Shim will be removed in 1.0')
logger.warning('octodns_easydns shimmed. Update your provider class to '
'octodns_easydns.EasyDnsProvider. '
'Shim will be removed in 1.0')
from octodns_easydns import EasyDnsProvider, EasyDNSProvider
EasyDnsProvider # pragma: no cover
EasyDNSProvider # pragma: no cover
octodns/provider/edgedns.py
+3 -3
View File
@@ -9,9 +9,9 @@ from logging import getLogger
logger = getLogger('Akamai')
try:
logger.warn('octodns_edgedns shimmed. Update your provider class to '
'octodns_edgedns.AkamaiProvider. '
'Shim will be removed in 1.0')
logger.warning('octodns_edgedns shimmed. Update your provider class to '
'octodns_edgedns.AkamaiProvider. '
'Shim will be removed in 1.0')
from octodns_edgedns import AkamaiProvider
AkamaiProvider # pragma: no cover
except ModuleNotFoundError:
octodns/provider/etc_hosts.py
+3 -3
View File
@@ -9,9 +9,9 @@ from logging import getLogger
logger = getLogger('EtcHosts')
try:
logger.warn('octodns_etchosts shimmed. Update your provider class to '
'octodns_etchosts.EtcHostsProvider. '
'Shim will be removed in 1.0')
logger.warning('octodns_etchosts shimmed. Update your provider class to '
'octodns_etchosts.EtcHostsProvider. '
'Shim will be removed in 1.0')
from octodns_etchosts import EtcHostsProvider
EtcHostsProvider # pragma: no cover
except ModuleNotFoundError:
octodns/provider/fastdns.py
+5 -5
View File
@@ -8,8 +8,8 @@ from __future__ import absolute_import, division, print_function, \
from logging import getLogger
logger = getLogger('Akamai')
logger.warn('AkamaiProvider has been moved into a seperate module, '
'octodns_edgedns is now required. Provider class should '
'be updated to octodns_edgedns.AkamaiProvider. See '
'https://github.com/octodns/octodns/README.md#updating-'
'to-use-extracted-providers for more information.')
logger.warning('AkamaiProvider has been moved into a seperate module, '
'octodns_edgedns is now required. Provider class should '
'be updated to octodns_edgedns.AkamaiProvider. See '
'https://github.com/octodns/octodns/README.md#updating-'
'to-use-extracted-providers for more information.')
octodns/provider/gandi.py
+3 -3
View File
@@ -9,9 +9,9 @@ from logging import getLogger
logger = getLogger('Gandi')
try:
logger.warn('octodns_gandi shimmed. Update your provider class to '
'octodns_gandi.GandiProvider. '
'Shim will be removed in 1.0')
logger.warning('octodns_gandi shimmed. Update your provider class to '
'octodns_gandi.GandiProvider. '
'Shim will be removed in 1.0')
from octodns_gandi import GandiProvider
GandiProvider # pragma: no cover
except ModuleNotFoundError:
octodns/provider/gcore.py
+3 -3
View File
@@ -9,9 +9,9 @@ from logging import getLogger
logger = getLogger('GCore')
try:
logger.warn('octodns_gcore shimmed. Update your provider class to '
'octodns_gcore.GCoreProvider. '
'Shim will be removed in 1.0')
logger.warning('octodns_gcore shimmed. Update your provider class to '
'octodns_gcore.GCoreProvider. '
'Shim will be removed in 1.0')
from octodns_gcore import GCoreProvider
GCoreProvider # pragma: no cover
except ModuleNotFoundError:
octodns/provider/googlecloud.py
+3 -3
View File
@@ -9,9 +9,9 @@ from logging import getLogger
logger = getLogger('GoogleCloud')
try:
logger.warn('octodns_googlecloud shimmed. Update your provider class to '
'octodns_googlecloud.GoogleCloudProvider. '
'Shim will be removed in 1.0')
logger.warning('octodns_googlecloud shimmed. Update your provider class '
'to octodns_googlecloud.GoogleCloudProvider. '
'Shim will be removed in 1.0')
from octodns_googlecloud import GoogleCloudProvider
GoogleCloudProvider # pragma: no cover
except ModuleNotFoundError:
octodns/provider/hetzner.py
+3 -3
View File
@@ -9,9 +9,9 @@ from logging import getLogger
logger = getLogger('Hetzner')
try:
logger.warn('octodns_hetzner shimmed. Update your provider class to '
'octodns_hetzner.HetznerProvider. '
'Shim will be removed in 1.0')
logger.warning('octodns_hetzner shimmed. Update your provider class to '
'octodns_hetzner.HetznerProvider. '
'Shim will be removed in 1.0')
from octodns_hetzner import HetznerProvider
HetznerProvider # pragma: no cover
except ModuleNotFoundError:
octodns/provider/mythicbeasts.py
+3 -3
View File
@@ -9,9 +9,9 @@ from logging import getLogger
logger = getLogger('MythicBeasts')
try:
logger.warn('octodns_mythicbeasts shimmed. Update your provider class to '
'octodns_mythicbeasts.MythicBeastsProvider. '
'Shim will be removed in 1.0')
logger.warning('octodns_mythicbeasts shimmed. Update your provider class '
'to octodns_mythicbeasts.MythicBeastsProvider. '
'Shim will be removed in 1.0')
from octodns_mythicbeasts import MythicBeastsProvider
MythicBeastsProvider # pragma: no cover
except ModuleNotFoundError:
octodns/provider/ns1.py
+3 -3
View File
@@ -9,9 +9,9 @@ from logging import getLogger
logger = getLogger('Ns1')
try:
logger.warn('octodns_ns1 shimmed. Update your provider class to '
'octodns_ns1.Ns1Provider. '
'Shim will be removed in 1.0')
logger.warning('octodns_ns1 shimmed. Update your provider class to '
'octodns_ns1.Ns1Provider. '
'Shim will be removed in 1.0')
from octodns_ns1 import Ns1Provider
Ns1Provider # pragma: no cover
except ModuleNotFoundError:
octodns/provider/powerdns.py
+3 -3
View File
@@ -9,9 +9,9 @@ from logging import getLogger
logger = getLogger('PowerDns')
try:
logger.warn('octodns_powerdns shimmed. Update your provider class to '
'octodns_powerdns.PowerDnsProvider. '
'Shim will be removed in 1.0')
logger.warning('octodns_powerdns shimmed. Update your provider class to '
'octodns_powerdns.PowerDnsProvider. '
'Shim will be removed in 1.0')
from octodns_powerdns import PowerDnsProvider, PowerDnsBaseProvider
PowerDnsProvider # pragma: no cover
PowerDnsBaseProvider # pragma: no cover
octodns/provider/route53.py
+3 -3
View File
@@ -9,9 +9,9 @@ from logging import getLogger
logger = getLogger('Route53')
try:
logger.warn('octodns_route53 shimmed. Update your provider class to '
'octodns_route53.Route53Provider. '
'Shim will be removed in 1.0')
logger.warning('octodns_route53 shimmed. Update your provider class to '
'octodns_route53.Route53Provider. '
'Shim will be removed in 1.0')
from octodns_route53 import Route53Provider
Route53Provider # pragma: no cover
except ModuleNotFoundError:
octodns/record/__init__.py
+2 -2
View File
@@ -116,7 +116,7 @@ class Record(EqualityTupleMixin):
pass
if reasons:
if lenient:
cls.log.warn(ValidationError.build_message(fqdn, reasons))
cls.log.warning(ValidationError.build_message(fqdn, reasons))
else:
raise ValidationError(fqdn, reasons)
return _class(zone, name, data, source=source)
@@ -429,7 +429,7 @@ class _DynamicPool(object):
if len(values) == 1:
weight = data['values'][0].get('weight', 1)
if weight != 1:
self.log.warn(
self.log.warning(
'Using weight=1 instead of %s for single-value pool %s',
weight, _id)
values[0]['weight'] = 1
octodns/record/geo.py
+3 -3
View File
@@ -57,7 +57,7 @@ class GeoCodes(object):
for continent, countries in geo_data.items():
if country in countries:
return f'{continent}-{country}'
cls.log.warn('country_to_code: unrecognized country "%s"', country)
cls.log.warning('country_to_code: unrecognized country "%s"', country)
return
@classmethod
@@ -66,8 +66,8 @@ class GeoCodes(object):
# NA-US, NA-CA
if (province not in geo_data['NA']['US']['provinces'] and
province not in geo_data['NA']['CA']['provinces']):
cls.log.warn('country_to_code: unrecognized province "%s"',
province)
cls.log.warning('country_to_code: unrecognized province "%s"',
province)
return
if province in geo_data['NA']['US']['provinces']:
country = 'US'
octodns/source/tinydns.py
+3 -1
View File
@@ -217,7 +217,9 @@ class TinyDnsBaseSource(BaseSource):
try:
zone.add_record(record, lenient=lenient)
except DuplicateRecordException:
self.log.warn(f'Duplicate PTR record for {addr}, skipping')
self.log.warning(
f'Duplicate PTR record for {addr}, skipping'
)
class TinyDnsFileSource(TinyDnsBaseSource):
tests/test_octodns_equality.py
+7 -7
View File
@@ -30,14 +30,14 @@ class TestEqualityTupleMixin(TestCase):
doesnt = Simple(2, 3, 4)
# equality
self.assertEquals(one, one)
self.assertEquals(one, same)
self.assertEquals(same, one)
self.assertEqual(one, one)
self.assertEqual(one, same)
self.assertEqual(same, one)
# only a & c are considered
self.assertEquals(one, matches)
self.assertEquals(matches, one)
self.assertNotEquals(one, doesnt)
self.assertNotEquals(doesnt, one)
self.assertEqual(one, matches)
self.assertEqual(matches, one)
self.assertNotEqual(one, doesnt)
self.assertNotEqual(doesnt, one)
# lt
self.assertTrue(one < doesnt)
tests/test_octodns_manager.py
+37 -37
View File
@@ -91,8 +91,8 @@ class TestManager(TestCase):
with self.assertRaises(ManagerException) as ctx:
name = 'bad-plan-output-missing-class.yaml'
Manager(get_config_filename(name)).sync()
self.assertEquals('plan_output bad is missing class',
str(ctx.exception))
self.assertEqual('plan_output bad is missing class',
str(ctx.exception))
def test_bad_plan_output_config(self):
with self.assertRaises(ManagerException) as ctx:
@@ -113,45 +113,45 @@ class TestManager(TestCase):
tc = Manager(get_config_filename('always-dry-run.yaml')) \
.sync(dry_run=False)
# only the stuff from subzone, unit.tests. is always-dry-run
self.assertEquals(3, tc)
self.assertEqual(3, tc)
def test_simple(self):
with TemporaryDirectory() as tmpdir:
environ['YAML_TMP_DIR'] = tmpdir.dirname
tc = Manager(get_config_filename('simple.yaml')) \
.sync(dry_run=False)
self.assertEquals(26, tc)
self.assertEqual(26, tc)
# try with just one of the zones
tc = Manager(get_config_filename('simple.yaml')) \
.sync(dry_run=False, eligible_zones=['unit.tests.'])
self.assertEquals(20, tc)
self.assertEqual(20, tc)
# the subzone, with 2 targets
tc = Manager(get_config_filename('simple.yaml')) \
.sync(dry_run=False, eligible_zones=['subzone.unit.tests.'])
self.assertEquals(6, tc)
self.assertEqual(6, tc)
# and finally the empty zone
tc = Manager(get_config_filename('simple.yaml')) \
.sync(dry_run=False, eligible_zones=['empty.'])
self.assertEquals(0, tc)
self.assertEqual(0, tc)
# Again with force
tc = Manager(get_config_filename('simple.yaml')) \
.sync(dry_run=False, force=True)
self.assertEquals(26, tc)
self.assertEqual(26, tc)
# Again with max_workers = 1
tc = Manager(get_config_filename('simple.yaml'), max_workers=1) \
.sync(dry_run=False, force=True)
self.assertEquals(26, tc)
self.assertEqual(26, tc)
# Include meta
tc = Manager(get_config_filename('simple.yaml'), max_workers=1,
include_meta=True) \
.sync(dry_run=False, force=True)
self.assertEquals(30, tc)
self.assertEqual(30, tc)
def test_eligible_sources(self):
with TemporaryDirectory() as tmpdir:
@@ -159,7 +159,7 @@ class TestManager(TestCase):
# Only allow a target that doesn't exist
tc = Manager(get_config_filename('simple.yaml')) \
.sync(eligible_sources=['foo'])
self.assertEquals(0, tc)
self.assertEqual(0, tc)
def test_eligible_targets(self):
with TemporaryDirectory() as tmpdir:
@@ -167,7 +167,7 @@ class TestManager(TestCase):
# Only allow a target that doesn't exist
tc = Manager(get_config_filename('simple.yaml')) \
.sync(eligible_targets=['foo'])
self.assertEquals(0, tc)
self.assertEqual(0, tc)
def test_aliases(self):
with TemporaryDirectory() as tmpdir:
@@ -175,31 +175,31 @@ class TestManager(TestCase):
# Alias zones with a valid target.
tc = Manager(get_config_filename('simple-alias-zone.yaml')) \
.sync()
self.assertEquals(0, tc)
self.assertEqual(0, tc)
# Alias zone with an invalid target.
with self.assertRaises(ManagerException) as ctx:
tc = Manager(get_config_filename('unknown-source-zone.yaml')) \
.sync()
self.assertEquals('Invalid alias zone alias.tests.: source zone '
'does-not-exists.tests. does not exist',
str(ctx.exception))
self.assertEqual('Invalid alias zone alias.tests.: source zone '
'does-not-exists.tests. does not exist',
str(ctx.exception))
# Alias zone that points to another alias zone.
with self.assertRaises(ManagerException) as ctx:
tc = Manager(get_config_filename('alias-zone-loop.yaml')) \
.sync()
self.assertEquals('Invalid alias zone alias-loop.tests.: source '
'zone alias.tests. is an alias zone',
str(ctx.exception))
self.assertEqual('Invalid alias zone alias-loop.tests.: source '
'zone alias.tests. is an alias zone',
str(ctx.exception))
# Sync an alias without the zone it refers to
with self.assertRaises(ManagerException) as ctx:
tc = Manager(get_config_filename('simple-alias-zone.yaml')) \
.sync(eligible_zones=["alias.tests."])
self.assertEquals('Zone alias.tests. cannot be sync without zone '
'unit.tests. sinced it is aliased',
str(ctx.exception))
self.assertEqual('Zone alias.tests. cannot be sync without zone '
'unit.tests. sinced it is aliased',
str(ctx.exception))
def test_compare(self):
with TemporaryDirectory() as tmpdir:
@@ -207,27 +207,27 @@ class TestManager(TestCase):
manager = Manager(get_config_filename('simple.yaml'))
# make sure this was pulled in from the config
self.assertEquals(2, manager._executor._max_workers)
self.assertEqual(2, manager._executor._max_workers)
changes = manager.compare(['in'], ['in'], 'unit.tests.')
self.assertEquals([], changes)
self.assertEqual([], changes)
# Create an empty unit.test zone config
with open(join(tmpdir.dirname, 'unit.tests.yaml'), 'w') as fh:
fh.write('---\n{}')
changes = manager.compare(['in'], ['dump'], 'unit.tests.')
self.assertEquals(20, len(changes))
self.assertEqual(20, len(changes))
# Compound sources with varying support
changes = manager.compare(['in', 'nosshfp'],
['dump'],
'unit.tests.')
self.assertEquals(19, len(changes))
self.assertEqual(19, len(changes))
with self.assertRaises(ManagerException) as ctx:
manager.compare(['nope'], ['dump'], 'unit.tests.')
self.assertEquals('Unknown source: nope', str(ctx.exception))
self.assertEqual('Unknown source: nope', str(ctx.exception))
def test_aggregate_target(self):
simple = SimpleProvider()
@@ -268,7 +268,7 @@ class TestManager(TestCase):
with self.assertRaises(ManagerException) as ctx:
manager.dump('unit.tests.', tmpdir.dirname, False, False,
'nope')
self.assertEquals('Unknown source: nope', str(ctx.exception))
self.assertEqual('Unknown source: nope', str(ctx.exception))
manager.dump('unit.tests.', tmpdir.dirname, False, False, 'in')
@@ -297,7 +297,7 @@ class TestManager(TestCase):
with self.assertRaises(ManagerException) as ctx:
manager.dump('unit.tests.', tmpdir.dirname, False, True,
'nope')
self.assertEquals('Unknown source: nope', str(ctx.exception))
self.assertEqual('Unknown source: nope', str(ctx.exception))
manager.dump('unit.tests.', tmpdir.dirname, False, True, 'in')
@@ -376,7 +376,7 @@ class TestManager(TestCase):
with self.assertRaises(TypeError) as ctx:
manager._populate_and_plan('unit.tests.', [], [OtherType()],
[])
self.assertEquals('something else', str(ctx.exception))
self.assertEqual('something else', str(ctx.exception))
def test_plan_processors_fallback(self):
with TemporaryDirectory() as tmpdir:
@@ -402,7 +402,7 @@ class TestManager(TestCase):
with self.assertRaises(TypeError) as ctx:
manager._populate_and_plan('unit.tests.', [], [],
[OtherType()])
self.assertEquals('something else', str(ctx.exception))
self.assertEqual('something else', str(ctx.exception))
@patch('octodns.manager.Manager._get_named_class')
def test_sync_passes_file_handle(self, mock):
@@ -425,7 +425,7 @@ class TestManager(TestCase):
def test_processor_config(self):
# Smoke test loading a valid config
manager = Manager(get_config_filename('processors.yaml'))
self.assertEquals(['noop'], list(manager.processors.keys()))
self.assertEqual(['noop'], list(manager.processors.keys()))
# This zone specifies a valid processor
manager.sync(['unit.tests.'])
@@ -469,7 +469,7 @@ class TestManager(TestCase):
plans, zone = manager._populate_and_plan('unit.tests.', [mock], [],
targets)
# Our mock was called and added the record
self.assertEquals(record, list(zone.records)[0])
self.assertEqual(record, list(zone.records)[0])
# We got a create for the thing added to the expected state (source)
self.assertIsInstance(plans[0][1].changes[0], Create)
@@ -515,10 +515,10 @@ class TestMainThreadExecutor(TestCase):
mte = MainThreadExecutor()
future = mte.submit(self.success, 42)
self.assertEquals(42, future.result())
self.assertEqual(42, future.result())
future = mte.submit(self.success, ret=43)
self.assertEquals(43, future.result())
self.assertEqual(43, future.result())
def test_exception(self):
mte = MainThreadExecutor()
@@ -527,12 +527,12 @@ class TestMainThreadExecutor(TestCase):
future = mte.submit(self.exception, e)
with self.assertRaises(Exception) as ctx:
future.result()
self.assertEquals(e, ctx.exception)
self.assertEqual(e, ctx.exception)
future = mte.submit(self.exception, e=e)
with self.assertRaises(Exception) as ctx:
future.result()
self.assertEquals(e, ctx.exception)
self.assertEqual(e, ctx.exception)
def success(self, ret):
return ret
tests/test_octodns_plan.py
+3 -3
View File
@@ -58,7 +58,7 @@ 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', str(ctx.exception))
self.assertEqual('Unsupported level: not-a-level', str(ctx.exception))
def test_create(self):
@@ -83,7 +83,7 @@ class TestPlanHtml(TestCase):
def test_empty(self):
out = StringIO()
PlanHtml('html').run([], fh=out)
self.assertEquals('<b>No changes were planned</b>', out.getvalue())
self.assertEqual('<b>No changes were planned</b>', out.getvalue())
def test_simple(self):
out = StringIO()
@@ -99,7 +99,7 @@ class TestPlanMarkdown(TestCase):
def test_empty(self):
out = StringIO()
PlanMarkdown('markdown').run([], fh=out)
self.assertEquals('## No changes were planned\n', out.getvalue())
self.assertEqual('## No changes were planned\n', out.getvalue())
def test_simple(self):
out = StringIO()
tests/test_octodns_processor_acme.py
+3 -3
View File
@@ -64,7 +64,7 @@ class TestAcmeMangingProcessor(TestCase):
source.add_record(records['managed'])
got = acme.process_source_zone(source)
self.assertEquals([
self.assertEqual([
'_acme-challenge.managed',
'_acme-challenge.not-txt',
'not-acme',
@@ -76,7 +76,7 @@ class TestAcmeMangingProcessor(TestCase):
break
self.assertTrue(managed)
# Ownership was marked with an extra value
self.assertEquals(['*octoDNS*', 'magic bit'], record.values)
self.assertEqual(['*octoDNS*', 'magic bit'], record.values)
existing = Zone(zone.name, [])
# Unrelated stuff that should be untouched
@@ -93,7 +93,7 @@ class TestAcmeMangingProcessor(TestCase):
existing.add_record(records['going-away'])
got = acme.process_target_zone(existing)
self.assertEquals([
self.assertEqual([
'_acme-challenge.going-away',
'_acme-challenge.managed',
'_acme-challenge.not-txt',
tests/test_octodns_processor_filter.py
+14 -14
View File
@@ -48,21 +48,21 @@ class TestTypeAllowListFilter(TestCase):
filter_a = TypeAllowlistFilter('only-a', set(('A')))
got = filter_a.process_source_zone(zone.copy())
self.assertEquals(['a', 'a2'], sorted([r.name for r in got.records]))
self.assertEqual(['a', 'a2'], sorted([r.name for r in got.records]))
filter_aaaa = TypeAllowlistFilter('only-aaaa', ('AAAA',))
got = filter_aaaa.process_source_zone(zone.copy())
self.assertEquals(['aaaa'], sorted([r.name for r in got.records]))
self.assertEqual(['aaaa'], sorted([r.name for r in got.records]))
filter_txt = TypeAllowlistFilter('only-txt', ['TXT'])
got = filter_txt.process_target_zone(zone.copy())
self.assertEquals(['txt', 'txt2'],
sorted([r.name for r in got.records]))
self.assertEqual(['txt', 'txt2'],
sorted([r.name for r in got.records]))
filter_a_aaaa = TypeAllowlistFilter('only-aaaa', set(('A', 'AAAA')))
got = filter_a_aaaa.process_target_zone(zone.copy())
self.assertEquals(['a', 'a2', 'aaaa'],
sorted([r.name for r in got.records]))
self.assertEqual(['a', 'a2', 'aaaa'],
sorted([r.name for r in got.records]))
class TestTypeRejectListFilter(TestCase):
@@ -71,20 +71,20 @@ class TestTypeRejectListFilter(TestCase):
filter_a = TypeRejectlistFilter('not-a', set(('A')))
got = filter_a.process_source_zone(zone.copy())
self.assertEquals(['aaaa', 'txt', 'txt2'],
sorted([r.name for r in got.records]))
self.assertEqual(['aaaa', 'txt', 'txt2'],
sorted([r.name for r in got.records]))
filter_aaaa = TypeRejectlistFilter('not-aaaa', ('AAAA',))
got = filter_aaaa.process_source_zone(zone.copy())
self.assertEquals(['a', 'a2', 'txt', 'txt2'],
sorted([r.name for r in got.records]))
self.assertEqual(['a', 'a2', 'txt', 'txt2'],
sorted([r.name for r in got.records]))
filter_txt = TypeRejectlistFilter('not-txt', ['TXT'])
got = filter_txt.process_target_zone(zone.copy())
self.assertEquals(['a', 'a2', 'aaaa'],
sorted([r.name for r in got.records]))
self.assertEqual(['a', 'a2', 'aaaa'],
sorted([r.name for r in got.records]))
filter_a_aaaa = TypeRejectlistFilter('not-a-aaaa', set(('A', 'AAAA')))
got = filter_a_aaaa.process_target_zone(zone.copy())
self.assertEquals(['txt', 'txt2'],
sorted([r.name for r in got.records]))
self.assertEqual(['txt', 'txt2'],
sorted([r.name for r in got.records]))
tests/test_octodns_processor_ownership.py
+8 -8
View File
@@ -56,7 +56,7 @@ class TestOwnershipProcessor(TestCase):
ownership = OwnershipProcessor('ownership')
got = ownership.process_source_zone(zone.copy())
self.assertEquals([
self.assertEqual([
'',
'*',
'_owner.a',
@@ -72,7 +72,7 @@ class TestOwnershipProcessor(TestCase):
found = False
for record in got.records:
if record.name.startswith(ownership.txt_name):
self.assertEquals([ownership.txt_value], record.values)
self.assertEqual([ownership.txt_value], record.values)
# test _is_ownership while we're in here
self.assertTrue(ownership._is_ownership(record))
found = True
@@ -92,12 +92,12 @@ class TestOwnershipProcessor(TestCase):
plan = provider.plan(ownership_added)
self.assertTrue(plan)
# Double the number of records
self.assertEquals(len(records) * 2, len(plan.changes))
self.assertEqual(len(records) * 2, len(plan.changes))
# Now process the plan, shouldn't make any changes, we're creating
# everything
got = ownership.process_plan(plan)
self.assertTrue(got)
self.assertEquals(len(records) * 2, len(got.changes))
self.assertEqual(len(records) * 2, len(got.changes))
# Something extra exists and doesn't have ownership TXT, leave it
# alone, we don't own it.
@@ -113,7 +113,7 @@ class TestOwnershipProcessor(TestCase):
# something we don't own
got = ownership.process_plan(plan)
self.assertTrue(got)
self.assertEquals(len(records) * 2, len(got.changes))
self.assertEqual(len(records) * 2, len(got.changes))
# Something extra exists and does have an ownership record so we will
# delete it...
@@ -123,7 +123,7 @@ class TestOwnershipProcessor(TestCase):
copy.add_record(record)
# New ownership, without the `the-a`
ownership_added = ownership.process_source_zone(copy)
self.assertEquals(len(records) * 2 - 2, len(ownership_added.records))
self.assertEqual(len(records) * 2 - 2, len(ownership_added.records))
plan = provider.plan(ownership_added)
# Fake the extra existing by adding the record, its ownership, and the
# two delete changes.
@@ -142,5 +142,5 @@ class TestOwnershipProcessor(TestCase):
# plan out, meaning the planned deletes were allowed to happen.
got = ownership.process_plan(plan)
self.assertTrue(got)
self.assertEquals(plan, got)
self.assertEquals(len(plan.changes), len(got.changes))
self.assertEqual(plan, got)
self.assertEqual(len(plan.changes), len(got.changes))
tests/test_octodns_provider_base.py
+20 -20
View File
@@ -77,16 +77,16 @@ class TestBaseProvider(TestCase):
def test_base_provider(self):
with self.assertRaises(NotImplementedError) as ctx:
BaseProvider('base')
self.assertEquals('Abstract base class, log property missing',
str(ctx.exception))
self.assertEqual('Abstract base class, log property missing',
str(ctx.exception))
class HasLog(BaseProvider):
log = getLogger('HasLog')
with self.assertRaises(NotImplementedError) as ctx:
HasLog('haslog')
self.assertEquals('Abstract base class, SUPPORTS_GEO property missing',
str(ctx.exception))
self.assertEqual('Abstract base class, SUPPORTS_GEO property missing',
str(ctx.exception))
class HasSupportsGeo(HasLog):
SUPPORTS_GEO = False
@@ -94,15 +94,15 @@ class TestBaseProvider(TestCase):
zone = Zone('unit.tests.', ['sub'])
with self.assertRaises(NotImplementedError) as ctx:
HasSupportsGeo('hassupportsgeo').populate(zone)
self.assertEquals('Abstract base class, SUPPORTS property missing',
str(ctx.exception))
self.assertEqual('Abstract base class, SUPPORTS property missing',
str(ctx.exception))
class HasSupports(HasSupportsGeo):
SUPPORTS = set(('A',))
with self.assertRaises(NotImplementedError) as ctx:
HasSupports('hassupports').populate(zone)
self.assertEquals('Abstract base class, populate method missing',
str(ctx.exception))
self.assertEqual('Abstract base class, populate method missing',
str(ctx.exception))
# SUPPORTS_DYNAMIC has a default/fallback
self.assertFalse(HasSupports('hassupports').SUPPORTS_DYNAMIC)
@@ -143,19 +143,19 @@ class TestBaseProvider(TestCase):
.supports(list(zone.records)[0]))
plan = HasPopulate('haspopulate').plan(zone)
self.assertEquals(3, len(plan.changes))
self.assertEqual(3, len(plan.changes))
with self.assertRaises(NotImplementedError) as ctx:
HasPopulate('haspopulate').apply(plan)
self.assertEquals('Abstract base class, _apply method missing',
str(ctx.exception))
self.assertEqual('Abstract base class, _apply method missing',
str(ctx.exception))
def test_plan(self):
ignored = Zone('unit.tests.', [])
# No change, thus no plan
provider = HelperProvider([])
self.assertEquals(None, provider.plan(ignored))
self.assertEqual(None, provider.plan(ignored))
record = Record.new(ignored, 'a', {
'ttl': 30,
@@ -165,7 +165,7 @@ class TestBaseProvider(TestCase):
provider = HelperProvider([Create(record)])
plan = provider.plan(ignored)
self.assertTrue(plan)
self.assertEquals(1, len(plan.changes))
self.assertEqual(1, len(plan.changes))
def test_plan_with_processors(self):
zone = Zone('unit.tests.', [])
@@ -181,11 +181,11 @@ class TestBaseProvider(TestCase):
tricky = TrickyProcessor('tricky', [record])
plan = provider.plan(zone, processors=[tricky])
self.assertTrue(plan)
self.assertEquals(1, len(plan.changes))
self.assertEqual(1, len(plan.changes))
self.assertIsInstance(plan.changes[0], Delete)
# Called processor stored its params
self.assertTrue(tricky.existing)
self.assertEquals(zone.name, tricky.existing.name)
self.assertEqual(zone.name, tricky.existing.name)
# Chain of processors happen one after the other
other = Record.new(zone, 'b', {
@@ -197,14 +197,14 @@ class TestBaseProvider(TestCase):
another = TrickyProcessor('tricky', [other])
plan = provider.plan(zone, processors=[tricky, another])
self.assertTrue(plan)
self.assertEquals(2, len(plan.changes))
self.assertEqual(2, len(plan.changes))
self.assertIsInstance(plan.changes[0], Delete)
self.assertIsInstance(plan.changes[1], Delete)
# 2nd processor stored its params, and we'll see the record the
# first one added
self.assertTrue(another.existing)
self.assertEquals(zone.name, another.existing.name)
self.assertEquals(1, len(another.existing.records))
self.assertEqual(zone.name, another.existing.name)
self.assertEqual(1, len(another.existing.records))
def test_apply(self):
ignored = Zone('unit.tests.', [])
@@ -219,7 +219,7 @@ class TestBaseProvider(TestCase):
provider.apply(plan)
provider.apply_disabled = False
self.assertEquals(1, provider.apply(plan))
self.assertEqual(1, provider.apply(plan))
def test_include_change(self):
zone = Zone('unit.tests.', [])
@@ -527,5 +527,5 @@ class TestBaseProvider(TestCase):
# Should log and not expect
with self.assertRaises(SupportsException) as ctx:
strict.supports_warn_or_except('Hello World!', 'Will not see')
self.assertEquals('minimal: Hello World!', str(ctx.exception))
self.assertEqual('minimal: Hello World!', str(ctx.exception))
strict.log.warning.assert_not_called()
tests/test_octodns_provider_rackspace.py
+4 -4
View File
@@ -62,7 +62,7 @@ class TestRackspaceProvider(TestCase):
with self.assertRaises(HTTPError) as ctx:
zone = Zone('unit.tests.', [])
self.provider.populate(zone)
self.assertEquals(502, ctx.exception.response.status_code)
self.assertEqual(502, ctx.exception.response.status_code)
self.assertTrue(mock.called_once)
def test_nonexistent_zone(self):
@@ -73,7 +73,7 @@ class TestRackspaceProvider(TestCase):
zone = Zone('unit.tests.', [])
exists = self.provider.populate(zone)
self.assertEquals(set(), zone.records)
self.assertEqual(set(), zone.records)
self.assertTrue(mock.called_once)
self.assertFalse(exists)
@@ -88,7 +88,7 @@ class TestRackspaceProvider(TestCase):
zone = Zone('unit.tests.', [])
self.provider.populate(zone)
self.assertEquals(5, len(zone.records))
self.assertEqual(5, len(zone.records))
def test_plan_disappearing_ns_records(self):
expected = Zone('unit.tests.', [])
@@ -112,7 +112,7 @@ class TestRackspaceProvider(TestCase):
self.assertTrue(plan.exists)
# OctoDNS does not propagate top-level NS records.
self.assertEquals(1, len(plan.changes))
self.assertEqual(1, len(plan.changes))
def test_fqdn_a_record(self):
expected = Zone('example.com.', [])
tests/test_octodns_provider_selectel.py
+10 -10
View File
@@ -206,7 +206,7 @@ class TestSelectelProvider(TestCase):
provider = SelectelProvider(123, 'secret_token')
provider.populate(zone)
self.assertEquals(self.expected, zone.records)
self.assertEqual(self.expected, zone.records)
@requests_mock.Mocker()
def test_populate_invalid_record(self, fake_http):
@@ -264,8 +264,8 @@ class TestSelectelProvider(TestCase):
zone.add_record(record)
plan = provider.plan(zone)
self.assertEquals(8, len(plan.changes))
self.assertEquals(8, provider.apply(plan))
self.assertEqual(8, len(plan.changes))
self.assertEqual(8, provider.apply(plan))
@requests_mock.Mocker()
def test_domain_list(self, fake_http):
@@ -277,7 +277,7 @@ class TestSelectelProvider(TestCase):
provider = SelectelProvider(123, 'test_token')
result = provider.domain_list()
self.assertEquals(result, expected)
self.assertEqual(result, expected)
@requests_mock.Mocker()
def test_authentication_fail(self, fake_http):
@@ -287,8 +287,8 @@ class TestSelectelProvider(TestCase):
with self.assertRaises(Exception) as ctx:
SelectelProvider(123, 'fail_token')
self.assertEquals(str(ctx.exception),
'Authorization failed. Invalid or empty token.')
self.assertEqual(str(ctx.exception),
'Authorization failed. Invalid or empty token.')
@requests_mock.Mocker()
def test_not_exist_domain(self, fake_http):
@@ -312,8 +312,8 @@ class TestSelectelProvider(TestCase):
zone.add_record(record)
plan = provider.plan(zone)
self.assertEquals(8, len(plan.changes))
self.assertEquals(8, provider.apply(plan))
self.assertEqual(8, len(plan.changes))
self.assertEqual(8, provider.apply(plan))
@requests_mock.Mocker()
def test_delete_no_exist_record(self, fake_http):
@@ -365,8 +365,8 @@ class TestSelectelProvider(TestCase):
zone.add_record(record)
plan = provider.plan(zone)
self.assertEquals(8, len(plan.changes))
self.assertEquals(8, provider.apply(plan))
self.assertEqual(8, len(plan.changes))
self.assertEqual(8, provider.apply(plan))
@requests_mock.Mocker()
def test_include_change_returns_false(self, fake_http):
tests/test_octodns_provider_transip.py
+6 -6
View File
@@ -66,7 +66,7 @@ class TestTransipProvider(TestCase):
with self.assertRaises(TransipConfigException) as ctx:
TransipProvider("test", "unittest")
self.assertEquals(
self.assertEqual(
"Missing `key` or `key_file` parameter in config",
str(ctx.exception),
)
@@ -365,7 +365,7 @@ class TestTransipProvider(TestCase):
{"name": "@", "expire": 300, "type": "A", "content": "1.2.3.5"},
{"name": "www", "expire": 300, "type": "A", "content": "2.2.3.6"},
]
self.assertEquals(
self.assertEqual(
sorted(seen_entries, key=itemgetter("name", "type", "expire")),
sorted(expected_entries, key=itemgetter("name", "type", "expire")),
)
@@ -408,12 +408,12 @@ class TestTransipProvider(TestCase):
class TestParseFQDN(TestCase):
def test_parse_fqdn(self):
zone = Zone("unit.tests.", [])
self.assertEquals("www.unit.tests.", _parse_to_fqdn("www", zone))
self.assertEquals(
self.assertEqual("www.unit.tests.", _parse_to_fqdn("www", zone))
self.assertEqual(
"www.unit.tests.", _parse_to_fqdn("www.unit.tests.", zone)
)
self.assertEquals(
self.assertEqual(
"www.sub.sub.sub.unit.tests.",
_parse_to_fqdn("www.sub.sub.sub", zone),
)
self.assertEquals("unit.tests.", _parse_to_fqdn("@", zone))
self.assertEqual("unit.tests.", _parse_to_fqdn("@", zone))
tests/test_octodns_provider_ultra.py
+24 -24
View File
@@ -44,7 +44,7 @@ class TestUltraProvider(TestCase):
text='{"errorCode": 60001}')
with self.assertRaises(Exception) as ctx:
UltraProvider('test', 'account', 'user', 'wrongpass')
self.assertEquals('Unauthorized', str(ctx.exception))
self.assertEqual('Unauthorized', str(ctx.exception))
# Good Auth
with requests_mock() as mock:
@@ -54,11 +54,11 @@ class TestUltraProvider(TestCase):
text='{"token type": "Bearer", "refresh_token": "abc", '
'"access_token":"123", "expires_in": "3600"}')
UltraProvider('test', 'account', 'user', 'rightpass')
self.assertEquals(1, mock.call_count)
self.assertEqual(1, mock.call_count)
expected_payload = "grant_type=password&username=user&"\
"password=rightpass"
self.assertEquals(parse_qs(mock.last_request.text),
parse_qs(expected_payload))
self.assertEqual(parse_qs(mock.last_request.text),
parse_qs(expected_payload))
def test_get_zones(self):
provider = _get_provider()
@@ -71,7 +71,7 @@ class TestUltraProvider(TestCase):
"errorMessage": "Authorization Header required"})
with self.assertRaises(HTTPError) as ctx:
zones = provider.zones
self.assertEquals(400, ctx.exception.response.status_code)
self.assertEqual(400, ctx.exception.response.status_code)
# Test no zones exist error
with requests_mock() as mock:
@@ -79,8 +79,8 @@ class TestUltraProvider(TestCase):
headers={'Authorization': 'Bearer 123'},
json=self.empty_body)
zones = provider.zones
self.assertEquals(1, mock.call_count)
self.assertEquals(list(), zones)
self.assertEqual(1, mock.call_count)
self.assertEqual(list(), zones)
# Reset zone cache so they are queried again
provider._zones = None
@@ -112,9 +112,9 @@ class TestUltraProvider(TestCase):
headers={'Authorization': 'Bearer 123'},
json=payload)
zones = provider.zones
self.assertEquals(1, mock.call_count)
self.assertEquals(1, len(zones))
self.assertEquals('testzone123.com.', zones[0])
self.assertEqual(1, mock.call_count)
self.assertEqual(1, len(zones))
self.assertEqual('testzone123.com.', zones[0])
# Test different paging behavior
provider._zones = None
@@ -132,7 +132,7 @@ class TestUltraProvider(TestCase):
"returnedCount": 5},
"zones": []})
zones = provider.zones
self.assertEquals(2, mock.call_count)
self.assertEqual(2, mock.call_count)
def test_request(self):
provider = _get_provider()
@@ -144,7 +144,7 @@ class TestUltraProvider(TestCase):
headers={'Authorization': 'Bearer 123'}, json={})
with self.assertRaises(Exception) as ctx:
provider._get(path)
self.assertEquals('Unauthorized', str(ctx.exception))
self.assertEqual('Unauthorized', str(ctx.exception))
# Test all GET patterns
with requests_mock() as mock:
@@ -227,11 +227,11 @@ class TestUltraProvider(TestCase):
zone = Zone('octodns1.test.', [])
self.assertTrue(provider.zone_records(zone))
self.assertEquals(mock.call_count, 2)
self.assertEqual(mock.call_count, 2)
# Populate the same zone again and confirm cache is hit
self.assertTrue(provider.zone_records(zone))
self.assertEquals(mock.call_count, 2)
self.assertEqual(mock.call_count, 2)
def test_populate(self):
provider = _get_provider()
@@ -242,13 +242,13 @@ class TestUltraProvider(TestCase):
zone = Zone('unit.tests.', [])
provider.populate(zone)
self.assertEquals(set(), zone.records)
self.assertEqual(set(), zone.records)
# re-populating the same non-existent zone uses cache and makes no
# calls
again = Zone('unit.tests.', [])
provider.populate(again)
self.assertEquals(set(), again.records)
self.assertEqual(set(), again.records)
# Test zones with data
provider._zones = None
@@ -272,9 +272,9 @@ class TestUltraProvider(TestCase):
zone = Zone('octodns1.test.', [])
self.assertTrue(provider.populate(zone))
self.assertEquals('octodns1.test.', zone.name)
self.assertEquals(12, len(zone.records))
self.assertEquals(4, mock.call_count)
self.assertEqual('octodns1.test.', zone.name)
self.assertEqual(12, len(zone.records))
self.assertEqual(4, mock.call_count)
def test_apply(self):
provider = _get_provider()
@@ -288,8 +288,8 @@ class TestUltraProvider(TestCase):
# non-existent zone, create everything
plan = provider.plan(self.expected)
self.assertEquals(15, len(plan.changes))
self.assertEquals(15, provider.apply(plan))
self.assertEqual(15, len(plan.changes))
self.assertEqual(15, provider.apply(plan))
self.assertFalse(plan.exists)
provider._request.assert_has_calls([
@@ -319,7 +319,7 @@ class TestUltraProvider(TestCase):
'p=A/kinda+of/long/string+with+numb3rs']}),
], True)
# expected number of total calls
self.assertEquals(17, provider._request.call_count)
self.assertEqual(17, provider._request.call_count)
# Create sample rrset payload to attempt to alter
page1 = json_load(open('tests/fixtures/ultra-records-page-1.json'))
@@ -351,8 +351,8 @@ class TestUltraProvider(TestCase):
}))
plan = provider.plan(wanted)
self.assertEquals(11, len(plan.changes))
self.assertEquals(11, provider.apply(plan))
self.assertEqual(11, len(plan.changes))
self.assertEqual(11, provider.apply(plan))
self.assertTrue(plan.exists)
provider._request.assert_has_calls([
tests/test_octodns_provider_yaml.py
+33 -33
View File
@@ -30,14 +30,14 @@ class TestYamlProvider(TestCase):
# With target we don't add anything
source.populate(zone, target=source)
self.assertEquals(0, len(zone.records))
self.assertEqual(0, len(zone.records))
# without it we see everything
source.populate(zone)
self.assertEquals(23, len(zone.records))
self.assertEqual(23, len(zone.records))
source.populate(dynamic_zone)
self.assertEquals(6, len(dynamic_zone.records))
self.assertEqual(6, len(dynamic_zone.records))
# Assumption here is that a clean round-trip means that everything
# worked as expected, data that went in came back out and could be
@@ -57,21 +57,21 @@ class TestYamlProvider(TestCase):
# We add everything
plan = target.plan(zone)
self.assertEquals(20, len([c for c in plan.changes
if isinstance(c, Create)]))
self.assertEqual(20, len([c for c in plan.changes
if isinstance(c, Create)]))
self.assertFalse(isfile(yaml_file))
# Now actually do it
self.assertEquals(20, target.apply(plan))
self.assertEqual(20, target.apply(plan))
self.assertTrue(isfile(yaml_file))
# Dynamic plan
plan = target.plan(dynamic_zone)
self.assertEquals(6, len([c for c in plan.changes
if isinstance(c, Create)]))
self.assertEqual(6, len([c for c in plan.changes
if isinstance(c, Create)]))
self.assertFalse(isfile(dynamic_yaml_file))
# Apply it
self.assertEquals(6, target.apply(plan))
self.assertEqual(6, target.apply(plan))
self.assertTrue(isfile(dynamic_yaml_file))
# There should be no changes after the round trip
@@ -86,8 +86,8 @@ class TestYamlProvider(TestCase):
# A 2nd sync should still create everything
plan = target.plan(zone)
self.assertEquals(20, len([c for c in plan.changes
if isinstance(c, Create)]))
self.assertEqual(20, len([c for c in plan.changes
if isinstance(c, Create)]))
with open(yaml_file) as fh:
data = safe_load(fh.read())
@@ -120,7 +120,7 @@ class TestYamlProvider(TestCase):
self.assertTrue('value' in data.pop('www.sub'))
# make sure nothing is left
self.assertEquals([], list(data.keys()))
self.assertEqual([], list(data.keys()))
with open(dynamic_yaml_file) as fh:
data = safe_load(fh.read())
@@ -153,7 +153,7 @@ class TestYamlProvider(TestCase):
# self.assertTrue('dynamic' in dyna)
# make sure nothing is left
self.assertEquals([], list(data.keys()))
self.assertEqual([], list(data.keys()))
def test_empty(self):
source = YamlProvider('test', join(dirname(__file__), 'config'))
@@ -162,7 +162,7 @@ class TestYamlProvider(TestCase):
# without it we see everything
source.populate(zone)
self.assertEquals(0, len(zone.records))
self.assertEqual(0, len(zone.records))
def test_unsorted(self):
source = YamlProvider('test', join(dirname(__file__), 'config'))
@@ -185,8 +185,8 @@ class TestYamlProvider(TestCase):
zone = Zone('unit.tests.', ['sub'])
with self.assertRaises(SubzoneRecordException) as ctx:
source.populate(zone)
self.assertEquals('Record www.sub.unit.tests. is under a managed '
'subzone', str(ctx.exception))
self.assertEqual('Record www.sub.unit.tests. is under a managed '
'subzone', str(ctx.exception))
class TestSplitYamlProvider(TestCase):
@@ -244,14 +244,14 @@ class TestSplitYamlProvider(TestCase):
# With target we don't add anything
source.populate(zone, target=source)
self.assertEquals(0, len(zone.records))
self.assertEqual(0, len(zone.records))
# without it we see everything
source.populate(zone)
self.assertEquals(20, len(zone.records))
self.assertEqual(20, len(zone.records))
source.populate(dynamic_zone)
self.assertEquals(5, len(dynamic_zone.records))
self.assertEqual(5, len(dynamic_zone.records))
with TemporaryDirectory() as td:
# Add some subdirs to make sure that it can create them
@@ -263,20 +263,20 @@ class TestSplitYamlProvider(TestCase):
# We add everything
plan = target.plan(zone)
self.assertEquals(17, len([c for c in plan.changes
if isinstance(c, Create)]))
self.assertEqual(17, len([c for c in plan.changes
if isinstance(c, Create)]))
self.assertFalse(isdir(zone_dir))
# Now actually do it
self.assertEquals(17, target.apply(plan))
self.assertEqual(17, target.apply(plan))
# Dynamic plan
plan = target.plan(dynamic_zone)
self.assertEquals(5, len([c for c in plan.changes
if isinstance(c, Create)]))
self.assertEqual(5, len([c for c in plan.changes
if isinstance(c, Create)]))
self.assertFalse(isdir(dynamic_zone_dir))
# Apply it
self.assertEquals(5, target.apply(plan))
self.assertEqual(5, target.apply(plan))
self.assertTrue(isdir(dynamic_zone_dir))
# There should be no changes after the round trip
@@ -291,8 +291,8 @@ class TestSplitYamlProvider(TestCase):
# A 2nd sync should still create everything
plan = target.plan(zone)
self.assertEquals(17, len([c for c in plan.changes
if isinstance(c, Create)]))
self.assertEqual(17, len([c for c in plan.changes
if isinstance(c, Create)]))
yaml_file = join(zone_dir, '$unit.tests.yaml')
self.assertTrue(isfile(yaml_file))
@@ -353,7 +353,7 @@ class TestSplitYamlProvider(TestCase):
# without it we see everything
source.populate(zone)
self.assertEquals(0, len(zone.records))
self.assertEqual(0, len(zone.records))
def test_unsorted(self):
source = SplitYamlProvider(
@@ -383,8 +383,8 @@ class TestSplitYamlProvider(TestCase):
zone = Zone('unit.tests.', ['sub'])
with self.assertRaises(SubzoneRecordException) as ctx:
source.populate(zone)
self.assertEquals('Record www.sub.unit.tests. is under a managed '
'subzone', str(ctx.exception))
self.assertEqual('Record www.sub.unit.tests. is under a managed '
'subzone', str(ctx.exception))
class TestOverridingYamlProvider(TestCase):
@@ -401,7 +401,7 @@ class TestOverridingYamlProvider(TestCase):
# Load the base, should see the 5 records
base.populate(zone)
got = {r.name: r for r in zone.records}
self.assertEquals(6, len(got))
self.assertEqual(6, len(got))
# We get the "dynamic" A from the base config
self.assertTrue('dynamic' in got['a'].data)
# No added
@@ -410,9 +410,9 @@ class TestOverridingYamlProvider(TestCase):
# Load the overrides, should replace one and add 1
override.populate(zone)
got = {r.name: r for r in zone.records}
self.assertEquals(7, len(got))
self.assertEqual(7, len(got))
# 'a' was replaced with a generic record
self.assertEquals({
self.assertEqual({
'ttl': 3600,
'values': ['4.4.4.4', '5.5.5.5']
}, got['a'].data)
tests/test_octodns_record.py
+435 -435
View File
File diff suppressed because it is too large Load Diff
tests/test_octodns_record_geo.py
+32 -32
View File
@@ -16,67 +16,67 @@ class TestRecordGeoCodes(TestCase):
prefix = 'xyz '
# All valid
self.assertEquals([], GeoCodes.validate('NA', prefix))
self.assertEquals([], GeoCodes.validate('NA-US', prefix))
self.assertEquals([], GeoCodes.validate('NA-US-OR', prefix))
self.assertEqual([], GeoCodes.validate('NA', prefix))
self.assertEqual([], GeoCodes.validate('NA-US', prefix))
self.assertEqual([], GeoCodes.validate('NA-US-OR', prefix))
# Just plain bad
self.assertEquals(['xyz invalid geo code "XX-YY-ZZ-AA"'],
GeoCodes.validate('XX-YY-ZZ-AA', prefix))
self.assertEquals(['xyz unknown continent code "X-Y-Z"'],
GeoCodes.validate('X-Y-Z', prefix))
self.assertEquals(['xyz unknown continent code "XXX-Y-Z"'],
GeoCodes.validate('XXX-Y-Z', prefix))
self.assertEqual(['xyz invalid geo code "XX-YY-ZZ-AA"'],
GeoCodes.validate('XX-YY-ZZ-AA', prefix))
self.assertEqual(['xyz unknown continent code "X-Y-Z"'],
GeoCodes.validate('X-Y-Z', prefix))
self.assertEqual(['xyz unknown continent code "XXX-Y-Z"'],
GeoCodes.validate('XXX-Y-Z', prefix))
# Bad continent
self.assertEquals(['xyz unknown continent code "XX"'],
GeoCodes.validate('XX', prefix))
self.assertEqual(['xyz unknown continent code "XX"'],
GeoCodes.validate('XX', prefix))
# Bad continent good country
self.assertEquals(['xyz unknown continent code "XX-US"'],
GeoCodes.validate('XX-US', prefix))
self.assertEqual(['xyz unknown continent code "XX-US"'],
GeoCodes.validate('XX-US', prefix))
# Bad continent good country and province
self.assertEquals(['xyz unknown continent code "XX-US-OR"'],
GeoCodes.validate('XX-US-OR', prefix))
self.assertEqual(['xyz unknown continent code "XX-US-OR"'],
GeoCodes.validate('XX-US-OR', prefix))
# Bad country, good continent
self.assertEquals(['xyz unknown country code "NA-XX"'],
GeoCodes.validate('NA-XX', prefix))
self.assertEqual(['xyz unknown country code "NA-XX"'],
GeoCodes.validate('NA-XX', prefix))
# Bad country, good continent and state
self.assertEquals(['xyz unknown country code "NA-XX-OR"'],
GeoCodes.validate('NA-XX-OR', prefix))
self.assertEqual(['xyz unknown country code "NA-XX-OR"'],
GeoCodes.validate('NA-XX-OR', prefix))
# Good country, good continent, but bad match
self.assertEquals(['xyz unknown country code "NA-GB"'],
GeoCodes.validate('NA-GB', prefix))
self.assertEqual(['xyz unknown country code "NA-GB"'],
GeoCodes.validate('NA-GB', prefix))
# Bad province code, good continent and country
self.assertEquals(['xyz unknown province code "NA-US-XX"'],
GeoCodes.validate('NA-US-XX', prefix))
self.assertEqual(['xyz unknown province code "NA-US-XX"'],
GeoCodes.validate('NA-US-XX', prefix))
def test_parse(self):
self.assertEquals({
self.assertEqual({
'continent_code': 'NA',
'country_code': None,
'province_code': None,
}, GeoCodes.parse('NA'))
self.assertEquals({
self.assertEqual({
'continent_code': 'NA',
'country_code': 'US',
'province_code': None,
}, GeoCodes.parse('NA-US'))
self.assertEquals({
self.assertEqual({
'continent_code': 'NA',
'country_code': 'US',
'province_code': 'CA',
}, GeoCodes.parse('NA-US-CA'))
def test_country_to_code(self):
self.assertEquals('NA-US', GeoCodes.country_to_code('US'))
self.assertEquals('EU-GB', GeoCodes.country_to_code('GB'))
self.assertEqual('NA-US', GeoCodes.country_to_code('US'))
self.assertEqual('EU-GB', GeoCodes.country_to_code('GB'))
self.assertFalse(GeoCodes.country_to_code('XX'))
def test_province_to_code(self):
self.assertEquals('NA-US-OR', GeoCodes.province_to_code('OR'))
self.assertEquals('NA-US-KY', GeoCodes.province_to_code('KY'))
self.assertEquals('NA-CA-AB', GeoCodes.province_to_code('AB'))
self.assertEquals('NA-CA-BC', GeoCodes.province_to_code('BC'))
self.assertEqual('NA-US-OR', GeoCodes.province_to_code('OR'))
self.assertEqual('NA-US-KY', GeoCodes.province_to_code('KY'))
self.assertEqual('NA-CA-AB', GeoCodes.province_to_code('AB'))
self.assertEqual('NA-CA-BC', GeoCodes.province_to_code('BC'))
self.assertFalse(GeoCodes.province_to_code('XX'))
tests/test_octodns_source_axfr.py
+14 -14
View File
@@ -35,13 +35,13 @@ class TestAxfrSource(TestCase):
]
self.source.populate(got)
self.assertEquals(15, len(got.records))
self.assertEqual(15, len(got.records))
with self.assertRaises(AxfrSourceZoneTransferFailed) as ctx:
zone = Zone('unit.tests.', [])
self.source.populate(zone)
self.assertEquals('Unable to Perform Zone Transfer',
str(ctx.exception))
self.assertEqual('Unable to Perform Zone Transfer',
str(ctx.exception))
class TestZoneFileSource(TestCase):
@@ -52,7 +52,7 @@ class TestZoneFileSource(TestCase):
# Load zonefiles with a specified file extension
valid = Zone('ext.unit.tests.', [])
source.populate(valid)
self.assertEquals(1, len(valid.records))
self.assertEqual(1, len(valid.records))
def test_zonefiles_without_extension(self):
# Windows doesn't let files end with a `.` so we add a .tst to them in
@@ -72,18 +72,18 @@ class TestZoneFileSource(TestCase):
# Load zonefiles without a specified file extension
valid = Zone('unit.tests.', [])
source.populate(valid)
self.assertEquals(15, len(valid.records))
self.assertEqual(15, len(valid.records))
def test_populate(self):
# Valid zone file in directory
valid = Zone('unit.tests.', [])
self.source.populate(valid)
self.assertEquals(15, len(valid.records))
self.assertEqual(15, len(valid.records))
# 2nd populate does not read file again
again = Zone('unit.tests.', [])
self.source.populate(again)
self.assertEquals(15, len(again.records))
self.assertEqual(15, len(again.records))
# bust the cache
del self.source._zone_records[valid.name]
@@ -91,24 +91,24 @@ class TestZoneFileSource(TestCase):
# No zone file in directory
missing = Zone('missing.zone.', [])
self.source.populate(missing)
self.assertEquals(0, len(missing.records))
self.assertEqual(0, len(missing.records))
# Zone file is not valid
with self.assertRaises(ZoneFileSourceLoadFailure) as ctx:
zone = Zone('invalid.zone.', [])
self.source.populate(zone)
self.assertEquals('The DNS zone has no NS RRset at its origin.',
str(ctx.exception))
self.assertEqual('The DNS zone has no NS RRset at its origin.',
str(ctx.exception))
# Records are not to RFC (lenient=False)
with self.assertRaises(ValidationError) as ctx:
zone = Zone('invalid.records.', [])
self.source.populate(zone)
self.assertEquals('Invalid record _invalid.invalid.records.\n'
' - invalid name for SRV record',
str(ctx.exception))
self.assertEqual('Invalid record _invalid.invalid.records.\n'
' - invalid name for SRV record',
str(ctx.exception))
# Records are not to RFC, but load anyhow (lenient=True)
invalid = Zone('invalid.records.', [])
self.source.populate(invalid, lenient=True)
self.assertEquals(12, len(invalid.records))
self.assertEqual(12, len(invalid.records))
tests/test_octodns_source_envvar.py
+8 -8
View File
@@ -14,11 +14,11 @@ class TestEnvVarSource(TestCase):
with self.assertRaises(EnvironmentVariableNotFoundException) as ctx:
source._read_variable()
msg = f'Unknown environment variable {envvar}'
self.assertEquals(msg, str(ctx.exception))
self.assertEqual(msg, str(ctx.exception))
with patch.dict('os.environ', {envvar: 'testvalue'}):
value = source._read_variable()
self.assertEquals(value, 'testvalue')
self.assertEqual(value, 'testvalue')
def test_populate(self):
envvar = 'TEST_VAR'
@@ -31,10 +31,10 @@ class TestEnvVarSource(TestCase):
with patch.dict('os.environ', {envvar: value}):
source.populate(zone)
self.assertEquals(1, len(zone.records))
self.assertEqual(1, len(zone.records))
record = list(zone.records)[0]
self.assertEquals(name, record.name)
self.assertEquals(f'{name}.{zone_name}', record.fqdn)
self.assertEquals('TXT', record._type)
self.assertEquals(1, len(record.values))
self.assertEquals(value, record.values[0])
self.assertEqual(name, record.name)
self.assertEqual(f'{name}.{zone_name}', record.fqdn)
self.assertEqual('TXT', record._type)
self.assertEqual(1, len(record.values))
self.assertEqual(value, record.values[0])
tests/test_octodns_source_tinydns.py
+8 -8
View File
@@ -20,7 +20,7 @@ class TestTinyDnsFileSource(TestCase):
def test_populate_normal(self):
got = Zone('example.com.', [])
self.source.populate(got)
self.assertEquals(17, len(got.records))
self.assertEqual(17, len(got.records))
expected = Zone('example.com.', [])
for name, data in (
@@ -121,12 +121,12 @@ class TestTinyDnsFileSource(TestCase):
expected.add_record(record)
changes = expected.changes(got, SimpleProvider())
self.assertEquals([], changes)
self.assertEqual([], changes)
def test_populate_normal_sub1(self):
got = Zone('asdf.subtest.com.', [])
self.source.populate(got)
self.assertEquals(1, len(got.records))
self.assertEqual(1, len(got.records))
expected = Zone('asdf.subtest.com.', [])
for name, data in (
@@ -140,12 +140,12 @@ class TestTinyDnsFileSource(TestCase):
expected.add_record(record)
changes = expected.changes(got, SimpleProvider())
self.assertEquals([], changes)
self.assertEqual([], changes)
def test_populate_normal_sub2(self):
got = Zone('blah-asdf.subtest.com.', [])
self.source.populate(got)
self.assertEquals(2, len(got.records))
self.assertEqual(2, len(got.records))
expected = Zone('sub-asdf.subtest.com.', [])
for name, data in (
@@ -164,7 +164,7 @@ class TestTinyDnsFileSource(TestCase):
expected.add_record(record)
changes = expected.changes(got, SimpleProvider())
self.assertEquals([], changes)
self.assertEqual([], changes)
def test_populate_in_addr_arpa(self):
@@ -198,9 +198,9 @@ class TestTinyDnsFileSource(TestCase):
expected.add_record(record)
changes = expected.changes(got, SimpleProvider())
self.assertEquals([], changes)
self.assertEqual([], changes)
def test_ignores_subs(self):
got = Zone('example.com.', ['sub'])
self.source.populate(got)
self.assertEquals(16, len(got.records))
self.assertEqual(16, len(got.records))
tests/test_octodns_yaml.py
+5 -5
View File
@@ -15,7 +15,7 @@ from octodns.yaml import safe_dump, safe_load
class TestYaml(TestCase):
def test_stuff(self):
self.assertEquals({
self.assertEqual({
1: 'a',
2: 'b',
'3': 'c',
@@ -29,7 +29,7 @@ class TestYaml(TestCase):
'11': e
'''))
self.assertEquals({
self.assertEqual({
'*.1.2': 'a',
'*.2.2': 'b',
'*.10.1': 'c',
@@ -57,8 +57,8 @@ class TestYaml(TestCase):
'*.11.1': 43,
'*.2.1': 44,
}, buf)
self.assertEquals("---\n'*.1.1': 42\n'*.2.1': 44\n'*.11.1': 43\n",
buf.getvalue())
self.assertEqual("---\n'*.1.1': 42\n'*.2.1': 44\n'*.11.1': 43\n",
buf.getvalue())
# hex sorting isn't ideal, not treated as hex, this make sure we don't
# change the behavior
@@ -67,4 +67,4 @@ class TestYaml(TestCase):
'45a03129': 42,
'45a0392a': 43,
}, buf)
self.assertEquals("---\n45a0392a: 43\n45a03129: 42\n", buf.getvalue())
self.assertEqual("---\n45a0392a: 43\n45a03129: 42\n", buf.getvalue())
tests/test_octodns_zone.py
+32 -32
View File
@@ -18,7 +18,7 @@ class TestZone(TestCase):
def test_lowering(self):
zone = Zone('UniT.TEsTs.', [])
self.assertEquals('unit.tests.', zone.name)
self.assertEqual('unit.tests.', zone.name)
def test_hostname_from_fqdn(self):
zone = Zone('unit.tests.', [])
@@ -32,7 +32,7 @@ class TestZone(TestCase):
('foo.unit.tests', 'foo.unit.tests.unit.tests.'),
('foo.unit.tests', 'foo.unit.tests.unit.tests'),
):
self.assertEquals(hostname, zone.hostname_from_fqdn(fqdn))
self.assertEqual(hostname, zone.hostname_from_fqdn(fqdn))
def test_add_record(self):
zone = Zone('unit.tests.', [])
@@ -42,21 +42,21 @@ class TestZone(TestCase):
c = ARecord(zone, 'a', {'ttl': 43, 'value': '2.2.2.2'})
zone.add_record(a)
self.assertEquals(zone.records, set([a]))
self.assertEqual(zone.records, set([a]))
# Can't add record with same name & type
with self.assertRaises(DuplicateRecordException) as ctx:
zone.add_record(a)
self.assertEquals('Duplicate record a.unit.tests., type A',
str(ctx.exception))
self.assertEquals(zone.records, set([a]))
self.assertEqual('Duplicate record a.unit.tests., type A',
str(ctx.exception))
self.assertEqual(zone.records, set([a]))
# can add duplicate with replace=True
zone.add_record(c, replace=True)
self.assertEquals('2.2.2.2', list(zone.records)[0].values[0])
self.assertEqual('2.2.2.2', list(zone.records)[0].values[0])
# Can add dup name, with different type
zone.add_record(b)
self.assertEquals(zone.records, set([a, b]))
self.assertEqual(zone.records, set([a, b]))
def test_changes(self):
before = Zone('unit.tests.', [])
@@ -78,17 +78,17 @@ class TestZone(TestCase):
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]))
self.assertEqual(after.records, set([a, c]))
changes = before.changes(after, target)
self.assertEquals(2, len(changes))
self.assertEqual(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.assertEqual(b, delete.existing)
self.assertFalse(delete.new)
self.assertEquals(c, create.new)
self.assertEqual(c, create.new)
self.assertFalse(create.existing)
delete.__repr__()
create.__repr__()
@@ -98,7 +98,7 @@ class TestZone(TestCase):
after.add_record(changed)
after.add_record(b)
changes = before.changes(after, target)
self.assertEquals(1, len(changes))
self.assertEqual(1, len(changes))
update = changes[0]
self.assertIsInstance(update, Update)
# Using changes here to get a full equality
@@ -126,12 +126,12 @@ class TestZone(TestCase):
# Only create the supported A, not the AAAA
changes = current.changes(desired, NoAaaaProvider())
self.assertEquals(1, len(changes))
self.assertEqual(1, len(changes))
self.assertIsInstance(changes[0], Create)
# Only delete the supported A, not the AAAA
changes = desired.changes(current, NoAaaaProvider())
self.assertEquals(1, len(changes))
self.assertEqual(1, len(changes))
self.assertIsInstance(changes[0], Delete)
def test_missing_dot(self):
@@ -149,7 +149,7 @@ class TestZone(TestCase):
'values': ['1.2.3.4.', '2.3.4.5.'],
})
zone.add_record(record)
self.assertEquals(set([record]), zone.records)
self.assertEqual(set([record]), zone.records)
# non-NS for exactly the sub is rejected
zone = Zone('unit.tests.', set(['sub', 'barred']))
@@ -163,7 +163,7 @@ class TestZone(TestCase):
self.assertTrue('not of type NS', str(ctx.exception))
# Can add it w/lenient
zone.add_record(record, lenient=True)
self.assertEquals(set([record]), zone.records)
self.assertEqual(set([record]), zone.records)
# NS for something below the sub is rejected
zone = Zone('unit.tests.', set(['sub', 'barred']))
@@ -177,7 +177,7 @@ class TestZone(TestCase):
self.assertTrue('under a managed sub-zone', str(ctx.exception))
# Can add it w/lenient
zone.add_record(record, lenient=True)
self.assertEquals(set([record]), zone.records)
self.assertEqual(set([record]), zone.records)
# A for something below the sub is rejected
zone = Zone('unit.tests.', set(['sub', 'barred']))
@@ -191,7 +191,7 @@ class TestZone(TestCase):
self.assertTrue('under a managed sub-zone', str(ctx.exception))
# Can add it w/lenient
zone.add_record(record, lenient=True)
self.assertEquals(set([record]), zone.records)
self.assertEqual(set([record]), zone.records)
def test_ignored_records(self):
zone_normal = Zone('unit.tests.', [])
@@ -243,18 +243,18 @@ class TestZone(TestCase):
zone.add_record(a)
with self.assertRaises(InvalidNodeException):
zone.add_record(cname)
self.assertEquals(set([a]), zone.records)
self.assertEqual(set([a]), zone.records)
zone.add_record(cname, lenient=True)
self.assertEquals(set([a, cname]), zone.records)
self.assertEqual(set([a, cname]), zone.records)
# add a to cname
zone = Zone('unit.tests.', [])
zone.add_record(cname)
with self.assertRaises(InvalidNodeException):
zone.add_record(a)
self.assertEquals(set([cname]), zone.records)
self.assertEqual(set([cname]), zone.records)
zone.add_record(a, lenient=True)
self.assertEquals(set([a, cname]), zone.records)
self.assertEqual(set([a, cname]), zone.records)
def test_excluded_records(self):
zone_normal = Zone('unit.tests.', [])
@@ -355,10 +355,10 @@ class TestZone(TestCase):
self.assertTrue(zone_missing.changes(zone_normal, provider))
self.assertFalse(zone_missing.changes(zone_included, provider))
def assertEqualsNameAndValues(self, a, b):
def assertEqualNameAndValues(self, a, b):
a = dict([(r.name, r.values[0]) for r in a])
b = dict([(r.name, r.values[0]) for r in b])
self.assertEquals(a, b)
self.assertEqual(a, b)
def test_copy(self):
zone = Zone('unit.tests.', [])
@@ -369,13 +369,13 @@ class TestZone(TestCase):
zone.add_record(b)
# Sanity check
self.assertEqualsNameAndValues(set((a, b)), zone.records)
self.assertEqualNameAndValues(set((a, b)), zone.records)
copy = zone.copy()
# We have an origin set and it is the source/original zone
self.assertEquals(zone, copy._origin)
self.assertEqual(zone, copy._origin)
# Our records are zone's records to start (references)
self.assertEqualsNameAndValues(zone.records, copy.records)
self.assertEqualNameAndValues(zone.records, copy.records)
# If we try and change something that's already there we realize and
# then get an error about a duplicate
@@ -384,25 +384,25 @@ class TestZone(TestCase):
copy.add_record(b_prime)
self.assertIsNone(copy._origin)
# Unchanged, straight copies
self.assertEqualsNameAndValues(zone.records, copy.records)
self.assertEqualNameAndValues(zone.records, copy.records)
# If we add with replace things will be realized and the record will
# have changed
copy = zone.copy()
copy.add_record(b_prime, replace=True)
self.assertIsNone(copy._origin)
self.assertEqualsNameAndValues(set((a, b_prime)), copy.records)
self.assertEqualNameAndValues(set((a, b_prime)), copy.records)
# If we add another record, things are reliazed and it has been added
copy = zone.copy()
c = ARecord(zone, 'c', {'ttl': 42, 'value': '1.1.1.3'})
copy.add_record(c)
self.assertEqualsNameAndValues(set((a, b, c)), copy.records)
self.assertEqualNameAndValues(set((a, b, c)), copy.records)
# If we remove a record, things are reliazed and it has been removed
copy = zone.copy()
copy.remove_record(a)
self.assertEqualsNameAndValues(set((b,)), copy.records)
self.assertEqualNameAndValues(set((b,)), copy.records)
# Re-realizing is a noop
copy = zone.copy()