Implement to/from rr text for MxValue as a POC

octodns/record/__init__.py

@@ -73,6 +73,10 @@ class RecordException(Exception):
     pass
 
 
+class RrParseError(RecordException):
+    pass
+
+
 class ValidationError(RecordException):
     @classmethod
     def build_message(cls, fqdn, reasons):
@@ -1254,12 +1258,32 @@ Record.register_type(LocRecord)
 
 
 class MxValue(EqualityTupleMixin, dict):
+    @classmethod
+    def parse_rr_text(self, value):
+        try:
+            preference, exchange = value.split(' ')
+        except ValueError:
+            raise RrParseError('failed to parse string value as RR text')
+        try:
+            preference = int(preference)
+        except ValueError:
+            pass
+        return {'preference': preference, 'exchange': exchange}
+
     @classmethod
     def validate(cls, data, _type):
         if not isinstance(data, (list, tuple)):
             data = (data,)
         reasons = []
         for value in data:
+            if isinstance(value, str):
+                # it's hopefully RR formatted, give parsing a try
+                try:
+                    value = cls.parse_rr_text(value)
+                except RrParseError as e:
+                    reasons.append(str(e))
+                    # not a dict so no point in continuing
+                    continue
             try:
                 try:
                     int(value['preference'])
@@ -1291,6 +1315,8 @@ class MxValue(EqualityTupleMixin, dict):
         return [cls(v) for v in values]
 
     def __init__(self, value):
+        if isinstance(value, str):
+            value = self.parse_rr_text(value)
         # RFC1035 says preference, half the providers use priority
         try:
             preference = value['preference']
@@ -1325,6 +1351,10 @@ class MxValue(EqualityTupleMixin, dict):
     def data(self):
         return self
 
+    @property
+    def rr_text(self):
+        return f'{self.preference} {self.exchange}'
+
     def __hash__(self):
         return hash((self.preference, self.exchange))
 

tests/test_octodns_record.py

@@ -32,6 +32,7 @@ from octodns.record import (
     PtrRecord,
     Record,
     RecordException,
+    RrParseError,
     SshfpRecord,
     SshfpValue,
     SpfRecord,
@@ -592,6 +593,68 @@ class TestRecord(TestCase):
         # __repr__ doesn't blow up
         a.__repr__()
 
+    def test_mx_rr_text(self):
+
+        # empty string won't parse
+        with self.assertRaises(RrParseError):
+            MxValue.parse_rr_text('')
+
+        # single word won't parse
+        with self.assertRaises(RrParseError):
+            MxValue.parse_rr_text('nope')
+
+        # 3rd word won't parse
+        with self.assertRaises(RrParseError):
+            MxValue.parse_rr_text('10 mx.unit.tests. another')
+
+        # preference not an int, will parse
+        self.assertEqual(
+            {'preference': 10, 'exchange': 'mx.unit.tests.'},
+            MxValue.parse_rr_text('10 mx.unit.tests.'),
+        )
+
+        # preference not an int
+        self.assertEqual(
+            {'preference': 'abc', 'exchange': 'mx.unit.tests.'},
+            MxValue.parse_rr_text('abc mx.unit.tests.'),
+        )
+
+        # make sure that validate is using parse_rr_text when passed string
+        # values
+        reasons = MxRecord.validate(
+            'mx', 'mx.unit.tests.', {'ttl': 32, 'value': ''}
+        )
+        self.assertEqual(['failed to parse string value as RR text'], reasons)
+        reasons = MxRecord.validate(
+            'mx', 'mx.unit.tests.', {'ttl': 32, 'values': ['nope']}
+        )
+        self.assertEqual(['failed to parse string value as RR text'], reasons)
+        reasons = MxRecord.validate(
+            'mx', 'mx.unit.tests.', {'ttl': 32, 'value': '10 mail.unit.tests.'}
+        )
+        self.assertFalse(reasons)
+
+        # make sure that the cstor is using parse_rr_text
+        zone = Zone('unit.tests.', [])
+        a = MxRecord(zone, 'mx', {'ttl': 32, 'value': '10 mail.unit.tests.'})
+        self.assertEqual(10, a.values[0].preference)
+        self.assertEqual('10 mail.unit.tests.', a.values[0].rr_text)
+        self.assertEqual('mail.unit.tests.', a.values[0].exchange)
+        a = MxRecord(
+            zone,
+            'mx',
+            {
+                'ttl': 32,
+                'values': ['11 mail1.unit.tests.', '12 mail2.unit.tests.'],
+            },
+        )
+        self.assertEqual(11, a.values[0].preference)
+        self.assertEqual('mail1.unit.tests.', a.values[0].exchange)
+        self.assertEqual('11 mail1.unit.tests.', a.values[0].rr_text)
+        self.assertEqual(12, a.values[1].preference)
+        self.assertEqual('mail2.unit.tests.', a.values[1].exchange)
+        self.assertEqual('12 mail2.unit.tests.', a.values[1].rr_text)
+
     def test_naptr(self):
         a_values = [
             NaptrValue(