tero@440: import ipaddr tero@477: import pvl.args tero@440: import unittest tero@440: tero@477: from pvl.hosts import config, dhcp, zone tero@477: from pvl.hosts.host import Host tero@440: from StringIO import StringIO tero@440: tero@447: class ConfFile(StringIO): tero@440: def __init__(self, name, buffer): tero@440: StringIO.__init__(self, buffer) tero@440: self.name = name tero@440: tero@440: class TestConfig(unittest.TestCase): tero@440: def setUp(self): tero@490: self.options = pvl.args.options( tero@490: hosts_charset = 'utf-8', tero@490: hosts_domain = None, tero@490: hosts_include = None, tero@490: ) tero@440: tero@461: def assertHostEqual(self, host, host_str, attrs): tero@461: self.assertEquals(str(host), host_str) tero@461: tero@461: for attr, value in attrs.iteritems(): tero@461: self.assertEquals(getattr(host, attr), value) tero@461: tero@447: def assertHostsEqual(self, hosts, expected): tero@447: for host, expect in zip(hosts, expected): tero@447: host_str, attrs = expect tero@441: tero@461: self.assertHostEqual(host, host_str, attrs) tero@441: tero@451: def testApplyHostsFileError(self): tero@440: with self.assertRaises(config.HostConfigError): tero@440: list(config.apply_hosts(self.options, ['nonexistant'])) tero@440: tero@440: def testApplyHosts(self): tero@447: conf_file = ConfFile('test', """ tero@440: [foo] tero@440: ip = 127.0.0.1 tero@440: tero@440: [bar] tero@440: ip = 127.0.0.2 tero@447: """) tero@447: expected = [ tero@447: ('foo@test', dict(ip=ipaddr.IPAddress('127.0.0.1'))), tero@447: ('bar@test', dict(ip=ipaddr.IPAddress('127.0.0.2'))), tero@447: ] tero@440: tero@447: self.assertHostsEqual(config.apply_hosts_file(self.options, conf_file), expected) tero@447: tero@451: def testApply(self): tero@451: self.assertHostsEqual(config.apply(self.options, ['etc/hosts/test']), [ tero@451: ('foo@test', dict( tero@451: ip = ipaddr.IPAddress('127.0.0.1'), tero@451: ethernet = {None: '00:11:22:33:44:55'}, tero@451: )), tero@451: ('bar@test', dict( tero@451: ip = ipaddr.IPAddress('127.0.0.2'), tero@451: ethernet = {None: '01:23:45:67:89:ab'}, tero@451: )), tero@451: ]) tero@451: tero@447: def testApplyHostsExpand(self): tero@449: self.assertHostsEqual(config.apply_host_config(self.options, 'asdf', 'asdf{1-3}', ip='10.100.100.$'), [ tero@447: ('asdf1@asdf', dict(ip=ipaddr.IPAddress('10.100.100.1'))), tero@447: ('asdf2@asdf', dict(ip=ipaddr.IPAddress('10.100.100.2'))), tero@447: ('asdf3@asdf', dict(ip=ipaddr.IPAddress('10.100.100.3'))), tero@447: ]) tero@461: tero@461: def testApplyHostConfigDict(self): tero@461: host = config.apply_host(self.options, 'foo', 'test', { tero@461: 'ethernet.eth0': '00:11:22:33:44:55', tero@461: }) tero@461: tero@461: self.assertHostEqual(host, 'foo@test', dict( tero@461: ethernet = { 'eth0': '00:11:22:33:44:55' } tero@461: )) tero@461: tero@451: def testApplyHostsConfigError(self): tero@451: with self.assertRaises(config.HostConfigError): tero@451: config.apply_host(self.options, 'foo', 'test', { tero@451: 'ethernet': 'foo', tero@451: 'ethernet.eth0': 'bar', tero@451: }) tero@447: tero@469: class TestZoneMixin(object): tero@463: def assertZoneEquals(self, rrs, expected): tero@463: gather = { } tero@463: tero@463: for rr in rrs: tero@463: key = (rr.name.lower(), rr.type.upper()) tero@463: tero@463: self.assertNotIn(key, gather) tero@463: tero@463: gather[key] = rr.data tero@463: tero@463: self.assertDictEqual(gather, expected) tero@463: tero@469: tero@469: class TestForwardZone(TestZoneMixin, unittest.TestCase): tero@469: def testHostOutOfOrigin(self): tero@477: h = Host('host', 'domain', ip=ipaddr.IPAddress('10.0.0.1')) tero@463: tero@463: self.assertZoneEquals(zone.host_forward(h, 'test'), { }) tero@463: tero@464: def testHostIP(self): tero@477: h = Host.build('host', 'domain', tero@463: ip = '192.0.2.1', tero@463: ip6 = '2001:db8::192.0.2.1', tero@463: ) tero@463: tero@463: self.assertZoneEquals(zone.host_forward(h, 'domain'), { tero@463: ('host', 'A'): ['192.0.2.1'], tero@463: ('host', 'AAAA'): ['2001:db8::c000:201'], tero@463: }) tero@465: tero@466: def testHostAlias(self): tero@477: h = Host.build('host', 'domain', tero@466: ip = '192.0.2.1', tero@466: alias = 'test *.test', tero@466: ) tero@466: tero@466: self.assertEquals(h.alias, ['test', '*.test']) tero@466: tero@466: self.assertZoneEquals(zone.host_forward(h, 'domain'), { tero@466: ('host', 'A'): ['192.0.2.1'], tero@466: ('test', 'CNAME'): ['host'], tero@466: ('*.test', 'CNAME'): ['host'], tero@466: }) tero@466: tero@468: def testHostAlias46(self): tero@477: h = Host.build('host', 'domain', tero@468: ip = '192.0.2.1', tero@468: ip6 = '2001:db8::192.0.2.1', tero@468: alias4 = 'test4', tero@468: alias6 = 'test6', tero@468: ) tero@468: tero@468: self.assertZoneEquals(zone.host_forward(h, 'domain'), { tero@468: ('host', 'A'): ['192.0.2.1'], tero@468: ('host', 'AAAA'): ['2001:db8::c000:201'], tero@468: ('test4', 'A'): ['192.0.2.1'], tero@468: ('test6', 'AAAA'): ['2001:db8::c000:201'], tero@468: }) tero@468: tero@468: def testHostAlias4Missing(self): tero@477: h = Host.build('host', 'domain', tero@468: ip6 = '2001:db8::192.0.2.1', tero@468: alias4 = 'test4', tero@468: alias6 = 'test6', tero@468: ) tero@468: tero@468: with self.assertRaises(zone.HostZoneError): tero@468: self.assertZoneEquals(zone.host_forward(h, 'domain'), { }) tero@468: tero@468: def testHostAlias6Missing(self): tero@477: h = Host.build('host', 'domain', tero@468: ip = '192.0.2.1', tero@468: alias4 = 'test4', tero@468: alias6 = 'test6', tero@468: ) tero@468: tero@468: with self.assertRaises(zone.HostZoneError): tero@468: self.assertZoneEquals(zone.host_forward(h, 'domain'), { }) tero@468: tero@469: def testHostDelegate(self): tero@477: h = Host.build('host', 'example.com', tero@469: forward = 'host.example.net', tero@469: ) tero@469: tero@469: self.assertZoneEquals(zone.host_forward(h, 'example.com'), { tero@469: ('host', 'CNAME'): ['host.example.net.'], tero@469: }) tero@469: tero@467: def testHostForwardAlias(self): tero@477: h = Host.build('host', 'domain', tero@467: forward = 'host.example.net', tero@467: alias = 'test', tero@467: ) tero@467: tero@467: self.assertZoneEquals(zone.host_forward(h, 'domain'), { tero@467: ('host', 'CNAME'): ['host.example.net.'], tero@467: ('test', 'CNAME'): ['host'], tero@467: }) tero@466: tero@470: def testHostLocation(self): tero@477: h = Host.build('host', 'domain', tero@470: ip = '192.0.2.1', tero@470: location = 'test', tero@470: ) tero@470: tero@470: self.assertEquals(h.location, ('test', 'domain')) tero@470: tero@470: self.assertZoneEquals(zone.host_forward(h, 'domain'), { tero@470: ('host', 'A'): ['192.0.2.1'], tero@470: ('test', 'CNAME'): ['host'], tero@470: }) tero@471: tero@471: def testHostLocationDomain(self): tero@477: h = Host.build('host', 'foo.domain', tero@471: ip = '192.0.2.1', tero@471: location = 'test@bar.domain', tero@471: ) tero@471: tero@471: self.assertEquals(h.location, ('test', 'bar.domain')) tero@471: tero@471: self.assertZoneEquals(zone.host_forward(h, 'domain'), { tero@471: ('host.foo', 'A'): ['192.0.2.1'], tero@471: ('test.bar', 'CNAME'): ['host.foo'], tero@471: }) tero@471: tero@471: def testHostLocationDomainOutOfOrigin(self): tero@477: h = Host.build('host', 'foo.domain', tero@471: ip = '192.0.2.1', tero@471: location = 'test@bar.domain', tero@471: ) tero@471: tero@471: self.assertEquals(h.location, ('test', 'bar.domain')) tero@471: tero@471: with self.assertRaises(zone.HostZoneError): tero@471: self.assertZoneEquals(zone.host_forward(h, 'foo.domain'), { tero@471: ('host', 'A'): ['192.0.2.1'], tero@471: }) tero@471: tero@471: # TODO tero@471: #self.assertZoneEquals(zone.host_forward(h, 'bar.domain'), { tero@471: # ('test', 'CNAME'): ['host.foo'], tero@471: #}) tero@472: tero@472: def testHostsForward(self): tero@472: hosts = [ tero@477: Host.build('foo', 'domain', tero@472: ip = '192.0.2.1', tero@472: ip6 = '2001:db8::192.0.2.1', tero@472: alias = 'test', tero@472: ), tero@477: Host.build('bar', 'domain', tero@472: ip = '192.0.2.2', tero@472: ) tero@472: ] tero@472: tero@489: self.assertZoneEquals(zone.apply_hosts_forward(hosts, 'domain'), { tero@472: ('foo', 'A'): ['192.0.2.1'], tero@472: ('foo', 'AAAA'): ['2001:db8::c000:201'], tero@472: ('test', 'CNAME'): ['foo'], tero@472: ('bar', 'A'): ['192.0.2.2'], tero@472: }) tero@472: tero@472: def testHostsConflict(self): tero@472: hosts = [ tero@477: Host.build('foo', 'domain', tero@472: ip = '192.0.2.1', tero@472: ), tero@477: Host.build('foo', 'domain', tero@472: ip = '192.0.2.2', tero@472: ) tero@472: ] tero@472: tero@472: with self.assertRaises(zone.HostZoneError): tero@489: self.assertZoneEquals(zone.apply_hosts_forward(hosts, 'domain'), { }) tero@472: tero@472: def testHostsAliasConflict(self): tero@472: hosts = [ tero@477: Host.build('foo', 'domain', tero@472: ip = '192.0.2.1', tero@472: ), tero@477: Host.build('bar', 'domain', tero@472: ip = '192.0.2.2', tero@472: alias = 'foo', tero@472: ) tero@472: ] tero@472: tero@472: # with A first tero@472: with self.assertRaises(zone.HostZoneError): tero@489: self.assertZoneEquals(zone.apply_hosts_forward(hosts, 'domain'), { }) tero@472: tero@472: # also with CNAME first tero@472: with self.assertRaises(zone.HostZoneError): tero@489: self.assertZoneEquals(zone.apply_hosts_forward(reversed(hosts), 'domain'), { }) tero@472: tero@472: def testHostsAlias4Conflict(self): tero@472: hosts = [ tero@477: Host.build('foo', 'domain', tero@472: ip = '192.0.2.1', tero@472: ), tero@477: Host.build('bar', 'domain', tero@472: ip = '192.0.2.2', tero@472: alias4 = 'foo', tero@472: ) tero@472: ] tero@472: tero@472: with self.assertRaises(zone.HostZoneError): tero@489: self.assertZoneEquals(zone.apply_hosts_forward(hosts, 'domain'), { }) tero@472: tero@472: tero@469: class TestReverseZone(TestZoneMixin, unittest.TestCase): tero@473: def testHostIP(self): tero@477: h = Host.build('host', 'domain', tero@473: ip = '192.0.2.1', tero@473: ip6 = '2001:db8::192.0.2.1', tero@473: ) tero@473: tero@473: self.assertZoneEquals((rr for ip, rr in zone.host_reverse(h, ipaddr.IPNetwork('192.0.2.1/24'))), { tero@473: ('1', 'PTR'): ['host.domain.'], tero@473: }) tero@473: tero@473: self.assertZoneEquals((rr for ip, rr in zone.host_reverse(h, ipaddr.IPNetwork('2001:db8::/64'))), { tero@473: ('1.0.2.0.0.0.0.c.0.0.0.0.0.0.0.0', 'PTR'): ['host.domain.'], tero@473: }) tero@473: tero@473: def testHostIP4(self): tero@477: h = Host.build('host', 'domain', tero@473: ip = '192.0.2.1', tero@473: ) tero@473: tero@473: self.assertZoneEquals((rr for ip, rr in zone.host_reverse(h, ipaddr.IPNetwork('192.0.2.1/24'))), { tero@473: ('1', 'PTR'): ['host.domain.'], tero@473: }) tero@474: tero@474: self.assertZoneEquals((rr for ip, rr in zone.host_reverse(h, ipaddr.IPNetwork('192.0.0.0/16'))), { tero@474: ('1.2', 'PTR'): ['host.domain.'], tero@474: }) tero@474: tero@474: self.assertZoneEquals((rr for ip, rr in zone.host_reverse(h, ipaddr.IPNetwork('192.0.0.0/12'))), { tero@474: ('1.2.0', 'PTR'): ['host.domain.'], tero@474: }) tero@473: tero@473: self.assertZoneEquals((rr for ip, rr in zone.host_reverse(h, ipaddr.IPNetwork('2001:db8::/64'))), { tero@473: tero@473: }) tero@473: tero@473: def testHostIP6(self): tero@477: h = Host.build('host', 'domain', tero@473: ip6 = '2001:db8::192.0.2.1', tero@473: ) tero@473: tero@473: self.assertZoneEquals((rr for ip, rr in zone.host_reverse(h, ipaddr.IPNetwork('192.0.2.1/24'))), { tero@473: }) tero@473: tero@473: self.assertZoneEquals((rr for ip, rr in zone.host_reverse(h, ipaddr.IPNetwork('2001:db8::/64'))), { tero@473: ('1.0.2.0.0.0.0.c.0.0.0.0.0.0.0.0', 'PTR'): ['host.domain.'], tero@473: }) tero@473: tero@473: def testHostIPOutOfPrefix(self): tero@477: h = Host.build('host', 'domain', tero@473: ip = '192.0.2.1', tero@473: ip6 = '2001:db8::192.0.2.1', tero@473: ) tero@473: tero@473: self.assertZoneEquals((rr for ip, rr in zone.host_reverse(h, ipaddr.IPNetwork('192.0.1.0/24'))), { tero@473: tero@473: }) tero@473: tero@473: self.assertZoneEquals((rr for ip, rr in zone.host_reverse(h, ipaddr.IPNetwork('2001:db8:1::/64'))), { tero@473: tero@473: }) tero@473: tero@463: def testHostDelegate(self): tero@477: h = Host.build('host', 'example.com', tero@463: ip = '192.0.2.1', tero@463: ip6 = '2001:db8::192.0.2.1', tero@465: forward = '', tero@463: reverse = '1.0/28.2.0.192.in-addr.arpa', tero@463: ) tero@463: tero@463: self.assertZoneEquals(zone.host_forward(h, 'example.com'), { tero@465: tero@463: }) tero@463: tero@463: self.assertZoneEquals((rr for ip, rr in zone.host_reverse(h, ipaddr.IPNetwork('192.0.2.1/24'))), { tero@463: ('1', 'CNAME'): ['1.0/28.2.0.192.in-addr.arpa.'], tero@463: }) tero@464: tero@464: self.assertZoneEquals((rr for ip, rr in zone.host_reverse(h, ipaddr.IPNetwork('2001:db8::/64'))), { tero@464: tero@464: }) tero@463: tero@474: def testHostsConflict(self): tero@474: hosts = [ tero@477: Host.build('foo', 'domain', tero@474: ip = '192.0.2.1', tero@474: ), tero@477: Host.build('bar', 'domain', tero@474: ip = '192.0.2.1', tero@474: ) tero@474: ] tero@474: tero@474: with self.assertRaises(zone.HostZoneError): tero@489: self.assertZoneEquals(zone.apply_hosts_reverse(hosts, ipaddr.IPNetwork('192.0.2.1/24')), { }) tero@474: tero@474: def testHostsGenerateUnknown(self): tero@474: hosts = [ tero@477: Host.build('foo', 'domain', tero@474: ip = '192.0.2.1', tero@474: ), tero@477: Host.build('bar', 'domain', tero@474: ip = '192.0.2.5', tero@474: ), tero@474: ] tero@474: tero@489: self.assertZoneEquals(zone.apply_hosts_reverse(hosts, ipaddr.IPNetwork('192.0.2.1/29'), tero@489: unknown_host = 'ufc', tero@489: unknown_domain = 'domain', tero@489: ), { tero@474: ('1', 'PTR'): ['foo.domain.'], tero@474: ('2', 'PTR'): ['ufc.domain.'], tero@474: ('3', 'PTR'): ['ufc.domain.'], tero@474: ('4', 'PTR'): ['ufc.domain.'], tero@474: ('5', 'PTR'): ['bar.domain.'], tero@474: ('6', 'PTR'): ['ufc.domain.'], tero@474: }) tero@469: tero@479: class TestDhcp(unittest.TestCase): tero@479: def assertBlocksEqual(self, blockdefs, expected): tero@479: for (_block, _items, _opts), (block, items, opts) in zip(blockdefs, expected): tero@479: self.assertEqual(_block, block) tero@479: self.assertItemsEqual(_items, items) tero@479: tero@479: if opts is not None: tero@479: self.assertEqual(_opts, opts) tero@479: tero@479: self.assertEqual(len(blockdefs), len(expected)) tero@479: tero@479: def testHost(self): tero@479: host = Host.build('foo', 'test', tero@479: ip = '192.0.2.1', tero@479: ethernet = '00:11:22:33:44:55', tero@492: owner = 'foo', tero@479: ) tero@479: tero@479: self.assertBlocksEqual(list(dhcp.dhcp_host(host)), [ tero@479: (('host', 'foo'), [ tero@479: ('option', 'host-name', "foo"), tero@479: ('fixed-address', '192.0.2.1'), tero@479: ('hardware', 'ethernet', '00:11:22:33:44:55'), tero@492: ], dict(comment="Owner: foo")) tero@492: ]) tero@492: tero@492: def testHostStatic(self): tero@492: host = Host.build('foo', 'test', tero@492: ip = '192.0.2.1', tero@492: ) tero@492: tero@492: self.assertBlocksEqual(list(dhcp.dhcp_host(host)), [ tero@492: tero@479: ]) tero@479: tero@480: def testHostDynamic(self): tero@480: host = Host.build('foo', 'test', tero@480: ethernet = '00:11:22:33:44:55', tero@480: ) tero@480: tero@480: self.assertBlocksEqual(list(dhcp.dhcp_host(host)), [ tero@480: (('host', 'foo'), [ tero@480: ('option', 'host-name', "foo"), tero@480: ('hardware', 'ethernet', '00:11:22:33:44:55'), tero@480: ], None) tero@480: ]) tero@480: tero@480: def testHostBoot(self): tero@491: hosts = [ tero@491: Host.build('foo1', 'test', tero@491: ethernet = '00:11:22:33:44:55', tero@491: boot = 'boot.lan:debian/wheezy/pxelinux.0', tero@491: ), tero@491: Host.build('foo2', 'test', tero@491: ethernet = '00:11:22:33:44:55', tero@491: boot = 'boot.lan:', tero@491: ), tero@491: Host.build('foo3', 'test', tero@491: ethernet = '00:11:22:33:44:55', tero@491: boot = '/debian/wheezy/pxelinux.0', tero@491: ), tero@491: ] tero@480: tero@491: self.assertBlocksEqual(list(dhcp.dhcp_hosts(hosts)), [ tero@491: (('host', 'foo1'), [ tero@491: ('option', 'host-name', "foo1"), tero@480: ('hardware', 'ethernet', '00:11:22:33:44:55'), tero@480: ('next-server', 'boot.lan'), tero@480: ('filename', 'debian/wheezy/pxelinux.0'), tero@491: ], None), tero@491: (('host', 'foo2'), [ tero@491: ('option', 'host-name', "foo2"), tero@491: ('hardware', 'ethernet', '00:11:22:33:44:55'), tero@491: ('next-server', 'boot.lan'), tero@491: ], None), tero@491: (('host', 'foo3'), [ tero@491: ('option', 'host-name', "foo3"), tero@491: ('hardware', 'ethernet', '00:11:22:33:44:55'), tero@491: ('filename', 'debian/wheezy/pxelinux.0'), tero@491: ], None), tero@480: ]) tero@480: tero@483: def testHosts(self): tero@483: hosts = [ tero@483: Host.build('foo', 'test', tero@483: ip = '192.0.2.1', tero@483: ethernet = '00:11:22:33:44:55', tero@483: ), tero@483: Host.build('bar', 'test', tero@483: ip = '192.0.2.2', tero@483: ethernet = '01:23:45:67:89:ab', tero@483: ), tero@483: ] tero@483: tero@483: self.assertBlocksEqual(list(dhcp.dhcp_hosts(hosts)), [ tero@483: (('host', 'foo'), [ tero@483: ('option', 'host-name', "foo"), tero@483: ('fixed-address', '192.0.2.1'), tero@483: ('hardware', 'ethernet', '00:11:22:33:44:55'), tero@483: ], None), tero@483: (('host', 'bar'), [ tero@483: ('option', 'host-name', "bar"), tero@483: ('fixed-address', '192.0.2.2'), tero@483: ('hardware', 'ethernet', '01:23:45:67:89:ab'), tero@483: ], None), tero@483: ]) tero@483: tero@483: def testHostConflict(self): tero@483: hosts = [ tero@483: Host.build('foo', 'test1', tero@483: ethernet = '00:11:22:33:44:55', tero@483: ), tero@483: Host.build('foo', 'test2', tero@483: ethernet = '01:23:45:67:89:ab', tero@483: ), tero@483: ] tero@483: tero@483: with self.assertRaises(dhcp.HostDHCPError): tero@483: list(dhcp.dhcp_hosts(hosts)) tero@483: tero@483: def testHostMultinet(self): tero@483: hosts = [ tero@483: Host.build('foo', 'test1', tero@483: ip = '192.0.1.1', tero@483: ethernet = { 'eth1': '00:11:22:33:44:55' }, tero@483: ), tero@483: Host.build('foo', 'test2', tero@483: ip = '192.0.2.1', tero@483: ethernet = { 'eth2': '01:23:45:67:89:ab' }, tero@483: ), tero@483: ] tero@483: tero@483: self.assertBlocksEqual(list(dhcp.dhcp_hosts(hosts)), [ tero@483: (('host', 'foo-eth1'), [ tero@483: ('option', 'host-name', "foo"), tero@483: ('fixed-address', '192.0.1.1'), tero@483: ('hardware', 'ethernet', '00:11:22:33:44:55'), tero@483: ], None), tero@483: (('host', 'foo-eth2'), [ tero@483: ('option', 'host-name', "foo"), tero@483: ('fixed-address', '192.0.2.1'), tero@483: ('hardware', 'ethernet', '01:23:45:67:89:ab'), tero@483: ], None), tero@483: ]) tero@483: tero@483: tero@440: if __name__ == '__main__': tero@440: unittest.main()