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

   160

    def testApplyIncludes(self):

   161

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

   162

                ('bar@test', dict(

   163

                    ip          = ipaddr.IPAddress('192.0.2.2'),

   164

                )),

   165

                ('foo@test', dict(

   166

                    ip          = ipaddr.IPAddress('192.0.2.1'),

   167

                )),

   168

        ])

   169

   170

    def testApplyIncludePath(self):

   171

        self.options.hosts_include = 'etc/hosts'

   172

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

   173

                ('quux@asdf.test', dict(

   174

                    ip          = ipaddr.IPAddress('192.0.2.5'),

   175

                )),

   176

                ('bar@test', dict(

   177

                    ip          = ipaddr.IPAddress('192.0.2.2'),

   178

                )),

   179

                ('foo@test', dict(

   180

                    ip          = ipaddr.IPAddress('192.0.2.1'),

   181

                )),

   182

        ])

   183

   184

    def testApply(self):

   185

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

   186

                ('foo@example.com', dict(

   187

                    ip          = ipaddr.IPAddress('192.0.2.1'),

   188

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

   189

                )),

   190

                ('bar@example.com', dict(

   191

                    ip          = ipaddr.IPAddress('192.0.2.2'),

   192

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

   193

                )),

   194

        ])

   195

   196

class TestZoneMixin(object):

   197

    def assertZoneEquals(self, rrs, expected):

   198

        """

   199

            Tests that the given list of ZoneRecords is equal to the given {(rr.name, rr.type): str(rr.data)} dict.

   200

   201

            Multiple records for the same name/type are gathered as a list. XXX: ordering

   202

        """

   203

   204

        gather = { }

   205

   206

        for rr in rrs:

   207

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

   208

            value = '\t'.join(rr.data)

   209

   210

            if key not in gather:

   211

                gather[key] = value

   212

            elif not isinstance(gather[key], list):

   213

                gather[key] = [gather[key], value]

   214

            else:

   215

                gather[key].append(value)

   216

   217

        self.assertDictEqual(gather, expected)

   218

   219

   220

class TestForwardZone(TestZoneMixin, unittest.TestCase):

   221

    def testHostOutOfOrigin(self):

   222

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

   223

                ip  = '10.0.0.1',

   224

        )

   225

   226

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

   227

   228

    def testHostIP(self):

   229

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

   230

                ip  = '192.0.2.1',

   231

                ip6 = '2001:db8::192.0.2.1',

   232

        )

   233

   234

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

   235

            ('host', 'A'): '192.0.2.1',

   236

            ('host', 'AAAA'): '2001:db8::c000:201',

   237

        })

   238

   239

    def testHostAlias(self):

   240

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

   241

                ip      = '192.0.2.1',

   242

                alias   = 'test *.test',

   243

        )

   244

   245

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

   246

   247

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

   248

            ('host', 'A'): '192.0.2.1',

   249

            ('test', 'CNAME'): 'host',

   250

            ('*.test', 'CNAME'): 'host',

   251

        })

   252

   253

    def testHostAlias46(self):

   254

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

   255

                ip      = '192.0.2.1',

   256

                ip6     = '2001:db8::192.0.2.1',

   257

                alias4  = 'test4',

   258

                alias6  = 'test6',

   259

        )

   260

   261

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

   262

            ('host', 'A'): '192.0.2.1',

   263

            ('host', 'AAAA'): '2001:db8::c000:201',

   264

            ('test4', 'A'): '192.0.2.1',

   265

            ('test6', 'AAAA'): '2001:db8::c000:201',

   266

        })

   267

   268

    def testHostAlias4Missing(self):

   269

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

   270

                ip6     = '2001:db8::192.0.2.1',

   271

                alias4  = 'test4',

   272

                alias6  = 'test6',

   273

        )

   274

   275

        with self.assertRaises(zone.HostZoneError):

   276

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

   277

   278

    def testHostAlias6Missing(self):

   279

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

   280

                ip      = '192.0.2.1',

   281

                alias4  = 'test4',

   282

                alias6  = 'test6',

   283

        )

   284

   285

        with self.assertRaises(zone.HostZoneError):

   286

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

   287

   288

    def testHostFQDN(self):

   289

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

   290

                ip          = '192.0.2.3',

   291

        )

   292

   293

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

   294

   295

        })

   296

   297

    def testHostDelegate(self):

   298

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

   299

                forward = 'host.example.net',

   300

        )

   301

   302

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

   303

            ('host', 'CNAME'): 'host.example.net.',

   304

        })

   305

   306

    def testHostForwardAlias(self):

   307

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

   308

                forward = 'host.example.net',

   309

                alias   = 'test',

   310

        )

   311

   312

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

   313

            ('host', 'CNAME'): 'host.example.net.',

   314

            ('test', 'CNAME'): 'host',

   315

        })

   316

   317

    def testHostLocation(self):

   318

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

   319

                ip          = '192.0.2.1',

   320

                location    = 'test',

   321

        )

   322

   323

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

   324

   325

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

   326

            ('host', 'A'): '192.0.2.1',

   327

            ('test', 'CNAME'): 'host',

   328

        })

   329

   330

    def testHostLocationDomain(self):

   331

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

   332

                ip          = '192.0.2.1',

   333

                location    = 'test@bar.domain',

   334

        )

   335

   336

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

   337

   338

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

   339

            ('host.foo', 'A'): '192.0.2.1',

   340

            ('test.bar', 'CNAME'): 'host.foo',

   341

        })

   342

   343

    def testHostLocationDomainOutOfOrigin(self):

   344

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

   345

                ip          = '192.0.2.1',

   346

                location    = 'test@bar.domain',

   347

        )

   348

   349

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

   350

   351

        with self.assertRaises(zone.HostZoneError):

   352

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

   353

                ('host', 'A'): '192.0.2.1',

   354

            })

   355

   356

        # TODO

   357

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

   358

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

   359

        #})

   360

   361

    def testHostsForward(self):

   362

        hosts = [

   363

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

   364

                    ip      = '192.0.2.1',

   365

                    ip6     = '2001:db8::192.0.2.1',

   366

                    alias   = 'test',

   367

                ),

   368

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

   369

                    ip      = '192.0.2.2',

   370

                ),

   371

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

   372

                    ip      = '192.0.2.3',

   373

                ),

   374

        ]

   375

   376

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

   377

   378

        # handle the $ORIGIN directive

   379

        rd = next(rrs)

   380

   381

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

   382

   383

        self.assertZoneEquals(rrs, {

   384

            ('foo', 'A'): '192.0.2.1',

   385

            ('foo', 'AAAA'): '2001:db8::c000:201',

   386

            ('test', 'CNAME'): 'foo',

   387

            ('bar', 'A'): '192.0.2.2',

   388

        })

   389

   390

    def testHostsMultiAlias(self):

   391

        hosts = [

   392

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

   393

                    ip      = '192.0.2.1',

   394

                    alias4  = 'test',

   395

                ),

   396

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

   397

                    ip      = '192.0.2.2',

   398

                    alias4  = 'test',

   399

                )

   400

        ]

   401

   402

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

   403

            ('foo', 'A'): '192.0.2.1',

   404

            ('bar', 'A'): '192.0.2.2',

   405

            ('test', 'A'): ['192.0.2.1', '192.0.2.2'],

   406

        })

   407

   408

    def testHostsConflict(self):

   409

        hosts = [

   410

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

   411

                    ip      = '192.0.2.1',

   412

                ),

   413

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

   414

                    ip      = '192.0.2.2',

   415

                )

   416

        ]

   417

   418

        with self.assertRaises(zone.HostZoneError):

   419

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

   420

   421

    def testHostsAliasConflict(self):

   422

        hosts = [

   423

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

   424

                    ip          = '192.0.2.1',

   425

                ),

   426

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

   427

                    ip          = '192.0.2.2',

   428

                    alias       = 'foo',

   429

                )

   430

        ]

   431

   432

        # with A first

   433

        with self.assertRaises(zone.HostZoneError):

   434

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

   435

   436

        # also with CNAME first

   437

        with self.assertRaises(zone.HostZoneError):

   438

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

   439

   440

    def testHostsAlias4Conflict(self):

   441

        hosts = [

   442

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

   443

                    ip          = '192.0.2.1',

   444

                ),

   445

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

   446

                    ip          = '192.0.2.2',

   447

                    alias4      = 'foo',

   448

                )

   449

        ]

   450

   451

        with self.assertRaises(zone.HostZoneError):

   452

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

   453

   454

   455

class TestReverseZone(TestZoneMixin, unittest.TestCase):

   456

    def testHostIP(self):

   457

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

   458

                ip  = '192.0.2.1',

   459

                ip6 = '2001:db8::192.0.2.1',

   460

        )

   461

   462

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

   463

            ('1', 'PTR'): 'host.domain.',

   464

        })

   465

   466

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

   467

            ('1.0.2.0.0.0.0.c.0.0.0.0.0.0.0.0', 'PTR'): 'host.domain.',

   468

        })

   469

   470

    def testHostIP4(self):

   471

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

   472

                ip  = '192.0.2.1',

   473

        )

   474

   475

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

   476

            ('1', 'PTR'): 'host.domain.',

   477

        })

   478

   479

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

   480

            ('1.2', 'PTR'): 'host.domain.',

   481

        })

   482

   483

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

   484

            ('1.2.0', 'PTR'): 'host.domain.',

   485

        })

   486

   487

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