wip stuff to store progress, queuing for 2.x

octodns/record/chunked.py

@@ -13,25 +13,16 @@ class _ChunkedValuesMixin(ValuesMixin):
     CHUNK_SIZE = 255
     _unescaped_semicolon_re = re.compile(r'\w;')
 
-    def chunked_value(self, value):
-        value = value.replace('"', '\\"')
-        vs = [
-            value[i : i + self.CHUNK_SIZE]
-            for i in range(0, len(value), self.CHUNK_SIZE)
-        ]
-        vs = '" "'.join(vs)
-        return self._value_type(f'"{vs}"')
-
     @property
     def chunked_values(self):
         values = []
         for v in self.values:
-            values.append(self.chunked_value(v))
+            values.append(v.rdata_text)
         return values
 
     @property
     def rr_values(self):
-        return self.chunked_values
+        return self.values
 
 
 def _parse(s, spec_unquoted=False, strict=False):
@@ -133,4 +124,11 @@ class _ChunkedValue(str):
 
     @property
     def rdata_text(self):
-        return self
+        # TODO: this needs to split & quote
+        val = self.replace('"', '\\"')
+        chunks = []
+        while val:
+            chunks.append(val[0 : _ChunkedValuesMixin.CHUNK_SIZE])
+            val = val[_ChunkedValuesMixin.CHUNK_SIZE :]
+        chunks = '" "'.join(chunks)
+        return f'"{chunks}"'

tests/test_octodns_processor_filter.py

@@ -233,10 +233,9 @@ class TestValueAllowListFilter(TestCase):
 
         self.assertEqual(6, len(self.zone.records))
         filtered = allows.process_source_zone(self.zone.copy())
-        self.assertEqual(2, len(filtered.records))
+        self.assertEqual(1, len(filtered.records))
         self.assertEqual(
-            ['good.exact', 'good.values'],
-            sorted([r.name for r in filtered.records]),
+            ['good.exact'], sorted([r.name for r in filtered.records])
         )
 
     def test_regex(self):
@@ -303,9 +302,15 @@ class TestValueRejectListFilter(TestCase):
 
         self.assertEqual(6, len(self.zone.records))
         filtered = rejects.process_source_zone(self.zone.copy())
-        self.assertEqual(4, len(filtered.records))
+        self.assertEqual(5, len(filtered.records))
         self.assertEqual(
-            ['bad.compare', 'bad.values', 'first.regex', 'second.regex'],
+            [
+                'bad.compare',
+                'bad.values',
+                'first.regex',
+                'good.values',
+                'second.regex',
+            ],
             sorted([r.name for r in filtered.records]),
         )
 

tests/test_octodns_record_chunked.py

@@ -61,7 +61,7 @@ class TestRecordChunked(TestCase):
 
         zone = Zone('unit.tests.', [])
         a = SpfRecord(zone, 'a', {'ttl': 42, 'value': 'some.target.'})
-        self.assertEqual('some.target.', a.values[0].rdata_text)
+        self.assertEqual('"some.target."', a.values[0].rdata_text)
 
 
 class TestChunkedValue(TestCase):
@@ -266,3 +266,54 @@ class TestChunkedValue(TestCase):
             'z after',
         ]
         self.assertEqual(expected, chunked)
+
+    def test_rdata_text(self):
+        value = _ChunkedValue('hello world')
+        self.assertEqual('"hello world"', value.rdata_text)
+
+        value = _ChunkedValue('hello " world')
+        self.assertEqual('"hello \\" world"', value.rdata_text)
+
+        # 254 chars
+        val = 'x' * 254
+        value = _ChunkedValue(val)
+        self.assertEqual(f'"{val}"', value.rdata_text)
+
+        # 255 chars
+        val = 'x' * 255
+        value = _ChunkedValue(val)
+        self.assertEqual(f'"{val}"', value.rdata_text)
+
+        # 256 chars, single split
+        val = 'x' * 256
+        value = _ChunkedValue(val)
+        got = value.rdata_text
+        expected = '"' + ('x' * 255) + '"'
+        self.assertEqual(expected, got[:257])
+        self.assertEqual(' "x"', got[257:])
+
+        val = (
+            'Lorem ipsum dolor sit amet, consectetur '
+            'adipiscing elit, sed do eiusmod tempor incididunt ut '
+            'labore et dolore magna aliqua. Ut enim ad minim veniam, '
+            'quis nostrud exercitation ullamco laboris nisi ut aliquip '
+            'ex ea commodo consequat. Duis aute irure dolor in '
+            'reprehenderit in voluptate velit esse cillum dolore eu '
+            'fugiat nulla pariatur. Excepteur sint occaecat cupidatat '
+            'non proident, sunt in culpa qui officia deserunt mollit '
+            'anim id est laborum.'
+        )
+        value = _ChunkedValue(val)
+        expected = (
+            '"Lorem ipsum dolor sit amet, consectetur '
+            'adipiscing elit, sed do eiusmod tempor incididunt ut '
+            'labore et dolore magna aliqua. Ut enim ad minim veniam, '
+            'quis nostrud exercitation ullamco laboris nisi ut aliquip '
+            'ex ea commodo consequat. Duis aute irure dolor i" "n '
+            'reprehenderit in voluptate velit esse cillum dolore eu '
+            'fugiat nulla pariatur. Excepteur sint occaecat cupidatat '
+            'non proident, sunt in culpa qui officia deserunt mollit '
+            'anim id est laborum."'
+        )
+        got = value.rdata_text
+        self.assertEqual(expected, got)