Merge branch 'main' into main

octodns/record/urlfwd.py

@@ -3,7 +3,8 @@
 #
 
 from ..equality import EqualityTupleMixin
-from .base import Record, ValuesMixin
+from .base import Record, ValuesMixin, unquote
+from .rr import RrParseError
 
 
 class UrlfwdValue(EqualityTupleMixin, dict):
@@ -11,6 +12,34 @@ class UrlfwdValue(EqualityTupleMixin, dict):
     VALID_MASKS = (0, 1, 2)
     VALID_QUERY = (0, 1)
 
+    @classmethod
+    def parse_rdata_text(self, value):
+        try:
+            path, target, code, masking, query = value.split(' ')
+        except ValueError:
+            raise RrParseError()
+        try:
+            code = int(code)
+        except ValueError:
+            pass
+        try:
+            masking = int(masking)
+        except ValueError:
+            pass
+        try:
+            query = int(query)
+        except ValueError:
+            pass
+        path = unquote(path)
+        target = unquote(target)
+        return {
+            'path': path,
+            'target': target,
+            'code': code,
+            'masking': masking,
+            'query': query,
+        }
+
     @classmethod
     def validate(cls, data, _type):
         reasons = []
@@ -99,6 +128,10 @@ class UrlfwdValue(EqualityTupleMixin, dict):
     def query(self, value):
         self['query'] = value
 
+    @property
+    def rdata_text(self):
+        return f'"{self.path}" "{self.target}" {self.code} {self.masking} {self.query}'
+
     def _equality_tuple(self):
         return (self.path, self.target, self.code, self.masking, self.query)
 

tests/test_octodns_record.py

@@ -846,3 +846,30 @@ class TestRecordValidation(TestCase):
             '<CnameRecord CNAME 43, pointer.unit.tests., unit.tests.>',
             record.__repr__(),
         )
+
+    def test_records_have_rdata_methods(self):
+        for _type, cls in Record.registered_types().items():
+            print(f'{_type} {cls}')
+            attr = 'parse_rdata_texts'
+            print(f'  {attr}')
+            method = getattr(cls, attr)
+            self.assertTrue(method, f'{_type}, {cls} has {attr}')
+            self.assertTrue(
+                callable(method), f'{_type}, {cls} {attr} is callable'
+            )
+
+            value_type = getattr(cls, '_value_type')
+            self.assertTrue(value_type, f'{_type}, {cls} has _value_type')
+
+            attr = 'parse_rdata_text'
+            print(f'  {attr}')
+            method = getattr(value_type, attr)
+            self.assertTrue(method, f'{_type}, {cls} has {attr}')
+            self.assertTrue(
+                callable(method), f'{_type}, {cls} {attr} is callable'
+            )
+
+            attr = 'rdata_text'
+            method = getattr(value_type, attr)
+            self.assertTrue(method, f'{_type}, {cls} has {attr}')
+            # this one is a @property so not callable

tests/test_octodns_record_urlfwd.py

@@ -8,6 +8,7 @@ from helpers import SimpleProvider
 
 from octodns.record import Record
 from octodns.record.exception import ValidationError
+from octodns.record.rr import RrParseError
 from octodns.record.urlfwd import UrlfwdRecord, UrlfwdValue
 from octodns.zone import Zone
 
@@ -389,3 +390,96 @@ class TestRecordUrlfwd(TestCase):
         self.assertEqual(
             ['unrecognized query setting "3"'], ctx.exception.reasons
         )
+
+    def test_urlfwd_value_rdata_text(self):
+        # empty string won't parse
+        with self.assertRaises(RrParseError):
+            UrlfwdValue.parse_rdata_text('')
+
+        # single word won't parse
+        with self.assertRaises(RrParseError):
+            UrlfwdValue.parse_rdata_text('nope')
+
+        # 2nd word won't parse
+        with self.assertRaises(RrParseError):
+            UrlfwdValue.parse_rdata_text('1 2')
+
+        # 3rd word won't parse
+        with self.assertRaises(RrParseError):
+            UrlfwdValue.parse_rdata_text('1 2 3')
+
+        # 4th word won't parse
+        with self.assertRaises(RrParseError):
+            UrlfwdValue.parse_rdata_text('1 2 3 4')
+
+        # 6th word won't parse
+        with self.assertRaises(RrParseError):
+            UrlfwdValue.parse_rdata_text('1 2 3 4 5 6')
+
+        # code, masking, and query not ints
+        self.assertEqual(
+            {
+                'path': 'one',
+                'target': 'urlfwd.unit.tests.',
+                'code': 'one',
+                'masking': 'two',
+                'query': 'three',
+            },
+            UrlfwdValue.parse_rdata_text(
+                'one urlfwd.unit.tests. one two three'
+            ),
+        )
+
+        # valid
+        self.assertEqual(
+            {
+                'path': 'one',
+                'target': 'urlfwd.unit.tests.',
+                'code': 1,
+                'masking': 2,
+                'query': 3,
+            },
+            UrlfwdValue.parse_rdata_text('one urlfwd.unit.tests. 1 2 3'),
+        )
+
+        # quoted
+        self.assertEqual(
+            {
+                'path': 'one',
+                'target': 'urlfwd.unit.tests.',
+                'code': 1,
+                'masking': 2,
+                'query': 3,
+            },
+            UrlfwdValue.parse_rdata_text('"one" "urlfwd.unit.tests." 1 2 3'),
+        )
+
+        zone = Zone('unit.tests.', [])
+        a = UrlfwdRecord(
+            zone,
+            'urlfwd',
+            {
+                'ttl': 32,
+                'value': {
+                    'path': 'one',
+                    'target': 'urlfwd.unit.tests.',
+                    'code': 1,
+                    'masking': 2,
+                    'query': 3,
+                },
+            },
+        )
+        self.assertEqual('one', a.values[0].path)
+        self.assertEqual('urlfwd.unit.tests.', a.values[0].target)
+        self.assertEqual(1, a.values[0].code)
+        self.assertEqual(2, a.values[0].masking)
+        self.assertEqual(3, a.values[0].query)
+
+        # both directions should match
+        rdata = '"one" "urlfwd.unit.tests." 1 2 3'
+        record = UrlfwdRecord(
+            zone,
+            'urlfwd',
+            {'ttl': 32, 'value': UrlfwdValue.parse_rdata_text(rdata)},
+        )
+        self.assertEqual(rdata, record.values[0].rdata_text)