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,

    22

        )

    23

    24

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

    25

        self.assertEquals(str(host), host_str)

    26

    27

        for attr, value in attrs.iteritems():

    28

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

    29

    30

    def assertHostsEqual(self, hosts, expected):

    31

        hosts = list(hosts)

    32

    33

        for host, expect in itertools.izip_longest(hosts, expected):

    34

            self.assertIsNotNone(host, expect)

    35

            self.assertIsNotNone(expect, host)

    36

    37

            host_str, attrs = expect

    38

    39

            self.assertHostEqual(host, host_str, attrs)

    40

    41

    def testApplyHostConfigDict(self):

    42

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

    43

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

    44

        })

    45

    46

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

    47

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

    48

        ))

    49

    50

    def testApplyHostConfigDictMulti(self):

    51

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

    52

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

    53

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

    54

        })

    55

    56

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

    57

                ethernet    = {

    58

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

    59

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

    60

                }

    61

        ))

    62

    63

    def testApplyHostsConfigErrorExtra(self):

    64

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

    65

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

    66

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

    67

        })

    68

    69

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

    70

                ethernet    = {

    71

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

    72

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

    73

                }

    74

        ))

    75

    76

    77

    def testApplyHostConfigExtensions(self):

    78

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

    79

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

    80

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

    81

        })

    82

    83

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

    84

                extensions = {

    85

                    'link': {

    86

                        '50': 'foo@test',

    87

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

    88

                    },

    89

                },

    90

        ))

    91

    92

    def testApplyHostFqdn(self):

    93

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

    94

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

    95

        ])

    96

    97

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

    98

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

    99

        ])

   100

   101

    def testApplyHostExpand(self):

   102

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

   103

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

   104

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

   105

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

   106

        ])

   107

   108

    def testApplyHostsFileError(self):

   109

        with self.assertRaises(config.HostConfigError):

   110

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

   111

   112

    def testApplyHostsConfig(self):

   113

        conf_file = ConfFile('test', """

   114

[foo]

   115

    ip = 127.0.0.1

   116

   117

[bar]

   118

    ip = 127.0.0.2

   119

        """)

   120

   121

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

   122

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

   123

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

   124

        ])

   125

   126

    def testApplyHostsConfigNested(self):

   127

        conf_file = ConfFile('test', """

   128

[asdf]

   129

    [[foo]]

   130

        ip = 127.0.0.1

   131

   132

[quux]

   133

    [[bar]]

   134

        ip = 127.0.0.2

   135

        """)

   136

   137

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

   138

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

   139

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

   140

        ])

   141

   142

    def testHostsConfigDdefaults(self):

   143

        hosts = config.apply_hosts_config(self.options, ConfFile('test', """

   144

boot.next-server = boot.lan

   145

   146

[foo]

   147

    ip = 192.0.2.1

   148

    ethernet.eth0 = 00:11:22:33:44:55

   149

    boot.filename = /pxelinux.0

   150

        """))

   151

   152

        self.assertHostsEqual(hosts, [

   153

                ('foo@test', dict(

   154

                    ip          = ipaddr.IPAddress('192.0.2.1'),

   155

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

   156

                    boot        = { 'next-server': 'boot.lan', 'filename': '/pxelinux.0' },

   157

                )),

   158

        ])

   159

   160

   161

   162

    def testApplyIncludes(self):

   163

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

   164

                ('bar@test', dict(

   165

                    ip          = ipaddr.IPAddress('192.0.2.2'),

   166

                )),

   167

                ('foo@test', dict(

   168

                    ip          = ipaddr.IPAddress('192.0.2.1'),

   169

                )),

   170

        ])

   171

   172

    def testApplyIncludesDefaults(self):

   173

        self.assertHostsEqual(config.apply_hosts_config(self.options, ConfFile('test', """

   174

boot.next-server = boot.lan

   175

   176

include = etc/hosts/test

   177

        """)), [

   178

                ('bar@test', dict(

   179

                    ip          = ipaddr.IPAddress('192.0.2.2'),

   180

                )),

   181

                ('foo@test', dict(

   182

                    ip          = ipaddr.IPAddress('192.0.2.1'),

   183

                )),

   184

        ])

   185

   186

   187

    def testApplyIncludePath(self):

   188

        self.options.hosts_include = 'etc/hosts'

   196

                ('quux@asdf.test', dict(

   197

                    ip          = ipaddr.IPAddress('192.0.2.5'),

   198

                )),

   199

                ('bar@test', dict(

   200

                    ip          = ipaddr.IPAddress('192.0.2.2'),

   201

                )),

   202

                ('foo@test', dict(

   203

                    ip          = ipaddr.IPAddress('192.0.2.1'),

   204

                )),

   205

        ])

   206

   216

    def testApply(self):

   217

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

   218

                ('foo@example.com', dict(

   219

                    ip          = ipaddr.IPAddress('192.0.2.1'),

   220

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

   221

                )),

   222

                ('bar@example.com', dict(

   223

                    ip          = ipaddr.IPAddress('192.0.2.2'),

   224

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

   225

                )),

   226

        ])

   227

   228

class TestZoneMixin(object):

   229

    def assertZoneEquals(self, rrs, expected):

   230

        """

   231

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

   232

   233

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

   234

        """

   235

   236

        gather = { }

   237

   238

        for rr in rrs:

   239

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

   240

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

   241

   242

            if key not in gather:

   243

                gather[key] = value

   244

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

   245

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

   246

            else:

   247

                gather[key].append(value)

   248

   249

        self.assertDictEqual(gather, expected)

   250

   251

   252

class TestForwardZone(TestZoneMixin, unittest.TestCase):

   253

    def testHostOutOfOrigin(self):

   254

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

   255

                ip  = '10.0.0.1',

   256

        )

   257

   258

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

   259

   260

    def testHostIP(self):

   261

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

   262

                ip  = '192.0.2.1',

   263

                ip6 = '2001:db8::192.0.2.1',

   264

        )

   265

   266

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

   267

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

   268

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

   269

        })

   270

   271

    def testHostAlias(self):

   272

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

   273

                ip      = '192.0.2.1',

   274

                alias   = 'test *.test',

   275

        )

   276

   277

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

   278

   279

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

   280

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

   281

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

   282

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

   283

        })

   284

   285

    def testHostAlias46(self):

   286

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

   287

                ip      = '192.0.2.1',

   288

                ip6     = '2001:db8::192.0.2.1',

   289

                alias4  = 'test4',

   290

                alias6  = 'test6',

   291

        )

   292

   293

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

   294

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

   295

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

   296

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

   297

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

   298

        })

   299

   300

    def testHostAlias4Missing(self):

   301

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

   302

                ip6     = '2001:db8::192.0.2.1',

   303

                alias4  = 'test4',

   304

                alias6  = 'test6',

   305

        )

   306

   307

        with self.assertRaises(zone.HostZoneError):

   308

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

   309

   310

    def testHostAlias6Missing(self):

   311

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

   312

                ip      = '192.0.2.1',

   313

                alias4  = 'test4',

   314

                alias6  = 'test6',

   315

        )

   316

   317

        with self.assertRaises(zone.HostZoneError):

   318

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

   319

   320

    def testHostFQDN(self):

   321

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

   322

                ip          = '192.0.2.3',

   323

        )

   324

   325

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

   326

   327

        })

   328

   329

    def testHostDelegate(self):

   330

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

   331

                forward = 'host.example.net',

   332

        )

   333

   334

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

   335

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

   336

        })

   337

   338

    def testHostForwardAlias(self):

   339

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

   340

                forward = 'host.example.net',

   341

                alias   = 'test',

   342

        )

   343

   344

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

   345

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

   346

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

   347

        })

   348

   349

    def testHostLocation(self):

   350

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

   351

                ip          = '192.0.2.1',

   352

                location    = 'test',

   353

        )

   354

   355

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

   356

   357

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

   358

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

   359

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

   360

        })

   361

   362

    def testHostLocationDomain(self):

   363

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

   364

                ip          = '192.0.2.1',

   365

                location    = 'test@bar.domain',

   366

        )

   367

   368

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

   369

   370

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

   371

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

   372

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

   373

        })

   374

   375

    def testHostLocationDomainOutOfOrigin(self):

   376

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

   377

                ip          = '192.0.2.1',

   378

                location    = 'test@bar.domain',

   379

        )

   380

   381

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

   382

   383

        with self.assertRaises(zone.HostZoneError):

   384

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

   385

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

   386

            })

   387

   388

        # TODO

   389

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

   390

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

   391

        #})

   392

   393

    def testHostsForward(self):

   394

        hosts = [

   395

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

   396

                    ip      = '192.0.2.1',

   397

                    ip6     = '2001:db8::192.0.2.1',

   398

                    alias   = 'test',

   399

                ),

   400

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

   401

                    ip      = '192.0.2.2',

   402

                ),

   403

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

   404

                    ip      = '192.0.2.3',

   405

                ),

   406

        ]

   407

   408

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

   409

   410

        # handle the $ORIGIN directive

   411

        rd = next(rrs)

   412

   413

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

   414

   415

        self.assertZoneEquals(rrs, {

   416

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

   417

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

   418

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

   419

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

   420

        })

   421

   422

    def testHostsMultiAlias(self):

   423

        hosts = [

   424

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

   425

                    ip      = '192.0.2.1',

   426

                    alias4  = 'test',

   427

                ),

   428

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

   429

                    ip      = '192.0.2.2',

   430

                    alias4  = 'test',

   431

                )

   432

        ]

   433

   434

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

   435

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

   436

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

   437

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

   438

        })

   439

   440

    def testHostsConflict(self):

   441

        hosts = [

   442

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

   443

                    ip      = '192.0.2.1',

   444

                ),

   445

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

   446

                    ip      = '192.0.2.2',

   447

                )

   448

        ]

   449

   450

        with self.assertRaises(zone.HostZoneError):

   451

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

   452

   453

    def testHostsAliasConflict(self):

   454

        hosts = [

   455

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

   456

                    ip          = '192.0.2.1',

   457

                ),

   458

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

   459

                    ip          = '192.0.2.2',

   460

                    alias       = 'foo',

   461

                )

   462

        ]

   463

   464

        # with A first

   465

        with self.assertRaises(zone.HostZoneError):

   466

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

   467

   468

        # also with CNAME first

   469

        with self.assertRaises(zone.HostZoneError):

   470

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

   471

   472

    def testHostsAlias4Conflict(self):

   473

        hosts = [

   474

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

   475

                    ip          = '192.0.2.1',

   476

                ),

   477

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

   478

                    ip          = '192.0.2.2',

   479

                    alias4      = 'foo',

   480

                )

   481

        ]

   482

   483

        with self.assertRaises(zone.HostZoneError):

   484

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

   485

   486

   487

class TestReverseZone(TestZoneMixin, unittest.TestCase):

   488