# # # from __future__ import absolute_import, division, print_function, \ unicode_literals from os.path import dirname, join from six import text_type from suds import WebFault from unittest import TestCase from octodns.provider.transip import TransipProvider from octodns.provider.yaml import YamlProvider from octodns.zone import Zone from transip.service.objects import DnsEntry class MockFault(object): faultstring = "" faultcode = "" def __init__(self, code, string, *args, **kwargs): self.faultstring = string self.faultcode = code class MockResponse(object): dnsEntries = [] class MockDomainService(object): def __init__(self, *args, **kwargs): self.mockupEntries = [] self.throw_auth_fault = False def mockup(self, records): provider = TransipProvider('', '', '') _dns_entries = [] for record in records: if record._type in provider.SUPPORTS: entries_for = getattr(provider, '_entries_for_{}'.format(record._type)) # Root records have '@' as name name = record.name if name == '': name = provider.ROOT_RECORD _dns_entries.extend(entries_for(name, record)) # Add a non-supported type # so it triggers the "is supported" (transip.py:115) check and # give 100% code coverage _dns_entries.append( DnsEntry('@', '3600', 'BOGUS', 'ns01.transip.nl.')) self.mockupEntries = _dns_entries # Skips authentication layer and returns the entries loaded by "Mockup" def get_info(self, domain_name): if self.throw_auth_fault: self.raiseInvalidAuth() # Special 'domain' to trigger error if str(domain_name) == str('notfound.unit.tests'): self.raiseZoneNotFound() result = MockResponse() result.dnsEntries = self.mockupEntries return result def set_dns_entries(self, domain_name, dns_entries): # Special 'domain' to trigger error if str(domain_name) == str('failsetdns.unit.tests'): self.raiseSaveError() return True def raiseZoneNotFound(self): fault = MockFault(str('102'), '102 is zone not found') document = {} raise WebFault(fault, document) def raiseInvalidAuth(self): fault = MockFault(str('200'), '200 is invalid auth') document = {} raise WebFault(fault, document) def raiseSaveError(self): fault = MockFault(str('200'), '202 random error') document = {} raise WebFault(fault, document) class TestTransipProvider(TestCase): bogus_key = str("""-----BEGIN RSA PRIVATE KEY----- MIIEowIBAAKCAQEA0U5HGCkLrz423IyUf3u4cKN2WrNz1x5KNr6PvH2M/zxas+zB elbxkdT3AQ+wmfcIvOuTmFRTHv35q2um1aBrPxVw+2s+lWo28VwIRttwIB1vIeWu lSBnkEZQRLyPI2tH0i5QoMX4CVPf9rvij3Uslimi84jdzDfPFIh6jZ6C8nLipOTG 0IMhge1ofVfB0oSy5H+7PYS2858QLAf5ruYbzbAxZRivS402wGmQ0d0Lc1KxraAj kiMM5yj/CkH/Vm2w9I6+tLFeASE4ub5HCP5G/ig4dbYtqZMQMpqyAbGxd5SOVtyn UHagAJUxf8DT3I8PyjEHjxdOPUsxNyRtepO/7QIDAQABAoIBAQC7fiZ7gxE/ezjD 2n6PsHFpHVTBLS2gzzZl0dCKZeFvJk6ODJDImaeuHhrh7X8ifMNsEI9XjnojMhl8 MGPzy88mZHugDNK0H8B19x5G8v1/Fz7dG5WHas660/HFkS+b59cfdXOugYiOOn9O 08HBBpLZNRUOmVUuQfQTjapSwGLG8PocgpyRD4zx0LnldnJcqYCxwCdev+AAsPnq ibNtOd/MYD37w9MEGcaxLE8wGgkv8yd97aTjkgE+tp4zsM4QE4Rag133tsLLNznT 4Qr/of15M3NW/DXq/fgctyRcJjZpU66eCXLCz2iRTnLyyxxDC2nwlxKbubV+lcS0 S4hbfd/BAoGBAO8jXxEaiybR0aIhhSR5esEc3ymo8R8vBN3ZMJ+vr5jEPXr/ZuFj /R4cZ2XV3VoQJG0pvIOYVPZ5DpJM7W+zSXtJ/7bLXy4Bnmh/rc+YYgC+AXQoLSil iD2OuB2xAzRAK71DVSO0kv8gEEXCersPT2i6+vC2GIlJvLcYbOdRKWGxAoGBAOAQ aJbRLtKujH+kMdoMI7tRlL8XwI+SZf0FcieEu//nFyerTePUhVgEtcE+7eQ7hyhG fIXUFx/wALySoqFzdJDLc8U8pTLhbUaoLOTjkwnCTKQVprhnISqQqqh/0U5u47IE RWzWKN6OHb0CezNTq80Dr6HoxmPCnJHBHn5LinT9AoGAQSpvZpbIIqz8pmTiBl2A QQ2gFpcuFeRXPClKYcmbXVLkuhbNL1BzEniFCLAt4LQTaRf9ghLJ3FyCxwVlkpHV zV4N6/8hkcTpKOraL38D/dXJSaEFJVVuee/hZl3tVJjEEpA9rDwx7ooLRSdJEJ6M ciq55UyKBSdt4KssSiDI2RECgYBL3mJ7xuLy5bWfNsrGiVvD/rC+L928/5ZXIXPw 26oI0Yfun7ulDH4GOroMcDF/GYT/Zzac3h7iapLlR0WYI47xxGI0A//wBZLJ3QIu krxkDo2C9e3Y/NqnHgsbOQR3aWbiDT4wxydZjIeXS3LKA2fl6Hyc90PN3cTEOb8I hq2gRQKBgEt0SxhhtyB93SjgTzmUZZ7PiEf0YJatfM6cevmjWHexrZH+x31PB72s fH2BQyTKKzoCLB1k/6HRaMnZdrWyWSZ7JKz3AHJ8+58d0Hr8LTrzDM1L6BbjeDct N4OiVz1I3rbZGYa396lpxO6ku8yCglisL1yrSP6DdEUp66ntpKVd -----END RSA PRIVATE KEY-----""") def make_expected(self): expected = Zone('unit.tests.', []) source = YamlProvider('test', join(dirname(__file__), 'config')) source.populate(expected) return expected def test_init(self): with self.assertRaises(Exception) as ctx: TransipProvider('test', 'unittest') self.assertEquals( str('Missing `key` or `key_file` parameter in config'), str(ctx.exception)) TransipProvider('test', 'unittest', key=self.bogus_key) # Existence and content of the key is tested in the SDK on client call TransipProvider('test', 'unittest', key_file='/fake/path') 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) self.assertEquals(str('WebFault'), str(ctx.exception.__class__.__name__)) self.assertEquals(str('200'), ctx.exception.fault.faultcode) # 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) self.assertEquals(str('TransipNewZoneException'), str(ctx.exception.__class__.__name__)) self.assertEquals( 'populate: (102) Transip used as target' + ' for non-existing zone: notfound.unit.tests.', text_type(ctx.exception)) # Happy Plan - Zone does not exists # 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) # Transip allows relative values for types like cname, mx. # Test is these are correctly appended with the domain provider._currentZone = zone self.assertEquals("www.unit.tests.", provider._parse_to_fqdn("www")) self.assertEquals("www.unit.tests.", provider._parse_to_fqdn("www.unit.tests.")) self.assertEquals("www.sub.sub.sub.unit.tests.", provider._parse_to_fqdn("www.sub.sub.sub")) self.assertEquals("unit.tests.", provider._parse_to_fqdn("@")) # 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): _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']) self.assertEqual(0, plan.change_counts['Update']) self.assertEqual(0, plan.change_counts['Delete']) return 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) self.assertEqual(changes, len(plan.changes)) # Test unhappy flow. Trigger 'not found error' in apply stage # This should normally not happen as populate will capture it first # but just in case. 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) # Changes should not be set due to an Exception self.assertEqual([], changes) self.assertEquals(str('WebFault'), str(ctx.exception.__class__.__name__)) self.assertEquals(str('102'), ctx.exception.fault.faultcode) # Test unhappy flow. Trigger a unrecoverable error while saving _expected = self.make_expected() # reset expected 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 = 'failsetdns.unit.tests.' changes = provider.apply(plan) # Changes should not be set due to an Exception self.assertEqual([], changes) self.assertEquals(str('TransipException'), str(ctx.exception.__class__.__name__))