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

    73

    74

    def testApplyHostConfigExtensions(self):

    75

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

    76

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

    77

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

    78

        })

    79

    80

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

    81

                extensions = {

    82

                    'link': {

    83

                        '50': 'foo@test',

    84

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

    85

                    },

    86

                },

    87

        ))

    88

    89

    def testApplyHostFqdn(self):

    90

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

    91

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

    92

        ])

    93

    94

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

    95

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

    96

        ])

    97

    98

    def testApplyHostExpand(self):

    99

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

   100

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

   101

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

   102

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

   103

        ])

   104

   105

    def testApplyHostsFileError(self):

   106

        with self.assertRaises(config.HostConfigError):

   107

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

   108

   109

    def testApplyHostsConfig(self):

   110

        conf_file = ConfFile('test', """

   111

[foo]

   112

    ip = 127.0.0.1

   113

   114

[bar]

   115

    ip = 127.0.0.2

   116

        """)

   117

   118

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

   119

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

   120

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

   121

        ])

   122

   123

    def testApplyHostsConfigNested(self):

   124

        conf_file = ConfFile('test', """

   125

[asdf]

   126

    [[foo]]

   127

        ip = 127.0.0.1

   128

   129

[quux]

   130

    [[bar]]

   131

        ip = 127.0.0.2

   132

        """)

   133

   134

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

   135

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

   136

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

   137

        ])

   138

   139

    def testApplyIncludes(self):

   140

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

   141

                ('foo@test', dict(

   142

                    ip          = ipaddr.IPAddress('192.0.2.1'),

   143

                )),

   144

                ('bar@test', dict(

   145

                    ip          = ipaddr.IPAddress('192.0.2.2'),

   146

                )),

   147

        ])

   148

   149

    def testApplyIncludePath(self):

   150

        self.options.hosts_include = 'etc/hosts'

   151

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

   152

                ('foo@test', dict(

   153

                    ip          = ipaddr.IPAddress('192.0.2.1'),

   154

                )),

   155

                ('bar@test', dict(

   156

                    ip          = ipaddr.IPAddress('192.0.2.2'),

   157

                )),

   158

                ('quux@asdf.test', dict(

   159

                    ip          = ipaddr.IPAddress('192.0.2.5'),

   160

                )),

   161

        ])

   162

   163

    def testApply(self):

   164

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

   165

                ('foo@example.com', dict(

   166

                    ip          = ipaddr.IPAddress('192.0.2.1'),

   167

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

   168

                )),

   169

                ('bar@example.com', dict(

   170

                    ip          = ipaddr.IPAddress('192.0.2.2'),

   171

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

   172

                )),

   173

        ])

   174

   175

class TestZoneMixin(object):

   176

    def assertZoneEquals(self, rrs, expected):

   177

        gather = { }

   178

   179

        for rr in rrs:

   180

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

   181

   182

            self.assertNotIn(key, gather)

   183

   184

            gather[key] = rr.data

   185

   186

        self.assertDictEqual(gather, expected)

   187

   188

   189

class TestForwardZone(TestZoneMixin, unittest.TestCase):

   190

    def testHostOutOfOrigin(self):

   191

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

   192

                ip  = '10.0.0.1',

   193

        )

   194

   195

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

   196

   197

    def testHostIP(self):

   198

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

   199

                ip  = '192.0.2.1',

   200

                ip6 = '2001:db8::192.0.2.1',

   201

        )

   202

   203

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

   204

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

   205

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

   206

        })

   207

   208

    def testHostAlias(self):

   209

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

   210

                ip      = '192.0.2.1',

   211

                alias   = 'test *.test',

   212

        )

   213

   214

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

   215

   216

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

   217

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

   218

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

   219

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

   220

        })

   221

   222

    def testHostAlias46(self):

   223

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

   224

                ip      = '192.0.2.1',

   225

                ip6     = '2001:db8::192.0.2.1',

   226

                alias4  = 'test4',

   227

                alias6  = 'test6',

   228

        )

   229

   230

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

   231

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

   232

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

   233

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

   234

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

   235

        })

   236

   237

    def testHostAlias4Missing(self):

   238

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

   239

                ip6     = '2001:db8::192.0.2.1',

   240

                alias4  = 'test4',

   241

                alias6  = 'test6',

   242

        )

   243

   244

        with self.assertRaises(zone.HostZoneError):

   245

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

   246

   247

    def testHostAlias6Missing(self):

   248

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

   249

                ip      = '192.0.2.1',

   250

                alias4  = 'test4',

   251

                alias6  = 'test6',

   252

        )

   253

   254

        with self.assertRaises(zone.HostZoneError):

   255

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

   256

   257

    def testHostFQDN(self):

   258

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

   259

                ip          = '192.0.2.3',

   260

        )

   261

   262

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

   263

   264

        })

   265

   266

    def testHostDelegate(self):

   267

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

   268

                forward = 'host.example.net',

   269

        )

   270

   271

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

   272

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

   273

        })

   274

   275

    def testHostForwardAlias(self):

   276

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

   277

                forward = 'host.example.net',

   278

                alias   = 'test',

   279

        )

   280

   281

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

   282

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

   283

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

   284

        })

   285

   286

    def testHostLocation(self):

   287

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

   288

                ip          = '192.0.2.1',

   289

                location    = 'test',

   290

        )

   291

   292

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

   293

   294

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

   295

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

   296

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

   297

        })

   298

   299

    def testHostLocationDomain(self):

   300

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

   301

                ip          = '192.0.2.1',

   302

                location    = 'test@bar.domain',

   303

        )

   304

   305

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

   306

   307

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

   308

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

   309

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

   310

        })

   311

   312

    def testHostLocationDomainOutOfOrigin(self):

   313

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

   314

                ip          = '192.0.2.1',

   315

                location    = 'test@bar.domain',

   316

        )

   317

   318

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

   319

   320

        with self.assertRaises(zone.HostZoneError):

   321

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

   322

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

   323

            })

   324

   325

        # TODO

   326

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

   327

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

   328

        #})

   329

   330

    def testHostsForward(self):

   331

        hosts = [

   332

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

   333

                    ip      = '192.0.2.1',

   334

                    ip6     = '2001:db8::192.0.2.1',

   335

                    alias   = 'test',

   336

                ),

   337

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

   338

                    ip      = '192.0.2.2',

   339

                ),

   340

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

   341

                    ip      = '192.0.2.3',

   342

                ),

   343

        ]

   344

   345

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

   346

   347

        # handle the $ORIGIN directive

   348

        rd = next(rrs)

   349

   350

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

   351

   352

        self.assertZoneEquals(rrs, {

   353

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

   354

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

   355

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

   356

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

   357

        })

   358

   359

    def testHostsConflict(self):

   360

        hosts = [

   361

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

   362

                    ip      = '192.0.2.1',

   363

                ),

   364

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

   365

                    ip      = '192.0.2.2',

   366

                )

   367

        ]

   368

   369

        with self.assertRaises(zone.HostZoneError):

   370

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

   371

   372

    def testHostsAliasConflict(self):

   373

        hosts = [

   374

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

   375

                    ip          = '192.0.2.1',

   376

                ),

   377

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

   378

                    ip          = '192.0.2.2',

   379

                    alias       = 'foo',

   380

                )

   381

        ]

   382

   383

        # with A first

   384

        with self.assertRaises(zone.HostZoneError):

   385

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

   386

   387

        # also with CNAME first

   388

        with self.assertRaises(zone.HostZoneError):

   389

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

   390

   391

    def testHostsAlias4Conflict(self):

   392

        hosts = [

   393

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

   394

                    ip          = '192.0.2.1',

   395

                ),

   396

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

   397

                    ip          = '192.0.2.2',

   398

                    alias4      = 'foo',

   399

                )

   400

        ]

   401

   402

        with self.assertRaises(zone.HostZoneError):

   403

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

   404

   405

   406

class TestReverseZone(TestZoneMixin, unittest.TestCase):

   407

    def testHostIP(self):

   408

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

   409

                ip  = '192.0.2.1',

   410

                ip6 = '2001:db8::192.0.2.1',

   411

        )

   412

   413

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

   414

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

   415

        })

   416

   417

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

   418

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

   419

        })

   420

   421

    def testHostIP4(self):

   422

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

   423

                ip  = '192.0.2.1',

   424

        )

   425

   426

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

   427

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

   428

        })

   429

   430

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

   431

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

   432

        })

   433

   434

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

   435

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

   436

        })

   437

   438

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

   439

   440

        })

   441

   442

    def testHostIP6(self):

   443

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

   444

                ip6 = '2001:db8::192.0.2.1',

   445

        )

   446

   447

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

   448

        })

   449

   450

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

   451

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

   452

        })

   453

   454

    def testHostIPOutOfPrefix(self):

   455

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

   456

                ip  = '192.0.2.1',

   457

                ip6 = '2001:db8::192.0.2.1',

   458

        )

   459

   460

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

   461

   462

        })

   463

   464

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

   465

   466

        })

   467

   468

    def testHostFQDN(self):

   469

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

   470

                ip          = '192.0.2.3',

   471

                ip6         = '2001:db8::192.0.2.3',

   472

        )

   473

   474

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

   475

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

   476

   477

        })

   478

   479

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

   480

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

   481

        })

   482

   483

    def testHostDelegate(self):

   484

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

   485

                ip      = '192.0.2.1',

   486

                ip6     = '2001:db8::192.0.2.1',

   487

                forward = '',

   488

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