mirror/github-octodns
1
0
Fork 0
mirror of https://github.com/github/octodns.git synced 2024-05-11 05:55:00 +00:00
Code Issues Releases Activity

Merge pull request #1071 from octodns/rrs-improvements

Browse Source 
Record RR handling improvements
This commit is contained in:
Ross McFarland
2023-09-25 13:19:32 -07:00
committed by GitHub
parent 4a7e488d2f 6b681ba8f5
commit 1c0f0b4a48
18 changed files with 131 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
CHANGELOG.md
View File

@@ -1,3 +1,8 @@
## v1.?.? - 2023-??-?? - ???
* Record.from_rrs supports `source` parameter
* *Record.parse_rdata_text unquotes any quoted (string) values
## v1.1.2 - 2023-09-20 - Bunch more bug fixes
* Fix crash bug when using the YamlProvider with a directory that contains a

12
octodns/record/base.py
View File

@@ -12,6 +12,12 @@ from .change import Update
from .exception import RecordException, ValidationError
def unquote(s):
if s and s[0] in ('"', "'"):
return s[1:-1]
return s
class Record(EqualityTupleMixin):
log = getLogger('Record')
@@ -113,7 +119,7 @@ class Record(EqualityTupleMixin):
return reasons
@classmethod
def from_rrs(cls, zone, rrs, lenient=False):
def from_rrs(cls, zone, rrs, lenient=False, source=None):
# group records by name & type so that multiple rdatas can be combined
# into a single record when needed
grouped = defaultdict(list)
@@ -128,7 +134,9 @@ class Record(EqualityTupleMixin):
name = zone.hostname_from_fqdn(rr.name)
_class = cls._CLASSES[rr._type]
data = _class.data_from_rrs(rrs)
record = Record.new(zone, name, data, lenient=lenient)
record = Record.new(
zone, name, data, lenient=lenient, source=source
)
records.append(record)
return records

4
octodns/record/caa.py
View File

@@ -3,7 +3,7 @@
#
from ..equality import EqualityTupleMixin
from .base import Record, ValuesMixin
from .base import Record, ValuesMixin, unquote
from .rr import RrParseError
@@ -20,6 +20,8 @@ class CaaValue(EqualityTupleMixin, dict):
flags = int(flags)
except ValueError:
pass
tag = unquote(tag)
value = unquote(value)
return {'flags': flags, 'tag': tag, 'value': value}
@classmethod

12
octodns/record/loc.py
View File

@@ -3,7 +3,7 @@
#
from ..equality import EqualityTupleMixin
from .base import Record, ValuesMixin
from .base import Record, ValuesMixin, unquote
from .rr import RrParseError
@@ -58,21 +58,23 @@ class LocValue(EqualityTupleMixin, dict):
except ValueError:
pass
try:
altitude = float(altitude)
altitude = float(unquote(altitude))
except ValueError:
pass
try:
size = float(size)
size = float(unquote(size))
except ValueError:
pass
try:
precision_horz = float(precision_horz)
precision_horz = float(unquote(precision_horz))
except ValueError:
pass
try:
precision_vert = float(precision_vert)
precision_vert = float(unquote(precision_vert))
except ValueError:
pass
lat_direction = unquote(lat_direction)
long_direction = unquote(long_direction)
return {
'lat_degrees': lat_degrees,
'lat_minutes': lat_minutes,

3
octodns/record/mx.py
View File

@@ -6,7 +6,7 @@ from fqdn import FQDN
from ..equality import EqualityTupleMixin
from ..idna import idna_encode
from .base import Record, ValuesMixin
from .base import Record, ValuesMixin, unquote
from .rr import RrParseError
@@ -21,6 +21,7 @@ class MxValue(EqualityTupleMixin, dict):
preference = int(preference)
except ValueError:
pass
exchange = unquote(exchange)
return {'preference': preference, 'exchange': exchange}
@classmethod

6
octodns/record/naptr.py
View File

@@ -3,7 +3,7 @@
#
from ..equality import EqualityTupleMixin
from .base import Record, ValuesMixin
from .base import Record, ValuesMixin, unquote
from .rr import RrParseError
@@ -28,6 +28,10 @@ class NaptrValue(EqualityTupleMixin, dict):
preference = int(preference)
except ValueError:
pass
flags = unquote(flags)
service = unquote(service)
regexp = unquote(regexp)
replacement = unquote(replacement)
return {
'order': order,
'preference': preference,

3
octodns/record/srv.py
View File

@@ -8,7 +8,7 @@ from fqdn import FQDN
from ..equality import EqualityTupleMixin
from ..idna import idna_encode
from .base import Record, ValuesMixin
from .base import Record, ValuesMixin, unquote
from .rr import RrParseError
@@ -31,6 +31,7 @@ class SrvValue(EqualityTupleMixin, dict):
port = int(port)
except ValueError:
pass
target = unquote(target)
return {
'priority': priority,
'weight': weight,

3
octodns/record/sshfp.py
View File

@@ -3,7 +3,7 @@
#
from ..equality import EqualityTupleMixin
from .base import Record, ValuesMixin
from .base import Record, ValuesMixin, unquote
from .rr import RrParseError
@@ -25,6 +25,7 @@ class SshfpValue(EqualityTupleMixin, dict):
fingerprint_type = int(fingerprint_type)
except ValueError:
pass
fingerprint = unquote(fingerprint)
return {
'algorithm': algorithm,
'fingerprint_type': fingerprint_type,

3
octodns/record/tlsa.py
View File

@@ -3,7 +3,7 @@
#
from ..equality import EqualityTupleMixin
from .base import Record, ValuesMixin
from .base import Record, ValuesMixin, unquote
from .rr import RrParseError
@@ -31,6 +31,7 @@ class TlsaValue(EqualityTupleMixin, dict):
matching_type = int(matching_type)
except ValueError:
pass
certificate_association_data = unquote(certificate_association_data)
return {
'certificate_usage': certificate_usage,
'selector': selector,

18
tests/test_octodns_record.py
View File

@@ -22,6 +22,7 @@ from octodns.record import (
ValidationError,
ValuesMixin,
)
from octodns.record.base import unquote
from octodns.yaml import ContextDict
from octodns.zone import Zone
@@ -159,10 +160,13 @@ class TestRecord(TestCase):
)
zone = Zone('unit.tests.', [])
records = {(r._type, r.name): r for r in Record.from_rrs(zone, rrs)}
records = {
(r._type, r.name): r for r in Record.from_rrs(zone, rrs, source=99)
}
record = records[('A', '')]
self.assertEqual(42, record.ttl)
self.assertEqual(['1.2.3.4', '2.3.4.5'], record.values)
self.assertEqual(99, record.source)
record = records[('AAAA', '')]
self.assertEqual(43, record.ttl)
self.assertEqual(['fc00::1', 'fc00::2'], record.values)
@@ -409,6 +413,18 @@ class TestRecord(TestCase):
record.rrs,
)
def test_unquote(self):
s = 'Hello "\'"World!'
single = f"'{s}'"
double = f'"{s}"'
self.assertEqual(s, unquote(s))
self.assertEqual(s, unquote(single))
self.assertEqual(s, unquote(double))
# edge cases
self.assertEqual(None, unquote(None))
self.assertEqual('', unquote(''))
class TestRecordValidation(TestCase):
zone = Zone('unit.tests.', [])

6
tests/test_octodns_record_caa.py
View File

@@ -105,6 +105,12 @@ class TestRecordCaa(TestCase):
CaaValue.parse_rdata_text('0 tag 99148c81'),
)
# quoted
self.assertEqual(
{'flags': 0, 'tag': 'tag', 'value': '99148c81'},
CaaValue.parse_rdata_text('0 "tag" "99148c81"'),
)
zone = Zone('unit.tests.', [])
a = CaaRecord(
zone,

2
tests/test_octodns_record_chunked.py
View File

@@ -21,6 +21,8 @@ class TestRecordChunked(TestCase):
'some.words.that.here',
'1.2.word.4',
'1.2.3.4',
# quotes are not removed
'"Hello World!"',
):
self.assertEqual(s, _ChunkedValue.parse_rdata_text(s))

22
tests/test_octodns_record_loc.py
View File

@@ -160,6 +160,26 @@ class TestRecordLoc(TestCase):
LocValue.parse_rdata_text(s),
)
# quoted
s = '0 1 2.2 "N" 3 4 5.5 "E" "6.6m" "7.7m" "8.8m" "9.9m"'
self.assertEqual(
{
'altitude': 6.6,
'lat_degrees': 0,
'lat_direction': 'N',
'lat_minutes': 1,
'lat_seconds': 2.2,
'long_degrees': 3,
'long_direction': 'E',
'long_minutes': 4,
'long_seconds': 5.5,
'precision_horz': 8.8,
'precision_vert': 9.9,
'size': 7.7,
},
LocValue.parse_rdata_text(s),
)
# make sure that the cstor is using parse_rdata_text
zone = Zone('unit.tests.', [])
a = LocRecord(
@@ -196,7 +216,7 @@ class TestRecordLoc(TestCase):
self.assertEqual(7.7, a.values[0].size)
self.assertEqual(8.8, a.values[0].precision_horz)
self.assertEqual(9.9, a.values[0].precision_vert)
self.assertEqual(s, a.values[0].rdata_text)
self.assertEqual(s.replace('"', ''), a.values[0].rdata_text)
def test_loc_value(self):
a = LocValue(

6
tests/test_octodns_record_mx.py
View File

@@ -92,6 +92,12 @@ class TestRecordMx(TestCase):
MxValue.parse_rdata_text('10 mx.unit.tests.'),
)
# quoted
self.assertEqual(
{'preference': 10, 'exchange': 'mx.unit.tests.'},
MxValue.parse_rdata_text('10 "mx.unit.tests."'),
)
zone = Zone('unit.tests.', [])
a = MxRecord(
zone,

13
tests/test_octodns_record_naptr.py
View File

@@ -346,6 +346,19 @@ class TestRecordNaptr(TestCase):
NaptrValue.parse_rdata_text('1 2 three four five six'),
)
# string fields are unquoted if needed
self.assertEqual(
{
'order': 1,
'preference': 2,
'flags': 'three',
'service': 'four',
'regexp': 'five',
'replacement': 'six',
},
NaptrValue.parse_rdata_text('1 2 "three" "four" "five" "six"'),
)
# make sure that the cstor is using parse_rdata_text
zone = Zone('unit.tests.', [])
a = NaptrRecord(

11
tests/test_octodns_record_srv.py
View File

@@ -123,6 +123,17 @@ class TestRecordSrv(TestCase):
SrvValue.parse_rdata_text('1 2 3 srv.unit.tests.'),
)
# quoted
self.assertEqual(
{
'priority': 1,
'weight': 2,
'port': 3,
'target': 'srv.unit.tests.',
},
SrvValue.parse_rdata_text('1 2 3 "srv.unit.tests."'),
)
zone = Zone('unit.tests.', [])
a = SrvRecord(
zone,

6
tests/test_octodns_record_sshfp.py
View File

@@ -113,6 +113,12 @@ class TestRecordSshfp(TestCase):
SshfpValue.parse_rdata_text('1 2 00479b27'),
)
# valid
self.assertEqual(
{'algorithm': 1, 'fingerprint_type': 2, 'fingerprint': '00479b27'},
SshfpValue.parse_rdata_text('1 2 "00479b27"'),
)
zone = Zone('unit.tests.', [])
a = SshfpRecord(
zone,

11
tests/test_octodns_record_tlsa.py
View File

@@ -160,6 +160,17 @@ class TestRecordTlsa(TestCase):
TlsaValue.parse_rdata_text('1 2 3 abcd'),
)
# valid
self.assertEqual(
{
'certificate_usage': 1,
'selector': 2,
'matching_type': 3,
'certificate_association_data': 'abcd',
},
TlsaValue.parse_rdata_text('1 2 3 "abcd"'),
)
zone = Zone('unit.tests.', [])
a = TlsaRecord(
zone,