pvl.hosts.config: support direct directory hosts, using the directory name as the parent
import ipaddr
import pvl.args
import unittest
from pvl.hosts import config, dhcp, zone
from pvl.hosts.host import Host
from StringIO import StringIO
class ConfFile(StringIO):
def __init__(self, name, buffer):
StringIO.__init__(self, buffer)
self.name = name
class TestConfig(unittest.TestCase):
def setUp(self):
self.options = pvl.args.options(
hosts_charset = 'utf-8',
hosts_domain = None,
hosts_include = None,
)
def assertHostEqual(self, host, host_str, attrs):
self.assertEquals(str(host), host_str)
for attr, value in attrs.iteritems():
self.assertEquals(getattr(host, attr), value)
def assertHostsEqual(self, hosts, expected):
hosts = list(hosts)
for host, expect in zip(hosts, expected):
host_str, attrs = expect
self.assertHostEqual(host, host_str, attrs)
self.assertEqual(len(hosts), len(expected))
def testApplyHostConfigDict(self):
host = config.apply_host('foo', 'test', {
'ethernet.eth0': '00:11:22:33:44:55',
})
self.assertHostEqual(host, 'foo@test', dict(
ethernet = { 'eth0': '00:11:22:33:44:55' }
))
def testApplyHostConfigDictMulti(self):
host = config.apply_host('foo', 'test', {
'ethernet.eth0': '00:11:22:33:44:55',
'ethernet.eth1': '00:11:22:33:44:66',
})
self.assertHostEqual(host, 'foo@test', dict(
ethernet = {
'eth0': '00:11:22:33:44:55',
'eth1': '00:11:22:33:44:66',
}
))
def testApplyHostsConfigErrorDict(self):
with self.assertRaises(ValueError):
config.apply_host('test', 'foo', {
'ethernet': 'foo',
'ethernet.eth0': 'bar',
})
def testApplyHostConfigExtensions(self):
host = config.apply_host('foo', 'test', {
'link:50': 'foo@test',
'link:uplink.49': 'bar@test',
})
self.assertHostEqual(host, 'foo@test', dict(
extensions = {
'link': {
'50': 'foo@test',
'uplink': { '49': 'bar@test' },
},
},
))
def testApplyHostFqdn(self):
self.assertHostsEqual(config.apply_hosts('test', 'asdf@foo.test', { }), [
('asdf@foo.test', dict()),
])
self.assertHostsEqual(config.apply_hosts('test', 'asdf.test2', { }), [
('asdf.test2@', dict()),
])
def testApplyHostExpand(self):
self.assertHostsEqual(config.apply_hosts('test', 'asdf{1-3}', { 'ip': '10.100.100.$' }), [
('asdf1@test', dict(ip=ipaddr.IPAddress('10.100.100.1'))),
('asdf2@test', dict(ip=ipaddr.IPAddress('10.100.100.2'))),
('asdf3@test', dict(ip=ipaddr.IPAddress('10.100.100.3'))),
])
def testApplyHostsFileError(self):
with self.assertRaises(config.HostConfigError):
list(config.apply_hosts_files(self.options, ['nonexistant']))
def testApplyHostsFile(self):
conf_file = ConfFile('test', """
[foo]
ip = 127.0.0.1
[bar]
ip = 127.0.0.2
""")
expected = [
('foo@test', dict(ip=ipaddr.IPAddress('127.0.0.1'))),
('bar@test', dict(ip=ipaddr.IPAddress('127.0.0.2'))),
]
self.assertHostsEqual(config.apply_hosts_config(self.options, conf_file), expected)
def testApplyIncludes(self):
self.assertHostsEqual(config.apply_hosts_files(self.options, ['etc/hosts/includes.test']), [
('foo@includes.test', dict(
ip = ipaddr.IPAddress('192.0.2.1'),
)),
('bar@includes.test', dict(
ip = ipaddr.IPAddress('192.0.2.2'),
)),
('quux@includes.test', dict(
ip = ipaddr.IPAddress('192.0.2.3'),
)),
])
def testApplyDirectory(self):
self.assertHostsEqual(config.apply_hosts_files(self.options, ['etc/hosts/included.test/']), [
('foo@included.test', dict(
ip = ipaddr.IPAddress('192.0.2.1'),
)),
('bar@included.test', dict(
ip = ipaddr.IPAddress('192.0.2.2'),
)),
])
def testApply(self):
self.assertHostsEqual(config.apply(self.options, ['etc/hosts/test']), [
('foo@test', dict(
ip = ipaddr.IPAddress('127.0.0.1'),
ethernet = {None: '00:11:22:33:44:55'},
)),
('bar@test', dict(
ip = ipaddr.IPAddress('127.0.0.2'),
ethernet = {None: '01:23:45:67:89:ab'},
)),
])
class TestZoneMixin(object):
def assertZoneEquals(self, rrs, expected):
gather = { }
for rr in rrs:
key = (rr.name.lower(), rr.type.upper())
self.assertNotIn(key, gather)
gather[key] = rr.data
self.assertDictEqual(gather, expected)
class TestForwardZone(TestZoneMixin, unittest.TestCase):
def testHostOutOfOrigin(self):
h = Host.build('host', 'domain',
ip = '10.0.0.1',
)
self.assertZoneEquals(zone.host_forward(h, 'test'), { })
def testHostIP(self):
h = Host.build('host', 'domain',
ip = '192.0.2.1',
ip6 = '2001:db8::192.0.2.1',
)
self.assertZoneEquals(zone.host_forward(h, 'domain'), {
('host', 'A'): ['192.0.2.1'],
('host', 'AAAA'): ['2001:db8::c000:201'],
})
def testHostAlias(self):
h = Host.build('host', 'domain',
ip = '192.0.2.1',
alias = 'test *.test',
)
self.assertEquals(h.alias, ['test', '*.test'])
self.assertZoneEquals(zone.host_forward(h, 'domain'), {
('host', 'A'): ['192.0.2.1'],
('test', 'CNAME'): ['host'],
('*.test', 'CNAME'): ['host'],
})
def testHostAlias46(self):
h = Host.build('host', 'domain',
ip = '192.0.2.1',
ip6 = '2001:db8::192.0.2.1',
alias4 = 'test4',
alias6 = 'test6',
)
self.assertZoneEquals(zone.host_forward(h, 'domain'), {
('host', 'A'): ['192.0.2.1'],
('host', 'AAAA'): ['2001:db8::c000:201'],
('test4', 'A'): ['192.0.2.1'],
('test6', 'AAAA'): ['2001:db8::c000:201'],
})
def testHostAlias4Missing(self):
h = Host.build('host', 'domain',
ip6 = '2001:db8::192.0.2.1',
alias4 = 'test4',
alias6 = 'test6',
)
with self.assertRaises(zone.HostZoneError):
self.assertZoneEquals(zone.host_forward(h, 'domain'), { })
def testHostAlias6Missing(self):
h = Host.build('host', 'domain',
ip = '192.0.2.1',
alias4 = 'test4',
alias6 = 'test6',
)
with self.assertRaises(zone.HostZoneError):
self.assertZoneEquals(zone.host_forward(h, 'domain'), { })
def testHostFQDN(self):
h = Host.build('host.example.net', None,
ip = '192.0.2.3',
)
self.assertZoneEquals(zone.host_forward(h, 'example.com'), {
})
def testHostDelegate(self):
h = Host.build('host', 'example.com',
forward = 'host.example.net',
)
self.assertZoneEquals(zone.host_forward(h, 'example.com'), {
('host', 'CNAME'): ['host.example.net.'],
})
def testHostForwardAlias(self):
h = Host.build('host', 'domain',
forward = 'host.example.net',
alias = 'test',
)
self.assertZoneEquals(zone.host_forward(h, 'domain'), {
('host', 'CNAME'): ['host.example.net.'],
('test', 'CNAME'): ['host'],
})
def testHostLocation(self):
h = Host.build('host', 'domain',
ip = '192.0.2.1',
location = 'test',
)
self.assertEquals(h.location, ('test', 'domain'))
self.assertZoneEquals(zone.host_forward(h, 'domain'), {
('host', 'A'): ['192.0.2.1'],
('test', 'CNAME'): ['host'],
})
def testHostLocationDomain(self):
h = Host.build('host', 'foo.domain',
ip = '192.0.2.1',
location = 'test@bar.domain',
)
self.assertEquals(h.location, ('test', 'bar.domain'))
self.assertZoneEquals(zone.host_forward(h, 'domain'), {
('host.foo', 'A'): ['192.0.2.1'],
('test.bar', 'CNAME'): ['host.foo'],
})
def testHostLocationDomainOutOfOrigin(self):
h = Host.build('host', 'foo.domain',
ip = '192.0.2.1',
location = 'test@bar.domain',
)
self.assertEquals(h.location, ('test', 'bar.domain'))
with self.assertRaises(zone.HostZoneError):
self.assertZoneEquals(zone.host_forward(h, 'foo.domain'), {
('host', 'A'): ['192.0.2.1'],
})
# TODO
#self.assertZoneEquals(zone.host_forward(h, 'bar.domain'), {
# ('test', 'CNAME'): ['host.foo'],
#})
def testHostsForward(self):
hosts = [
Host.build('foo', 'domain',
ip = '192.0.2.1',
ip6 = '2001:db8::192.0.2.1',
alias = 'test',
),
Host.build('bar', 'domain',
ip = '192.0.2.2',
),
Host.build('quux', 'example',
ip = '192.0.2.3',
),
]
rrs = zone.apply_hosts_forward(hosts, 'domain', add_origin=True)
# handle the $ORIGIN directive
rd = next(rrs)
self.assertEquals(unicode(rd), '$ORIGIN\tdomain.')
self.assertZoneEquals(rrs, {
('foo', 'A'): ['192.0.2.1'],
('foo', 'AAAA'): ['2001:db8::c000:201'],
('test', 'CNAME'): ['foo'],
('bar', 'A'): ['192.0.2.2'],
})
def testHostsConflict(self):
hosts = [
Host.build('foo', 'domain',
ip = '192.0.2.1',
),
Host.build('foo', 'domain',
ip = '192.0.2.2',
)
]
with self.assertRaises(zone.HostZoneError):
self.assertZoneEquals(zone.apply_hosts_forward(hosts, 'domain'), { })
def testHostsAliasConflict(self):
hosts = [
Host.build('foo', 'domain',
ip = '192.0.2.1',
),
Host.build('bar', 'domain',
ip = '192.0.2.2',
alias = 'foo',
)
]
# with A first
with self.assertRaises(zone.HostZoneError):
self.assertZoneEquals(zone.apply_hosts_forward(hosts, 'domain'), { })
# also with CNAME first
with self.assertRaises(zone.HostZoneError):
self.assertZoneEquals(zone.apply_hosts_forward(reversed(hosts), 'domain'), { })
def testHostsAlias4Conflict(self):
hosts = [
Host.build('foo', 'domain',
ip = '192.0.2.1',
),
Host.build('bar', 'domain',
ip = '192.0.2.2',
alias4 = 'foo',
)
]
with self.assertRaises(zone.HostZoneError):
self.assertZoneEquals(zone.apply_hosts_forward(hosts, 'domain'), { })
class TestReverseZone(TestZoneMixin, unittest.TestCase):
def testHostIP(self):
h = Host.build('host', 'domain',
ip = '192.0.2.1',
ip6 = '2001:db8::192.0.2.1',
)
self.assertZoneEquals((rr for ip, rr in zone.host_reverse(h, ipaddr.IPNetwork('192.0.2.1/24'))), {
('1', 'PTR'): ['host.domain.'],
})
self.assertZoneEquals((rr for ip, rr in zone.host_reverse(h, ipaddr.IPNetwork('2001:db8::/64'))), {
('1.0.2.0.0.0.0.c.0.0.0.0.0.0.0.0', 'PTR'): ['host.domain.'],
})
def testHostIP4(self):
h = Host.build('host', 'domain',
ip = '192.0.2.1',
)
self.assertZoneEquals((rr for ip, rr in zone.host_reverse(h, ipaddr.IPNetwork('192.0.2.1/24'))), {
('1', 'PTR'): ['host.domain.'],
})
self.assertZoneEquals((rr for ip, rr in zone.host_reverse(h, ipaddr.IPNetwork('192.0.0.0/16'))), {
('1.2', 'PTR'): ['host.domain.'],
})
self.assertZoneEquals((rr for ip, rr in zone.host_reverse(h, ipaddr.IPNetwork('192.0.0.0/12'))), {
('1.2.0', 'PTR'): ['host.domain.'],
})
self.assertZoneEquals((rr for ip, rr in zone.host_reverse(h, ipaddr.IPNetwork('2001:db8::/64'))), {
})
def testHostIP6(self):
h = Host.build('host', 'domain',
ip6 = '2001:db8::192.0.2.1',
)
self.assertZoneEquals((rr for ip, rr in zone.host_reverse(h, ipaddr.IPNetwork('192.0.2.1/24'))), {
})
self.assertZoneEquals((rr for ip, rr in zone.host_reverse(h, ipaddr.IPNetwork('2001:db8::/64'))), {
('1.0.2.0.0.0.0.c.0.0.0.0.0.0.0.0', 'PTR'): ['host.domain.'],
})
def testHostIPOutOfPrefix(self):
h = Host.build('host', 'domain',
ip = '192.0.2.1',
ip6 = '2001:db8::192.0.2.1',
)
self.assertZoneEquals((rr for ip, rr in zone.host_reverse(h, ipaddr.IPNetwork('192.0.1.0/24'))), {
})
self.assertZoneEquals((rr for ip, rr in zone.host_reverse(h, ipaddr.IPNetwork('2001:db8:1::/64'))), {
})
def testHostFQDN(self):
h = Host.build('host.example.net', None,
ip = '192.0.2.3',
ip6 = '2001:db8::192.0.2.3',
)
self.assertZoneEquals((rr for ip, rr in zone.host_reverse(h, ipaddr.IPNetwork('192.0.2.1/24'))), {
('3', 'PTR'): ['host.example.net.'],
})
self.assertZoneEquals((rr for ip, rr in zone.host_reverse(h, ipaddr.IPNetwork('2001:db8::/64'))), {
('3.0.2.0.0.0.0.c.0.0.0.0.0.0.0.0', 'PTR'): ['host.example.net.'],
})
def testHostDelegate(self):
h = Host.build('host', 'example.com',
ip = '192.0.2.1',
ip6 = '2001:db8::192.0.2.1',
forward = '',
reverse = '1.0/28.2.0.192.in-addr.arpa',
)
self.assertZoneEquals(zone.host_forward(h, 'example.com'), {
})
self.assertZoneEquals((rr for ip, rr in zone.host_reverse(h, ipaddr.IPNetwork('192.0.2.1/24'))), {
('1', 'CNAME'): ['1.0/28.2.0.192.in-addr.arpa.'],
})
self.assertZoneEquals((rr for ip, rr in zone.host_reverse(h, ipaddr.IPNetwork('2001:db8::/64'))), {
})
def testHosts(self):
hosts = [
Host.build('foo', 'domain',
ip = '192.0.2.1',
),
Host.build('bar', 'domain',
ip = '192.0.2.2',
)
]
self.assertZoneEquals(zone.apply_hosts_reverse(hosts, ipaddr.IPNetwork('192.0.2.1/24')), {
('1', 'PTR'): ['foo.domain.'],
('2', 'PTR'): ['bar.domain.'],
})
# in ip order
self.assertZoneEquals(zone.apply_hosts_reverse(reversed(hosts), ipaddr.IPNetwork('192.0.2.1/24')), {
('1', 'PTR'): ['foo.domain.'],
('2', 'PTR'): ['bar.domain.'],
})
def testHostsConflict(self):
hosts = [
Host.build('foo', 'domain',
ip = '192.0.2.1',
),
Host.build('bar', 'domain',
ip = '192.0.2.1',
)
]
with self.assertRaises(zone.HostZoneError):
self.assertZoneEquals(zone.apply_hosts_reverse(hosts, ipaddr.IPNetwork('192.0.2.1/24')), { })
def testHostsGenerateUnknown(self):
hosts = [
Host.build('foo', 'domain',
ip = '192.0.2.1',
),
Host.build('bar', 'domain',
ip = '192.0.2.5',
),
]
self.assertZoneEquals(zone.apply_hosts_reverse(hosts, ipaddr.IPNetwork('192.0.2.1/29'),
unknown_host = 'ufc',
unknown_domain = 'domain',
), {
('1', 'PTR'): ['foo.domain.'],
('2', 'PTR'): ['ufc.domain.'],
('3', 'PTR'): ['ufc.domain.'],
('4', 'PTR'): ['ufc.domain.'],
('5', 'PTR'): ['bar.domain.'],
('6', 'PTR'): ['ufc.domain.'],
})
class TestDhcp(unittest.TestCase):
def assertBlocksEqual(self, blockdefs, expected):
for (_block, _items, _opts), (block, items, opts) in zip(blockdefs, expected):
self.assertEqual(_block, block)
self.assertItemsEqual(_items, items)
if opts is not None:
self.assertEqual(_opts, opts)
self.assertEqual(len(blockdefs), len(expected))
def testHost(self):
host = Host.build('foo', 'test',
ip = '192.0.2.1',
ethernet = '00:11:22:33:44:55',
owner = 'foo',
)
self.assertBlocksEqual(list(dhcp.dhcp_host(host)), [
(('host', 'foo'), [
('option', 'host-name', "foo"),
('fixed-address', '192.0.2.1'),
('hardware', 'ethernet', '00:11:22:33:44:55'),
], dict(comment="Owner: foo"))
])
def testHostStatic(self):
host = Host.build('foo', 'test',
ip = '192.0.2.1',
)
self.assertBlocksEqual(list(dhcp.dhcp_host(host)), [
])
def testHostDynamic(self):
host = Host.build('foo', 'test',
ethernet = '00:11:22:33:44:55',
)
self.assertBlocksEqual(list(dhcp.dhcp_host(host)), [
(('host', 'foo'), [
('option', 'host-name', "foo"),
('hardware', 'ethernet', '00:11:22:33:44:55'),
], None)
])
def testHostBoot(self):
hosts = [
Host.build('foo1', 'test',
ethernet = '00:11:22:33:44:55',
boot = 'boot.lan:debian/wheezy/pxelinux.0',
),
Host.build('foo2', 'test',
ethernet = '00:11:22:33:44:55',
boot = 'boot.lan:',
),
Host.build('foo3', 'test',
ethernet = '00:11:22:33:44:55',
boot = '/debian/wheezy/pxelinux.0',
),
]
self.assertBlocksEqual(list(dhcp.dhcp_hosts(hosts)), [
(('host', 'foo1'), [
('option', 'host-name', "foo1"),
('hardware', 'ethernet', '00:11:22:33:44:55'),
('next-server', 'boot.lan'),
('filename', 'debian/wheezy/pxelinux.0'),
], None),
(('host', 'foo2'), [
('option', 'host-name', "foo2"),
('hardware', 'ethernet', '00:11:22:33:44:55'),
('next-server', 'boot.lan'),
], None),
(('host', 'foo3'), [
('option', 'host-name', "foo3"),
('hardware', 'ethernet', '00:11:22:33:44:55'),
('filename', 'debian/wheezy/pxelinux.0'),
], None),
])
def testHosts(self):
hosts = [
Host.build('foo', 'test',
ip = '192.0.2.1',
ethernet = '00:11:22:33:44:55',
),
Host.build('bar', 'test',
ip = '192.0.2.2',
ethernet = '01:23:45:67:89:ab',
),
]
self.assertBlocksEqual(list(dhcp.dhcp_hosts(hosts)), [
(('host', 'foo'), [
('option', 'host-name', "foo"),
('fixed-address', '192.0.2.1'),
('hardware', 'ethernet', '00:11:22:33:44:55'),
], None),
(('host', 'bar'), [
('option', 'host-name', "bar"),
('fixed-address', '192.0.2.2'),
('hardware', 'ethernet', '01:23:45:67:89:ab'),
], None),
])
def testHostConflict(self):
hosts = [
Host.build('foo', 'test1',
ethernet = '00:11:22:33:44:55',
),
Host.build('foo', 'test2',
ethernet = '01:23:45:67:89:ab',
),
]
with self.assertRaises(dhcp.HostDHCPError):
list(dhcp.dhcp_hosts(hosts))
def testHostMultinet(self):
hosts = [
Host.build('foo', 'test1',
ip = '192.0.1.1',
ethernet = { 'eth1': '00:11:22:33:44:55' },
),
Host.build('foo', 'test2',
ip = '192.0.2.1',
ethernet = { 'eth2': '01:23:45:67:89:ab' },
),
]
self.assertBlocksEqual(list(dhcp.dhcp_hosts(hosts)), [
(('host', 'foo-eth1'), [
('option', 'host-name', "foo"),
('fixed-address', '192.0.1.1'),
('hardware', 'ethernet', '00:11:22:33:44:55'),
], None),
(('host', 'foo-eth2'), [
('option', 'host-name', "foo"),
('fixed-address', '192.0.2.1'),
('hardware', 'ethernet', '01:23:45:67:89:ab'),
], None),
])
if __name__ == '__main__':
unittest.main()