wire = dns.wiredata.maybe_wrap(wire)
cls = get_rdata_class(rdclass, rdtype)
return cls.from_wire(rdclass, rdtype, wire, current, rdlen, origin)
+
+
+class RdatatypeExists(dns.exception.DNSException):
+ """DNS rdatatype already exists."""
+ supp_kwargs = set(['rdclass', 'rdtype'])
+ fmt = "The rdata type with class {rdclass} and rdtype {rdtype} " + \
+ "already exists."
+
+
+def register_type(implementation, rdtype, rdtype_text, is_singleton=False,
+ rdclass=dns.rdataclass.IN):
+ """Dynamically register a module to handle an rdatatype.
+
+ *implementation*, a module implementing the type in the usual dnspython
+ way.
+
+ *rdtype*, an ``int``, the rdatatype to register.
+
+ *rdtype_text*, a ``text``, the textual form of the rdatatype.
+
+ *is_singleton*, a ``bool``, indicating if the type is a singleton (i.e.
+ RRsets of the type can have only one member.)
+
+ *rdclass*, the rdataclass of the type, or ``dns.rdataclass.ANY`` if
+ it applies to all classes.
+ """
+
+ existing_cls = get_rdata_class(rdclass, rdtype)
+ if existing_cls != GenericRdata:
+ raise RdatatypeExists(rdclass=rdclass, rdtype=rdtype)
+ _rdata_modules[(rdclass, rdtype)] = implementation
+ dns.rdatatype.register_type(rdtype, rdtype_text, is_singleton)
if rdtype in _singletons:
return True
return False
+
+
+def register_type(rdtype, rdtype_text, is_singleton=False):
+ """Dynamically register an rdatatype.
+
+ *rdtype*, an ``int``, the rdatatype to register.
+
+ *rdtype_text*, a ``text``, the textual form of the rdatatype.
+
+ *is_singleton*, a ``bool``, indicating if the type is a singleton (i.e.
+ RRsets of the type can have only one member.)
+ """
+
+ _by_text[rdtype_text] = rdtype
+ _by_value[rdtype] = rdtype_text
+ if is_singleton:
+ _singletons[rdtype] = True
import dns.rdataclass
import dns.rdatatype
+import ttxt_module
+
class RdataTestCase(unittest.TestCase):
def test_str(self):
rdata = dns.rdata.from_text(dns.rdataclass.IN, dns.rdatatype.A, u"1.2.3.4")
self.failUnless(rdata.address == "1.2.3.4")
+ def test_module_registration(self):
+ TTXT = 64001
+ dns.rdata.register_type(ttxt_module, TTXT, 'TTXT')
+ rdata = dns.rdata.from_text(dns.rdataclass.IN, TTXT, 'hello world')
+ self.failUnless(rdata.strings == [b'hello', b'world'])
+ self.failUnless(dns.rdatatype.to_text(TTXT) == 'TTXT')
+ self.failUnless(dns.rdatatype.from_text('TTXT') == TTXT)
+
+ def test_module_reregistration(self):
+ def bad():
+ TTXTTWO = dns.rdatatype.TXT
+ dns.rdata.register_type(ttxt_module, TTXTTWO, 'TTXTTWO')
+ self.failUnlessRaises(dns.rdata.RdatatypeExists, bad)
+
if __name__ == '__main__':
unittest.main()
--- /dev/null
+import dns.rdtypes.txtbase
+
+class TTXT(dns.rdtypes.txtbase.TXTBase):
+ """Test TXT-like record"""