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

        hosts = list(hosts)

    30

    31

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

    32

            host_str, attrs = expect

    33

    34

            self.assertHostEqual(host, host_str, attrs)

    35

    36

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

    37

    38

    def testApplyHostConfigDict(self):

    39

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

    40

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

    41

        })

    42

    43

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

    44

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

    45

        ))

    46

    47

    def testApplyHostConfigDictMulti(self):

    48

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

    49

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

    50

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

    51

        })

    52

    53

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

    54

                ethernet    = {

    55

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

    56

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

    57

                }

    58

        ))

    59

    60

    def testApplyHostsConfigErrorDict(self):

    61

        with self.assertRaises(ValueError):

    62

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

    63

                'ethernet': 'foo',

    64

                'ethernet.eth0': 'bar',

    65

            })

    66

    67

    def testApplyHostConfigExtensions(self):

    68

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

    69

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

    70

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

    71

        })

    72

    73

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

    74

                extensions = {

    75

                    'link': {

    76

                        '50': 'foo@test',

    77

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

    78

                    },

    79

                },

    80

        ))

    81

    82

    def testApplyHostFqdn(self):

    83

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

    84

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

    85

        ])

    86

    87

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

    88

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

    89

        ])

    90

    91

    def testApplyHostExpand(self):

    92

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

    93

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

    94

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

    95

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

    96

        ])

    97

    98

    def testApplyHostsFileError(self):

    99

        with self.assertRaises(config.HostConfigError):

   100

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

   101

   102

    def testApplyHostsConfig(self):

   103

        conf_file = ConfFile('test', """

   104

[foo]

   105

    ip = 127.0.0.1

   106

   107

[bar]

   108

    ip = 127.0.0.2

   109

        """)

   110

   111

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

   112

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

   113

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

   114

        ])

   115

   116

    def testApplyHostsConfigNested(self):

   117

        conf_file = ConfFile('test', """

   118

[asdf]

   119

    [[foo]]

   120

        ip = 127.0.0.1

   121

   122

[quux]

   123

    [[bar]]

   124

        ip = 127.0.0.2

   125

        """)

   126

   127

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

   128

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

   129

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

   130

        ])

   131

   132

    def testApplyIncludes(self):

   133

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

   134

                ('foo@test', dict(

   135

                    ip          = ipaddr.IPAddress('192.0.2.1'),

   136

                )),

   137

                ('bar@test', dict(

   138

                    ip          = ipaddr.IPAddress('192.0.2.2'),

   139

                )),

   140

        ])

   141

   142

    def testApplyDirectory(self):

   143

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

   144

                ('quux@asdf.test', dict(

   145

                    ip          = ipaddr.IPAddress('192.0.2.5'),

   146

                )),

   147

                ('foo@test', dict(

   148

                    ip          = ipaddr.IPAddress('192.0.2.1'),

   149

                )),

   150

                ('bar@test', dict(

   151

                    ip          = ipaddr.IPAddress('192.0.2.2'),

   152

                )),

   153

        ])

   154

   169

    def testApply(self):

   170

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

   171

                ('foo@example.com', dict(

   172

                    ip          = ipaddr.IPAddress('192.0.2.1'),

   173

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

   174

                )),

   175

                ('bar@example.com', dict(

   176

                    ip          = ipaddr.IPAddress('192.0.2.2'),

   177

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

   178

                )),

   179

        ])

   180

   181

class TestZoneMixin(object):

   182

    def assertZoneEquals(self, rrs, expected):

   183

        gather = { }

   184

   185

        for rr in rrs:

   186

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

   187

   188

            self.assertNotIn(key, gather)

   189

   190

            gather[key] = rr.data

   191

   192

        self.assertDictEqual(gather, expected)

   193

   194

   195

class TestForwardZone(TestZoneMixin, unittest.TestCase):

   196

    def testHostOutOfOrigin(self):

   197

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

   198

                ip  = '10.0.0.1',

   199

        )

   200

   201

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

   202

   203

    def testHostIP(self):

   204

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

   205

                ip  = '192.0.2.1',

   206

                ip6 = '2001:db8::192.0.2.1',

   207

        )

   208

   209

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

   210

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

   211

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

   212

        })

   213

   214

    def testHostAlias(self):

   215

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

   216

                ip      = '192.0.2.1',

   217

                alias   = 'test *.test',

   218

        )

   219

   220

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

   221

   222

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

   223

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

   224

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

   225

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

   226

        })

   227

   228

    def testHostAlias46(self):

   229

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

   230

                ip      = '192.0.2.1',

   231

                ip6     = '2001:db8::192.0.2.1',

   232

                alias4  = 'test4',

   233

                alias6  = 'test6',

   234

        )

   235

   236

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

   237

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

   238

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

   239

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

   240

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

   241

        })

   242

   243

    def testHostAlias4Missing(self):

   244

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

   245

                ip6     = '2001:db8::192.0.2.1',

   246

                alias4  = 'test4',

   247

                alias6  = 'test6',

   248

        )

   249

   250

        with self.assertRaises(zone.HostZoneError):

   251

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

   252

   253

    def testHostAlias6Missing(self):

   254

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

   255

                ip      = '192.0.2.1',

   256

                alias4  = 'test4',

   257

                alias6  = 'test6',

   258

        )

   259

   260

        with self.assertRaises(zone.HostZoneError):

   261

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

   262

   263

    def testHostFQDN(self):

   264

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

   265

                ip          = '192.0.2.3',

   266

        )

   267

   268

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

   269

   270

        })

   271

   272

    def testHostDelegate(self):

   273

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

   274

                forward = 'host.example.net',

   275

        )

   276

   277

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

   278

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

   279

        })

   280

   281

    def testHostForwardAlias(self):

   282

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

   283

                forward = 'host.example.net',

   284

                alias   = 'test',

   285

        )

   286

   287

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

   288

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

   289

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

   290

        })

   291

   292

    def testHostLocation(self):

   293

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

   294

                ip          = '192.0.2.1',

   295

                location    = 'test',

   296

        )

   297

   298

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

   299

   300

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

   301

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

   302

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

   303

        })

   304

   305

    def testHostLocationDomain(self):

   306

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

   307

                ip          = '192.0.2.1',

   308

                location    = 'test@bar.domain',

   309

        )

   310

   311

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

   312

   313

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

   314

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

   315

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

   316

        })

   317

   318

    def testHostLocationDomainOutOfOrigin(self):

   319

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

   320

                ip          = '192.0.2.1',

   321

                location    = 'test@bar.domain',

   322

        )

   323

   324

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

   325

   326

        with self.assertRaises(zone.HostZoneError):

   327

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

   328

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

   329

            })

   330

   331

        # TODO

   332

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

   333

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

   334

        #})

   335

   336

    def testHostsForward(self):

   337

        hosts = [

   338

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

   339

                    ip      = '192.0.2.1',

   340

                    ip6     = '2001:db8::192.0.2.1',

   341

                    alias   = 'test',

   342

                ),

   343

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

   344

                    ip      = '192.0.2.2',

   345

                ),

   346

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

   347

                    ip      = '192.0.2.3',

   348

                ),

   349

        ]

   350

   351

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

   352

   353

        # handle the $ORIGIN directive

   354

        rd = next(rrs)

   355

   356

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

   357

   358

        self.assertZoneEquals(rrs, {

   359

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

   360

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

   361

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

   362

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

   363

        })

   364

   365

    def testHostsConflict(self):

   366

        hosts = [

   367

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

   368

                    ip      = '192.0.2.1',

   369

                ),

   370

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

   371

                    ip      = '192.0.2.2',

   372

                )

   373

        ]

   374

   375

        with self.assertRaises(zone.HostZoneError):

   376

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

   377

   378

    def testHostsAliasConflict(self):

   379

        hosts = [

   380

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

   381

                    ip          = '192.0.2.1',

   382

                ),

   383

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

   384

                    ip          = '192.0.2.2',

   385

                    alias       = 'foo',

   386

                )

   387

        ]

   388

   389

        # with A first

   390

        with self.assertRaises(zone.HostZoneError):

   391

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

   392

   393

        # also with CNAME first

   394

        with self.assertRaises(zone.HostZoneError):

   395

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

   396

   397

    def testHostsAlias4Conflict(self):

   398

        hosts = [

   399

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

   400

                    ip          = '192.0.2.1',

   401

                ),

   402

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

   403

                    ip          = '192.0.2.2',

   404

                    alias4      = 'foo',

   405

                )

   406

        ]

   407

   408

        with self.assertRaises(zone.HostZoneError):

   409

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

   410

   411

   412

class TestReverseZone(TestZoneMixin, unittest.TestCase):

   413

    def testHostIP(self):

   414

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

   415

                ip  = '192.0.2.1',

   416

                ip6 = '2001:db8::192.0.2.1',

   417

        )

   418

   419

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

   420

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

   421

        })

   422

   423

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

   424

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

   425

        })

   426

   427

    def testHostIP4(self):

   428

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

   429

                ip  = '192.0.2.1',

   430

        )

   431

   432

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

   433

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

   434

        })

   435

   436

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

   437

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

   438

        })

   439

   440

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

   441

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

   442

        })

   443

   444

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

   445

   446

        })

   447

   448

    def testHostIP6(self):

   449

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

   450

                ip6 = '2001:db8::192.0.2.1',

   451

        )

   452

   453

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

   454

        })

   455

   456

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

   457

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

   458

        })

   459

   460

    def testHostIPOutOfPrefix(self):

   461

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

   462

                ip  = '192.0.2.1',

   463

                ip6 = '2001:db8::192.0.2.1',

   464

        )

   465

   466

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

   467

   468

        })

   469

   470

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

   471

   472

        })

   473

   474

    def testHostFQDN(self):

   475

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

   476

                ip          = '192.0.2.3',

   477

                ip6         = '2001:db8::192.0.2.3',

   478

        )

   479

   480

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

   481

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

   482

   483

        })

   484

   485

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

   486

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

   487

        })

   488