1

import ipaddr

     2

import itertools

     3

import pvl.args

     4

import unittest

     5

     6

from pvl.hosts import config, dhcp, zone

     7

from pvl.hosts.host import Host

     8

from StringIO import StringIO

     9

    10

class ConfFile(StringIO):

    11

    def __init__(self, name, buffer):

    12

        StringIO.__init__(self, buffer)

    13

        self.name = name

    14

    15

class TestConfig(unittest.TestCase):

    16

    def setUp(self):

    17

        self.options = pvl.args.options(

    18

                hosts_charset   = 'utf-8',

    19

                hosts_domain    = None,

    20

                hosts_include   = None,

    21

        )

    22

    23

    def assertHostEqual(self, host, host_str, attrs):

    24

        self.assertEquals(str(host), host_str)

    25

    26

        for attr, value in attrs.iteritems():

    27

            self.assertEquals(getattr(host, attr), value)

    28

    29

    def assertHostsEqual(self, hosts, expected):

    30

        hosts = list(hosts)

    31

    32

        for host, expect in zip(hosts, expected):

    33

            host_str, attrs = expect

    34

    35

            self.assertHostEqual(host, host_str, attrs)

    36

    37

        self.assertEqual(len(hosts), len(expected))

    38

    39

    def testApplyHostConfigDict(self):

    40

        host = config.apply_host('foo', 'test', {

    41

            'ethernet.eth0': '00:11:22:33:44:55',

    42

        })

    43

    44

        self.assertHostEqual(host, 'foo@test', dict(

    45

                ethernet    = { 'eth0': '00:11:22:33:44:55' }

    46

        ))

    47

    48

    def testApplyHostConfigDictMulti(self):

    49

        host = config.apply_host('foo', 'test', {

    50

            'ethernet.eth0': '00:11:22:33:44:55',

    51

            'ethernet.eth1': '00:11:22:33:44:66',

    52

        })

    53

    54

        self.assertHostEqual(host, 'foo@test', dict(

    55

                ethernet    = {

    56

                    'eth0': '00:11:22:33:44:55',

    57

                    'eth1': '00:11:22:33:44:66',

    58

                }

    59

        ))

    60

    61

    def testApplyHostsConfigErrorExtra(self):

    62

        host = config.apply_host('foo', 'test', {

    63

            'ethernet': '00:11:22:33:44:55',

    64

            'ethernet.eth1': '00:11:22:33:44:66',

    65

        })

    66

    67

        self.assertHostEqual(host, 'foo@test', dict(

    68

                ethernet    = {

    69

                    None:   '00:11:22:33:44:55',

    70

                    'eth1': '00:11:22:33:44:66',

    71

                }

    72

        ))

    73

    74

    75

    def testApplyHostConfigExtensions(self):

    76

        host = config.apply_host('foo', 'test', {

    77

            'link:50':          'foo@test',

    78

            'link:uplink.49':   'bar@test',

    79

        })

    80

    81

        self.assertHostEqual(host, 'foo@test', dict(

    82

                extensions = {

    83

                    'link': {

    84

                        '50': 'foo@test',

    85

                        'uplink': { '49': 'bar@test' },

    86

                    },

    87

                },

    88

        ))

    89

    90

    def testApplyHostFqdn(self):

    91

        self.assertHostsEqual(config.apply_hosts('test', 'asdf@foo.test', { }), [

    92

                ('asdf@foo.test', dict()),

    93

        ])

    94

    95

        self.assertHostsEqual(config.apply_hosts('test', 'asdf.test2', { }), [

    96

                ('asdf.test2@', dict()),

    97

        ])

    98

    99

    def testApplyHostExpand(self):

   100

        self.assertHostsEqual(config.apply_hosts('test', 'asdf{1-3}', { 'ip': '10.100.100.$' }), [

   101

                ('asdf1@test', dict(ip=ipaddr.IPAddress('10.100.100.1'))),

   102

                ('asdf2@test', dict(ip=ipaddr.IPAddress('10.100.100.2'))),

   103

                ('asdf3@test', dict(ip=ipaddr.IPAddress('10.100.100.3'))),

   104

        ])

   105

   106

    def testApplyHostsFileError(self):

   107

        with self.assertRaises(config.HostConfigError):

   108

            list(config.apply_hosts_files(self.options, ['nonexistant']))

   109

   110

    def testApplyHostsConfig(self):

   111

        conf_file = ConfFile('test', """

   112

[foo]

   113

    ip = 127.0.0.1

   114

   115

[bar]

   116

    ip = 127.0.0.2

   117

        """)

   118

   119

        self.assertHostsEqual(config.apply_hosts_config(self.options, conf_file), [

   120

                ('foo@test', dict(ip=ipaddr.IPAddress('127.0.0.1'))),

   121

                ('bar@test', dict(ip=ipaddr.IPAddress('127.0.0.2'))),

   122

        ])

   123

   124

    def testApplyHostsConfigNested(self):

   125

        conf_file = ConfFile('test', """

   126

[asdf]

   127

    [[foo]]

   128

        ip = 127.0.0.1

   129

   130

[quux]

   131

    [[bar]]

   132

        ip = 127.0.0.2

   133

        """)

   134

   135

        self.assertHostsEqual(config.apply_hosts_config(self.options, conf_file), [

   136

                ('foo@asdf.test', dict(ip=ipaddr.IPAddress('127.0.0.1'))),

   137

                ('bar@quux.test', dict(ip=ipaddr.IPAddress('127.0.0.2'))),

   138

        ])

   139

   140

    def testApplyIncludes(self):

   141

        self.assertHostsEqual(config.apply_hosts_files(self.options, ['etc/hosts/test']), [

   142

                ('bar@test', dict(

   143

                    ip          = ipaddr.IPAddress('192.0.2.2'),

   144

                )),

   145

                ('foo@test', dict(

   146

                    ip          = ipaddr.IPAddress('192.0.2.1'),

   147

                )),

   148

        ])

   149

   150

    def testApplyIncludePath(self):

   151

        self.options.hosts_include = 'etc/hosts'

   152

        self.assertHostsEqual(config.apply_hosts_files(self.options, ['etc/zones/forward/test']), [

   153

                ('quux@asdf.test', dict(

   154

                    ip          = ipaddr.IPAddress('192.0.2.5'),

   155

                )),

   156

                ('bar@test', dict(

   157

                    ip          = ipaddr.IPAddress('192.0.2.2'),

   158

                )),

   159

                ('foo@test', dict(

   160

                    ip          = ipaddr.IPAddress('192.0.2.1'),

   161

                )),

   162

        ])

   163

   164

    def testApply(self):

   165

        self.assertHostsEqual(config.apply(self.options, ['etc/hosts/example.com']), [

   166

                ('foo@example.com', dict(

   167

                    ip          = ipaddr.IPAddress('192.0.2.1'),

   168

                    ethernet    = {None: '00:11:22:33:44:55'},

   169

                )),

   170

                ('bar@example.com', dict(

   171

                    ip          = ipaddr.IPAddress('192.0.2.2'),

   172

                    ethernet    = {None: '01:23:45:67:89:ab'},

   173

                )),

   174

        ])

   175

   176

class TestZoneMixin(object):

   177

    def assertZoneEquals(self, rrs, expected):

   184

        gather = { }

   185

   186

        for rr in rrs:

   187

            key = (rr.name.lower(), rr.type.upper())

   196

   197

        self.assertDictEqual(gather, expected)

   198

   199

   200

class TestForwardZone(TestZoneMixin, unittest.TestCase):

   201

    def testHostOutOfOrigin(self):

   202

        h = Host.build('host', 'domain', 

   203

                ip  = '10.0.0.1',

   204

        )

   205

   206

        self.assertZoneEquals(zone.host_forward(h, 'test'), { })

   207

   208

    def testHostIP(self):

   209

        h = Host.build('host', 'domain',

   210

                ip  = '192.0.2.1',

   211

                ip6 = '2001:db8::192.0.2.1',

   212

        )

   213

   214

        self.assertZoneEquals(zone.host_forward(h, 'domain'), {

   217

        })

   218

   219

    def testHostAlias(self):

   220

        h = Host.build('host', 'domain',

   221

                ip      = '192.0.2.1',

   222

                alias   = 'test *.test',

   223

        )

   224

   225

        self.assertEquals(h.alias, ['test', '*.test'])

   226

   227

        self.assertZoneEquals(zone.host_forward(h, 'domain'), {

   231

        })

   232

   233

    def testHostAlias46(self):

   234

        h = Host.build('host', 'domain',

   235

                ip      = '192.0.2.1',

   236

                ip6     = '2001:db8::192.0.2.1',

   237

                alias4  = 'test4',

   238

                alias6  = 'test6',

   239

        )

   240

   241

        self.assertZoneEquals(zone.host_forward(h, 'domain'), {

   246

        })

   247

   248

    def testHostAlias4Missing(self):

   249

        h = Host.build('host', 'domain',

   250

                ip6     = '2001:db8::192.0.2.1',

   251

                alias4  = 'test4',

   252

                alias6  = 'test6',

   253

        )

   254

   255

        with self.assertRaises(zone.HostZoneError):

   256

            self.assertZoneEquals(zone.host_forward(h, 'domain'), { })

   257

   258

    def testHostAlias6Missing(self):

   259

        h = Host.build('host', 'domain',

   260

                ip      = '192.0.2.1',

   261

                alias4  = 'test4',

   262

                alias6  = 'test6',

   263

        )

   264

   265

        with self.assertRaises(zone.HostZoneError):

   266

            self.assertZoneEquals(zone.host_forward(h, 'domain'), { })

   267

   268

    def testHostFQDN(self):

   269

        h = Host.build('host.example.net', None,

   270

                ip          = '192.0.2.3',

   271

        )

   272

   273

        self.assertZoneEquals(zone.host_forward(h, 'example.com'), {

   274

   275

        })

   276

   277

    def testHostDelegate(self):

   278

        h = Host.build('host', 'example.com',

   279

                forward = 'host.example.net',

   280

        )

   281

   282

        self.assertZoneEquals(zone.host_forward(h, 'example.com'), {

   284

        })

   285

   286

    def testHostForwardAlias(self):

   287

        h = Host.build('host', 'domain',

   288

                forward = 'host.example.net',

   289

                alias   = 'test',

   290

        )

   291

   292

        self.assertZoneEquals(zone.host_forward(h, 'domain'), {

   295

        })

   296

   297

    def testHostLocation(self):

   298

        h = Host.build('host', 'domain',

   299

                ip          = '192.0.2.1',

   300

                location    = 'test',

   301

        )

   302

   303

        self.assertEquals(h.location, ('test', 'domain'))

   304

   305

        self.assertZoneEquals(zone.host_forward(h, 'domain'), {

   308

        })

   309

   310

    def testHostLocationDomain(self):

   311

        h = Host.build('host', 'foo.domain',

   312

                ip          = '192.0.2.1',

   313

                location    = 'test@bar.domain',

   314

        )

   315

   316

        self.assertEquals(h.location, ('test', 'bar.domain'))

   317

   318

        self.assertZoneEquals(zone.host_forward(h, 'domain'), {

   321

        })

   322

   323

    def testHostLocationDomainOutOfOrigin(self):

   324

        h = Host.build('host', 'foo.domain',

   325

                ip          = '192.0.2.1',

   326

                location    = 'test@bar.domain',

   327

        )

   328

   329

        self.assertEquals(h.location, ('test', 'bar.domain'))

   330

   331

        with self.assertRaises(zone.HostZoneError):

   332

            self.assertZoneEquals(zone.host_forward(h, 'foo.domain'), {

   334

            })

   335

   336

        # TODO

   337

        #self.assertZoneEquals(zone.host_forward(h, 'bar.domain'), {

   338

        #    ('test', 'CNAME'): ['host.foo'],

   339

        #})

   340

   341

    def testHostsForward(self):

   342

        hosts = [

   343

                Host.build('foo', 'domain',

   344

                    ip      = '192.0.2.1',

   345

                    ip6     = '2001:db8::192.0.2.1',

   346

                    alias   = 'test',

   347

                ),

   348

                Host.build('bar', 'domain',

   349

                    ip      = '192.0.2.2',

   350

                ),

   351

                Host.build('quux', 'example',

   352

                    ip      = '192.0.2.3',

   353

                ),

   354

        ]

   355

   356

        rrs = zone.apply_hosts_forward(hosts, 'domain', add_origin=True)

   357

   358

        # handle the $ORIGIN directive

   359

        rd = next(rrs)

   360

   361

        self.assertEquals(unicode(rd), '$ORIGIN\tdomain.')

   362

   363

        self.assertZoneEquals(rrs, {

   386

        })

   387

   388

    def testHostsConflict(self):

   389

        hosts = [

   390

                Host.build('foo', 'domain',

   391

                    ip      = '192.0.2.1',

   392

                ),

   393

                Host.build('foo', 'domain',

   394

                    ip      = '192.0.2.2',

   395

                )

   396

        ]

   397

   398

        with self.assertRaises(zone.HostZoneError):

   399

            self.assertZoneEquals(zone.apply_hosts_forward(hosts, 'domain', check_conflicts=True), { })

   400

   401

    def testHostsAliasConflict(self):

   402

        hosts = [

   403

                Host.build('foo', 'domain',

   404

                    ip          = '192.0.2.1',

   405

                ),

   406

                Host.build('bar', 'domain',

   407

                    ip          = '192.0.2.2',

   408

                    alias       = 'foo',

   409

                )

   410

        ]

   411

   412

        # with A first

   413

        with self.assertRaises(zone.HostZoneError):

   414

            self.assertZoneEquals(zone.apply_hosts_forward(hosts, 'domain'), { })

   415

   416

        # also with CNAME first

   417

        with self.assertRaises(zone.HostZoneError):

   418

            self.assertZoneEquals(zone.apply_hosts_forward(reversed(hosts), 'domain'), { })

   419

   420

    def testHostsAlias4Conflict(self):

   421

        hosts = [

   422

                Host.build('foo', 'domain',

   423

                    ip          = '192.0.2.1',

   424

                ),

   425

                Host.build('bar', 'domain',

   426

                    ip          = '192.0.2.2',

   427

                    alias4      = 'foo',

   428

                )

   429

        ]

   430

   431

        with self.assertRaises(zone.HostZoneError):

   432

            self.assertZoneEquals(zone.apply_hosts_forward(hosts, 'domain', check_conflicts=True), { })

   433

   434

   435

class TestReverseZone(TestZoneMixin, unittest.TestCase):

   436

    def testHostIP(self):

   437

        h = Host.build('host', 'domain',

   438

                ip  = '192.0.2.1',

   439

                ip6 = '2001:db8::192.0.2.1',

   440

        )

   441

   442

        self.assertZoneEquals((rr for ip, rr in zone.host_reverse(h, ipaddr.IPNetwork('192.0.2.1/24'))), {

   444

        })

   445

   446

        self.assertZoneEquals((rr for ip, rr in zone.host_reverse(h, ipaddr.IPNetwork('2001:db8::/64'))), {

   448

        })

   449

   450

    def testHostIP4(self):

   451

        h = Host.build('host', 'domain',

   452

                ip  = '192.0.2.1',

   453

        )

   454

   455

        self.assertZoneEquals((rr for ip, rr in zone.host_reverse(h, ipaddr.IPNetwork('192.0.2.1/24'))), {

   457

        })

   458

   459

        self.assertZoneEquals((rr for ip, rr in zone.host_reverse(h, ipaddr.IPNetwork('192.0.0.0/16'))), {

   461

        })

   462

   463

        self.assertZoneEquals((rr for ip, rr in zone.host_reverse(h, ipaddr.IPNetwork('192.0.0.0/12'))), {

   465

        })

   466

   467

        self.assertZoneEquals((rr for ip, rr in zone.host_reverse(h, ipaddr.IPNetwork('2001:db8::/64'))), {

   468

   469

        })

   470

   471

    def testHostIP6(self):

   472

        h = Host.build('host', 'domain',

   473

                ip6 = '2001:db8::192.0.2.1',

   474

        )

   475

   476

        self.assertZoneEquals((rr for ip, rr in zone.host_reverse(h, ipaddr.IPNetwork('192.0.2.1/24'))), {

   477

        })

   478

   479

        self.assertZoneEquals((rr for ip, rr in zone.host_reverse(h, ipaddr.IPNetwork('2001:db8::/64'))), {

   481

        })

   482

   483

    def testHostIPOutOfPrefix(self):

   484

        h = Host.build('host', 'domain',

   485

                ip  = '192.0.2.1',

   486

                ip6 = '2001:db8::192.0.2.1',

   487