Testing for manager's use of list_zones for dynamically configured zones

octodns/manager.py

@@ -528,7 +528,7 @@ class Manager(object):
             for source in sources:
                 if not hasattr(source, 'list_zones'):
                     raise ManagerException(
-                        f'dynamic zone={name} includes a source that does not support `list_zones`',
+                        f'dynamic zone={name} includes a source that does not support `list_zones`'
                     )
                 for zone_name in source.list_zones():
                     if zone_name in zones:

tests/config/dynamic-config-no-list-zones.yaml

@@ -0,0 +1,21 @@
+providers:
+  in:
+    # does not support list_zones
+    class: helpers.SimpleProvider
+
+  dump:
+    class: octodns.provider.yaml.YamlProvider
+    directory: env/YAML_TMP_DIR
+
+zones:
+  '*':
+    sources:
+    - in
+    targets:
+    - dump
+
+  subzone.unit.tests.:
+    sources:
+    - in
+    targets:
+    - dump

tests/config/dynamic-config.yaml

@@ -0,0 +1,21 @@
+providers:
+  in:
+    class: octodns.provider.yaml.YamlProvider
+    directory: tests/config
+
+  dump:
+    class: octodns.provider.yaml.YamlProvider
+    directory: env/YAML_TMP_DIR
+
+zones:
+  '*':
+    sources:
+    - in
+    targets:
+    - dump
+
+  subzone.unit.tests.:
+    sources:
+    - in
+    targets:
+    - dump

tests/test_octodns_manager.py

@@ -961,6 +961,38 @@ class TestManager(TestCase):
             tc = manager.sync(dry_run=False)
             self.assertEqual(26, tc)
 
+    def test_dynamic_config(self):
+        with TemporaryDirectory() as tmpdir:
+            environ['YAML_TMP_DIR'] = tmpdir.dirname
+
+            manager = Manager(get_config_filename('dynamic-config.yaml'))
+
+            # just unit.tests. which should have been dynamically configured via
+            # list_zones
+            self.assertEqual(
+                23, manager.sync(eligible_zones=['unit.tests.'], dry_run=False)
+            )
+
+            # just subzone.unit.tests. which was explicitly configured
+            self.assertEqual(
+                3,
+                manager.sync(
+                    eligible_zones=['subzone.unit.tests.'], dry_run=False
+                ),
+            )
+
+            # should sync everything across all zones, total of 32 records
+            self.assertEqual(32, manager.sync(dry_run=False))
+
+    def test_dynamic_config_unsupported_zone(self):
+        manager = Manager(
+            get_config_filename('dynamic-config-no-list-zones.yaml')
+        )
+
+        with self.assertRaises(ManagerException) as ctx:
+            manager.sync()
+        self.assertTrue('does not support `list_zones`' in str(ctx.exception))
+
 
 class TestMainThreadExecutor(TestCase):
     def test_success(self):