summaryrefslogtreecommitdiff
path: root/examples
diff options
context:
space:
mode:
authorBob Halley <halley@dnspython.org>2022-03-15 08:37:20 -0700
committerBob Halley <halley@dnspython.org>2022-03-15 08:37:20 -0700
commitb1d2332687adbecc0acbb4e623124f783f859d9e (patch)
tree5318d5ecc0dd35e0a6922380cd60f9d9caa9ad34 /examples
parent08f8bde64e8679d5e4f0b129292461de152ba32b (diff)
downloaddnspython-b1d2332687adbecc0acbb4e623124f783f859d9e.tar.gz
black autoformatting
Diffstat (limited to 'examples')
-rw-r--r--examples/async_dns.py21
-rwxr-xr-xexamples/ddns.py10
-rwxr-xr-xexamples/doh-json.py48
-rwxr-xr-xexamples/doh.py11
-rwxr-xr-xexamples/e164.py1
-rwxr-xr-xexamples/ecs.py8
-rwxr-xr-xexamples/edns.py56
-rw-r--r--examples/edns_resolver.py33
-rwxr-xr-xexamples/mx.py4
-rwxr-xr-xexamples/name.py16
-rw-r--r--examples/query_specific.py30
-rw-r--r--examples/receive_notify.py11
-rwxr-xr-xexamples/reverse.py9
-rwxr-xr-xexamples/reverse_name.py1
-rwxr-xr-xexamples/xfr.py6
-rwxr-xr-xexamples/zonediff.py217
16 files changed, 293 insertions, 189 deletions
diff --git a/examples/async_dns.py b/examples/async_dns.py
index c42defc..f7e3fe5 100644
--- a/examples/async_dns.py
+++ b/examples/async_dns.py
@@ -1,4 +1,3 @@
-
import sys
import trio
@@ -7,24 +6,26 @@ import dns.message
import dns.asyncquery
import dns.asyncresolver
+
async def main():
if len(sys.argv) > 1:
host = sys.argv[0]
else:
- host = 'www.dnspython.org'
- q = dns.message.make_query(host, 'A')
- r = await dns.asyncquery.udp(q, '8.8.8.8')
+ host = "www.dnspython.org"
+ q = dns.message.make_query(host, "A")
+ r = await dns.asyncquery.udp(q, "8.8.8.8")
print(r)
- q = dns.message.make_query(host, 'A')
- r = await dns.asyncquery.tcp(q, '8.8.8.8')
+ q = dns.message.make_query(host, "A")
+ r = await dns.asyncquery.tcp(q, "8.8.8.8")
print(r)
- q = dns.message.make_query(host, 'A')
- r = await dns.asyncquery.tls(q, '8.8.8.8')
+ q = dns.message.make_query(host, "A")
+ r = await dns.asyncquery.tls(q, "8.8.8.8")
print(r)
- a = await dns.asyncresolver.resolve(host, 'A')
+ a = await dns.asyncresolver.resolve(host, "A")
print(a.response)
zn = await dns.asyncresolver.zone_for_name(host)
print(zn)
-if __name__ == '__main__':
+
+if __name__ == "__main__":
trio.run(main)
diff --git a/examples/ddns.py b/examples/ddns.py
index c584f42..154ab3d 100755
--- a/examples/ddns.py
+++ b/examples/ddns.py
@@ -35,17 +35,15 @@ import dns.tsigkeyring
# Replace the keyname and secret with appropriate values for your
# configuration.
#
-keyring = dns.tsigkeyring.from_text({
- 'keyname.' : 'NjHwPsMKjdN++dOfE5iAiQ=='
- })
+keyring = dns.tsigkeyring.from_text({"keyname.": "NjHwPsMKjdN++dOfE5iAiQ=="})
#
# Replace "example." with your domain, and "host" with your hostname.
#
-update = dns.update.Update('example.', keyring=keyring)
-update.replace('host', 300, 'A', sys.argv[1])
+update = dns.update.Update("example.", keyring=keyring)
+update.replace("host", 300, "A", sys.argv[1])
#
# Replace "10.0.0.1" with the IP address of your master server.
#
-response = dns.query.tcp(update, '10.0.0.1', timeout=10)
+response = dns.query.tcp(update, "10.0.0.1", timeout=10)
diff --git a/examples/doh-json.py b/examples/doh-json.py
index 8cfe1b0..e9fa087 100755
--- a/examples/doh-json.py
+++ b/examples/doh-json.py
@@ -24,26 +24,27 @@ import dns.rdatatype
# "simple" below means "simple python data types", i.e. things made of
# combinations of dictionaries, lists, strings, and numbers.
+
def make_rr(simple, rdata):
csimple = copy.copy(simple)
- csimple['data'] = rdata.to_text()
+ csimple["data"] = rdata.to_text()
return csimple
+
def flatten_rrset(rrs):
simple = {
- 'name': str(rrs.name),
- 'type': rrs.rdtype,
+ "name": str(rrs.name),
+ "type": rrs.rdtype,
}
if len(rrs) > 0:
- simple['TTL'] = rrs.ttl
+ simple["TTL"] = rrs.ttl
return [make_rr(simple, rdata) for rdata in rrs]
else:
return [simple]
+
def to_doh_simple(message):
- simple = {
- 'Status': message.rcode()
- }
+ simple = {"Status": message.rcode()}
for f in dns.flags.Flag:
if f != dns.flags.Flag.AA and f != dns.flags.Flag.QR:
# DoH JSON doesn't need AA and omits it. DoH JSON is only
@@ -57,6 +58,7 @@ def to_doh_simple(message):
# we don't encode the ecs_client_subnet field
return simple
+
def from_doh_simple(simple, add_qr=False):
message = dns.message.QueryMessage()
flags = 0
@@ -66,27 +68,35 @@ def from_doh_simple(simple, add_qr=False):
if add_qr: # QR is implied
flags |= dns.flags.QR
message.flags = flags
- message.set_rcode(simple.get('Status', 0))
+ message.set_rcode(simple.get("Status", 0))
for i, sn in enumerate(dns.message.MessageSection):
rr_list = simple.get(sn.name.title(), [])
for rr in rr_list:
- rdtype = dns.rdatatype.RdataType(rr['type'])
- rrs = message.find_rrset(i, dns.name.from_text(rr['name']),
- dns.rdataclass.IN, rdtype,
- create=True)
- if 'data' in rr:
- rrs.add(dns.rdata.from_text(dns.rdataclass.IN, rdtype,
- rr['data']), rr.get('TTL', 0))
+ rdtype = dns.rdatatype.RdataType(rr["type"])
+ rrs = message.find_rrset(
+ i,
+ dns.name.from_text(rr["name"]),
+ dns.rdataclass.IN,
+ rdtype,
+ create=True,
+ )
+ if "data" in rr:
+ rrs.add(
+ dns.rdata.from_text(dns.rdataclass.IN, rdtype, rr["data"]),
+ rr.get("TTL", 0),
+ )
# we don't decode the ecs_client_subnet field
return message
-a = dns.resolver.resolve('www.dnspython.org', 'a')
+a = dns.resolver.resolve("www.dnspython.org", "a")
p = to_doh_simple(a.response)
print(json.dumps(p, indent=4))
-response = requests.get('https://dns.google/resolve?', verify=True,
- params={'name': 'www.dnspython.org',
- 'type': 1})
+response = requests.get(
+ "https://dns.google/resolve?",
+ verify=True,
+ params={"name": "www.dnspython.org", "type": 1},
+)
p = json.loads(response.text)
m = from_doh_simple(p, True)
print(m)
diff --git a/examples/doh.py b/examples/doh.py
index e789bf1..17787ed 100755
--- a/examples/doh.py
+++ b/examples/doh.py
@@ -13,8 +13,8 @@ import dns.rdatatype
def main():
- where = '1.1.1.1'
- qname = 'example.com.'
+ where = "1.1.1.1"
+ qname = "example.com."
# one method is to use context manager, session will automatically close
with requests.sessions.Session() as session:
q = dns.message.make_query(qname, dns.rdatatype.A)
@@ -24,8 +24,8 @@ def main():
# ... do more lookups
- where = 'https://dns.google/dns-query'
- qname = 'example.net.'
+ where = "https://dns.google/dns-query"
+ qname = "example.net."
# second method, close session manually
session = requests.sessions.Session()
q = dns.message.make_query(qname, dns.rdatatype.A)
@@ -38,5 +38,6 @@ def main():
# close the session when you're done
session.close()
-if __name__ == '__main__':
+
+if __name__ == "__main__":
main()
diff --git a/examples/e164.py b/examples/e164.py
index 6d9e872..8b677bf 100755
--- a/examples/e164.py
+++ b/examples/e164.py
@@ -1,6 +1,7 @@
#!/usr/bin/env python3
import dns.e164
+
n = dns.e164.from_e164("+1 555 1212")
print(n)
print(dns.e164.to_e164(n))
diff --git a/examples/ecs.py b/examples/ecs.py
index f7b31d8..d5a84f2 100755
--- a/examples/ecs.py
+++ b/examples/ecs.py
@@ -1,15 +1,13 @@
-
import dns.edns
import dns.message
import dns.query
# This example demonstrates how to use the EDNS client subnet option
-ADDRESS = '0.0.0.0' # replace this with the address you want to check
+ADDRESS = "0.0.0.0" # replace this with the address you want to check
PREFIX = 0 # replace this with a prefix length (typically 24 for IPv4)
ecs = dns.edns.ECSOption(ADDRESS, PREFIX)
-q = dns.message.make_query('www.google.com', 'A', use_edns=0, options=[ecs])
-r = dns.query.udp(q, '8.8.8.8')
+q = dns.message.make_query("www.google.com", "A", use_edns=0, options=[ecs])
+r = dns.query.udp(q, "8.8.8.8")
print(r)
-
diff --git a/examples/edns.py b/examples/edns.py
index a130f85..0566bfb 100755
--- a/examples/edns.py
+++ b/examples/edns.py
@@ -5,10 +5,10 @@ import dns.message
import dns.query
import dns.resolver
-n = '.'
+n = "."
t = dns.rdatatype.SOA
-l = '199.7.83.42' # Address of l.root-servers.net
-i = '149.20.1.73' # Address of ns1.isc.org, for COOKIEs
+l = "199.7.83.42" # Address of l.root-servers.net
+i = "149.20.1.73" # Address of ns1.isc.org, for COOKIEs
q_list = []
@@ -16,7 +16,7 @@ q_list = []
q_list.append((l, dns.message.make_query(n, t)))
# The same query, but with EDNS0 turned on with no options
-q_list.append((l,dns.message.make_query(n, t, use_edns=0)))
+q_list.append((l, dns.message.make_query(n, t, use_edns=0)))
# Use use_edns() to specify EDNS0 options, such as buffer size
this_q = dns.message.make_query(n, t)
@@ -25,28 +25,46 @@ q_list.append((l, this_q))
# With an NSID option
# use_edns=0 is not needed if options are specified)
-q_list.append((l, dns.message.make_query(n, t,\
- options=[dns.edns.GenericOption(dns.edns.OptionType.NSID, b'')])))
+q_list.append(
+ (
+ l,
+ dns.message.make_query(
+ n, t, options=[dns.edns.GenericOption(dns.edns.OptionType.NSID, b"")]
+ ),
+ )
+)
# With an NSID option, but with use_edns() to specify the options
this_q = dns.message.make_query(n, t)
-this_q.use_edns(0, options=[dns.edns.GenericOption(dns.edns.OptionType.NSID, b'')])
+this_q.use_edns(0, options=[dns.edns.GenericOption(dns.edns.OptionType.NSID, b"")])
q_list.append((l, this_q))
# With a COOKIE
-q_list.append((i, dns.message.make_query(n, t,\
- options=[dns.edns.GenericOption(dns.edns.OptionType.COOKIE, b'0xfe11ac99bebe3322')])))
+q_list.append(
+ (
+ i,
+ dns.message.make_query(
+ n,
+ t,
+ options=[
+ dns.edns.GenericOption(
+ dns.edns.OptionType.COOKIE, b"0xfe11ac99bebe3322"
+ )
+ ],
+ ),
+ )
+)
# With an ECS option using dns.edns.ECSOption to form the option
-q_list.append((l, dns.message.make_query(n, t,\
- options=[dns.edns.ECSOption('192.168.0.0', 20)])))
+q_list.append(
+ (l, dns.message.make_query(n, t, options=[dns.edns.ECSOption("192.168.0.0", 20)]))
+)
for (addr, q) in q_list:
- r = dns.query.udp(q, addr)
- if not r.options:
- print('No EDNS options returned')
- else:
- for o in r.options:
- print(o.otype.value, o.data)
- print()
-
+ r = dns.query.udp(q, addr)
+ if not r.options:
+ print("No EDNS options returned")
+ else:
+ for o in r.options:
+ print(o.otype.value, o.data)
+ print()
diff --git a/examples/edns_resolver.py b/examples/edns_resolver.py
index fe5cc0f..6edf4a9 100644
--- a/examples/edns_resolver.py
+++ b/examples/edns_resolver.py
@@ -5,10 +5,10 @@ import dns.message
import dns.query
import dns.resolver
-n = '.'
+n = "."
t = dns.rdatatype.SOA
-l = 'google.com' # Address of l.root-servers.net, '199.7.83.42'
-i = 'ns1.isc.org' # Address of ns1.isc.org, for COOKIEs, '149.20.1.73'
+l = "google.com" # Address of l.root-servers.net, '199.7.83.42'
+i = "ns1.isc.org" # Address of ns1.isc.org, for COOKIEs, '149.20.1.73'
o_list = []
@@ -22,21 +22,32 @@ o_list.append((l, dict(options=[])))
o_list.append((l, dict(payload=2000)))
# With an NSID option, but with use_edns() to specify the options
-edns_kwargs = dict(edns=0, options=[
- dns.edns.GenericOption(dns.edns.OptionType.NSID, b'')])
+edns_kwargs = dict(
+ edns=0, options=[dns.edns.GenericOption(dns.edns.OptionType.NSID, b"")]
+)
o_list.append((l, edns_kwargs))
# With a COOKIE
-o_list.append((i, dict(options=[
- dns.edns.GenericOption(dns.edns.OptionType.COOKIE, b'0xfe11ac99bebe3322')])))
+o_list.append(
+ (
+ i,
+ dict(
+ options=[
+ dns.edns.GenericOption(
+ dns.edns.OptionType.COOKIE, b"0xfe11ac99bebe3322"
+ )
+ ]
+ ),
+ )
+)
# With an ECS option using cloudflare dns address
-o_list.append((l, dict(options=[dns.edns.ECSOption('1.1.1.1', 24)])))
+o_list.append((l, dict(options=[dns.edns.ECSOption("1.1.1.1", 24)])))
# With an ECS option using the current machine address
import urllib.request
-external_ip = urllib.request.urlopen('https://ident.me').read().decode('utf8')
+external_ip = urllib.request.urlopen("https://ident.me").read().decode("utf8")
o_list.append((l, dict(options=[dns.edns.ECSOption(external_ip, 24)])))
@@ -45,5 +56,5 @@ aresolver = dns.resolver.Resolver()
for (addr, edns_kwargs) in o_list:
if edns_kwargs:
aresolver.use_edns(**edns_kwargs)
- aresolver.nameservers = ['8.8.8.8']
- print(list(aresolver.resolve(addr, 'A')))
+ aresolver.nameservers = ["8.8.8.8"]
+ print(list(aresolver.resolve(addr, "A")))
diff --git a/examples/mx.py b/examples/mx.py
index 2c310ea..5e9075b 100755
--- a/examples/mx.py
+++ b/examples/mx.py
@@ -2,6 +2,6 @@
import dns.resolver
-answers = dns.resolver.resolve('nominum.com', 'MX')
+answers = dns.resolver.resolve("nominum.com", "MX")
for rdata in answers:
- print('Host', rdata.exchange, 'has preference', rdata.preference)
+ print("Host", rdata.exchange, "has preference", rdata.preference)
diff --git a/examples/name.py b/examples/name.py
index 614fdbc..ff687e8 100755
--- a/examples/name.py
+++ b/examples/name.py
@@ -2,12 +2,12 @@
import dns.name
-n = dns.name.from_text('www.dnspython.org')
-o = dns.name.from_text('dnspython.org')
-print(n.is_subdomain(o)) # True
-print(n.is_superdomain(o)) # False
-print(n > o) # True
-rel = n.relativize(o) # rel is the relative name www
+n = dns.name.from_text("www.dnspython.org")
+o = dns.name.from_text("dnspython.org")
+print(n.is_subdomain(o)) # True
+print(n.is_superdomain(o)) # False
+print(n > o) # True
+rel = n.relativize(o) # rel is the relative name www
n2 = rel + o
-print(n2 == n) # True
-print(n.labels) # ['www', 'dnspython', 'org', '']
+print(n2 == n) # True
+print(n.labels) # ['www', 'dnspython', 'org', '']
diff --git a/examples/query_specific.py b/examples/query_specific.py
index c82207c..73dc351 100644
--- a/examples/query_specific.py
+++ b/examples/query_specific.py
@@ -9,30 +9,30 @@ import dns.query
# This way is just like nslookup/dig:
-qname = dns.name.from_text('amazon.com')
+qname = dns.name.from_text("amazon.com")
q = dns.message.make_query(qname, dns.rdatatype.NS)
-print('The query is:')
+print("The query is:")
print(q)
-print('')
-r = dns.query.udp(q, '8.8.8.8')
-print('The response is:')
+print("")
+r = dns.query.udp(q, "8.8.8.8")
+print("The response is:")
print(r)
-print('')
-print('The nameservers are:')
+print("")
+print("The nameservers are:")
ns_rrset = r.find_rrset(r.answer, qname, dns.rdataclass.IN, dns.rdatatype.NS)
for rr in ns_rrset:
print(rr.target)
-print('')
-print('')
+print("")
+print("")
# A higher-level way
import dns.resolver
resolver = dns.resolver.Resolver(configure=False)
-resolver.nameservers = ['8.8.8.8']
-answer = resolver.resolve('amazon.com', 'NS')
-print('The nameservers are:')
+resolver.nameservers = ["8.8.8.8"]
+answer = resolver.resolve("amazon.com", "NS")
+print("The nameservers are:")
for rr in answer:
print(rr.target)
@@ -42,7 +42,7 @@ for rr in answer:
# This sends a query with RD=0 for the root SOA RRset to the IP address
# for l.root-servers.net.
-q = dns.message.make_query('.', dns.rdatatype.SOA, flags=0)
-r = dns.query.udp(q, '199.7.83.42')
-print('\nThe flags in the response are {}'.format(dns.flags.to_text(r.flags)))
+q = dns.message.make_query(".", dns.rdatatype.SOA, flags=0)
+r = dns.query.udp(q, "199.7.83.42")
+print("\nThe flags in the response are {}".format(dns.flags.to_text(r.flags)))
print('The SOA in the response is "{}"'.format((r.answer)[0][0]))
diff --git a/examples/receive_notify.py b/examples/receive_notify.py
index c41b336..97d01f3 100644
--- a/examples/receive_notify.py
+++ b/examples/receive_notify.py
@@ -13,7 +13,7 @@ import dns.name
from typing import cast
-address = '127.0.0.1'
+address = "127.0.0.1"
port = 53535
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
@@ -23,16 +23,17 @@ while True:
notify = dns.message.from_wire(wire)
try:
- soa = notify.find_rrset(notify.answer, notify.question[0].name,
- dns.rdataclass.IN, dns.rdatatype.SOA)
+ soa = notify.find_rrset(
+ notify.answer, notify.question[0].name, dns.rdataclass.IN, dns.rdatatype.SOA
+ )
# Do something with the SOA RR here
- print('The serial number for', soa.name, 'is', soa[0].serial)
+ print("The serial number for", soa.name, "is", soa[0].serial)
except KeyError:
# No SOA RR in the answer section.
pass
- response = dns.message.make_response(notify) # type: dns.message.Message
+ response = dns.message.make_response(notify) # type: dns.message.Message
response.flags |= dns.flags.AA
wire = response.to_wire(cast(dns.name.Name, response))
s.sendto(wire, address)
diff --git a/examples/reverse.py b/examples/reverse.py
index 83b99b7..5829c68 100755
--- a/examples/reverse.py
+++ b/examples/reverse.py
@@ -20,14 +20,13 @@ import dns.zone
import dns.ipv4
import os.path
import sys
-from typing import Dict, List # pylint: disable=unused-import
+from typing import Dict, List # pylint: disable=unused-import
-reverse_map = {} # type: Dict[str, List[str]]
+reverse_map = {} # type: Dict[str, List[str]]
for filename in sys.argv[1:]:
- zone = dns.zone.from_file(filename, os.path.basename(filename),
- relativize=False)
- for (name, ttl, rdata) in zone.iterate_rdatas('A'):
+ zone = dns.zone.from_file(filename, os.path.basename(filename), relativize=False)
+ for (name, ttl, rdata) in zone.iterate_rdatas("A"):
print(type(rdata))
try:
reverse_map[rdata.address].append(name.to_text())
diff --git a/examples/reverse_name.py b/examples/reverse_name.py
index 02b2e51..ec7fe1c 100755
--- a/examples/reverse_name.py
+++ b/examples/reverse_name.py
@@ -1,6 +1,7 @@
#!/usr/bin/env python3
import dns.reversename
+
n = dns.reversename.from_address("127.0.0.1")
print(n)
print(dns.reversename.to_address(n))
diff --git a/examples/xfr.py b/examples/xfr.py
index a20cae3..1c8175e 100755
--- a/examples/xfr.py
+++ b/examples/xfr.py
@@ -4,9 +4,9 @@ import dns.query
import dns.resolver
import dns.zone
-soa_answer = dns.resolver.resolve('dnspython.org', 'SOA')
-master_answer = dns.resolver.resolve(soa_answer[0].mname, 'A')
+soa_answer = dns.resolver.resolve("dnspython.org", "SOA")
+master_answer = dns.resolver.resolve(soa_answer[0].mname, "A")
-z = dns.zone.from_xfr(dns.query.xfr(master_answer[0].address, 'dnspython.org'))
+z = dns.zone.from_xfr(dns.query.xfr(master_answer[0].address, "dnspython.org"))
for n in sorted(z.nodes.keys()):
print(z[n].to_text(n))
diff --git a/examples/zonediff.py b/examples/zonediff.py
index 164bf2b..2957f87 100755
--- a/examples/zonediff.py
+++ b/examples/zonediff.py
@@ -21,9 +21,9 @@
# OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
"""See diff_zones.__doc__ for more information"""
-from typing import cast, Union, Any # pylint: disable=unused-import
+from typing import cast, Union, Any # pylint: disable=unused-import
-__all__ = ['diff_zones', 'format_changes_plain', 'format_changes_html']
+__all__ = ["diff_zones", "format_changes_plain", "format_changes_html"]
try:
import dns.zone
@@ -32,11 +32,12 @@ except ImportError:
raise SystemExit("Please install dnspython")
-def diff_zones(zone1, # type: dns.zone.Zone
- zone2, # type: dns.zone.Zone
- ignore_ttl=False,
- ignore_soa=False
- ): # type: (...) -> list
+def diff_zones(
+ zone1, # type: dns.zone.Zone
+ zone2, # type: dns.zone.Zone
+ ignore_ttl=False,
+ ignore_soa=False,
+): # type: (...) -> list
"""diff_zones(zone1, zone2, ignore_ttl=False, ignore_soa=False) -> changes
Compares two dns.zone.Zone objects and returns a list of all changes
in the format (name, oldnode, newnode).
@@ -67,11 +68,13 @@ def diff_zones(zone1, # type: dns.zone.Zone
changes.append((str(name), n3, n4))
return changes
-def _nodes_differ(n1, # type: dns.node.Node
- n2, # type: dns.node.Node
- ignore_ttl, # type: bool
- ignore_soa # type: bool
- ): # type: (...) -> bool
+
+def _nodes_differ(
+ n1, # type: dns.node.Node
+ n2, # type: dns.node.Node
+ ignore_ttl, # type: bool
+ ignore_soa, # type: bool
+): # type: (...) -> bool
if ignore_soa or not ignore_ttl:
# Compare datasets directly
for r in n1.rdatasets:
@@ -91,11 +94,13 @@ def _nodes_differ(n1, # type: dns.node.Node
else:
return n1 != n2
-def format_changes_plain(oldf, # type: str
- newf, # type: str
- changes, # type: list
- ignore_ttl=False
- ): # type: (...) -> str
+
+def format_changes_plain(
+ oldf, # type: str
+ newf, # type: str
+ changes, # type: list
+ ignore_ttl=False,
+): # type: (...) -> str
"""format_changes(oldfile, newfile, changes, ignore_ttl=False) -> str
Given 2 filenames and a list of changes from diff_zones, produce diff-like
output. If ignore_ttl is True, TTL-only changes are not displayed"""
@@ -105,35 +110,37 @@ def format_changes_plain(oldf, # type: str
ret += "@ %s\n" % name
if not old:
for r in new.rdatasets:
- ret += "+ %s\n" % str(r).replace('\n', '\n+ ')
+ ret += "+ %s\n" % str(r).replace("\n", "\n+ ")
elif not new:
for r in old.rdatasets:
- ret += "- %s\n" % str(r).replace('\n', '\n+ ')
+ ret += "- %s\n" % str(r).replace("\n", "\n+ ")
else:
for r in old.rdatasets:
if r not in new.rdatasets or (
- r.ttl != new.find_rdataset(r.rdclass, r.rdtype).ttl and
- not ignore_ttl
+ r.ttl != new.find_rdataset(r.rdclass, r.rdtype).ttl
+ and not ignore_ttl
):
- ret += "- %s\n" % str(r).replace('\n', '\n+ ')
+ ret += "- %s\n" % str(r).replace("\n", "\n+ ")
for r in new.rdatasets:
if r not in old.rdatasets or (
- r.ttl != old.find_rdataset(r.rdclass, r.rdtype).ttl and
- not ignore_ttl
+ r.ttl != old.find_rdataset(r.rdclass, r.rdtype).ttl
+ and not ignore_ttl
):
- ret += "+ %s\n" % str(r).replace('\n', '\n+ ')
+ ret += "+ %s\n" % str(r).replace("\n", "\n+ ")
return ret
-def format_changes_html(oldf, # type: str
- newf, # type: str
- changes, # type: list
- ignore_ttl=False
- ): # type: (...) -> str
+
+def format_changes_html(
+ oldf, # type: str
+ newf, # type: str
+ changes, # type: list
+ ignore_ttl=False,
+): # type: (...) -> str
"""format_changes(oldfile, newfile, changes, ignore_ttl=False) -> str
Given 2 filenames and a list of changes from diff_zones, produce nice html
output. If ignore_ttl is True, TTL-only changes are not displayed"""
- ret = '''<table class="zonediff">
+ ret = """<table class="zonediff">
<thead>
<tr>
<th>&nbsp;</th>
@@ -141,7 +148,10 @@ def format_changes_html(oldf, # type: str
<th class="new">%s</th>
</tr>
</thead>
- <tbody>\n''' % (oldf, newf)
+ <tbody>\n""" % (
+ oldf,
+ newf,
+ )
for name, old, new in changes:
ret += ' <tr class="rdata">\n <td class="rdname">%s</td>\n' % name
@@ -150,36 +160,36 @@ def format_changes_html(oldf, # type: str
ret += (
' <td class="old">&nbsp;</td>\n'
' <td class="new">%s</td>\n'
- ) % str(r).replace('\n', '<br />')
+ ) % str(r).replace("\n", "<br />")
elif not new:
for r in old.rdatasets:
ret += (
' <td class="old">%s</td>\n'
' <td class="new">&nbsp;</td>\n'
- ) % str(r).replace('\n', '<br />')
+ ) % str(r).replace("\n", "<br />")
else:
ret += ' <td class="old">'
for r in old.rdatasets:
if r not in new.rdatasets or (
- r.ttl != new.find_rdataset(r.rdclass, r.rdtype).ttl and
- not ignore_ttl
+ r.ttl != new.find_rdataset(r.rdclass, r.rdtype).ttl
+ and not ignore_ttl
):
- ret += str(r).replace('\n', '<br />')
- ret += '</td>\n'
+ ret += str(r).replace("\n", "<br />")
+ ret += "</td>\n"
ret += ' <td class="new">'
for r in new.rdatasets:
if r not in old.rdatasets or (
- r.ttl != old.find_rdataset(r.rdclass, r.rdtype).ttl and
- not ignore_ttl
+ r.ttl != old.find_rdataset(r.rdclass, r.rdtype).ttl
+ and not ignore_ttl
):
- ret += str(r).replace('\n', '<br />')
- ret += '</td>\n'
- ret += ' </tr>\n'
- return ret + ' </tbody>\n</table>'
+ ret += str(r).replace("\n", "<br />")
+ ret += "</td>\n"
+ ret += " </tr>\n"
+ return ret + " </tbody>\n</table>"
# Make this module usable as a script too.
-def main(): # type: () -> None
+def main(): # type: () -> None
import argparse
import subprocess
import sys
@@ -191,24 +201,66 @@ def main(): # type: () -> None
The differences shown will be logical differences, not textual differences.
"""
p = argparse.ArgumentParser(usage=usage)
- p.add_argument('-s', '--ignore-soa', action="store_true", default=False, dest="ignore_soa",
- help="Ignore SOA-only changes to records")
- p.add_argument('-t', '--ignore-ttl', action="store_true", default=False, dest="ignore_ttl",
- help="Ignore TTL-only changes to Rdata")
- p.add_argument('-T', '--traceback', action="store_true", default=False, dest="tracebacks",
- help="Show python tracebacks when errors occur")
- p.add_argument('-H', '--html', action="store_true", default=False, dest="html",
- help="Print HTML output")
- p.add_argument('-g', '--git', action="store_true", default=False, dest="use_git",
- help="Use git revisions instead of real files")
- p.add_argument('-b', '--bzr', action="store_true", default=False, dest="use_bzr",
- help="Use bzr revisions instead of real files")
- p.add_argument('-r', '--rcs', action="store_true", default=False, dest="use_rcs",
- help="Use rcs revisions instead of real files")
+ p.add_argument(
+ "-s",
+ "--ignore-soa",
+ action="store_true",
+ default=False,
+ dest="ignore_soa",
+ help="Ignore SOA-only changes to records",
+ )
+ p.add_argument(
+ "-t",
+ "--ignore-ttl",
+ action="store_true",
+ default=False,
+ dest="ignore_ttl",
+ help="Ignore TTL-only changes to Rdata",
+ )
+ p.add_argument(
+ "-T",
+ "--traceback",
+ action="store_true",
+ default=False,
+ dest="tracebacks",
+ help="Show python tracebacks when errors occur",
+ )
+ p.add_argument(
+ "-H",
+ "--html",
+ action="store_true",
+ default=False,
+ dest="html",
+ help="Print HTML output",
+ )
+ p.add_argument(
+ "-g",
+ "--git",
+ action="store_true",
+ default=False,
+ dest="use_git",
+ help="Use git revisions instead of real files",
+ )
+ p.add_argument(
+ "-b",
+ "--bzr",
+ action="store_true",
+ default=False,
+ dest="use_bzr",
+ help="Use bzr revisions instead of real files",
+ )
+ p.add_argument(
+ "-r",
+ "--rcs",
+ action="store_true",
+ default=False,
+ dest="use_rcs",
+ help="Use rcs revisions instead of real files",
+ )
opts, args = p.parse_args()
opts.use_vc = opts.use_git or opts.use_bzr or opts.use_rcs
- def _open(what, err): # type: (Union[list,str], str) -> Any
+ def _open(what, err): # type: (Union[list,str], str) -> Any
if isinstance(what, list):
# Must be a list, open subprocess
try:
@@ -224,7 +276,7 @@ The differences shown will be logical differences, not textual differences.
else:
# Open as normal file
try:
- return open(what, 'rb')
+ return open(what, "rb")
except IOError:
sys.stderr.write(err + "\n")
if opts.tracebacks:
@@ -254,23 +306,35 @@ The differences shown will be logical differences, not textual differences.
old, new = None, None
oldz, newz = None, None
if opts.use_bzr:
- old = _open(["bzr", "cat", "-r" + oldr, filename],
- "Unable to retrieve revision {} of {}".format(oldr, filename))
+ old = _open(
+ ["bzr", "cat", "-r" + oldr, filename],
+ "Unable to retrieve revision {} of {}".format(oldr, filename),
+ )
if newr is not None:
- new = _open(["bzr", "cat", "-r" + newr, filename],
- "Unable to retrieve revision {} of {}".format(newr, filename))
+ new = _open(
+ ["bzr", "cat", "-r" + newr, filename],
+ "Unable to retrieve revision {} of {}".format(newr, filename),
+ )
elif opts.use_git:
- old = _open(["git", "show", oldn],
- "Unable to retrieve revision {} of {}".format(oldr, filename))
+ old = _open(
+ ["git", "show", oldn],
+ "Unable to retrieve revision {} of {}".format(oldr, filename),
+ )
if newr is not None:
- new = _open(["git", "show", newn],
- "Unable to retrieve revision {} of {}".format(newr, filename))
+ new = _open(
+ ["git", "show", newn],
+ "Unable to retrieve revision {} of {}".format(newr, filename),
+ )
elif opts.use_rcs:
- old = _open(["co", "-q", "-p", "-r" + oldr, filename],
- "Unable to retrieve revision {} of {}".format(oldr, filename))
+ old = _open(
+ ["co", "-q", "-p", "-r" + oldr, filename],
+ "Unable to retrieve revision {} of {}".format(oldr, filename),
+ )
if newr is not None:
- new = _open(["co", "-q", "-p", "-r" + newr, filename],
- "Unable to retrieve revision {} of {}".format(newr, filename))
+ new = _open(
+ ["co", "-q", "-p", "-r" + newr, filename],
+ "Unable to retrieve revision {} of {}".format(newr, filename),
+ )
if not opts.use_vc:
old = _open(oldn, "Unable to open %s" % oldn)
if not opts.use_vc or newr is None:
@@ -281,13 +345,13 @@ The differences shown will be logical differences, not textual differences.
# Parse the zones
try:
- oldz = dns.zone.from_file(old, origin='.', check_origin=False)
+ oldz = dns.zone.from_file(old, origin=".", check_origin=False)
except dns.exception.DNSException:
sys.stderr.write("Incorrect zonefile: %s\n" % old)
if opts.tracebacks:
traceback.print_exc()
try:
- newz = dns.zone.from_file(new, origin='.', check_origin=False)
+ newz = dns.zone.from_file(new, origin=".", check_origin=False)
except dns.exception.DNSException:
sys.stderr.write("Incorrect zonefile: %s\n" % new)
if opts.tracebacks:
@@ -306,5 +370,6 @@ The differences shown will be logical differences, not textual differences.
print(format_changes_plain(oldn, newn, changes, opts.ignore_ttl))
sys.exit(1)
-if __name__ == '__main__':
+
+if __name__ == "__main__":
main()