1

import ipaddr

     2

import pvl.args

     3

import unittest

     4

     5

from pvl.hosts import config, dhcp, zone

     6

from pvl.hosts.host import Host

     7

from StringIO import StringIO

     8

     9

class Options(object):

    10

    hosts_charset   = 'utf-8'

    11

    hosts_domain    = None

    12

    hosts_include   = None

    13

    14

class ConfFile(StringIO):

    15

    def __init__(self, name, buffer):

    16

        StringIO.__init__(self, buffer)

    17

        self.name = name

    18

    19

class TestConfig(unittest.TestCase):

    20

    def setUp(self):

    21

        self.options = Options()

    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

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

    31

            host_str, attrs = expect

    32

    33

            self.assertHostEqual(host, host_str, attrs)

    34

    35

    def testApplyHostsFileError(self):

    36

        with self.assertRaises(config.HostConfigError):

    37

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

    38

    39

    def testApplyHosts(self):

    40

        conf_file = ConfFile('test', """

    41

[foo]

    42

    ip = 127.0.0.1

    43

    44

[bar]

    45

    ip = 127.0.0.2

    46

        """)

    47

        expected = [

    48

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

    49

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

    50

        ]

    51

    52

        self.assertHostsEqual(config.apply_hosts_file(self.options, conf_file), expected)

    53

    54

    def testApply(self):

    55

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

    56

                ('foo@test', dict(

    57

                    ip          = ipaddr.IPAddress('127.0.0.1'),

    58

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

    59

                )),

    60

                ('bar@test', dict(

    61

                    ip          = ipaddr.IPAddress('127.0.0.2'),

    62

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

    63

                )),

    64

        ])

    65

    66

    def testApplyHostsExpand(self):

    67

        self.assertHostsEqual(config.apply_host_config(self.options, 'asdf', 'asdf{1-3}', ip='10.100.100.$'), [

    68

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

    69

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

    70

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

    71

        ])

    72

    73

    def testApplyHostConfigDict(self):

    74

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

    75

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

    76

        })

    77

    78

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

    79

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

    80

        ))

    81

    82

    def testApplyHostsConfigError(self):

    83

        with self.assertRaises(config.HostConfigError):

    84

            config.apply_host(self.options, 'foo', 'test', {

    85

                'ethernet': 'foo',

    86

                'ethernet.eth0': 'bar',

    87

            })

    88

    89

class TestZoneMixin(object):

    90

    def assertZoneEquals(self, rrs, expected):

    91

        gather = { }

    92

    93

        for rr in rrs:

    94

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

    95

    96

            self.assertNotIn(key, gather)

    97

    98

            gather[key] = rr.data

    99

   100

        self.assertDictEqual(gather, expected)

   101

   102

   103

class TestForwardZone(TestZoneMixin, unittest.TestCase):

   104

    def setUp(self):

   105

        self.options = Options()

   106

        self.options.add_origin = False

   107

   108

    def testResolve(self):

   109

        self.assertEquals(zone.resolve(None, None, 'host'), 'host.')

   110

        self.assertEquals(zone.resolve(None, 'domain', 'host'), 'host.domain.')

   111

   112

        with self.assertRaises(zone.HostZoneError):

   113

            zone.resolve('origin', 'domain', 'host')

   114

   115

        self.assertEquals(zone.resolve('domain', 'domain', 'host'), 'host')

   116

        self.assertEquals(zone.resolve('origin', 'domain.origin', 'host'), 'host.domain')

   117

   118

        with self.assertRaises(zone.HostZoneError):

   119

            zone.resolve('origin', 'domainorigin', 'host')

   120

   121

        with self.assertRaises(zone.HostZoneError):

   122

            zone.resolve('origin', None, 'host.domain')

   123

   124

    def testHostOutOfOrigin(self):

   125

        h = Host('host', 'domain', ip=ipaddr.IPAddress('10.0.0.1'))

   126

   127

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

   128

   129

    def testHostIP(self):

   130

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

   131

                ip  = '192.0.2.1',

   132

                ip6 = '2001:db8::192.0.2.1',

   133

        )

   134

   135

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

   136

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

   137

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

   138

        })

   139

   140

    def testHostAlias(self):

   141

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

   142

                ip      = '192.0.2.1',

   143

                alias   = 'test *.test',

   144

        )

   145

   146

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

   147

   148

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

   149

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

   150

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

   151

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

   152

        })

   153

   154

    def testHostAlias46(self):

   155

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

   156

                ip      = '192.0.2.1',

   157

                ip6     = '2001:db8::192.0.2.1',

   158

                alias4  = 'test4',

   159

                alias6  = 'test6',

   160

        )

   161

   162

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

   163

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

   164

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

   165

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

   166

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

   167

        })

   168

   169

    def testHostAlias4Missing(self):

   170

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

   171

                ip6     = '2001:db8::192.0.2.1',

   172

                alias4  = 'test4',

   173

                alias6  = 'test6',

   174

        )

   175

   176

        with self.assertRaises(zone.HostZoneError):

   177

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

   178

   179

    def testHostAlias6Missing(self):

   180

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

   181

                ip      = '192.0.2.1',

   182

                alias4  = 'test4',

   183

                alias6  = 'test6',

   184

        )

   185

   186

        with self.assertRaises(zone.HostZoneError):

   187

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

   188

   189

    def testHostDelegate(self):

   190

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

   191

                forward = 'host.example.net',

   192

        )

   193

   194

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

   195

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

   196

        })

   197

   198

    def testHostForwardAlias(self):

   199

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

   200

                forward = 'host.example.net',

   201

                alias   = 'test',

   202

        )

   203

   204

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

   205

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

   206

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

   207

        })

   208

   209

    def testHostLocation(self):

   210

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

   211

                ip          = '192.0.2.1',

   212

                location    = 'test',

   213

        )

   214

   215

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

   216

   217

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

   218

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

   219

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

   220

        })

   221

   222

    def testHostLocationDomain(self):

   223

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

   224

                ip          = '192.0.2.1',

   225

                location    = 'test@bar.domain',

   226

        )

   227

   228

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

   229

   230

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

   231

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

   232

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

   233

        })

   234

   235

    def testHostLocationDomainOutOfOrigin(self):

   236

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

   237

                ip          = '192.0.2.1',

   238

                location    = 'test@bar.domain',

   239

        )

   240

   241

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

   242

   243

        with self.assertRaises(zone.HostZoneError):

   244

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

   245

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

   246

            })

   247

   248

        # TODO

   249

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

   250

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

   251

        #})

   252

   253

    def testHostsForward(self):

   254

        hosts = [

   255

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

   256

                    ip      = '192.0.2.1',

   257

                    ip6     = '2001:db8::192.0.2.1',

   258

                    alias   = 'test',

   259

                ),

   260

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

   261

                    ip      = '192.0.2.2',

   262

                )

   263

        ]

   264

   265

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

   266

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

   267

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

   268

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

   269

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

   270

        })

   271

   272

    def testHostsConflict(self):

   273

        hosts = [

   274

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

   275

                    ip      = '192.0.2.1',

   276

                ),

   277

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

   278

                    ip      = '192.0.2.2',

   279

                )

   280

        ]

   281

   282

        with self.assertRaises(zone.HostZoneError):

   283

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

   284

   285

    def testHostsAliasConflict(self):

   286

        hosts = [

   287

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

   288

                    ip          = '192.0.2.1',

   289

                ),

   290

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

   291

                    ip          = '192.0.2.2',

   292

                    alias       = 'foo',

   293

                )

   294

        ]

   295

   296

        # with A first

   297

        with self.assertRaises(zone.HostZoneError):

   298

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

   299

   300

        # also with CNAME first

   301

        with self.assertRaises(zone.HostZoneError):

   302

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

   303

   304

    def testHostsAlias4Conflict(self):

   305

        hosts = [

   306

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

   307

                    ip          = '192.0.2.1',

   308

                ),

   309

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

   310

                    ip          = '192.0.2.2',

   311

                    alias4      = 'foo',

   312

                )

   313

        ]

   314

   315

        with self.assertRaises(zone.HostZoneError):

   316

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

   317

   318

   319

class TestReverseZone(TestZoneMixin, unittest.TestCase):

   320

    def setUp(self):

   321

        self.options = Options()

   322

        self.options.unknown_host = False

   323

   324

    def testHostIP(self):

   325

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

   326

                ip  = '192.0.2.1',

   327

                ip6 = '2001:db8::192.0.2.1',

   328

        )

   329

   330

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

   331

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

   332

        })

   333

   334

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

   335

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

   336

        })

   337

   338

    def testHostIP4(self):

   339

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

   340

                ip  = '192.0.2.1',

   341

        )

   342

   343

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

   344

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

   345

        })

   346

   347

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

   348

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

   349

        })

   350

   351

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

   352

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

   353

        })

   354

   355

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

   356

   357

        })

   358

   359

    def testHostIP6(self):

   360

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

   361

                ip6 = '2001:db8::192.0.2.1',

   362

        )

   363

   364

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

   365

        })

   366

   367

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

   368

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

   369

        })

   370

   371

    def testHostIPOutOfPrefix(self):

   372

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

   373

                ip  = '192.0.2.1',

   374

                ip6 = '2001:db8::192.0.2.1',

   375

        )

   376

   377

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

   378

   379

        })

   380

   381

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

   382

   383

        })

   384

   385

    def testHostDelegate(self):

   386

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

   387

                ip      = '192.0.2.1',

   388

                ip6     = '2001:db8::192.0.2.1',

   389

                forward = '',

   390

                reverse = '1.0/28.2.0.192.in-addr.arpa',

   391

        )

   392

   393

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

   394

   395

        })

   396

   397

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

   398

            ('1', 'CNAME'): ['1.0/28.2.0.192.in-addr.arpa.'],

   399

        })

   400

   401

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

   402

   403

        })

   404

   405

    def testHostsConflict(self):

   406

        hosts = [

   407

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

   408

                    ip      = '192.0.2.1',

   409

                ),

   410

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

   411

                    ip      = '192.0.2.1',

   412

                )

   413

        ]

   414

   415

        with self.assertRaises(zone.HostZoneError):

   416

            self.assertZoneEquals(zone.apply_hosts_reverse(self.options, hosts, ipaddr.IPNetwork('192.0.2.1/24')), { })

   417

   418

    def testHostsGenerateUnknown(self):

   419

        hosts = [

   420

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

   421

                    ip      = '192.0.2.1',

   422

                ),

   423

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

   424

                    ip      = '192.0.2.5',

   425

                ),

   426

        ]

   427

   428

        self.options.unknown_host = 'ufc'

   429

        self.options.hosts_domain = 'domain'

   430

        self.assertZoneEquals(zone.apply_hosts_reverse(self.options, hosts, ipaddr.IPNetwork('192.0.2.1/29')), {

   431

            ('1', 'PTR'): ['foo.domain.'],

   432

            ('2', 'PTR'): ['ufc.domain.'],

   433

            ('3', 'PTR'): ['ufc.domain.'],

   434

            ('4', 'PTR'): ['ufc.domain.'],

   435

            ('5', 'PTR'): ['bar.domain.'],

   436

            ('6', 'PTR'): ['ufc.domain.'],

   437

        })

   438

   439

class TestDhcp(unittest.TestCase):

   440

    def setUp(self):

   441

        self.options = pvl.args.options(

   442

                hosts_charset   = 'utf-8',

   443

                hosts_domain    = None,

   444

                hosts_include   = None,

   445

        )

   446

   447

    def assertBlocksEqual(self, blockdefs, expected):

   448

        for (_block, _items, _opts), (block, items, opts) in zip(blockdefs, expected):

   449

            self.assertEqual(_block, block)

   450

            self.assertItemsEqual(_items, items)

   451

   452

            if opts is not None:

   453

                self.assertEqual(_opts, opts)

   454

   455

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

   456

   457

    def testHost(self):

   458

        host = Host.build('foo', 'test',

   459

                ip          = '192.0.2.1',

   460

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

   461

        )

   462

   463

        self.assertBlocksEqual(list(dhcp.dhcp_host(host)), [

   464

            (('host', 'foo'), [

   465

                ('option', 'host-name', "foo"),

   466

                ('fixed-address', '192.0.2.1'),

   467

                ('hardware', 'ethernet', '00:11:22:33:44:55'),

   468

            ], None)

   469

        ])

   470

   471

    def testHostDynamic(self):

   472

        host = Host.build('foo', 'test',

   473

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

   474

        )

   475

   476

        self.assertBlocksEqual(list(dhcp.dhcp_host(host)), [

   477

            (('host', 'foo'), [

   478

                ('option', 'host-name', "foo"),

   479

                ('hardware', 'ethernet', '00:11:22:33:44:55'),

   480

            ], None)

   481

        ])

   482

   483

    def testHostBoot(self):

   484

        host = Host.build('foo', 'test',

   485

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

   486

                boot        = 'boot.lan:debian/wheezy/pxelinux.0',

   487

        )

   488

   489

        self.assertBlocksEqual(list(dhcp.dhcp_host(host)), [

   490

            (('host', 'foo'), [

   491

                ('option', 'host-name', "foo"),