107 lines
2.6 KiB
Python
107 lines
2.6 KiB
Python
import unittest
|
|
|
|
from dns_sync.adapter_support import (
|
|
require_action_endpoint,
|
|
require_desired,
|
|
require_endpoint_plugin,
|
|
)
|
|
from dns_sync.adapters import (
|
|
Endpoint,
|
|
NormalizedObject,
|
|
PlanAction,
|
|
PlanOperation,
|
|
)
|
|
|
|
|
|
class AdapterSupportTests(unittest.TestCase):
|
|
def test_endpoint_plugin_accepts_matching_plugin(self):
|
|
endpoint = Endpoint("npm", "node.example:81")
|
|
|
|
require_endpoint_plugin(endpoint, "npm")
|
|
|
|
|
|
def test_endpoint_plugin_rejects_wrong_plugin(self):
|
|
endpoint = Endpoint("npm", "node.example:81")
|
|
|
|
with self.assertRaisesRegex(
|
|
RuntimeError,
|
|
"cannot use endpoint",
|
|
):
|
|
require_endpoint_plugin(
|
|
endpoint,
|
|
"verizon",
|
|
)
|
|
|
|
|
|
def test_endpoint_plugin_rejects_empty_expected_name(self):
|
|
endpoint = Endpoint("npm", "node.example:81")
|
|
|
|
with self.assertRaisesRegex(
|
|
ValueError,
|
|
"cannot be empty",
|
|
):
|
|
require_endpoint_plugin(endpoint, " ")
|
|
|
|
|
|
def test_action_endpoint_accepts_matching_endpoint(self):
|
|
endpoint = Endpoint("npm", "node.example:81")
|
|
action = PlanAction(
|
|
endpoint_id=endpoint.identity,
|
|
operation=PlanOperation.CREATE,
|
|
)
|
|
|
|
require_action_endpoint(action, endpoint)
|
|
|
|
|
|
def test_action_endpoint_rejects_other_endpoint(self):
|
|
endpoint = Endpoint("npm", "node.example:81")
|
|
action = PlanAction(
|
|
endpoint_id="npm://other.example:81",
|
|
operation=PlanOperation.CREATE,
|
|
)
|
|
|
|
with self.assertRaisesRegex(
|
|
RuntimeError,
|
|
"does not belong",
|
|
):
|
|
require_action_endpoint(action, endpoint)
|
|
|
|
|
|
def test_require_desired_returns_desired_object(self):
|
|
desired = NormalizedObject(
|
|
object_type="example",
|
|
logical_key="object",
|
|
payload={},
|
|
)
|
|
|
|
action = PlanAction(
|
|
endpoint_id="example://endpoint",
|
|
operation=PlanOperation.CREATE,
|
|
desired=desired,
|
|
)
|
|
|
|
self.assertIs(
|
|
require_desired(action),
|
|
desired,
|
|
)
|
|
|
|
|
|
def test_require_desired_fails_closed(self):
|
|
action = PlanAction(
|
|
endpoint_id="example://endpoint",
|
|
operation=PlanOperation.CREATE,
|
|
)
|
|
|
|
with self.assertRaisesRegex(
|
|
RuntimeError,
|
|
"Test mutation requires desired state",
|
|
):
|
|
require_desired(
|
|
action,
|
|
context="Test mutation",
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|