337 lines
9.5 KiB
Python
337 lines
9.5 KiB
Python
import os
|
|
import unittest
|
|
from unittest.mock import Mock, patch
|
|
|
|
from dns_sync.adapters import (
|
|
DNS_RECORDS,
|
|
Capability,
|
|
Endpoint,
|
|
CapabilityOperation,
|
|
NormalizedObject,
|
|
PlanAction,
|
|
PlanOperation,
|
|
)
|
|
from dns_sync.config import AppConfig
|
|
from external_plugins.verizon.client import VerizonRecord
|
|
from dns_sync.registry import (
|
|
configured_adapter_endpoints,
|
|
load_adapter,
|
|
)
|
|
|
|
|
|
class VerizonAdapterTests(unittest.TestCase):
|
|
def config(self):
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"DNS_TARGET_IP": "192.168.1.101",
|
|
"NPM_ENDPOINTS": "192.168.1.3",
|
|
"VERIZON_ENDPOINTS": "192.168.1.1",
|
|
"ENABLED_PLUGINS": "npm,verizon",
|
|
"DNS_SYNC_PLUGIN_PATH": "external_plugins",
|
|
},
|
|
clear=True,
|
|
):
|
|
return AppConfig.from_env()
|
|
|
|
def adapter(self, client):
|
|
config = self.config()
|
|
|
|
import dns_sync.registry as registry
|
|
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"ENABLED_PLUGINS": "npm,verizon",
|
|
"DNS_SYNC_PLUGIN_PATH": "external_plugins",
|
|
},
|
|
clear=False,
|
|
):
|
|
module = registry._load_external_adapter_module(
|
|
"verizon"
|
|
)
|
|
|
|
if module is None:
|
|
raise AssertionError(
|
|
"external Verizon adapter was not found"
|
|
)
|
|
|
|
with patch.object(
|
|
module,
|
|
"VerizonClient",
|
|
return_value=client,
|
|
):
|
|
endpoint = configured_adapter_endpoints(
|
|
"verizon",
|
|
config,
|
|
)[0]
|
|
|
|
return load_adapter(
|
|
"verizon",
|
|
config,
|
|
endpoint,
|
|
)
|
|
|
|
@staticmethod
|
|
def dns_object(
|
|
hostname="app.example.com",
|
|
ip="192.168.1.101",
|
|
wildcard=False,
|
|
):
|
|
return NormalizedObject(
|
|
object_type=DNS_RECORDS,
|
|
logical_key=hostname,
|
|
payload={
|
|
"hostname": hostname,
|
|
"ip": ip,
|
|
"record_type": "A",
|
|
"wildcard": wildcard,
|
|
},
|
|
)
|
|
|
|
def test_adapter_rejects_endpoint_for_other_plugin(self):
|
|
config = self.config()
|
|
endpoint = Endpoint(
|
|
"npm",
|
|
"192.168.1.3:81",
|
|
)
|
|
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"ENABLED_PLUGINS": "npm,verizon",
|
|
"DNS_SYNC_PLUGIN_PATH": "external_plugins",
|
|
},
|
|
clear=False,
|
|
):
|
|
with self.assertRaisesRegex(
|
|
RuntimeError,
|
|
"cannot use endpoint",
|
|
):
|
|
load_adapter(
|
|
"verizon",
|
|
config,
|
|
endpoint,
|
|
)
|
|
|
|
|
|
def test_capabilities_cover_dns_discovery_apply_and_verify(self):
|
|
adapter = self.adapter(Mock())
|
|
|
|
self.assertEqual(
|
|
adapter.capabilities,
|
|
frozenset(
|
|
{
|
|
Capability(
|
|
DNS_RECORDS,
|
|
CapabilityOperation.DISCOVER,
|
|
),
|
|
Capability(
|
|
DNS_RECORDS,
|
|
CapabilityOperation.APPLY,
|
|
),
|
|
Capability(
|
|
DNS_RECORDS,
|
|
CapabilityOperation.VERIFY,
|
|
),
|
|
}
|
|
),
|
|
)
|
|
|
|
def test_client_uses_explicit_endpoint(self):
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"DNS_TARGET_IP": "192.168.1.101",
|
|
"NPM_ENDPOINTS": "192.168.1.3",
|
|
"VERIZON_ENDPOINTS": "192.168.1.1:8443",
|
|
"ENABLED_PLUGINS": "npm,verizon",
|
|
"DNS_SYNC_PLUGIN_PATH": "external_plugins",
|
|
"ROUTER_URL": "https://192.168.1.254",
|
|
"ROUTER_PASSWORD": "not-a-real-password",
|
|
},
|
|
clear=True,
|
|
):
|
|
config = AppConfig.from_env()
|
|
endpoint = configured_adapter_endpoints(
|
|
"verizon",
|
|
config,
|
|
)[0]
|
|
|
|
adapter = load_adapter(
|
|
"verizon",
|
|
config,
|
|
endpoint,
|
|
)
|
|
|
|
self.assertEqual(
|
|
adapter.endpoint.identity,
|
|
"verizon://192.168.1.1:8443",
|
|
)
|
|
self.assertEqual(
|
|
adapter.client.router_url,
|
|
"https://192.168.1.1:8443",
|
|
)
|
|
|
|
def test_connect_delegates_to_client(self):
|
|
client = Mock()
|
|
adapter = self.adapter(client)
|
|
|
|
adapter.connect()
|
|
|
|
client.connect.assert_called_once_with()
|
|
|
|
def test_discovers_dns_records_with_slot_local_keys(self):
|
|
client = Mock()
|
|
client.list_records.return_value = {
|
|
"second.example.com": VerizonRecord(
|
|
hostname="second.example.com",
|
|
ip="192.168.1.101",
|
|
metadata={"slot": 42},
|
|
),
|
|
"first.example.com": VerizonRecord(
|
|
hostname="first.example.com",
|
|
ip="192.168.1.101",
|
|
metadata={"slot": 41},
|
|
),
|
|
}
|
|
adapter = self.adapter(client)
|
|
|
|
discovered = adapter.discover(DNS_RECORDS)
|
|
|
|
self.assertEqual(
|
|
tuple(item.local_key for item in discovered),
|
|
("41", "42"),
|
|
)
|
|
self.assertEqual(
|
|
discovered[0].payload,
|
|
{
|
|
"hostname": "first.example.com",
|
|
"ip": "192.168.1.101",
|
|
"record_type": "A",
|
|
},
|
|
)
|
|
|
|
def test_create_and_update_delegate_without_backend_ip(self):
|
|
client = Mock()
|
|
client.add_record.return_value = "created"
|
|
client.update_record.return_value = "updated"
|
|
client.list_records.return_value = {
|
|
"app.example.com": VerizonRecord(
|
|
hostname="app.example.com",
|
|
ip="192.168.1.101",
|
|
metadata={"slot": 41},
|
|
),
|
|
}
|
|
adapter = self.adapter(client)
|
|
desired = self.dns_object(ip="192.168.1.101")
|
|
|
|
create_result = adapter.apply(
|
|
PlanAction(
|
|
endpoint_id=adapter.endpoint.identity,
|
|
operation=PlanOperation.CREATE,
|
|
desired=desired,
|
|
)
|
|
)
|
|
update_result = adapter.apply(
|
|
PlanAction(
|
|
endpoint_id=adapter.endpoint.identity,
|
|
operation=PlanOperation.UPDATE,
|
|
desired=desired,
|
|
)
|
|
)
|
|
|
|
self.assertEqual(
|
|
create_result.object.local_key,
|
|
"41",
|
|
)
|
|
self.assertEqual(
|
|
create_result.detail,
|
|
"created",
|
|
)
|
|
self.assertEqual(
|
|
update_result.object.local_key,
|
|
"41",
|
|
)
|
|
self.assertEqual(
|
|
update_result.detail,
|
|
"updated",
|
|
)
|
|
self.assertEqual(
|
|
client.add_record.call_args.args[0].ip,
|
|
"192.168.1.101",
|
|
)
|
|
self.assertEqual(
|
|
client.update_record.call_args.args[0].ip,
|
|
"192.168.1.101",
|
|
)
|
|
|
|
def test_noop_does_not_mutate_router(self):
|
|
client = Mock()
|
|
adapter = self.adapter(client)
|
|
|
|
result = adapter.apply(
|
|
PlanAction(
|
|
endpoint_id=adapter.endpoint.identity,
|
|
operation=PlanOperation.NOOP,
|
|
desired=self.dns_object(),
|
|
)
|
|
)
|
|
|
|
self.assertIsNone(result)
|
|
client.add_record.assert_not_called()
|
|
client.update_record.assert_not_called()
|
|
|
|
def test_verify_delegates_expected_dns_record(self):
|
|
client = Mock()
|
|
client.verify_record.return_value = True
|
|
adapter = self.adapter(client)
|
|
|
|
verified = adapter.verify(self.dns_object())
|
|
|
|
self.assertTrue(verified)
|
|
record = client.verify_record.call_args.args[0]
|
|
self.assertEqual(record.hostname, "app.example.com")
|
|
self.assertEqual(record.ip, "192.168.1.101")
|
|
|
|
def test_unsupported_objects_actions_and_wildcards_fail(self):
|
|
client = Mock()
|
|
adapter = self.adapter(client)
|
|
|
|
with self.assertRaisesRegex(RuntimeError, "cannot discover"):
|
|
adapter.discover("certificates")
|
|
|
|
with self.assertRaisesRegex(RuntimeError, "does not belong"):
|
|
adapter.apply(
|
|
PlanAction(
|
|
endpoint_id="verizon://different:443",
|
|
operation=PlanOperation.CREATE,
|
|
desired=self.dns_object(),
|
|
)
|
|
)
|
|
|
|
with self.assertRaisesRegex(RuntimeError, "wildcard"):
|
|
adapter.apply(
|
|
PlanAction(
|
|
endpoint_id=adapter.endpoint.identity,
|
|
operation=PlanOperation.CREATE,
|
|
desired=self.dns_object(
|
|
hostname="*.example.com",
|
|
wildcard=True,
|
|
),
|
|
)
|
|
)
|
|
|
|
with self.assertRaisesRegex(RuntimeError, "delete"):
|
|
adapter.apply(
|
|
PlanAction(
|
|
endpoint_id=adapter.endpoint.identity,
|
|
operation=PlanOperation.DELETE,
|
|
desired=self.dns_object(),
|
|
)
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|