From d374dc355bf7b788d975979fa00a87c716fe63f3 Mon Sep 17 00:00:00 2001 From: Ross McFarland Date: Thu, 2 Sep 2021 13:00:38 -0700 Subject: [PATCH] Use patch to mock in transip tests, rework bits to make more mockable --- octodns/provider/transip.py | 34 ++++++++-------------- tests/test_octodns_provider_transip.py | 39 +++++++++++++++++--------- 2 files changed, 37 insertions(+), 36 deletions(-) diff --git a/octodns/provider/transip.py b/octodns/provider/transip.py index 67011f9..d3e2018 100644 --- a/octodns/provider/transip.py +++ b/octodns/provider/transip.py @@ -63,31 +63,21 @@ class TransipProvider(BaseProvider): account) super(TransipProvider, self).__init__(id, *args, **kwargs) - if key is None and key_file is None: + if key_file is not None: + self._client = self._domain_service(account, + private_key_file=key_file) + elif key is not None: + self._client = self._domain_service(account, private_key=key) + else: raise TransipConfigException( 'Missing `key` or `key_file` parameter in config' ) - self.account = account - self.key = key - self.key_file = key_file - - self._client = None self._currentZone = {} - @property - def client(self): - # This can't happen in __init__ b/c it makes network calls during the - # construction of the object and that before the tests have had a - # chance to install the mock client - if self._client is None: - if self.key_file is not None: - self._client = DomainService(self.account, - private_key_file=self.key_file) - else: # we checked key in __init__ so can assume it's not None - self._client = DomainService(self.account, - private_key=self.key) - return self._client + def _domain_service(self, *args, **kwargs): + 'This exists only for mocking purposes' + return DomainService(*args, **kwargs) def populate(self, zone, target=False, lenient=False): @@ -98,7 +88,7 @@ class TransipProvider(BaseProvider): before = len(zone.records) try: - zoneInfo = self.client.get_info(zone.name[:-1]) + zoneInfo = self._client.get_info(zone.name[:-1]) except WebFault as e: if e.fault.faultcode == '102' and target is False: # Zone not found in account, and not a target so just @@ -148,7 +138,7 @@ class TransipProvider(BaseProvider): self._currentZone = plan.desired try: - self.client.get_info(plan.desired.name[:-1]) + self._client.get_info(plan.desired.name[:-1]) except WebFault as e: self.log.exception('_apply: get_info failed') raise e @@ -167,7 +157,7 @@ class TransipProvider(BaseProvider): _dns_entries.extend(entries_for(name, record)) try: - self.client.set_dns_entries(plan.desired.name[:-1], _dns_entries) + self._client.set_dns_entries(plan.desired.name[:-1], _dns_entries) except WebFault as e: self.log.warning(('_apply: Set DNS returned ' + 'one or more errors: {}').format( diff --git a/tests/test_octodns_provider_transip.py b/tests/test_octodns_provider_transip.py index 4817f7d..8a2e11a 100644 --- a/tests/test_octodns_provider_transip.py +++ b/tests/test_octodns_provider_transip.py @@ -10,6 +10,7 @@ from six import text_type from suds import WebFault +from mock import patch from unittest import TestCase from octodns.provider.transip import TransipProvider @@ -136,7 +137,11 @@ N4OiVz1I3rbZGYa396lpxO6ku8yCglisL1yrSP6DdEUp66ntpKVd source.populate(expected) return expected - def test_init(self): + @patch('octodns.provider.transip.TransipProvider._domain_service', + return_value=MockDomainService()) + def test_init(self, _): + + # No key nor key_file with self.assertRaises(Exception) as ctx: TransipProvider('test', 'unittest') @@ -144,19 +149,26 @@ N4OiVz1I3rbZGYa396lpxO6ku8yCglisL1yrSP6DdEUp66ntpKVd str('Missing `key` or `key_file` parameter in config'), str(ctx.exception)) + # With key TransipProvider('test', 'unittest', key=self.bogus_key) - # Existence and content of the key is tested in the SDK on client call + # With key_file TransipProvider('test', 'unittest', key_file='/fake/path') - def test_populate(self): + @patch('suds.client.Client.__init__', new=lambda *args, **kwargs: None) + def test_domain_service(self): + # Special case smoke test for DomainService to get coverage + TransipProvider('test', 'unittest', key=self.bogus_key) + + @patch('octodns.provider.transip.TransipProvider._domain_service', + return_value=MockDomainService()) + def test_populate(self, _): _expected = self.make_expected() # Unhappy Plan - Not authenticated # Live test against API, will fail in an unauthorized error with self.assertRaises(WebFault) as ctx: provider = TransipProvider('test', 'unittest', self.bogus_key) - provider._client = MockDomainService('unittest', self.bogus_key) provider._client.throw_auth_fault = True zone = Zone('unit.tests.', []) provider.populate(zone, True) @@ -166,12 +178,14 @@ N4OiVz1I3rbZGYa396lpxO6ku8yCglisL1yrSP6DdEUp66ntpKVd self.assertEquals(str('200'), ctx.exception.fault.faultcode) + # No more auth problems + provider._client.throw_auth_fault = False + # Unhappy Plan - Zone does not exists # Will trigger an exception if provider is used as a target for a # non-existing zone with self.assertRaises(Exception) as ctx: provider = TransipProvider('test', 'unittest', self.bogus_key) - provider._client = MockDomainService('unittest', self.bogus_key) zone = Zone('notfound.unit.tests.', []) provider.populate(zone, True) @@ -187,13 +201,11 @@ N4OiVz1I3rbZGYa396lpxO6ku8yCglisL1yrSP6DdEUp66ntpKVd # Won't trigger an exception if provider is NOT used as a target for a # non-existing zone. provider = TransipProvider('test', 'unittest', self.bogus_key) - provider._client = MockDomainService('unittest', self.bogus_key) zone = Zone('notfound.unit.tests.', []) provider.populate(zone, False) # Happy Plan - Populate with mockup records provider = TransipProvider('test', 'unittest', self.bogus_key) - provider._client = MockDomainService('unittest', self.bogus_key) provider._client.mockup(_expected.records) zone = Zone('unit.tests.', []) provider.populate(zone, False) @@ -211,19 +223,19 @@ N4OiVz1I3rbZGYa396lpxO6ku8yCglisL1yrSP6DdEUp66ntpKVd # Happy Plan - Even if the zone has no records the zone should exist provider = TransipProvider('test', 'unittest', self.bogus_key) - provider._client = MockDomainService('unittest', self.bogus_key) zone = Zone('unit.tests.', []) exists = provider.populate(zone, True) self.assertTrue(exists, 'populate should return true') return - def test_plan(self): + @patch('octodns.provider.transip.TransipProvider._domain_service', + return_value=MockDomainService()) + def test_plan(self, _): _expected = self.make_expected() # Test Happy plan, only create provider = TransipProvider('test', 'unittest', self.bogus_key) - provider._client = MockDomainService('unittest', self.bogus_key) plan = provider.plan(_expected) self.assertEqual(15, plan.change_counts['Create']) @@ -232,12 +244,13 @@ N4OiVz1I3rbZGYa396lpxO6ku8yCglisL1yrSP6DdEUp66ntpKVd return - def test_apply(self): + @patch('octodns.provider.transip.TransipProvider._domain_service', + return_value=MockDomainService()) + def test_apply(self, _): _expected = self.make_expected() # Test happy flow. Create all supoorted records provider = TransipProvider('test', 'unittest', self.bogus_key) - provider._client = MockDomainService('unittest', self.bogus_key) plan = provider.plan(_expected) self.assertEqual(15, len(plan.changes)) changes = provider.apply(plan) @@ -249,7 +262,6 @@ N4OiVz1I3rbZGYa396lpxO6ku8yCglisL1yrSP6DdEUp66ntpKVd changes = [] # reset changes with self.assertRaises(Exception) as ctx: provider = TransipProvider('test', 'unittest', self.bogus_key) - provider._client = MockDomainService('unittest', self.bogus_key) plan = provider.plan(_expected) plan.desired.name = 'notfound.unit.tests.' changes = provider.apply(plan) @@ -268,7 +280,6 @@ N4OiVz1I3rbZGYa396lpxO6ku8yCglisL1yrSP6DdEUp66ntpKVd with self.assertRaises(Exception) as ctx: provider = TransipProvider('test', 'unittest', self.bogus_key) - provider._client = MockDomainService('unittest', self.bogus_key) plan = provider.plan(_expected) plan.desired.name = 'failsetdns.unit.tests.' changes = provider.apply(plan)