1

import ipaddr

     2

import itertools

     3

import pvl.args

     4

import unittest

     5

     6

from pvl.hosts import config, dhcp, zone

     7

from pvl.hosts.host import Host

     8

from StringIO import StringIO

     9

    10

class ConfFile(StringIO):

    11

    def __init__(self, name, buffer):

    12

        StringIO.__init__(self, buffer)

    13

        self.name = name

    14

    15

class TestConfig(unittest.TestCase):

    16

    def setUp(self):

    17

        self.options = pvl.args.options(

    18

                hosts_charset   = 'utf-8',

    19

                hosts_domain    = None,

    20

                hosts_include   = None,

    21

        )

    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

        hosts = list(hosts)

    31

    32

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

    33

            host_str, attrs = expect

    34

    35

            self.assertHostEqual(host, host_str, attrs)

    36

    37

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

    38

    39

    def testApplyHostConfigDict(self):

    40

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

    41

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

    42

        })

    43

    44

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

    45

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

    46

        ))

    47

    48

    def testApplyHostConfigDictMulti(self):

    49

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

    50

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

    51

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

    52

        })

    53

    54

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

    55

                ethernet    = {

    56

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

    57

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

    58

                }

    59

        ))

    60

    61

    def testApplyHostsConfigErrorExtra(self):

    62

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

    63

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

    64

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

    65

        })

    66

    67

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

    68

                ethernet    = {

    69

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

    70

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

    71

                }

    72

        ))

    73

    74

    75

    def testApplyHostConfigExtensions(self):

    76

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

    77

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

    78

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

    79

        })

    80

    81

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

    82

                extensions = {

    83

                    'link': {

    84

                        '50': 'foo@test',

    85

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

    86

                    },

    87

                },

    88

        ))

    89

    90

    def testApplyHostFqdn(self):

    91

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

    92

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

    93

        ])

    94

    95

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

    96

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

    97

        ])

    98

    99

    def testApplyHostExpand(self):

   100

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

   101

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

   102

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

   103

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

   104

        ])

   105

   106

    def testApplyHostsFileError(self):

   107

        with self.assertRaises(config.HostConfigError):

   108

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

   109

   110

    def testApplyHostsConfig(self):

   111

        conf_file = ConfFile('test', """

   112

[foo]

   113

    ip = 127.0.0.1

   114

   115

[bar]

   116

    ip = 127.0.0.2

   117

        """)

   118

   119

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

   120

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

   121

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

   122

        ])

   123

   124

    def testApplyHostsConfigNested(self):

   125

        conf_file = ConfFile('test', """

   126

[asdf]

   127

    [[foo]]

   128

        ip = 127.0.0.1

   129

   130

[quux]

   131

    [[bar]]

   132

        ip = 127.0.0.2

   133

        """)

   134

   135

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

   136

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

   137

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

   138

        ])

   139

   140

    def testApplyIncludes(self):

   141

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

   142

                ('bar@test', dict(

   143

                    ip          = ipaddr.IPAddress('192.0.2.2'),

   144

                )),

   145

                ('foo@test', dict(

   146

                    ip          = ipaddr.IPAddress('192.0.2.1'),

   147

                )),

   148

        ])

   149

   150

    def testApplyIncludePath(self):

   151

        self.options.hosts_include = 'etc/hosts'

   152

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

   153

                ('quux@asdf.test', dict(

   154

                    ip          = ipaddr.IPAddress('192.0.2.5'),

   155

                )),

   156

                ('bar@test', dict(

   157

                    ip          = ipaddr.IPAddress('192.0.2.2'),

   158

                )),

   159

                ('foo@test', dict(

   160

                    ip          = ipaddr.IPAddress('192.0.2.1'),

   161

                )),

   162

        ])

   163

   164

    def testApply(self):

   165

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

   166

                ('foo@example.com', dict(

   167

                    ip          = ipaddr.IPAddress('192.0.2.1'),

   168

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

   169

                )),

   170

                ('bar@example.com', dict(

   171

                    ip          = ipaddr.IPAddress('192.0.2.2'),

   172

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

   173

                )),

   174

        ])

   175

   176

class TestZoneMixin(object):

   177

    def assertZoneEquals(self, rrs, expected):

   178

        gather = { }

   179

   180

        for rr in rrs:

   181

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

   182

   183

            self.assertNotIn(key, gather)

   184

   185

            gather[key] = rr.data

   186

   187

        self.assertDictEqual(gather, expected)

   188

   189

   190

class TestForwardZone(TestZoneMixin, unittest.TestCase):

   191

    def testHostOutOfOrigin(self):

   192

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

   193

                ip  = '10.0.0.1',

   194

        )

   195

   196

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

   197

   198

    def testHostIP(self):

   199

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

   200

                ip  = '192.0.2.1',

   201

                ip6 = '2001:db8::192.0.2.1',

   202

        )

   203

   204

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

   205

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

   206

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

   207

        })

   208

   209

    def testHostAlias(self):

   210

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

   211

                ip      = '192.0.2.1',

   212

                alias   = 'test *.test',

   213

        )

   214

   215

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

   216

   217

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

   218

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

   219

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

   220

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

   221

        })

   222

   223

    def testHostAlias46(self):

   224

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

   225

                ip      = '192.0.2.1',

   226

                ip6     = '2001:db8::192.0.2.1',

   227

                alias4  = 'test4',

   228

                alias6  = 'test6',

   229

        )

   230

   231

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

   232

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

   233

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

   234

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

   235

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

   236

        })

   237

   238

    def testHostAlias4Missing(self):

   239

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

   240

                ip6     = '2001:db8::192.0.2.1',

   241

                alias4  = 'test4',

   242

                alias6  = 'test6',

   243

        )

   244

   245

        with self.assertRaises(zone.HostZoneError):

   246

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

   247

   248

    def testHostAlias6Missing(self):

   249

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

   250

                ip      = '192.0.2.1',

   251

                alias4  = 'test4',

   252

                alias6  = 'test6',

   253

        )

   254

   255

        with self.assertRaises(zone.HostZoneError):

   256

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

   257

   258

    def testHostFQDN(self):

   259

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

   260

                ip          = '192.0.2.3',

   261

        )

   262

   263

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

   264

   265

        })

   266

   267

    def testHostDelegate(self):

   268

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

   269

                forward = 'host.example.net',

   270

        )

   271

   272

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

   273

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

   274

        })

   275

   276

    def testHostForwardAlias(self):

   277

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

   278

                forward = 'host.example.net',

   279

                alias   = 'test',

   280

        )

   281

   282

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

   283

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

   284

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

   285

        })

   286

   287

    def testHostLocation(self):

   288

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

   289

                ip          = '192.0.2.1',

   290

                location    = 'test',

   291

        )

   292

   293

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

   294

   295

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

   296

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

   297

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

   298

        })

   299

   300

    def testHostLocationDomain(self):

   301

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

   302

                ip          = '192.0.2.1',

   303

                location    = 'test@bar.domain',

   304

        )

   305

   306

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

   307

   308

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

   309

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

   310

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

   311

        })

   312

   313

    def testHostLocationDomainOutOfOrigin(self):

   314

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

   315

                ip          = '192.0.2.1',

   316

                location    = 'test@bar.domain',

   317

        )

   318

   319

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

   320

   321

        with self.assertRaises(zone.HostZoneError):

   322

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

   323

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

   324

            })

   325

   326

        # TODO

   327

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

   328

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

   329

        #})

   330

   331

    def testHostsForward(self):

   332

        hosts = [

   333

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

   334

                    ip      = '192.0.2.1',

   335

                    ip6     = '2001:db8::192.0.2.1',

   336

                    alias   = 'test',

   337

                ),

   338

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

   339

                    ip      = '192.0.2.2',

   340

                ),

   341

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

   342

                    ip      = '192.0.2.3',

   343

                ),

   344

        ]

   345

   346

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

   347

   348

        # handle the $ORIGIN directive

   349

        rd = next(rrs)

   350

   351

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

   352

   353

        self.assertZoneEquals(rrs, {

   354

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

   355

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

   356

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

   357

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

   358

        })

   359

   360

    def testHostsConflict(self):

   361

        hosts = [

   362

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

   363

                    ip      = '192.0.2.1',

   364

                ),

   365

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

   366

                    ip      = '192.0.2.2',

   367

                )

   368

        ]

   369

   370

        with self.assertRaises(zone.HostZoneError):

   371

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

   372

   373

    def testHostsAliasConflict(self):

   374

        hosts = [

   375

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

   376

                    ip          = '192.0.2.1',

   377

                ),

   378

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

   379

                    ip          = '192.0.2.2',

   380

                    alias       = 'foo',

   381

                )

   382

        ]

   383

   384

        # with A first

   385

        with self.assertRaises(zone.HostZoneError):

   386

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

   387

   388

        # also with CNAME first

   389

        with self.assertRaises(zone.HostZoneError):

   390

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

   391

   392

    def testHostsAlias4Conflict(self):

   393

        hosts = [

   394

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

   395

                    ip          = '192.0.2.1',

   396

                ),

   397

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

   398

                    ip          = '192.0.2.2',

   399

                    alias4      = 'foo',

   400

                )

   401

        ]

   402

   403

        with self.assertRaises(zone.HostZoneError):

   404

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

   405

   406

   407

class TestReverseZone(TestZoneMixin, unittest.TestCase):

   408

    def testHostIP(self):

   409

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

   410

                ip  = '192.0.2.1',

   411

                ip6 = '2001:db8::192.0.2.1',

   412

        )

   413

   414

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

   415

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

   416

        })

   417

   418

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

   419

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

   420

        })

   421

   422

    def testHostIP4(self):

   423

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

   424

                ip  = '192.0.2.1',

   425

        )

   426

   427

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

   428

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

   429

        })

   430

   431

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

   432

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

   433

        })

   434

   435

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

   436

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

   437

        })

   438

   439

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

   440

   441

        })

   442

   443

    def testHostIP6(self):

   444

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

   445

                ip6 = '2001:db8::192.0.2.1',

   446

        )

   447

   448

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

   449

        })

   450

   451

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

   452

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

   453

        })

   454

   455

    def testHostIPOutOfPrefix(self):

   456

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

   457

                ip  = '192.0.2.1',

   458

                ip6 = '2001:db8::192.0.2.1',

   459

        )

   460

   461

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

   462

   463

        })

   464

   465

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

   466

   467

        })

   468

   469

    def testHostFQDN(self):

   470

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

   471

                ip          = '192.0.2.3',

   472

                ip6         = '2001:db8::192.0.2.3',

   473

        )

   474

   475

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

   476

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

   477

   478

        })

   479

   480

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

   481

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

   482

        })

   483

   484

    def testHostDelegate(self):

   485

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

   486

                ip      = '192.0.2.1',

   487

                ip6     = '2001:db8::192.0.2.1',

   488

                forward = '',