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

                hosts_include_trace = 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):

   108

        ])

   109

   110

    def testApplyHostsFileError(self):

   111

        with self.assertRaises(config.HostConfigError):

   112

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

   113

   114

    def testApplyHostsConfig(self):

   115

        conf_file = ConfFile('test', """

   116

[foo]

   117

    ip = 127.0.0.1

   118

   119

[bar]

   120

    ip = 127.0.0.2

   121

        """)

   122

   123

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

   126

        ])

   127

   128

    def testApplyHostsConfigNested(self):

   129

        conf_file = ConfFile('test', """

   130

[asdf]

   131

    [[foo]]

   132

        ip = 127.0.0.1

   133

   134

[quux]

   135

    [[bar]]

   136

        ip = 127.0.0.2

   137

        """)

   138

   139

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

   142

        ])

   143

   144

    def testHostsConfigDdefaults(self):

   145

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

   146

boot.next-server = boot.lan

   147

   148

[foo]

   149

    ip = 192.0.2.1

   150

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

   151

    boot.filename = /pxelinux.0

   152

        """))

   153

   154

        self.assertHostsEqual(hosts, [

   155

                ('foo@test', dict(

   157

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

   158

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

   159

                )),

   160

        ])

   161

   162

   163

   164

    def testApplyIncludes(self):

   165

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

   166

                ('bar@test', dict(

   168

                )),

   169

                ('foo@test', dict(

   171

                )),

   172

        ])

   173

   174

    def testApplyIncludesDefaults(self):

   175

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

   176

boot.next-server = boot.lan

   177

   178

include = etc/hosts/test

   179

        """)), [

   180

                ('bar@test', dict(

   182

                )),

   183

                ('foo@test', dict(

   185

                )),

   186

        ])

   187

   188

   189

    def testApplyIncludePath(self):

   190

        self.options.hosts_include = 'etc/hosts'

   191

        include_trace = [ ]

   192

   193

        hosts = list(config.apply_hosts_files(self.options, ['etc/zones/forward/test'],

   194

            include_trace   = include_trace,

   195

        ))

   196

   197

        self.assertHostsEqual(hosts, [

   198

                ('quux@asdf.test', dict(

   200

                )),

   201

                ('bar@test', dict(

   203

                )),

   204

                ('foo@test', dict(

   206

                )),

   207

        ])

   208

   209

        self.assertEqual(include_trace, [

   210

            'etc/zones/forward/test',

   211

            'etc/zones/forward/test/asdf.test',

   212

            'etc/zones/forward/test/test',

   213

            'etc/hosts/test.d/',

   214

            'etc/hosts/test.d/bar',

   215

            'etc/hosts/test.d/foo',

   216

        ])

   217

   218

    def testApply(self):

   219

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

   220

                ('foo@example.com', dict(

   222

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

   223

                )),

   224

                ('bar@example.com', dict(

   226

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

   227

                )),

   228

        ])

   229

   230

class TestZoneMixin(object):

   231

    def assertZoneEquals(self, rrs, expected):

   232

        """

   233

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

   234

   235

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

   236

        """

   237

   238

        gather = { }

   239

   240

        for rr in rrs:

   241

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

   242

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

   243

   244

            if key not in gather:

   245

                gather[key] = value

   246

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

   247

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

   248

            else:

   249

                gather[key].append(value)

   250

   251

        self.assertDictEqual(gather, expected)

   252

   253

   254

class TestForwardZone(TestZoneMixin, unittest.TestCase):

   255

    def testHostOutOfOrigin(self):

   256

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

   257

                ip  = '10.0.0.1',

   258

        )

   259

   260

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

   261

   262

    def testHostIP(self):

   263

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

   264

                ip  = '192.0.2.1',

   265

                ip6 = '2001:db8::192.0.2.1',

   266

        )

   267

   268

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

   269

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

   270

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

   271

        })

   272

   273

    def testHostAlias(self):

   274

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

   275

                ip      = '192.0.2.1',

   276

                alias   = 'test *.test',

   277

        )

   278

   279

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

   280

   281

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

   282

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

   283

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

   284

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

   285

        })

   286

   287

    def testHostAlias46(self):

   288

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

   289

                ip      = '192.0.2.1',

   290

                ip6     = '2001:db8::192.0.2.1',

   291

                alias4  = 'test4',

   292

                alias6  = 'test6',

   293

        )

   294

   295

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

   296

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

   297

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

   298

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

   299

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

   300

        })

   301

   302

    def testHostAlias4Missing(self):

   303

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

   304

                ip6     = '2001:db8::192.0.2.1',

   305

                alias4  = 'test4',

   306

                alias6  = 'test6',

   307

        )

   308

   309

        with self.assertRaises(zone.HostZoneError):

   310

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

   311

   312

    def testHostAlias6Missing(self):

   313

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

   314

                ip      = '192.0.2.1',

   315

                alias4  = 'test4',

   316

                alias6  = 'test6',

   317

        )

   318

   319

        with self.assertRaises(zone.HostZoneError):

   320

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

   321

   322

    def testHostFQDN(self):

   323

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

   324

                ip          = '192.0.2.3',

   325

        )

   326

   327

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

   328

   329

        })

   330

   331

    def testHostDelegate(self):

   332

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

   333

                forward = 'host.example.net',

   334

        )

   335

   336

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

   337

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

   338

        })

   339

   340

    def testHostForwardAlias(self):

   341

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

   342

                forward = 'host.example.net',

   343

                alias   = 'test',

   344

        )

   345

   346

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

   347

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

   348

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

   349

        })

   350

   351

    def testHostLocation(self):

   352

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

   353

                ip          = '192.0.2.1',

   354

                location    = 'test',

   355

        )

   356

   357

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

   358

   359

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

   360

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

   361

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

   362

        })

   363

   364

    def testHostLocationDomain(self):

   365

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

   366

                ip          = '192.0.2.1',

   367

                location    = 'test@bar.domain',

   368

        )

   369

   370

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

   371

   372

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

   373

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

   374

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

   375

        })

   376

   377

    def testHostLocationDomainOutOfOrigin(self):

   378

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

   379

                ip          = '192.0.2.1',

   380

                location    = 'test@bar.domain',

   381

        )

   382

   383

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

   384

   385

        with self.assertRaises(zone.HostZoneError):

   386

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

   387

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

   388

            })

   389

   390

        # TODO

   391

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

   392

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

   393

        #})

   394

   395

    def testHostsForward(self):

   396

        hosts = [

   397

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

   398

                    ip      = '192.0.2.1',

   399

                    ip6     = '2001:db8::192.0.2.1',

   400

                    alias   = 'test',

   401

                ),

   402

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

   403

                    ip      = '192.0.2.2',

   404

                ),

   405

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

   406

                    ip      = '192.0.2.3',

   407

                ),

   408

        ]

   409

   410

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

   411

   412

        # handle the $ORIGIN directive

   413

        rd = next(rrs)

   414

   415

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

   416

   417

        self.assertZoneEquals(rrs, {

   418

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

   419

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

   420

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

   421

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

   422

        })

   423

   424

    def testHostsMultiAlias(self):

   425

        hosts = [

   426

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

   427

                    ip      = '192.0.2.1',

   428

                    alias4  = 'test',

   429

                ),

   430

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

   431

                    ip      = '192.0.2.2',

   432

                    alias4  = 'test',

   433

                )

   434

        ]

   435

   436

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

   437

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

   438

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

   439

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

   440

        })

   441

   442

    def testHostsConflict(self):

   443

        hosts = [

   444

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

   445

                    ip      = '192.0.2.1',

   446

                ),

   447

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

   448

                    ip      = '192.0.2.2',

   449

                )

   450

        ]

   451

   452

        with self.assertRaises(zone.HostZoneError):

   453

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

   454

   455

    def testHostsAliasConflict(self):

   456

        hosts = [

   457

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

   458

                    ip          = '192.0.2.1',

   459

                ),

   460

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

   461

                    ip          = '192.0.2.2',

   462

                    alias       = 'foo',

   463

                )

   464

        ]

   465

   466

        # with A first

   467

        with self.assertRaises(zone.HostZoneError):

   468

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

   469

   470

        # also with CNAME first

   471

        with self.assertRaises(zone.HostZoneError):

   472

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

   473

   474

    def testHostsAlias4Conflict(self):

   475

        hosts = [

   476

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

   477

                    ip          = '192.0.2.1',

   478

                ),

   479

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

   480

                    ip          = '192.0.2.2',

   481

                    alias4      = 'foo',

   482

                )

   483

        ]

   484

   485

        with self.assertRaises(zone.HostZoneError):

   486

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

   487