1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
|
# Copyright 2014 Hewlett-Packard Development Company, L.P.
#
# Author: Endre Karlson <endre.karlson@hpe.com>
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import mock
from dns import zone as dnszone
import dns.message
import dns.rdatatype
import dns.rcode
from designate import dnsutils
from designate.tests import TestCase
SAMPLES = {
("cname.example.com.", "CNAME"): {
"records": ["example.com."],
},
("_http._tcp.example.com.", "SRV"): {
"records": [
"10 0 80 192.0.0.4.example.com.",
"10 5 80 192.0.0.5.example.com."
],
},
("ipv4.example.com.", "A"): {
"ttl": 300,
"records": ["192.0.0.1"]
},
("delegation.example.com.", "NS"): {
"records": ["ns1.example.com."]
},
("ipv6.example.com.", "AAAA"): {
"records": ["fd00::1"],
},
("example.com.", "SOA"): {
"records": [
"ns1.example.com. nsadmin.example.com."
" 2013091101 7200 3600 2419200 10800"
],
"ttl": 600
},
("example.com.", "MX"): {
"records": [
"5 192.0.0.2.example.com.",
'10 192.0.0.3.example.com.'
]
},
("example.com.", "TXT"): {
"records": ['"abc" "def"']
},
("example.com.", "SPF"): {
"records": ['"v=spf1 mx a"']
},
("example.com.", "NS"): {
"records": [
'ns1.example.com.',
'ns2.example.com.'
]
}
}
class TestUtils(TestCase):
def test_parse_zone(self):
zone_file = self.get_zonefile_fixture()
dnspython_zone = dnszone.from_text(
zone_file,
# Don't relativize, otherwise we end up with '@' record names.
relativize=False,
# Dont check origin, we allow missing NS records (missing SOA
# records are taken care of in _create_zone).
check_origin=False)
zone = dnsutils.from_dnspython_zone(dnspython_zone)
for rrset in zone.recordsets:
k = (rrset.name, rrset.type)
self.assertIn(k, SAMPLES)
sample_ttl = SAMPLES[k].get('ttl', None)
if rrset.obj_attr_is_set('ttl') or sample_ttl is not None:
self.assertEqual(rrset.ttl, sample_ttl)
self.assertEqual(len(SAMPLES[k]['records']), len(rrset.records))
for r in rrset.records:
self.assertIn(r.data, SAMPLES[k]['records'])
self.assertEqual(len(SAMPLES), len(zone.recordsets))
self.assertEqual('example.com.', zone.name)
def test_zone_lock(self):
# Initialize a ZoneLock
lock = dnsutils.ZoneLock(0.1)
# Ensure there's no lock for different zones
for zone_name in ['foo.com.', 'bar.com.', 'example.com.']:
self.assertTrue(lock.acquire(zone_name))
# Ensure a lock for successive calls for the same zone
self.assertTrue(lock.acquire('example2.com.'))
self.assertFalse(lock.acquire('example2.com.'))
# Acquire, release, and reacquire
self.assertTrue(lock.acquire('example3.com.'))
lock.release('example3.com.')
self.assertTrue(lock.acquire('example3.com.'))
def test_limit_notify_middleware(self):
# Set the delay
self.config(notify_delay=.1,
group='service:agent')
# Initialize the middlware
placeholder_app = None
middleware = dnsutils.LimitNotifyMiddleware(placeholder_app)
# Prepare a NOTIFY
zone_name = 'example.com.'
notify = dns.message.make_query(zone_name, dns.rdatatype.SOA)
notify.flags = 0
notify.set_opcode(dns.opcode.NOTIFY)
notify.flags |= dns.flags.AA
# Send the NOTIFY through the middleware
# No problem, middleware should return None to pass it on
self.assertIsNone(middleware.process_request(notify))
@mock.patch('designate.dnsutils.ZoneLock.acquire', return_value=False)
def test_limit_notify_middleware_no_acquire(self, acquire):
# Set the delay
self.config(notify_delay=.1,
group='service:agent')
# Initialize the middlware
placeholder_app = None
middleware = dnsutils.LimitNotifyMiddleware(placeholder_app)
# Prepare a NOTIFY
zone_name = 'example.com.'
notify = dns.message.make_query(zone_name, dns.rdatatype.SOA)
notify.flags = 0
notify.set_opcode(dns.opcode.NOTIFY)
notify.flags |= dns.flags.AA
# Make a response object to match the middleware's return
response = dns.message.make_response(notify)
# Provide an authoritative answer
response.flags |= dns.flags.AA
# Send the NOTIFY through the middleware
# Lock can't be acquired, a NOTIFY is already being worked on
# so just return what would have come back for a successful NOTIFY
# This needs to be a one item tuple for the serialization middleware
self.assertEqual(middleware.process_request(notify), (response,))
|
