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 itertools.izip_longest(hosts, expected):

    33

            self.assertIsNotNone(host, expect)

    34

            self.assertIsNotNone(expect, host)

    35

    36

            host_str, attrs = expect

    37

    38

            self.assertHostEqual(host, host_str, attrs)

    39

    40

    def testApplyHostConfigDict(self):

    41

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

    42

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

    43

        })

    44

    45

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

    46

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

    47

        ))

    48

    49

    def testApplyHostConfigDictMulti(self):

    50

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

    51

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

    52

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

    53

        })

    54

    55

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

    56

                ethernet    = {

    57

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

    58

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

    59

                }

    60

        ))

    61

    62

    def testApplyHostsConfigErrorExtra(self):

    63

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

    64

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

    65

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

    66

        })

    67

    68

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

    69

                ethernet    = {

    70

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

    71

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

    72

                }

    73

        ))

    74

    75

    76

    def testApplyHostConfigExtensions(self):

    77

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

    78

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

    79

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

    80

        })

    81

    82

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

    83

                extensions = {

    84

                    'link': {

    85

                        '50': 'foo@test',

    86

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

    87

                    },

    88

                },

    89

        ))

    90

    91

    def testApplyHostFqdn(self):

    92

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

    93

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

    94

        ])

    95

    96

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

    97

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

    98

        ])

    99

   100

    def testApplyHostExpand(self):

   101

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

   102

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

   103

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

   104

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

   105

        ])

   106

   107

    def testApplyHostsFileError(self):

   108

        with self.assertRaises(config.HostConfigError):

   109

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

   110

   111

    def testApplyHostsConfig(self):

   112

        conf_file = ConfFile('test', """

   113

[foo]

   114

    ip = 127.0.0.1

   115

   116

[bar]

   117

    ip = 127.0.0.2

   118

        """)

   119

   120

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

   121

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

   122

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

   123

        ])

   124

   125

    def testApplyHostsConfigNested(self):

   126

        conf_file = ConfFile('test', """

   127

[asdf]

   128

    [[foo]]

   129

        ip = 127.0.0.1

   130

   131

[quux]

   132

    [[bar]]

   133

        ip = 127.0.0.2

   134

        """)

   135

   136

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

   137

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

   138

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

   139

        ])

   140

   141

    def testHostsConfigDdefaults(self):

   142

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

   143

boot.next-server = boot.lan

   144

   145

[foo]

   146

    ip = 192.0.2.1

   147

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

   148

    boot.filename = /pxelinux.0

   149

        """))

   150

   151

        self.assertHostsEqual(hosts, [

   152

                ('foo@test', dict(

   153

                    ip          = ipaddr.IPAddress('192.0.2.1'),

   154

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

   155

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

   156

                )),

   157

        ])

   158

   159

   160

   161

    def testApplyIncludes(self):

   162

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

   163

                ('bar@test', dict(

   164

                    ip          = ipaddr.IPAddress('192.0.2.2'),

   165

                )),

   166

                ('foo@test', dict(

   167

                    ip          = ipaddr.IPAddress('192.0.2.1'),

   168

                )),

   169

        ])

   170

   171

    def testApplyIncludesDefaults(self):

   172

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

   173

boot.next-server = boot.lan

   174

   175

include = etc/hosts/test

   176

        """)), [

   177

                ('bar@test', dict(

   178

                    ip          = ipaddr.IPAddress('192.0.2.2'),

   179

                )),

   180

                ('foo@test', dict(

   181

                    ip          = ipaddr.IPAddress('192.0.2.1'),

   182

                )),

   183

        ])

   184

   185

   186

    def testApplyIncludePath(self):

   187

        self.options.hosts_include = 'etc/hosts'

   188

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

   189

                ('quux@asdf.test', dict(

   190

                    ip          = ipaddr.IPAddress('192.0.2.5'),

   191

                )),

   192

                ('bar@test', dict(

   193

                    ip          = ipaddr.IPAddress('192.0.2.2'),

   194

                )),

   195

                ('foo@test', dict(

   196

                    ip          = ipaddr.IPAddress('192.0.2.1'),

   197

                )),

   198

        ])

   199

   200

    def testApply(self):

   201

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

   202

                ('foo@example.com', dict(

   203

                    ip          = ipaddr.IPAddress('192.0.2.1'),

   204

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

   205

                )),

   206

                ('bar@example.com', dict(

   207

                    ip          = ipaddr.IPAddress('192.0.2.2'),

   208

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

   209

                )),

   210

        ])

   211

   212

class TestZoneMixin(object):

   213

    def assertZoneEquals(self, rrs, expected):

   214

        """

   215

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

   216

   217

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

   218

        """

   219

   220

        gather = { }

   221

   222

        for rr in rrs:

   223

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

   224

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

   225

   226

            if key not in gather:

   227

                gather[key] = value

   228

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

   229

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

   230

            else:

   231

                gather[key].append(value)

   232

   233

        self.assertDictEqual(gather, expected)

   234

   235

   236

class TestForwardZone(TestZoneMixin, unittest.TestCase):

   237

    def testHostOutOfOrigin(self):

   238

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

   239

                ip  = '10.0.0.1',

   240

        )

   241

   242

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

   243

   244

    def testHostIP(self):

   245

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

   246

                ip  = '192.0.2.1',

   247

                ip6 = '2001:db8::192.0.2.1',

   248

        )

   249

   250

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

   251

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

   252

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

   253

        })

   254

   255

    def testHostAlias(self):

   256

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

   257

                ip      = '192.0.2.1',

   258

                alias   = 'test *.test',

   259

        )

   260

   261

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

   262

   263

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

   264

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

   265

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

   266

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

   267

        })

   268

   269

    def testHostAlias46(self):

   270

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

   271

                ip      = '192.0.2.1',

   272

                ip6     = '2001:db8::192.0.2.1',

   273

                alias4  = 'test4',

   274

                alias6  = 'test6',

   275

        )

   276

   277

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

   278

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

   279

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

   280

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

   281

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

   282

        })

   283

   284

    def testHostAlias4Missing(self):

   285

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

   286

                ip6     = '2001:db8::192.0.2.1',

   287

                alias4  = 'test4',

   288

                alias6  = 'test6',

   289

        )

   290

   291

        with self.assertRaises(zone.HostZoneError):

   292

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

   293

   294

    def testHostAlias6Missing(self):

   295

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

   296

                ip      = '192.0.2.1',

   297

                alias4  = 'test4',

   298

                alias6  = 'test6',

   299

        )

   300

   301

        with self.assertRaises(zone.HostZoneError):

   302

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

   303

   304

    def testHostFQDN(self):

   305

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

   306

                ip          = '192.0.2.3',

   307

        )

   308

   309

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

   310

   311

        })

   312

   313

    def testHostDelegate(self):

   314

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

   315

                forward = 'host.example.net',

   316

        )

   317

   318

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

   319

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

   320

        })

   321

   322

    def testHostForwardAlias(self):

   323

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

   324

                forward = 'host.example.net',

   325

                alias   = 'test',

   326

        )

   327

   328

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

   329

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

   330

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

   331

        })

   332

   333

    def testHostLocation(self):

   334

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

   335

                ip          = '192.0.2.1',

   336

                location    = 'test',

   337

        )

   338

   339

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

   340

   341

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

   342

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

   343

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

   344

        })

   345

   346

    def testHostLocationDomain(self):

   347

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

   348

                ip          = '192.0.2.1',

   349

                location    = 'test@bar.domain',

   350

        )

   351

   352

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

   353

   354

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

   355

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

   356

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

   357

        })

   358

   359

    def testHostLocationDomainOutOfOrigin(self):

   360

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

   361

                ip          = '192.0.2.1',

   362

                location    = 'test@bar.domain',

   363

        )

   364

   365

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

   366

   367

        with self.assertRaises(zone.HostZoneError):

   368

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

   369

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

   370

            })

   371

   372

        # TODO

   373

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

   374

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

   375

        #})

   376

   377

    def testHostsForward(self):

   378

        hosts = [

   379

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

   380

                    ip      = '192.0.2.1',

   381

                    ip6     = '2001:db8::192.0.2.1',

   382

                    alias   = 'test',

   383

                ),

   384

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

   385

                    ip      = '192.0.2.2',

   386

                ),

   387

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

   388

                    ip      = '192.0.2.3',

   389

                ),

   390

        ]

   391

   392

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

   393

   394

        # handle the $ORIGIN directive

   395

        rd = next(rrs)

   396

   397

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

   398

   399

        self.assertZoneEquals(rrs, {

   400

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

   401

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

   402

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

   403

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

   404

        })

   405

   406

    def testHostsMultiAlias(self):

   407

        hosts = [

   408

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

   409

                    ip      = '192.0.2.1',

   410

                    alias4  = 'test',

   411

                ),

   412

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

   413

                    ip      = '192.0.2.2',

   414

                    alias4  = 'test',

   415

                )

   416

        ]

   417

   418

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

   419

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

   420

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

   421

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

   422

        })

   423

   424

    def testHostsConflict(self):

   425

        hosts = [

   426

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

   427

                    ip      = '192.0.2.1',

   428

                ),

   429

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

   430

                    ip      = '192.0.2.2',

   431

                )

   432

        ]

   433

   434

        with self.assertRaises(zone.HostZoneError):

   435

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

   436

   437

    def testHostsAliasConflict(self):

   438

        hosts = [

   439

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

   440

                    ip          = '192.0.2.1',

   441

                ),

   442

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

   443

                    ip          = '192.0.2.2',

   444

                    alias       = 'foo',

   445

                )

   446

        ]

   447

   448

        # with A first

   449

        with self.assertRaises(zone.HostZoneError):

   450

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

   451

   452

        # also with CNAME first

   453

        with self.assertRaises(zone.HostZoneError):

   454

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

   455

   456

    def testHostsAlias4Conflict(self):

   457

        hosts = [

   458

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

   459

                    ip          = '192.0.2.1',

   460

                ),

   461

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

   462

                    ip          = '192.0.2.2',

   463

                    alias4      = 'foo',

   464

                )

   465

        ]

   466

   467

        with self.assertRaises(zone.HostZoneError):

   468

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

   469

   470

   471

class TestReverseZone(TestZoneMixin, unittest.TestCase):

   472

    def testHostIP(self):

   473

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

   474

                ip  = '192.0.2.1',

   475

                ip6 = '2001:db8::192.0.2.1',

   476

        )

   477

   478

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

   479

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

   480

        })

   481

   482

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

   483

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

   484

        })

   485

   486

    def testHostIP4(self):

   487

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