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 testHostOutOfOrigin(self):

   109

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

   110

   111

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

   112

   113

    def testHostIP(self):

   114

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

   115

                ip  = '192.0.2.1',

   116

                ip6 = '2001:db8::192.0.2.1',

   117

        )

   118

   119

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

   120

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

   121

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

   122

        })

   123

   124

    def testHostAlias(self):

   125

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

   126

                ip      = '192.0.2.1',

   127

                alias   = 'test *.test',

   128

        )

   129

   130

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

   131

   132

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

   133

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

   134

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

   135

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

   136

        })

   137

   138

    def testHostAlias46(self):

   139

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

   140

                ip      = '192.0.2.1',

   141

                ip6     = '2001:db8::192.0.2.1',

   142

                alias4  = 'test4',

   143

                alias6  = 'test6',

   144

        )

   145

   146

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

   147

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

   148

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

   149

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

   150

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

   151

        })

   152

   153

    def testHostAlias4Missing(self):

   154

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

   155

                ip6     = '2001:db8::192.0.2.1',

   156

                alias4  = 'test4',

   157

                alias6  = 'test6',

   158

        )

   159

   160

        with self.assertRaises(zone.HostZoneError):

   161

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

   162

   163

    def testHostAlias6Missing(self):

   164

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

   165

                ip      = '192.0.2.1',

   166

                alias4  = 'test4',

   167

                alias6  = 'test6',

   168

        )

   169

   170

        with self.assertRaises(zone.HostZoneError):

   171

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

   172

   173

    def testHostDelegate(self):

   174

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

   175

                forward = 'host.example.net',

   176

        )

   177

   178

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

   179

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

   180

        })

   181

   182

    def testHostForwardAlias(self):

   183

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

   184

                forward = 'host.example.net',

   185

                alias   = 'test',

   186

        )

   187

   188

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

   189

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

   190

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

   191

        })

   192

   193

    def testHostLocation(self):

   194

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

   195

                ip          = '192.0.2.1',

   196

                location    = 'test',

   197

        )

   198

   199

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

   200

   201

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

   202

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

   203

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

   204

        })

   205

   206

    def testHostLocationDomain(self):

   207

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

   208

                ip          = '192.0.2.1',

   209

                location    = 'test@bar.domain',

   210

        )

   211

   212

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

   213

   214

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

   215

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

   216

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

   217

        })

   218

   219

    def testHostLocationDomainOutOfOrigin(self):

   220

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

   221

                ip          = '192.0.2.1',

   222

                location    = 'test@bar.domain',

   223

        )

   224

   225

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

   226

   227

        with self.assertRaises(zone.HostZoneError):

   228

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

   229

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

   230

            })

   231

   232

        # TODO

   233

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

   234

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

   235

        #})

   236

   237

    def testHostsForward(self):

   238

        hosts = [

   239

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

   240

                    ip      = '192.0.2.1',

   241

                    ip6     = '2001:db8::192.0.2.1',

   242

                    alias   = 'test',

   243

                ),

   244

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

   245

                    ip      = '192.0.2.2',

   246

                )

   247

        ]

   248

   250

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

   251

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

   252

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

   253

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

   254

        })

   255

   256

    def testHostsConflict(self):

   257

        hosts = [

   258

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

   259

                    ip      = '192.0.2.1',

   260

                ),

   261

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

   262

                    ip      = '192.0.2.2',

   263

                )

   264

        ]

   265

   266

        with self.assertRaises(zone.HostZoneError):

   268

   269

    def testHostsAliasConflict(self):

   270

        hosts = [

   271

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

   272

                    ip          = '192.0.2.1',

   273

                ),

   274

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

   275

                    ip          = '192.0.2.2',

   276

                    alias       = 'foo',

   277

                )

   278

        ]

   279

   280

        # with A first

   281

        with self.assertRaises(zone.HostZoneError):

   283

   284

        # also with CNAME first

   285

        with self.assertRaises(zone.HostZoneError):

   287

   288

    def testHostsAlias4Conflict(self):

   289

        hosts = [

   290

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

   291

                    ip          = '192.0.2.1',

   292

                ),

   293

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

   294

                    ip          = '192.0.2.2',

   295

                    alias4      = 'foo',

   296

                )

   297

        ]

   298

   299

        with self.assertRaises(zone.HostZoneError):

   301

   302

   303

class TestReverseZone(TestZoneMixin, unittest.TestCase):

   304

    def setUp(self):

   305

        self.options = Options()

   306

        self.options.unknown_host = False

   307

   308

    def testHostIP(self):

   309

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

   310

                ip  = '192.0.2.1',

   311

                ip6 = '2001:db8::192.0.2.1',

   312

        )

   313

   314

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

   315

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

   316

        })

   317

   318

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

   319

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

   320

        })

   321

   322

    def testHostIP4(self):

   323

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

   324

                ip  = '192.0.2.1',

   325

        )

   326

   327

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

   328

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

   329

        })

   330

   331

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

   332

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

   333

        })

   334

   335

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

   336

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

   337

        })

   338

   339

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

   340

   341

        })

   342

   343

    def testHostIP6(self):

   344

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

   345

                ip6 = '2001:db8::192.0.2.1',

   346

        )

   347

   348

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

   349

        })

   350

   351

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

   352

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

   353

        })

   354

   355

    def testHostIPOutOfPrefix(self):

   356

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

   357

                ip  = '192.0.2.1',

   358

                ip6 = '2001:db8::192.0.2.1',

   359

        )

   360

   361

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

   362

   363

        })

   364

   365

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

   366

   367

        })

   368

   369

    def testHostDelegate(self):

   370

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

   371

                ip      = '192.0.2.1',

   372

                ip6     = '2001:db8::192.0.2.1',

   373

                forward = '',

   374

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

   375

        )

   376

   377

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

   378

   379

        })

   380

   381

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

   382

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

   383

        })

   384

   385

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

   386

   387

        })

   388

   389

    def testHostsConflict(self):

   390

        hosts = [

   391

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

   392

                    ip      = '192.0.2.1',

   393

                ),

   394

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

   395

                    ip      = '192.0.2.1',

   396

                )

   397

        ]

   398

   399

        with self.assertRaises(zone.HostZoneError):

   401

   402

    def testHostsGenerateUnknown(self):

   403

        hosts = [

   404

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

   405

                    ip      = '192.0.2.1',

   406

                ),

   407

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

   408

                    ip      = '192.0.2.5',

   409

                ),

   410

        ]

   411

   416

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

   417

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

   418

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

   419

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

   420

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

   421

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

   422

        })

   423

   424

class TestDhcp(unittest.TestCase):

   425

    def setUp(self):

   426

        self.options = pvl.args.options(

   427

                hosts_charset   = 'utf-8',

   428

                hosts_domain    = None,

   429

                hosts_include   = None,

   430

        )

   431

   432

    def assertBlocksEqual(self, blockdefs, expected):

   433

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

   434

            self.assertEqual(_block, block)

   435

            self.assertItemsEqual(_items, items)

   436

   437

            if opts is not None:

   438

                self.assertEqual(_opts, opts)

   439

   440

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

   441

   442

    def testHost(self):

   443

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

   444

                ip          = '192.0.2.1',

   445

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

   446

        )

   447

   448

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

   449

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

   450

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

   451

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

   452

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

   453

            ], None)

   454

        ])

   455

   456

    def testHostDynamic(self):

   457

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

   458

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

   459

        )

   460

   461

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

   462

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

   463

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

   464

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

   465

            ], None)

   466

        ])

   467

   468

    def testHostBoot(self):

   469

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

   470

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

   471

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

   472

        )

   473

   474

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

   475

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

   476

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

   477

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

   478

                ('next-server', 'boot.lan'),

   479

                ('filename', 'debian/wheezy/pxelinux.0'),

   480

            ], None)

   481

        ])

   482

   483

    def testHosts(self):

   484

        hosts = [

   485

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

   486

                        ip          = '192.0.2.1',

   487

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

   488

                ),

   489

                Host.build('bar', 'test',

   490

                        ip          = '192.0.2.2',