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

    35

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

    36

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

    37

        })

    38

    39

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

    40

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

    41

        ))

    42

    43

    def testApplyHostConfigDictMulti(self):

    44

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

    45

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

    46

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

    47

        })

    48

    49

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

    50

                ethernet    = {

    51

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

    52

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

    53

                }

    54

        ))

    55

    56

    def testApplyHostsConfigErrorDict(self):

    57

        with self.assertRaises(ValueError):

    58

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

    59

                'ethernet': 'foo',

    60

                'ethernet.eth0': 'bar',

    61

            })

    62

    63

    def testApplyHostConfigExtensions(self):

    64

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

    65

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

    66

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

    67

        })

    68

    69

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

    70

                extensions = {

    71

                    'link': {

    72

                        '50': 'foo@test',

    73

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

    74

                    },

    75

                },

    76

        ))

    77

    78

    def testApplyHostFqdn(self):

    79

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

    80

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

    81

        ])

    82

    83

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

    84

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

    85

        ])

    86

    87

    def testApplyHostExpand(self):

    88

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

    89

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

    90

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

    91

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

    92

        ])

    93

    94

    def testApplyHostsFileError(self):

    95

        with self.assertRaises(config.HostConfigError):

    96

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

    97

    99

        conf_file = ConfFile('test', """

   100

[foo]

   101

    ip = 127.0.0.1

   102

   103

[bar]

   104

    ip = 127.0.0.2

   105

        """)

   106

        expected = [

   107

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

   108

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

   109

        ]

   110

   111

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

   112

   123

    def testApply(self):

   124

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

   125

                ('foo@test', dict(

   126

                    ip          = ipaddr.IPAddress('127.0.0.1'),

   127

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

   128

                )),

   129

                ('bar@test', dict(

   130

                    ip          = ipaddr.IPAddress('127.0.0.2'),

   131

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

   132

                )),

   133

        ])

   134

   135

class TestZoneMixin(object):

   136

    def assertZoneEquals(self, rrs, expected):

   137

        gather = { }

   138

   139

        for rr in rrs:

   140

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

   141

   142

            self.assertNotIn(key, gather)

   143

   144

            gather[key] = rr.data

   145

   146

        self.assertDictEqual(gather, expected)

   147

   148

   149

class TestForwardZone(TestZoneMixin, unittest.TestCase):

   150

    def testHostOutOfOrigin(self):

   151

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

   152

                ip  = '10.0.0.1',

   153

        )

   154

   155

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

   156

   157

    def testHostIP(self):

   158

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

   159

                ip  = '192.0.2.1',

   160

                ip6 = '2001:db8::192.0.2.1',

   161

        )

   162

   163

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

   164

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

   165

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

   166

        })

   167

   168

    def testHostAlias(self):

   169

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

   170

                ip      = '192.0.2.1',

   171

                alias   = 'test *.test',

   172

        )

   173

   174

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

   175

   176

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

   177

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

   178

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

   179

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

   180

        })

   181

   182

    def testHostAlias46(self):

   183

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

   184

                ip      = '192.0.2.1',

   185

                ip6     = '2001:db8::192.0.2.1',

   186

                alias4  = 'test4',

   187

                alias6  = 'test6',

   188

        )

   189

   190

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

   191

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

   192

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

   193

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

   194

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

   195

        })

   196

   197

    def testHostAlias4Missing(self):

   198

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

   199

                ip6     = '2001:db8::192.0.2.1',

   200

                alias4  = 'test4',

   201

                alias6  = 'test6',

   202

        )

   203

   204

        with self.assertRaises(zone.HostZoneError):

   205

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

   206

   207

    def testHostAlias6Missing(self):

   208

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

   209

                ip      = '192.0.2.1',

   210

                alias4  = 'test4',

   211

                alias6  = 'test6',

   212

        )

   213

   214

        with self.assertRaises(zone.HostZoneError):

   215

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

   216

   217

    def testHostFQDN(self):

   218

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

   219

                ip          = '192.0.2.3',

   220

        )

   221

   222

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

   223

   224

        })

   225

   226

    def testHostDelegate(self):

   227

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

   228

                forward = 'host.example.net',

   229

        )

   230

   231

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

   232

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

   233

        })

   234

   235

    def testHostForwardAlias(self):

   236

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

   237

                forward = 'host.example.net',

   238

                alias   = 'test',

   239

        )

   240

   241

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

   242

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

   243

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

   244

        })

   245

   246

    def testHostLocation(self):

   247

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

   248

                ip          = '192.0.2.1',

   249

                location    = 'test',

   250

        )

   251

   252

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

   253

   254

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

   255

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

   256

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

   257

        })

   258

   259

    def testHostLocationDomain(self):

   260

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

   261

                ip          = '192.0.2.1',

   262

                location    = 'test@bar.domain',

   263

        )

   264

   265

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

   266

   267

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

   268

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

   269

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

   270

        })

   271

   272

    def testHostLocationDomainOutOfOrigin(self):

   273

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

   274

                ip          = '192.0.2.1',

   275

                location    = 'test@bar.domain',

   276

        )

   277

   278

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

   279

   280

        with self.assertRaises(zone.HostZoneError):

   281

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

   282

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

   283

            })

   284

   285

        # TODO

   286

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

   287

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

   288

        #})

   289

   290

    def testHostsForward(self):

   291

        hosts = [

   292

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

   293

                    ip      = '192.0.2.1',

   294

                    ip6     = '2001:db8::192.0.2.1',

   295

                    alias   = 'test',

   296

                ),

   297

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

   298

                    ip      = '192.0.2.2',

   299

                ),

   300

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

   301

                    ip      = '192.0.2.3',

   302

                ),

   303

        ]

   304

   305

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

   306

   307

        # handle the $ORIGIN directive

   308

        rd = next(rrs)

   309

   310

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

   311

   312

        self.assertZoneEquals(rrs, {

   313

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

   314

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

   315

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

   316

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

   317

        })

   318

   319

    def testHostsConflict(self):

   320

        hosts = [

   321

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

   322

                    ip      = '192.0.2.1',

   323

                ),

   324

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

   325

                    ip      = '192.0.2.2',

   326

                )

   327

        ]

   328

   329

        with self.assertRaises(zone.HostZoneError):

   330

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

   331

   332

    def testHostsAliasConflict(self):

   333

        hosts = [

   334

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

   335

                    ip          = '192.0.2.1',

   336

                ),

   337

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

   338

                    ip          = '192.0.2.2',

   339

                    alias       = 'foo',

   340

                )

   341

        ]

   342

   343

        # with A first

   344

        with self.assertRaises(zone.HostZoneError):

   345

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

   346

   347

        # also with CNAME first

   348

        with self.assertRaises(zone.HostZoneError):

   349

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

   350

   351

    def testHostsAlias4Conflict(self):

   352

        hosts = [

   353

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

   354

                    ip          = '192.0.2.1',

   355

                ),

   356

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

   357

                    ip          = '192.0.2.2',

   358

                    alias4      = 'foo',

   359

                )

   360

        ]

   361

   362

        with self.assertRaises(zone.HostZoneError):

   363

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

   364

   365

   366

class TestReverseZone(TestZoneMixin, unittest.TestCase):

   367

    def testHostIP(self):

   368

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

   369

                ip  = '192.0.2.1',

   370

                ip6 = '2001:db8::192.0.2.1',

   371

        )

   372

   373

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

   374

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

   375

        })

   376

   377

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

   378

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

   379

        })

   380

   381

    def testHostIP4(self):

   382

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

   383

                ip  = '192.0.2.1',

   384

        )

   385

   386

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

   387

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

   388

        })

   389

   390

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

   391

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

   392

        })

   393

   394

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

   395

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

   396

        })

   397

   398

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

   399

   400

        })

   401

   402

    def testHostIP6(self):

   403

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

   404

                ip6 = '2001:db8::192.0.2.1',

   405

        )

   406

   407

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

   408

        })

   409

   410

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

   411

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

   412

        })

   413

   414

    def testHostIPOutOfPrefix(self):

   415

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

   416

                ip  = '192.0.2.1',

   417

                ip6 = '2001:db8::192.0.2.1',

   418

        )

   419

   420

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

   421

   422

        })

   423

   424

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

   425

   426

        })

   427

   428

    def testHostFQDN(self):

   429

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

   430

                ip          = '192.0.2.3',

   431

                ip6         = '2001:db8::192.0.2.3',

   432

        )

   433

   434

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

   435

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

   436

   437

        })

   438

   439

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

   440

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

   441

        })

   442

   443

    def testHostDelegate(self):

   444

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

   445

                ip      = '192.0.2.1',

   446

                ip6     = '2001:db8::192.0.2.1',

   447

                forward = '',

   448

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

   449

        )

   450

   451

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

   452

   453

        })

   454

   455

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

   456

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

   457

        })

   458

   459

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

   460

   461

        })

   462

   463

    def testHosts(self):

   464

        hosts = [

   465

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

   466

                    ip      = '192.0.2.1',

   467

                ),

   468

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

   469

                    ip      = '192.0.2.2',

   470

                )

   471

        ]

   472

   473

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

   474

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

   475

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

   476

        })

   477

   478

        # in ip order

   479

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

   480

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

   481

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

   482

        })

   483

   484

    def testHostsConflict(self):

   485

        hosts = [

   486

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

   487

                    ip      = '192.0.2.1',

   488

                ),

   489

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