/usr/lib/python3/dist-packages/twisted/names/test
Edit: /usr/lib/python3/dist-packages/twisted/names/test/test_dns.py (163995B)
# test-case-name: twisted.names.test.test_dns
# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.
"""
Tests for twisted.names.dns.
"""
import struct
from io import BytesIO
from typing import cast
from zope.interface.verify import verifyClass
from twisted.internet import address, task
from twisted.internet.error import CannotListenError, ConnectionDone
from twisted.names import dns
from twisted.python.failure import Failure
from twisted.python.util import FancyEqMixin, FancyStrMixin
from twisted.test import proto_helpers
from twisted.test.testutils import ComparisonTestsMixin
from twisted.trial import unittest
RECORD_TYPES = [
dns.Record_NS,
dns.Record_MD,
dns.Record_MF,
dns.Record_CNAME,
dns.Record_MB,
dns.Record_MG,
dns.Record_MR,
dns.Record_PTR,
dns.Record_DNAME,
dns.Record_A,
dns.Record_SOA,
dns.Record_NULL,
dns.Record_WKS,
dns.Record_SRV,
dns.Record_AFSDB,
dns.Record_RP,
dns.Record_HINFO,
dns.Record_MINFO,
dns.Record_MX,
dns.Record_TXT,
dns.Record_AAAA,
dns.Record_A6,
dns.Record_NAPTR,
dns.Record_SSHFP,
dns.Record_TSIG,
dns.UnknownRecord,
]
class DomainStringTests(unittest.SynchronousTestCase):
def test_bytes(self):
"""
L{dns.domainString} returns L{bytes} unchanged.
"""
self.assertEqual(
b"twistedmatrix.com",
dns.domainString(b"twistedmatrix.com"),
)
def test_native(self):
"""
L{dns.domainString} converts a native string to L{bytes}
if necessary.
"""
self.assertEqual(b"example.com", dns.domainString("example.com"))
def test_text(self):
"""
L{dns.domainString} always converts a unicode string to L{bytes}.
"""
self.assertEqual(b"foo.example", dns.domainString("foo.example"))
def test_idna(self):
"""
L{dns.domainString} encodes Unicode using IDNA.
"""
self.assertEqual(b"xn--fwg.test", dns.domainString("\u203D.test"))
def test_nonsense(self):
"""
L{dns.domainString} encodes Unicode using IDNA.
"""
self.assertRaises(TypeError, dns.domainString, 9000)
self.assertRaises(TypeError, dns.domainString, dns.Name("bar.example"))
class Ord2ByteTests(unittest.TestCase):
"""
Tests for L{dns._ord2bytes}.
"""
def test_ord2byte(self):
"""
L{dns._ord2byte} accepts an integer and returns a byte string of length
one with an ordinal value equal to the given integer.
"""
self.assertEqual(b"\x10", dns._ord2bytes(0x10))
class Str2TimeTests(unittest.TestCase):
"""
Tests for L{dns.str2name}.
"""
def test_nonString(self):
"""
When passed a non-string object, L{dns.str2name} returns it unmodified.
"""
time = object()
self.assertIs(time, dns.str2time(time))
def test_seconds(self):
"""
Passed a string giving a number of seconds, L{dns.str2time} returns the
number of seconds represented. For example, C{"10S"} represents C{10}
seconds.
"""
self.assertEqual(10, dns.str2time("10S"))
def test_minutes(self):
"""
Like C{test_seconds}, but for the C{"M"} suffix which multiplies the
time value by C{60} (the number of seconds in a minute!).
"""
self.assertEqual(2 * 60, dns.str2time("2M"))
def test_hours(self):
"""
Like C{test_seconds}, but for the C{"H"} suffix which multiplies the
time value by C{3600}, the number of seconds in an hour.
"""
self.assertEqual(3 * 3600, dns.str2time("3H"))
def test_days(self):
"""
Like L{test_seconds}, but for the C{"D"} suffix which multiplies the
time value by C{86400}, the number of seconds in a day.
"""
self.assertEqual(4 * 86400, dns.str2time("4D"))
def test_weeks(self):
"""
Like L{test_seconds}, but for the C{"W"} suffix which multiplies the
time value by C{604800}, the number of seconds in a week.
"""
self.assertEqual(5 * 604800, dns.str2time("5W"))
def test_years(self):
"""
Like L{test_seconds}, but for the C{"Y"} suffix which multiplies the
time value by C{31536000}, the number of seconds in a year.
"""
self.assertEqual(6 * 31536000, dns.str2time("6Y"))
def test_invalidPrefix(self):
"""
If a non-integer prefix is given, L{dns.str2time} raises L{ValueError}.
"""
self.assertRaises(ValueError, dns.str2time, "fooS")
class NameTests(unittest.TestCase):
"""
Tests for L{Name}, the representation of a single domain name with support
for encoding into and decoding from DNS message format.
"""
def test_nonStringName(self):
"""
When constructed with a name which is neither C{bytes} nor C{str},
L{Name} raises L{TypeError}.
"""
self.assertRaises(TypeError, dns.Name, 123)
self.assertRaises(TypeError, dns.Name, object())
self.assertRaises(TypeError, dns.Name, [])
def test_unicodeName(self):
"""
L{dns.Name} automatically encodes unicode domain name using C{idna}
encoding.
"""
name = dns.Name("\u00e9chec.example.org")
self.assertIsInstance(name.name, bytes)
self.assertEqual(b"xn--chec-9oa.example.org", name.name)
def test_decode(self):
"""
L{Name.decode} populates the L{Name} instance with name information read
from the file-like object passed to it.
"""
n = dns.Name()
n.decode(BytesIO(b"\x07example\x03com\x00"))
self.assertEqual(n.name, b"example.com")
def test_encode(self):
"""
L{Name.encode} encodes its name information and writes it to the
file-like object passed to it.
"""
name = dns.Name(b"foo.example.com")
stream = BytesIO()
name.encode(stream)
self.assertEqual(stream.getvalue(), b"\x03foo\x07example\x03com\x00")
def test_encodeWithCompression(self):
"""
If a compression dictionary is passed to it, L{Name.encode} uses offset
information from it to encode its name with references to existing
labels in the stream instead of including another copy of them in the
output. It also updates the compression dictionary with the location of
the name it writes to the stream.
"""
name = dns.Name(b"foo.example.com")
compression = {b"example.com": 0x17}
# Some bytes already encoded into the stream for this message
previous = b"some prefix to change .tell()"
stream = BytesIO()
stream.write(previous)
# The position at which the encoded form of this new name will appear in
# the stream.
expected = len(previous) + dns.Message.headerSize
name.encode(stream, compression)
self.assertEqual(b"\x03foo\xc0\x17", stream.getvalue()[len(previous) :])
self.assertEqual(
{b"example.com": 0x17, b"foo.example.com": expected}, compression
)
def test_unknown(self):
"""
A resource record of unknown type and class is parsed into an
L{UnknownRecord} instance with its data preserved, and an
L{UnknownRecord} instance is serialized to a string equal to the one it
was parsed from.
"""
wire = (
b"\x01\x00" # Message ID
b"\x00" # answer bit, opCode nibble, auth bit, trunc bit, recursive
# bit
b"\x00" # recursion bit, empty bit, authenticData bit,
# checkingDisabled bit, response code nibble
b"\x00\x01" # number of queries
b"\x00\x01" # number of answers
b"\x00\x00" # number of authorities
b"\x00\x01" # number of additionals
# query
b"\x03foo\x03bar\x00" # foo.bar
b"\xde\xad" # type=0xdead
b"\xbe\xef" # cls=0xbeef
# 1st answer
b"\xc0\x0c" # foo.bar - compressed
b"\xde\xad" # type=0xdead
b"\xbe\xef" # cls=0xbeef
b"\x00\x00\x01\x01" # ttl=257
b"\x00\x08somedata" # some payload data
# 1st additional
b"\x03baz\x03ban\x00" # baz.ban
b"\x00\x01" # type=A
b"\x00\x01" # cls=IN
b"\x00\x00\x01\x01" # ttl=257
b"\x00\x04" # len=4
b"\x01\x02\x03\x04" # 1.2.3.4
)
msg = dns.Message()
msg.fromStr(wire)
self.assertEqual(
msg.queries,
[
dns.Query(b"foo.bar", type=0xDEAD, cls=0xBEEF),
],
)
self.assertEqual(
msg.answers,
[
dns.RRHeader(
b"foo.bar",
type=0xDEAD,
cls=0xBEEF,
ttl=257,
payload=dns.UnknownRecord(b"somedata", ttl=257),
),
],
)
self.assertEqual(
msg.additional,
[
dns.RRHeader(
b"baz.ban",
type=dns.A,
cls=dns.IN,
ttl=257,
payload=dns.Record_A("1.2.3.4", ttl=257),
),
],
)
enc = msg.toStr()
self.assertEqual(enc, wire)
def test_decodeWithCompression(self):
"""
If the leading byte of an encoded label (in bytes read from a stream
passed to L{Name.decode}) has its two high bits set, the next byte is
treated as a pointer to another label in the stream and that label is
included in the name being decoded.
"""
# Slightly modified version of the example from RFC 1035, section 4.1.4.
stream = BytesIO(
b"x" * 20 + b"\x01f\x03isi\x04arpa\x00"
b"\x03foo\xc0\x14"
b"\x03bar\xc0\x20"
)
stream.seek(20)
name = dns.Name()
name.decode(stream)
# Verify we found the first name in the stream and that the stream
# position is left at the first byte after the decoded name.
self.assertEqual(b"f.isi.arpa", name.name)
self.assertEqual(32, stream.tell())
# Get the second name from the stream and make the same assertions.
name.decode(stream)
self.assertEqual(name.name, b"foo.f.isi.arpa")
self.assertEqual(38, stream.tell())
# Get the third and final name
name.decode(stream)
self.assertEqual(name.name, b"bar.foo.f.isi.arpa")
self.assertEqual(44, stream.tell())
def test_rejectCompressionLoop(self):
"""
L{Name.decode} raises L{ValueError} if the stream passed to it includes
a compression pointer which forms a loop, causing the name to be
undecodable.
"""
name = dns.Name()
stream = BytesIO(b"\xc0\x00")
self.assertRaises(ValueError, name.decode, stream)
def test_rejectTooManyCompressionPointers(self):
"""
L{Name.decode} raises L{dns.DNSDecodeError} when it would have to
follow more than L{Name.maxCompressionPointers} compression
pointers to finish decoding a name.
"""
# Four distinct pointers chained end-to-end, terminated by a zero
# label byte. With maxCompressionPointers of three the fourth
# dereference must trip the safety limit.
payload = b"\xc0\x02\xc0\x04\xc0\x06\xc0\x08\x00"
name = dns.Name()
name.maxCompressionPointers = 3
self.assertRaises(
dns.DNSDecodeError, name.decode, BytesIO(payload)
)
def test_decodeRecoversAfterDNSDecodeError(self):
"""
After L{Name.decode} raises L{dns.DNSDecodeError}, subsequent
L{Name.decode} calls continue to work. No residual
compression-pointer counter leaks across calls, so a legitimate
name decoded right after a hostile one still succeeds.
"""
# First, force a DNSDecodeError by decoding a payload that
# exceeds the configured limit.
hostile = dns.Name()
hostile.maxCompressionPointers = 3
self.assertRaises(
dns.DNSDecodeError,
hostile.decode,
BytesIO(b"\xc0\x02\xc0\x04\xc0\x06\xc0\x08\x00"),
)
# Then prove the process has not been poisoned: a legitimate
# name still decodes normally, both with a fresh instance and
# with the instance that just errored.
stream = BytesIO()
dns.Name(b"example.org").encode(stream)
fresh = dns.Name()
stream.seek(0)
fresh.decode(stream)
self.assertEqual(fresh.name, b"example.org")
stream.seek(0)
hostile.decode(stream)
self.assertEqual(hostile.name, b"example.org")
def test_equality(self):
"""
L{Name} instances are equal as long as they have the same value for
L{Name.name}, regardless of the case.
"""
name1 = dns.Name(b"foo.bar")
name2 = dns.Name(b"foo.bar")
self.assertEqual(name1, name2)
name3 = dns.Name(b"fOO.bar")
self.assertEqual(name1, name3)
def test_inequality(self):
"""
L{Name} instances are not equal as long as they have different
L{Name.name} attributes.
"""
name1 = dns.Name(b"foo.bar")
name2 = dns.Name(b"bar.foo")
self.assertNotEqual(name1, name2)
class RoundtripDNSTests(unittest.TestCase):
"""
Encoding and then decoding various objects.
"""
names = [b"example.org", b"go-away.fish.tv", b"23strikesback.net"]
def test_name(self):
for n in self.names:
# encode the name
f = BytesIO()
dns.Name(n).encode(f)
# decode the name
f.seek(0, 0)
result = dns.Name()
result.decode(f)
self.assertEqual(result.name, n)
def test_query(self):
"""
L{dns.Query.encode} returns a byte string representing the fields of the
query which can be decoded into a new L{dns.Query} instance using
L{dns.Query.decode}.
"""
for n in self.names:
for dnstype in range(1, 17):
for dnscls in range(1, 5):
# encode the query
f = BytesIO()
dns.Query(n, dnstype, dnscls).encode(f)
# decode the result
f.seek(0, 0)
result = dns.Query()
result.decode(f)
self.assertEqual(result.name.name, n)
self.assertEqual(result.type, dnstype)
self.assertEqual(result.cls, dnscls)
def test_resourceRecordHeader(self):
"""
L{dns.RRHeader.encode} encodes the record header's information and
writes it to the file-like object passed to it and
L{dns.RRHeader.decode} reads from a file-like object to re-construct a
L{dns.RRHeader} instance.
"""
# encode the RR
f = BytesIO()
dns.RRHeader(b"test.org", 3, 4, 17).encode(f)
# decode the result
f.seek(0, 0)
result = dns.RRHeader()
result.decode(f)
self.assertEqual(result.name, dns.Name(b"test.org"))
self.assertEqual(result.type, 3)
self.assertEqual(result.cls, 4)
self.assertEqual(result.ttl, 17)
def test_resourceRecordHeaderTypeMismatch(self):
"""
L{RRHeader()} raises L{ValueError} when the given type and the type
of the payload don't match.
"""
with self.assertRaisesRegex(ValueError, r"Payload type \(AAAA\) .* type \(A\)"):
dns.RRHeader(type=dns.A, payload=dns.Record_AAAA())
def test_resources(self):
"""
L{dns.SimpleRecord.encode} encodes the record's name information and
writes it to the file-like object passed to it and
L{dns.SimpleRecord.decode} reads from a file-like object to re-construct
a L{dns.SimpleRecord} instance.
"""
names = (
b"this.are.test.name",
b"will.compress.will.this.will.name.will.hopefully",
b"test.CASE.preSErVatIOn.YeAH",
b"a.s.h.o.r.t.c.a.s.e.t.o.t.e.s.t",
b"singleton",
)
for s in names:
f = BytesIO()
dns.SimpleRecord(s).encode(f)
f.seek(0, 0)
result = dns.SimpleRecord()
result.decode(f)
self.assertEqual(result.name, dns.Name(s))
def test_hashable(self):
"""
Instances of all record types are hashable.
"""
for k in RECORD_TYPES:
k1, k2 = k(), k()
hk1 = hash(k1)
hk2 = hash(k2)
self.assertEqual(hk1, hk2, f"{hk1} != {hk2} (for {k})")
def test_Charstr(self):
"""
Test L{dns.Charstr} encode and decode.
"""
for n in self.names:
# encode the name
f = BytesIO()
dns.Charstr(n).encode(f)
# decode the name
f.seek(0, 0)
result = dns.Charstr()
result.decode(f)
self.assertEqual(result.string, n)
def _recordRoundtripTest(self, record):
"""
Assert that encoding C{record} and then decoding the resulting bytes
creates a record which compares equal to C{record}.
@type record: L{dns.IEncodable}
@param record: A record instance to encode
"""
stream = BytesIO()
record.encode(stream)
length = stream.tell()
stream.seek(0, 0)
replica = record.__class__()
replica.decode(stream, length)
self.assertEqual(record, replica)
def assertEncodedFormat(self, expectedEncoding, record):
"""
Assert that encoding C{record} produces the expected bytes.
@type record: L{dns.IEncodable}
@param record: A record instance to encode
@type expectedEncoding: L{bytes}
@param expectedEncoding: The value which C{record.encode()}
should produce.
"""
stream = BytesIO()
record.encode(stream)
self.assertEqual(stream.getvalue(), expectedEncoding)
def test_SOA(self):
"""
The byte stream written by L{dns.Record_SOA.encode} can be used by
L{dns.Record_SOA.decode} to reconstruct the state of the original
L{dns.Record_SOA} instance.
"""
self._recordRoundtripTest(
dns.Record_SOA(
mname=b"foo",
rname=b"bar",
serial=12,
refresh=34,
retry=56,
expire=78,
minimum=90,
)
)
def test_A(self):
"""
The byte stream written by L{dns.Record_A.encode} can be used by
L{dns.Record_A.decode} to reconstruct the state of the original
L{dns.Record_A} instance.
"""
self._recordRoundtripTest(dns.Record_A("1.2.3.4"))
def test_NULL(self):
"""
The byte stream written by L{dns.Record_NULL.encode} can be used by
L{dns.Record_NULL.decode} to reconstruct the state of the original
L{dns.Record_NULL} instance.
"""
self._recordRoundtripTest(dns.Record_NULL(b"foo bar"))
def test_WKS(self):
"""
The byte stream written by L{dns.Record_WKS.encode} can be used by
L{dns.Record_WKS.decode} to reconstruct the state of the original
L{dns.Record_WKS} instance.
"""
self._recordRoundtripTest(dns.Record_WKS("1.2.3.4", 3, b"xyz"))
def test_AAAA(self):
"""
The byte stream written by L{dns.Record_AAAA.encode} can be used by
L{dns.Record_AAAA.decode} to reconstruct the state of the original
L{dns.Record_AAAA} instance.
"""
self._recordRoundtripTest(dns.Record_AAAA("::1"))
def test_A6(self):
"""
The byte stream written by L{dns.Record_A6.encode} can be used by
L{dns.Record_A6.decode} to reconstruct the state of the original
L{dns.Record_A6} instance.
"""
self._recordRoundtripTest(dns.Record_A6(8, "::1:2", b"foo"))
def test_SRV(self):
"""
The byte stream written by L{dns.Record_SRV.encode} can be used by
L{dns.Record_SRV.decode} to reconstruct the state of the original
L{dns.Record_SRV} instance.
"""
self._recordRoundtripTest(
dns.Record_SRV(priority=1, weight=2, port=3, target=b"example.com")
)
def test_SSHFP(self):
"""
The byte stream written by L{dns.Record_SSHFP.encode} can be used by
L{dns.Record_SSHFP.decode} to reconstruct the state of the original
L{dns.Record_SSHFP} instance.
"""
fp = (
b"\xda\x39\xa3\xee\x5e\x6b\x4b\x0d"
+ b"\x32\x55\xbf\xef\x95\x60\x18\x90\xaf\xd8\x07\x09"
)
rr = dns.Record_SSHFP(
algorithm=dns.Record_SSHFP.ALGORITHM_DSS,
fingerprintType=dns.Record_SSHFP.FINGERPRINT_TYPE_SHA1,
fingerprint=fp,
)
self._recordRoundtripTest(rr)
self.assertEncodedFormat(b"\x02\x01" + fp, rr)
def test_NAPTR(self):
"""
Test L{dns.Record_NAPTR} encode and decode.
"""
naptrs = [
(100, 10, b"u", b"sip+E2U", b"!^.*$!sip:information@domain.tld!", b""),
(100, 50, b"s", b"http+I2L+I2C+I2R", b"", b"_http._tcp.gatech.edu"),
]
for (order, preference, flags, service, regexp, replacement) in naptrs:
rin = dns.Record_NAPTR(
order, preference, flags, service, regexp, replacement
)
e = BytesIO()
rin.encode(e)
e.seek(0, 0)
rout = dns.Record_NAPTR()
rout.decode(e)
self.assertEqual(rin.order, rout.order)
self.assertEqual(rin.preference, rout.preference)
self.assertEqual(rin.flags, rout.flags)
self.assertEqual(rin.service, rout.service)
self.assertEqual(rin.regexp, rout.regexp)
self.assertEqual(rin.replacement.name, rout.replacement.name)
self.assertEqual(rin.ttl, rout.ttl)
def test_AFSDB(self):
"""
The byte stream written by L{dns.Record_AFSDB.encode} can be used by
L{dns.Record_AFSDB.decode} to reconstruct the state of the original
L{dns.Record_AFSDB} instance.
"""
self._recordRoundtripTest(dns.Record_AFSDB(subtype=3, hostname=b"example.com"))
def test_RP(self):
"""
The byte stream written by L{dns.Record_RP.encode} can be used by
L{dns.Record_RP.decode} to reconstruct the state of the original
L{dns.Record_RP} instance.
"""
self._recordRoundtripTest(
dns.Record_RP(mbox=b"alice.example.com", txt=b"example.com")
)
def test_HINFO(self):
"""
The byte stream written by L{dns.Record_HINFO.encode} can be used by
L{dns.Record_HINFO.decode} to reconstruct the state of the original
L{dns.Record_HINFO} instance.
"""
self._recordRoundtripTest(dns.Record_HINFO(cpu=b"fast", os=b"great"))
def test_MINFO(self):
"""
The byte stream written by L{dns.Record_MINFO.encode} can be used by
L{dns.Record_MINFO.decode} to reconstruct the state of the original
L{dns.Record_MINFO} instance.
"""
self._recordRoundtripTest(dns.Record_MINFO(rmailbx=b"foo", emailbx=b"bar"))
def test_MX(self):
"""
The byte stream written by L{dns.Record_MX.encode} can be used by
L{dns.Record_MX.decode} to reconstruct the state of the original
L{dns.Record_MX} instance.
"""
self._recordRoundtripTest(dns.Record_MX(preference=1, name=b"example.com"))
def test_TSIG(self):
"""
The byte stream written by L{dns.Record_TSIG.encode} can be used by
L{dns.Record_TSIG.decode} to reconstruct the state of the original
L{dns.Record_TSIG} instance.
"""
mac = b"\x00\x01\x02\x03\x10\x11\x12\x13" b"\x20\x21\x22\x23\x30\x31\x32\x33"
rr = dns.Record_TSIG(
algorithm="hmac-md5.sig-alg.reg.int",
timeSigned=1515548975,
originalID=42,
fudge=5,
MAC=mac,
)
self._recordRoundtripTest(rr)
rdata = (
b"\x08hmac-md5\x07sig-alg\x03reg\x03int\x00"
b"\x00\x00\x5a\x55\x71\x2f\x00\x05\x00\x10"
+ mac
+ b"\x00\x2A\x00\x00\x00\x00"
)
self.assertEncodedFormat(rdata, rr)
rr = dns.Record_TSIG(
algorithm="hmac-sha256",
timeSigned=4511798055, # More than 32 bits
originalID=65535,
error=dns.EBADTIME,
otherData=b"\x80\x00\x00\x00\x00\x08",
MAC=mac,
)
self._recordRoundtripTest(rr)
rdata = (
b"\x0Bhmac-sha256\x00"
b"\x00\x01\x0c\xec\x93\x27\x00\x05\x00\x10"
+ mac
+ b"\xff\xff\x00\x12\x00\x06"
b"\x80\x00\x00\x00\x00\x08"
)
self.assertEncodedFormat(rdata, rr)
def test_TXT(self):
"""
The byte stream written by L{dns.Record_TXT.encode} can be used by
L{dns.Record_TXT.decode} to reconstruct the state of the original
L{dns.Record_TXT} instance.
"""
self._recordRoundtripTest(dns.Record_TXT(b"foo", b"bar"))
MESSAGE_AUTHENTIC_DATA_BYTES = (
b"\x00\x00" # ID
b"\x00" #
b"\x20" # RA, Z, AD=1, CD, RCODE
b"\x00\x00" # Query count
b"\x00\x00" # Answer count
b"\x00\x00" # Authority count
b"\x00\x00" # Additional count
)
MESSAGE_CHECKING_DISABLED_BYTES = (
b"\x00\x00" # ID
b"\x00" #
b"\x10" # RA, Z, AD, CD=1, RCODE
b"\x00\x00" # Query count
b"\x00\x00" # Answer count
b"\x00\x00" # Authority count
b"\x00\x00" # Additional count
)
class MessageTests(unittest.SynchronousTestCase):
"""
Tests for L{twisted.names.dns.Message}.
"""
def test_authenticDataDefault(self):
"""
L{dns.Message.authenticData} has default value 0.
"""
self.assertEqual(dns.Message().authenticData, 0)
def test_rejectCompressionPointerFlood(self):
"""
L{Message.decode} installs a shared compression-pointer counter and
raises L{dns.DNSDecodeError} when the aggregate number of pointer
dereferences across every record in the message exceeds
L{dns.Message.maxCompressionPointers}.
"""
chainLength = 100
numRecords = 8000
header = struct.pack(
"!H2B4H", 0x1234, 0x80, 0x00, 0, numRecords, 0, 0
)
# Long compression chain inside the RDATA of an unknown
# record so that subsequent records can aim pointers at it.
owner = b"\x04rrrr\x00"
chainBase = len(header) + len(owner) + 10
chain = bytearray()
for i in range(chainLength):
chain += struct.pack("!H", 0xC000 | (chainBase + 2 * (i + 1)))
chain += b"\x04test\x00"
firstRecord = (
owner
+ struct.pack("!HHIH", 999, 1, 0, len(chain))
+ bytes(chain)
)
followupRecord = (
struct.pack("!H", 0xC000 | chainBase)
+ struct.pack("!HHIH", 1, 1, 0, 4)
+ b"\x00\x00\x00\x00"
)
payload = header + firstRecord + followupRecord * (numRecords - 1)
message = dns.Message()
self.assertRaises(dns.DNSDecodeError, message.decode, BytesIO(payload))
def test_authenticDataOverride(self):
"""
L{dns.Message.__init__} accepts a C{authenticData} argument which
is assigned to L{dns.Message.authenticData}.
"""
self.assertEqual(dns.Message(authenticData=1).authenticData, 1)
def test_authenticDataEncode(self):
"""
L{dns.Message.toStr} encodes L{dns.Message.authenticData} into
byte4 of the byte string.
"""
self.assertEqual(
dns.Message(authenticData=1).toStr(), MESSAGE_AUTHENTIC_DATA_BYTES
)
def test_authenticDataDecode(self):
"""
L{dns.Message.fromStr} decodes byte4 and assigns bit3 to
L{dns.Message.authenticData}.
"""
m = dns.Message()
m.fromStr(MESSAGE_AUTHENTIC_DATA_BYTES)
self.assertEqual(m.authenticData, 1)
def test_checkingDisabledDefault(self):
"""
L{dns.Message.checkingDisabled} has default value 0.
"""
self.assertEqual(dns.Message().checkingDisabled, 0)
def test_checkingDisabledOverride(self):
"""
L{dns.Message.__init__} accepts a C{checkingDisabled} argument which
is assigned to L{dns.Message.checkingDisabled}.
"""
self.assertEqual(dns.Message(checkingDisabled=1).checkingDisabled, 1)
def test_checkingDisabledEncode(self):
"""
L{dns.Message.toStr} encodes L{dns.Message.checkingDisabled} into
byte4 of the byte string.
"""
self.assertEqual(
dns.Message(checkingDisabled=1).toStr(), MESSAGE_CHECKING_DISABLED_BYTES
)
def test_checkingDisabledDecode(self):
"""
L{dns.Message.fromStr} decodes byte4 and assigns bit4 to
L{dns.Message.checkingDisabled}.
"""
m = dns.Message()
m.fromStr(MESSAGE_CHECKING_DISABLED_BYTES)
self.assertEqual(m.checkingDisabled, 1)
def test_reprDefaults(self):
"""
L{dns.Message.__repr__} omits field values and sections which are
identical to their defaults. The id field value is always shown.
"""
self.assertEqual("
", repr(dns.Message()))
def test_reprFlagsIfSet(self):
"""
L{dns.Message.__repr__} displays flags if they are L{True}.
"""
m = dns.Message(
answer=True,
auth=True,
trunc=True,
recDes=True,
recAv=True,
authenticData=True,
checkingDisabled=True,
)
self.assertEqual(
"",
repr(m),
)
def test_reprNonDefautFields(self):
"""
L{dns.Message.__repr__} displays field values if they differ from their
defaults.
"""
m = dns.Message(id=10, opCode=20, rCode=30, maxSize=40)
self.assertEqual(
"",
repr(m),
)
def test_reprNonDefaultSections(self):
"""
L{dns.Message.__repr__} displays sections which differ from their
defaults.
"""
m = dns.Message()
m.queries = [1, 2, 3]
m.answers = [4, 5, 6]
m.authority = [7, 8, 9]
m.additional = [10, 11, 12]
self.assertEqual(
"",
repr(m),
)
def test_emptyMessage(self):
"""
Test that a message which has been truncated causes an EOFError to
be raised when it is parsed.
"""
msg = dns.Message()
self.assertRaises(EOFError, msg.fromStr, b"")
def test_emptyQuery(self):
"""
Test that bytes representing an empty query message can be decoded
as such.
"""
msg = dns.Message()
msg.fromStr(
b"\x01\x00" # Message ID
b"\x00" # answer bit, opCode nibble, auth bit, trunc bit, recursive bit
b"\x00" # recursion bit, empty bit, authenticData bit,
# checkingDisabled bit, response code nibble
b"\x00\x00" # number of queries
b"\x00\x00" # number of answers
b"\x00\x00" # number of authorities
b"\x00\x00" # number of additionals
)
self.assertEqual(msg.id, 256)
self.assertFalse(msg.answer, "Message was not supposed to be an answer.")
self.assertEqual(msg.opCode, dns.OP_QUERY)
self.assertFalse(msg.auth, "Message was not supposed to be authoritative.")
self.assertFalse(msg.trunc, "Message was not supposed to be truncated.")
self.assertEqual(msg.queries, [])
self.assertEqual(msg.answers, [])
self.assertEqual(msg.authority, [])
self.assertEqual(msg.additional, [])
def test_NULL(self):
"""
A I{NULL} record with an arbitrary payload can be encoded and decoded as
part of a L{dns.Message}.
"""
bytes = b"".join([dns._ord2bytes(i) for i in range(256)])
rec = dns.Record_NULL(bytes)
rr = dns.RRHeader(b"testname", dns.NULL, payload=rec)
msg1 = dns.Message()
msg1.answers.append(rr)
s = BytesIO()
msg1.encode(s)
s.seek(0, 0)
msg2 = dns.Message()
msg2.decode(s)
self.assertIsInstance(msg2.answers[0].payload, dns.Record_NULL)
self.assertEqual(msg2.answers[0].payload.payload, bytes)
def test_lookupRecordTypeDefault(self):
"""
L{Message.lookupRecordType} returns C{dns.UnknownRecord} if it is
called with an integer which doesn't correspond to any known record
type.
"""
# 65280 is the first value in the range reserved for private
# use, so it shouldn't ever conflict with an officially
# allocated value.
self.assertIs(dns.Message().lookupRecordType(65280), dns.UnknownRecord)
def test_nonAuthoritativeMessage(self):
"""
The L{RRHeader} instances created by L{Message} from a non-authoritative
message are marked as not authoritative.
"""
buf = BytesIO()
answer = dns.RRHeader(payload=dns.Record_A("1.2.3.4", ttl=0))
answer.encode(buf)
message = dns.Message()
message.fromStr(
b"\x01\x00" # Message ID
# answer bit, opCode nibble, auth bit, trunc bit, recursive bit
b"\x00"
# recursion bit, empty bit, authenticData bit,
# checkingDisabled bit, response code nibble
b"\x00"
b"\x00\x00" # number of queries
b"\x00\x01" # number of answers
b"\x00\x00" # number of authorities
b"\x00\x00" + buf.getvalue() # number of additionals
)
self.assertEqual(message.answers, [answer])
self.assertFalse(message.answers[0].auth)
def test_authoritativeMessage(self):
"""
The L{RRHeader} instances created by L{Message} from an authoritative
message are marked as authoritative.
"""
buf = BytesIO()
answer = dns.RRHeader(payload=dns.Record_A("1.2.3.4", ttl=0))
answer.encode(buf)
message = dns.Message()
message.fromStr(
b"\x01\x00" # Message ID
# answer bit, opCode nibble, auth bit, trunc bit, recursive bit
b"\x04"
# recursion bit, empty bit, authenticData bit,
# checkingDisabled bit, response code nibble
b"\x00"
b"\x00\x00" # number of queries
b"\x00\x01" # number of answers
b"\x00\x00" # number of authorities
b"\x00\x00" + buf.getvalue() # number of additionals
)
answer.auth = True
self.assertEqual(message.answers, [answer])
self.assertTrue(message.answers[0].auth)
class MessageComparisonTests(ComparisonTestsMixin, unittest.SynchronousTestCase):
"""
Tests for the rich comparison of L{dns.Message} instances.
"""
def messageFactory(self, *args, **kwargs):
"""
Create a L{dns.Message}.
The L{dns.Message} constructor doesn't accept C{queries}, C{answers},
C{authority}, C{additional} arguments, so we extract them from the
kwargs supplied to this factory function and assign them to the message.
@param args: Positional arguments.
@param kwargs: Keyword arguments.
@return: A L{dns.Message} instance.
"""
queries = kwargs.pop("queries", [])
answers = kwargs.pop("answers", [])
authority = kwargs.pop("authority", [])
additional = kwargs.pop("additional", [])
m = dns.Message(**kwargs)
if queries:
m.queries = queries
if answers:
m.answers = answers
if authority:
m.authority = authority
if additional:
m.additional = additional
return m
def test_id(self):
"""
Two L{dns.Message} instances compare equal if they have the same id
value.
"""
self.assertNormalEqualityImplementation(
self.messageFactory(id=10),
self.messageFactory(id=10),
self.messageFactory(id=20),
)
def test_answer(self):
"""
Two L{dns.Message} instances compare equal if they have the same answer
flag.
"""
self.assertNormalEqualityImplementation(
self.messageFactory(answer=1),
self.messageFactory(answer=1),
self.messageFactory(answer=0),
)
def test_opCode(self):
"""
Two L{dns.Message} instances compare equal if they have the same opCode
value.
"""
self.assertNormalEqualityImplementation(
self.messageFactory(opCode=10),
self.messageFactory(opCode=10),
self.messageFactory(opCode=20),
)
def test_recDes(self):
"""
Two L{dns.Message} instances compare equal if they have the same recDes
flag.
"""
self.assertNormalEqualityImplementation(
self.messageFactory(recDes=1),
self.messageFactory(recDes=1),
self.messageFactory(recDes=0),
)
def test_recAv(self):
"""
Two L{dns.Message} instances compare equal if they have the same recAv
flag.
"""
self.assertNormalEqualityImplementation(
self.messageFactory(recAv=1),
self.messageFactory(recAv=1),
self.messageFactory(recAv=0),
)
def test_auth(self):
"""
Two L{dns.Message} instances compare equal if they have the same auth
flag.
"""
self.assertNormalEqualityImplementation(
self.messageFactory(auth=1),
self.messageFactory(auth=1),
self.messageFactory(auth=0),
)
def test_rCode(self):
"""
Two L{dns.Message} instances compare equal if they have the same rCode
value.
"""
self.assertNormalEqualityImplementation(
self.messageFactory(rCode=10),
self.messageFactory(rCode=10),
self.messageFactory(rCode=20),
)
def test_trunc(self):
"""
Two L{dns.Message} instances compare equal if they have the same trunc
flag.
"""
self.assertNormalEqualityImplementation(
self.messageFactory(trunc=1),
self.messageFactory(trunc=1),
self.messageFactory(trunc=0),
)
def test_maxSize(self):
"""
Two L{dns.Message} instances compare equal if they have the same
maxSize value.
"""
self.assertNormalEqualityImplementation(
self.messageFactory(maxSize=10),
self.messageFactory(maxSize=10),
self.messageFactory(maxSize=20),
)
def test_authenticData(self):
"""
Two L{dns.Message} instances compare equal if they have the same
authenticData flag.
"""
self.assertNormalEqualityImplementation(
self.messageFactory(authenticData=1),
self.messageFactory(authenticData=1),
self.messageFactory(authenticData=0),
)
def test_checkingDisabled(self):
"""
Two L{dns.Message} instances compare equal if they have the same
checkingDisabled flag.
"""
self.assertNormalEqualityImplementation(
self.messageFactory(checkingDisabled=1),
self.messageFactory(checkingDisabled=1),
self.messageFactory(checkingDisabled=0),
)
def test_queries(self):
"""
Two L{dns.Message} instances compare equal if they have the same
queries.
"""
self.assertNormalEqualityImplementation(
self.messageFactory(queries=[dns.Query(b"example.com")]),
self.messageFactory(queries=[dns.Query(b"example.com")]),
self.messageFactory(queries=[dns.Query(b"example.org")]),
)
def test_answers(self):
"""
Two L{dns.Message} instances compare equal if they have the same
answers.
"""
self.assertNormalEqualityImplementation(
self.messageFactory(
answers=[dns.RRHeader(b"example.com", payload=dns.Record_A("1.2.3.4"))]
),
self.messageFactory(
answers=[dns.RRHeader(b"example.com", payload=dns.Record_A("1.2.3.4"))]
),
self.messageFactory(
answers=[dns.RRHeader(b"example.org", payload=dns.Record_A("4.3.2.1"))]
),
)
def test_authority(self):
"""
Two L{dns.Message} instances compare equal if they have the same
authority records.
"""
self.assertNormalEqualityImplementation(
self.messageFactory(
authority=[
dns.RRHeader(b"example.com", type=dns.SOA, payload=dns.Record_SOA())
]
),
self.messageFactory(
authority=[
dns.RRHeader(b"example.com", type=dns.SOA, payload=dns.Record_SOA())
]
),
self.messageFactory(
authority=[
dns.RRHeader(b"example.org", type=dns.SOA, payload=dns.Record_SOA())
]
),
)
def test_additional(self):
"""
Two L{dns.Message} instances compare equal if they have the same
additional records.
"""
self.assertNormalEqualityImplementation(
self.messageFactory(
additional=[
dns.RRHeader(b"example.com", payload=dns.Record_A("1.2.3.4"))
]
),
self.messageFactory(
additional=[
dns.RRHeader(b"example.com", payload=dns.Record_A("1.2.3.4"))
]
),
self.messageFactory(
additional=[
dns.RRHeader(b"example.org", payload=dns.Record_A("1.2.3.4"))
]
),
)
class TestController:
"""
Pretend to be a DNS query processor for a DNSDatagramProtocol.
@ivar messages: the list of received messages.
@type messages: C{list} of (msg, protocol, address)
"""
def __init__(self):
"""
Initialize the controller: create a list of messages.
"""
self.messages = []
def messageReceived(self, msg, proto, addr=None):
"""
Save the message so that it can be checked during the tests.
"""
self.messages.append((msg, proto, addr))
class DatagramProtocolTests(unittest.TestCase):
"""
Test various aspects of L{dns.DNSDatagramProtocol}.
"""
def setUp(self):
"""
Create a L{dns.DNSDatagramProtocol} with a deterministic clock.
"""
self.clock = task.Clock()
self.controller = TestController()
self.proto = dns.DNSDatagramProtocol(self.controller)
transport = proto_helpers.FakeDatagramTransport()
self.proto.makeConnection(transport)
self.proto.callLater = self.clock.callLater
def test_truncatedPacket(self):
"""
Test that when a short datagram is received, datagramReceived does
not raise an exception while processing it.
"""
self.proto.datagramReceived(b"", address.IPv4Address("UDP", "127.0.0.1", 12345))
self.assertEqual(self.controller.messages, [])
def test_simpleQuery(self):
"""
Test content received after a query.
"""
d = self.proto.query(("127.0.0.1", 21345), [dns.Query(b"foo")])
self.assertEqual(len(self.proto.liveMessages.keys()), 1)
m = dns.Message()
m.id = next(iter(self.proto.liveMessages.keys()))
m.answers = [dns.RRHeader(payload=dns.Record_A(address="1.2.3.4"))]
def cb(result):
self.assertEqual(result.answers[0].payload.dottedQuad(), "1.2.3.4")
d.addCallback(cb)
self.proto.datagramReceived(m.toStr(), ("127.0.0.1", 21345))
return d
def test_queryTimeout(self):
"""
Test that query timeouts after some seconds.
"""
d = self.proto.query(("127.0.0.1", 21345), [dns.Query(b"foo")])
self.assertEqual(len(self.proto.liveMessages), 1)
self.clock.advance(10)
self.assertFailure(d, dns.DNSQueryTimeoutError)
self.assertEqual(len(self.proto.liveMessages), 0)
return d
def test_writeError(self):
"""
Exceptions raised by the transport's write method should be turned into
C{Failure}s passed to errbacks of the C{Deferred} returned by
L{DNSDatagramProtocol.query}.
"""
def writeError(message, addr):
raise RuntimeError("bar")
self.proto.transport.write = writeError
d = self.proto.query(("127.0.0.1", 21345), [dns.Query(b"foo")])
return self.assertFailure(d, RuntimeError)
def test_listenError(self):
"""
Exception L{CannotListenError} raised by C{listenUDP} should be turned
into a C{Failure} passed to errback of the C{Deferred} returned by
L{DNSDatagramProtocol.query}.
"""
def startListeningError():
raise CannotListenError(None, None, None)
self.proto.startListening = startListeningError
# Clean up transport so that the protocol calls startListening again
self.proto.transport = None
d = self.proto.query(("127.0.0.1", 21345), [dns.Query(b"foo")])
return self.assertFailure(d, CannotListenError)
def test_receiveMessageNotInLiveMessages(self):
"""
When receiving a message whose id is not in
L{DNSDatagramProtocol.liveMessages} or L{DNSDatagramProtocol.resends},
the message will be received by L{DNSDatagramProtocol.controller}.
"""
message = dns.Message()
message.id = 1
message.answers = [dns.RRHeader(payload=dns.Record_A(address="1.2.3.4"))]
self.proto.datagramReceived(message.toStr(), ("127.0.0.1", 21345))
self.assertEqual(self.controller.messages[-1][0].toStr(), message.toStr())
class TestTCPController(TestController):
"""
Pretend to be a DNS query processor for a DNSProtocol.
@ivar connections: A list of L{DNSProtocol} instances which have
notified this controller that they are connected and have not
yet notified it that their connection has been lost.
"""
def __init__(self):
TestController.__init__(self)
self.connections = []
def connectionMade(self, proto):
self.connections.append(proto)
def connectionLost(self, proto):
self.connections.remove(proto)
class DNSProtocolTests(unittest.TestCase):
"""
Test various aspects of L{dns.DNSProtocol}.
"""
def setUp(self):
"""
Create a L{dns.DNSProtocol} with a deterministic clock.
"""
self.clock = task.Clock()
self.controller = TestTCPController()
self.proto = dns.DNSProtocol(self.controller)
self.proto.makeConnection(proto_helpers.StringTransport())
self.proto.callLater = self.clock.callLater
def test_connectionTracking(self):
"""
L{dns.DNSProtocol} calls its controller's C{connectionMade}
method with itself when it is connected to a transport and its
controller's C{connectionLost} method when it is disconnected.
"""
self.assertEqual(self.controller.connections, [self.proto])
self.proto.connectionLost(Failure(ConnectionDone("Fake Connection Done")))
self.assertEqual(self.controller.connections, [])
def test_queryTimeout(self):
"""
Test that query timeouts after some seconds.
"""
d = self.proto.query([dns.Query(b"foo")])
self.assertEqual(len(self.proto.liveMessages), 1)
self.clock.advance(60)
self.assertFailure(d, dns.DNSQueryTimeoutError)
self.assertEqual(len(self.proto.liveMessages), 0)
return d
def test_simpleQuery(self):
"""
Test content received after a query.
"""
d = self.proto.query([dns.Query(b"foo")])
self.assertEqual(len(self.proto.liveMessages.keys()), 1)
m = dns.Message()
m.id = next(iter(self.proto.liveMessages.keys()))
m.answers = [dns.RRHeader(payload=dns.Record_A(address="1.2.3.4"))]
def cb(result):
self.assertEqual(result.answers[0].payload.dottedQuad(), "1.2.3.4")
d.addCallback(cb)
s = m.toStr()
s = struct.pack("!H", len(s)) + s
self.proto.dataReceived(s)
return d
def test_writeError(self):
"""
Exceptions raised by the transport's write method should be turned into
C{Failure}s passed to errbacks of the C{Deferred} returned by
L{DNSProtocol.query}.
"""
def writeError(message):
raise RuntimeError("bar")
self.proto.transport.write = writeError
d = self.proto.query([dns.Query(b"foo")])
return self.assertFailure(d, RuntimeError)
def test_receiveMessageNotInLiveMessages(self):
"""
When receiving a message whose id is not in L{DNSProtocol.liveMessages}
the message will be received by L{DNSProtocol.controller}.
"""
message = dns.Message()
message.id = 1
message.answers = [dns.RRHeader(payload=dns.Record_A(address="1.2.3.4"))]
string = message.toStr()
string = struct.pack("!H", len(string)) + string
self.proto.dataReceived(string)
self.assertEqual(self.controller.messages[-1][0].toStr(), message.toStr())
class ReprTests(unittest.TestCase):
"""
Tests for the C{__repr__} implementation of record classes.
"""
def test_ns(self):
"""
The repr of a L{dns.Record_NS} instance includes the name of the
nameserver and the TTL of the record.
"""
self.assertEqual(
repr(dns.Record_NS(b"example.com", 4321)), ""
)
def test_md(self):
"""
The repr of a L{dns.Record_MD} instance includes the name of the
mail destination and the TTL of the record.
"""
self.assertEqual(
repr(dns.Record_MD(b"example.com", 4321)), ""
)
def test_mf(self):
"""
The repr of a L{dns.Record_MF} instance includes the name of the
mail forwarder and the TTL of the record.
"""
self.assertEqual(
repr(dns.Record_MF(b"example.com", 4321)), ""
)
def test_cname(self):
"""
The repr of a L{dns.Record_CNAME} instance includes the name of the
mail forwarder and the TTL of the record.
"""
self.assertEqual(
repr(dns.Record_CNAME(b"example.com", 4321)),
"",
)
def test_mb(self):
"""
The repr of a L{dns.Record_MB} instance includes the name of the
mailbox and the TTL of the record.
"""
self.assertEqual(
repr(dns.Record_MB(b"example.com", 4321)), ""
)
def test_mg(self):
"""
The repr of a L{dns.Record_MG} instance includes the name of the
mail group member and the TTL of the record.
"""
self.assertEqual(
repr(dns.Record_MG(b"example.com", 4321)), ""
)
def test_mr(self):
"""
The repr of a L{dns.Record_MR} instance includes the name of the
mail rename domain and the TTL of the record.
"""
self.assertEqual(
repr(dns.Record_MR(b"example.com", 4321)), ""
)
def test_ptr(self):
"""
The repr of a L{dns.Record_PTR} instance includes the name of the
pointer and the TTL of the record.
"""
self.assertEqual(
repr(dns.Record_PTR(b"example.com", 4321)),
"",
)
def test_dname(self):
"""
The repr of a L{dns.Record_DNAME} instance includes the name of the
non-terminal DNS name redirection and the TTL of the record.
"""
self.assertEqual(
repr(dns.Record_DNAME(b"example.com", 4321)),
"",
)
def test_a(self):
"""
The repr of a L{dns.Record_A} instance includes the dotted-quad
string representation of the address it is for and the TTL of the
record.
"""
self.assertEqual(
repr(dns.Record_A("1.2.3.4", 567)), ""
)
def test_soa(self):
"""
The repr of a L{dns.Record_SOA} instance includes all of the
authority fields.
"""
self.assertEqual(
repr(
dns.Record_SOA(
mname=b"mName",
rname=b"rName",
serial=123,
refresh=456,
retry=789,
expire=10,
minimum=11,
ttl=12,
)
),
"",
)
def test_null(self):
"""
The repr of a L{dns.Record_NULL} instance includes the repr of its
payload and the TTL of the record.
"""
self.assertEqual(
repr(dns.Record_NULL(b"abcd", 123)), ""
)
def test_wks(self):
"""
The repr of a L{dns.Record_WKS} instance includes the dotted-quad
string representation of the address it is for, the IP protocol
number it is for, and the TTL of the record.
"""
self.assertEqual(
repr(dns.Record_WKS("2.3.4.5", 7, ttl=8)),
"",
)
def test_aaaa(self):
"""
The repr of a L{dns.Record_AAAA} instance includes the colon-separated
hex string representation of the address it is for and the TTL of the
record.
"""
self.assertEqual(
repr(dns.Record_AAAA("8765::1234", ttl=10)),
"",
)
def test_a6(self):
"""
The repr of a L{dns.Record_A6} instance includes the colon-separated
hex string representation of the address it is for and the TTL of the
record.
"""
self.assertEqual(
repr(dns.Record_A6(0, "1234::5678", b"foo.bar", ttl=10)),
"",
)
def test_srv(self):
"""
The repr of a L{dns.Record_SRV} instance includes the name and port of
the target and the priority, weight, and TTL of the record.
"""
self.assertEqual(
repr(dns.Record_SRV(1, 2, 3, b"example.org", 4)),
"",
)
def test_naptr(self):
"""
The repr of a L{dns.Record_NAPTR} instance includes the order,
preference, flags, service, regular expression, replacement, and TTL of
the record.
"""
record = dns.Record_NAPTR(5, 9, b"S", b"http", b"/foo/bar/i", b"baz", 3)
self.assertEqual(
repr(record),
"",
)
def test_afsdb(self):
"""
The repr of a L{dns.Record_AFSDB} instance includes the subtype,
hostname, and TTL of the record.
"""
self.assertEqual(
repr(dns.Record_AFSDB(3, b"example.org", 5)),
"",
)
def test_rp(self):
"""
The repr of a L{dns.Record_RP} instance includes the mbox, txt, and TTL
fields of the record.
"""
self.assertEqual(
repr(dns.Record_RP(b"alice.example.com", b"admin.example.com", 3)),
"