Add ICMP & UDP healthcheck protocols

CHANGELOG.md

@@ -1,5 +1,6 @@
 ## v1.?.? - 2024-??-?? - ???
 
+* ICMP & UDP healthcheck protocol support added
 * Improved handling of present, but empty/None config file values.
 * Add PlanJson plan_output support
 * Include `record_type` in Change data

octodns/record/base.py

@@ -113,7 +113,9 @@ class Record(EqualityTupleMixin):
             if data['octodns']['healthcheck']['protocol'] not in (
                 'HTTP',
                 'HTTPS',
+                'ICMP',
                 'TCP',
+                'UDP',
             ):
                 reasons.append('invalid healthcheck protocol')
         except KeyError:
@@ -224,14 +226,16 @@ class Record(EqualityTupleMixin):
 
     def healthcheck_host(self, value=None):
         healthcheck = self.octodns.get('healthcheck', {})
-        if healthcheck.get('protocol', None) == 'TCP':
+        protocol = self.healthcheck_protocol
+        if protocol not in ('HTTP', 'HTTPS'):
             return None
         return healthcheck.get('host', self.fqdn[:-1]) or value
 
     @property
     def healthcheck_path(self):
         healthcheck = self.octodns.get('healthcheck', {})
-        if healthcheck.get('protocol', None) == 'TCP':
+        protocol = self.healthcheck_protocol
+        if protocol not in ('HTTP', 'HTTPS'):
             return None
         try:
             return healthcheck['path']
@@ -247,6 +251,8 @@ class Record(EqualityTupleMixin):
 
     @property
     def healthcheck_port(self):
+        if self.healthcheck_protocol == 'ICMP':
+            return None
         try:
             return int(self.octodns['healthcheck']['port'])
         except KeyError:

tests/test_octodns_record_dynamic.py

@@ -82,6 +82,7 @@ class TestRecordDynamic(TestCase):
         )
         self.assertEqual('1.2.3.4', new.healthcheck_host(value="1.2.3.4"))
 
+        # defaults
         new = Record.new(
             self.zone, 'a', {'ttl': 44, 'type': 'A', 'value': '1.2.3.4'}
         )
@@ -90,6 +91,44 @@ class TestRecordDynamic(TestCase):
         self.assertEqual('HTTPS', new.healthcheck_protocol)
         self.assertEqual(443, new.healthcheck_port)
 
+    def test_healthcheck_icmp(self):
+        new = Record.new(
+            self.zone,
+            'a',
+            {
+                'ttl': 44,
+                'type': 'A',
+                'value': '1.2.3.4',
+                'octodns': {
+                    'healthcheck': {
+                        'path': '/ignored',
+                        'host': 'completely.ignored',
+                        'protocol': 'ICMP',
+                        'port': -99,
+                    }
+                },
+            },
+        )
+        self.assertIsNone(new.healthcheck_path)
+        self.assertIsNone(new.healthcheck_host())
+        self.assertEqual('ICMP', new.healthcheck_protocol)
+        self.assertIsNone(new.healthcheck_port)
+
+        new = Record.new(
+            self.zone,
+            'a',
+            {
+                'ttl': 44,
+                'type': 'A',
+                'value': '1.2.3.4',
+                'octodns': {'healthcheck': {'protocol': 'ICMP'}},
+            },
+        )
+        self.assertIsNone(new.healthcheck_path)
+        self.assertIsNone(new.healthcheck_host())
+        self.assertEqual('ICMP', new.healthcheck_protocol)
+        self.assertIsNone(new.healthcheck_port)
+
     def test_healthcheck_tcp(self):
         new = Record.new(
             self.zone,
@@ -128,6 +167,44 @@ class TestRecordDynamic(TestCase):
         self.assertEqual('TCP', new.healthcheck_protocol)
         self.assertEqual(443, new.healthcheck_port)
 
+    def test_healthcheck_udp(self):
+        new = Record.new(
+            self.zone,
+            'a',
+            {
+                'ttl': 44,
+                'type': 'A',
+                'value': '1.2.3.4',
+                'octodns': {
+                    'healthcheck': {
+                        'path': '/ignored',
+                        'host': 'completely.ignored',
+                        'protocol': 'UDP',
+                        'port': 8081,
+                    }
+                },
+            },
+        )
+        self.assertIsNone(new.healthcheck_path)
+        self.assertIsNone(new.healthcheck_host())
+        self.assertEqual('UDP', new.healthcheck_protocol)
+        self.assertEqual(8081, new.healthcheck_port)
+
+        new = Record.new(
+            self.zone,
+            'a',
+            {
+                'ttl': 44,
+                'type': 'A',
+                'value': '1.2.3.4',
+                'octodns': {'healthcheck': {'protocol': 'UDP'}},
+            },
+        )
+        self.assertIsNone(new.healthcheck_path)
+        self.assertIsNone(new.healthcheck_host())
+        self.assertEqual('UDP', new.healthcheck_protocol)
+        self.assertEqual(443, new.healthcheck_port)
+
     def test_simple_a_weighted(self):
         a_data = {
             'dynamic': {