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 ConfFile(StringIO):

    10

    def __init__(self, name, buffer):

    11

        StringIO.__init__(self, buffer)

    12

        self.name = name

    13

    14

class TestConfig(unittest.TestCase):

    15

    def setUp(self):

    16

        self.options = pvl.args.options(

    17

                hosts_charset   = 'utf-8',

    18

                hosts_domain    = None,

    19

                hosts_include   = None,

    20

        )

    21

    22

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

    23

        self.assertEquals(str(host), host_str)

    24

    25

        for attr, value in attrs.iteritems():

    26

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

    27

    28

    def assertHostsEqual(self, hosts, expected):

    29

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

    30

            host_str, attrs = expect

    31

    32

            self.assertHostEqual(host, host_str, attrs)

    33

    34

    def testApplyHostsFileError(self):

    35

        with self.assertRaises(config.HostConfigError):

    37

    38

    def testApplyHosts(self):

    39

        conf_file = ConfFile('test', """

    40

[foo]

    41

    ip = 127.0.0.1

    42

    43

[bar]

    44

    ip = 127.0.0.2

    45

        """)

    46

        expected = [

    47

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

    48

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

    49

        ]

    50

    51

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

    52

    53

    def testApply(self):

    54

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

    55

                ('foo@test', dict(

    56

                    ip          = ipaddr.IPAddress('127.0.0.1'),

    57

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

    58

                )),

    59

                ('bar@test', dict(

    60

                    ip          = ipaddr.IPAddress('127.0.0.2'),

    61

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

    62

                )),

    63

        ])

    64

    65

    def testApplyHostFqdn(self):

    67

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

    68

        ])

    69

    71

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

    72

        ])

    73

    74

    def testApplyHostExpand(self):

    79

        ])

    80

    81

    def testApplyHostConfigDict(self):

    83

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

    84

        })

    85

    86

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

    87

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

    88

        ))

    89

    90

    def testApplyHostConfigDictMulti(self):

    92

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

    93

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

    94

        })

    95

    96

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

    97

                ethernet    = {

    98

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

    99

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

   100

                }

   101

        ))

   102

   106

                'ethernet': 'foo',

   107

                'ethernet.eth0': 'bar',

   108

            })

   109

   110

    def testApplyHostConfigExtensions(self):

   112

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

   113

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

   114

        })

   115

   116

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

   117

                extensions = {

   118

                    'link': {

   119

                        '50': 'foo@test',

   120

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

   121

                    },

   122

                },

   123

        ))

   124

   125

   126

class TestZoneMixin(object):

   127

    def assertZoneEquals(self, rrs, expected):

   128

        gather = { }

   129

   130

        for rr in rrs:

   131

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

   132

   133

            self.assertNotIn(key, gather)

   134

   135

            gather[key] = rr.data

   136

   137

        self.assertDictEqual(gather, expected)

   138

   139

   140

class TestForwardZone(TestZoneMixin, unittest.TestCase):

   141

    def testHostOutOfOrigin(self):

   145

   146

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

   147

   148

    def testHostIP(self):

   149

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

   150

                ip  = '192.0.2.1',

   151

                ip6 = '2001:db8::192.0.2.1',

   152

        )

   153

   154

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

   155

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

   156

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

   157

        })

   158

   159

    def testHostAlias(self):

   160

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

   161

                ip      = '192.0.2.1',

   162

                alias   = 'test *.test',

   163

        )

   164

   165

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

   166

   167

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

   168

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

   169

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

   170

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

   171

        })

   172

   173

    def testHostAlias46(self):

   174

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

   175

                ip      = '192.0.2.1',

   176

                ip6     = '2001:db8::192.0.2.1',

   177

                alias4  = 'test4',

   178

                alias6  = 'test6',

   179

        )

   180

   181

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

   182

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

   183

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

   184

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

   185

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

   186

        })

   187

   188

    def testHostAlias4Missing(self):

   189

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

   190

                ip6     = '2001:db8::192.0.2.1',

   191

                alias4  = 'test4',

   192

                alias6  = 'test6',

   193

        )

   194

   195

        with self.assertRaises(zone.HostZoneError):

   196

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

   197

   198

    def testHostAlias6Missing(self):

   199

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

   200

                ip      = '192.0.2.1',

   201

                alias4  = 'test4',

   202

                alias6  = 'test6',

   203

        )

   204

   205

        with self.assertRaises(zone.HostZoneError):

   206

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

   207

   208

    def testHostFQDN(self):

   209

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

   210

                ip          = '192.0.2.3',

   211

        )

   212

   213

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

   214

   215

        })

   216

   217

    def testHostDelegate(self):

   218

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

   219

                forward = 'host.example.net',

   220

        )

   221

   222

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

   223

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

   224

        })

   225

   226

    def testHostForwardAlias(self):

   227

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

   228

                forward = 'host.example.net',

   229

                alias   = 'test',

   230

        )

   231

   232

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

   233

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

   234

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

   235

        })

   236

   237

    def testHostLocation(self):

   238

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

   239

                ip          = '192.0.2.1',

   240

                location    = 'test',

   241

        )

   242

   243

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

   244

   245

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

   246

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

   247

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

   248

        })

   249

   250

    def testHostLocationDomain(self):

   251

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

   252

                ip          = '192.0.2.1',

   253

                location    = 'test@bar.domain',

   254

        )

   255

   256

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

   257

   258

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

   259

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

   260

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

   261

        })

   262

   263

    def testHostLocationDomainOutOfOrigin(self):

   264

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

   265

                ip          = '192.0.2.1',

   266

                location    = 'test@bar.domain',

   267

        )

   268

   269

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

   270

   271

        with self.assertRaises(zone.HostZoneError):

   272

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

   273

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

   274

            })

   275

   276

        # TODO

   277

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

   278

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

   279

        #})

   280

   281

    def testHostsForward(self):

   282

        hosts = [

   283

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

   284

                    ip      = '192.0.2.1',

   285

                    ip6     = '2001:db8::192.0.2.1',

   286

                    alias   = 'test',

   287

                ),

   288

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

   289

                    ip      = '192.0.2.2',

   290

                ),

   291

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

   292

                    ip      = '192.0.2.3',

   293

                ),

   294

        ]

   295

   296

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

   297

   298

        # handle the $ORIGIN directive

   299

        rd = next(rrs)

   300

   301

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

   302

   303

        self.assertZoneEquals(rrs, {

   304

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

   305

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

   306

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

   307

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

   308

        })

   309

   310

    def testHostsConflict(self):

   311

        hosts = [

   312

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

   313

                    ip      = '192.0.2.1',

   314

                ),

   315

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

   316

                    ip      = '192.0.2.2',

   317

                )

   318

        ]

   319

   320

        with self.assertRaises(zone.HostZoneError):

   321

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

   322

   323

    def testHostsAliasConflict(self):

   324

        hosts = [

   325

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

   326

                    ip          = '192.0.2.1',

   327

                ),

   328

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

   329

                    ip          = '192.0.2.2',

   330

                    alias       = 'foo',

   331

                )

   332

        ]

   333

   334

        # with A first

   335

        with self.assertRaises(zone.HostZoneError):

   336

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

   337

   338

        # also with CNAME first

   339

        with self.assertRaises(zone.HostZoneError):

   340

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

   341

   342

    def testHostsAlias4Conflict(self):

   343

        hosts = [

   344

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

   345

                    ip          = '192.0.2.1',

   346

                ),

   347

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

   348

                    ip          = '192.0.2.2',

   349

                    alias4      = 'foo',

   350

                )

   351

        ]

   352

   353

        with self.assertRaises(zone.HostZoneError):

   354

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

   355

   356

   357

class TestReverseZone(TestZoneMixin, unittest.TestCase):

   358

    def testHostIP(self):

   359

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

   360

                ip  = '192.0.2.1',

   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

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

   366

        })

   367

   368

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

   369

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

   370

        })

   371

   372

    def testHostIP4(self):

   373

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

   374

                ip  = '192.0.2.1',

   375

        )

   376

   377

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

   378

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

   379

        })

   380

   381

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

   382

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

   383

        })

   384

   385

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

   386

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

   387

        })

   388

   389

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

   390

   391

        })

   392

   393

    def testHostIP6(self):

   394

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

   395

                ip6 = '2001:db8::192.0.2.1',

   396

        )

   397

   398

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

   399

        })

   400

   401

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

   402

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

   403

        })

   404

   405

    def testHostIPOutOfPrefix(self):

   406

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

   407

                ip  = '192.0.2.1',

   408

                ip6 = '2001:db8::192.0.2.1',

   409

        )

   410

   411

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

   412

   413

        })

   414

   415

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

   416

   417

        })

   418

   419

    def testHostFQDN(self):

   420

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

   421

                ip          = '192.0.2.3',

   422

                ip6         = '2001:db8::192.0.2.3',

   423

        )

   424

   425

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

   426

            ('3', 'PTR'): ['host.example.net.'],

   427

   428

        })

   429

   430

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

   431

            ('3.0.2.0.0.0.0.c.0.0.0.0.0.0.0.0', 'PTR'): ['host.example.net.'],

   432

        })

   433

   434

    def testHostDelegate(self):

   435

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

   436

                ip      = '192.0.2.1',

   437

                ip6     = '2001:db8::192.0.2.1',

   438

                forward = '',

   439

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

   440

        )

   441

   442

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

   443

   444

        })

   445

   446

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

   447

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

   448

        })

   449

   450

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

   451

   452

        })

   453

   454

    def testHosts(self):

   455

        hosts = [

   456

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

   457

                    ip      = '192.0.2.1',

   458

                ),

   459

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

   460

                    ip      = '192.0.2.2',

   461

                )

   462

        ]

   463

   464

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

   465

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

   466

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

   467

        })

   468

   469

        # in ip order

   470

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

   471

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

   472

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

   473

        })

   474

   475

    def testHostsConflict(self):

   476

        hosts = [

   477

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

   478

                    ip      = '192.0.2.1',

   479

                ),

   480

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

   481

                    ip      = '192.0.2.1',

   482

                )

   483

        ]

   484

   485

        with self.assertRaises(zone.HostZoneError):

   486

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

   487

   488

    def testHostsGenerateUnknown(self):

   489

        hosts = [