104 lines
3.2 KiB
Python
104 lines
3.2 KiB
Python
import unittest
|
|
|
|
from dns_sync.adapters import (
|
|
CERTIFICATES,
|
|
DNS_RECORDS,
|
|
PROXY_HOSTS,
|
|
Capability,
|
|
CapabilityOperation,
|
|
Endpoint,
|
|
EndpointPlan,
|
|
NormalizedObject,
|
|
Observation,
|
|
PlanAction,
|
|
PlanOperation,
|
|
)
|
|
|
|
|
|
class EndpointTests(unittest.TestCase):
|
|
def test_identity_uses_plugin_and_address_not_label(self):
|
|
eve = Endpoint("NPM", "192.168.1.3:81/", label="EVE")
|
|
renamed = Endpoint("npm", "192.168.1.3:81", label="anything")
|
|
|
|
self.assertEqual(eve.identity, "npm://192.168.1.3:81")
|
|
self.assertEqual(eve.identity, renamed.identity)
|
|
|
|
def test_empty_endpoint_fields_are_rejected(self):
|
|
with self.assertRaises(ValueError):
|
|
Endpoint("", "192.168.1.3:81")
|
|
|
|
with self.assertRaises(ValueError):
|
|
Endpoint("npm", "")
|
|
|
|
|
|
class CapabilityTests(unittest.TestCase):
|
|
def test_capabilities_are_object_type_specific(self):
|
|
capabilities = frozenset(
|
|
{
|
|
Capability(PROXY_HOSTS, CapabilityOperation.DISCOVER),
|
|
Capability(PROXY_HOSTS, CapabilityOperation.APPLY),
|
|
Capability(CERTIFICATES, CapabilityOperation.DISCOVER),
|
|
Capability(CERTIFICATES, CapabilityOperation.APPLY),
|
|
Capability(DNS_RECORDS, CapabilityOperation.DISCOVER),
|
|
}
|
|
)
|
|
|
|
self.assertIn(
|
|
Capability(CERTIFICATES, CapabilityOperation.APPLY),
|
|
capabilities,
|
|
)
|
|
self.assertNotIn(
|
|
Capability(DNS_RECORDS, CapabilityOperation.APPLY),
|
|
capabilities,
|
|
)
|
|
|
|
|
|
class ObservationAndPlanTests(unittest.TestCase):
|
|
def test_local_objects_can_share_one_logical_identity(self):
|
|
first = NormalizedObject(
|
|
object_type=CERTIFICATES,
|
|
logical_key="chaptarr.loxthelion.com",
|
|
local_key="63",
|
|
payload={"expires_on": "2026-10-01T00:00:00+00:00"},
|
|
)
|
|
second = NormalizedObject(
|
|
object_type=CERTIFICATES,
|
|
logical_key="chaptarr.loxthelion.com",
|
|
local_key="67",
|
|
payload={"expires_on": "2026-11-01T00:00:00+00:00"},
|
|
)
|
|
|
|
observations = [
|
|
Observation("npm://192.168.1.3:81", first),
|
|
Observation("npm://192.168.1.3:81", second),
|
|
]
|
|
|
|
self.assertEqual(
|
|
observations[0].object.logical_key,
|
|
observations[1].object.logical_key,
|
|
)
|
|
self.assertNotEqual(
|
|
observations[0].object.local_key,
|
|
observations[1].object.local_key,
|
|
)
|
|
|
|
def test_plan_is_scoped_to_one_endpoint(self):
|
|
desired = NormalizedObject(
|
|
object_type=DNS_RECORDS,
|
|
logical_key="example.loxthelion.com",
|
|
payload={
|
|
"hostname": "example.loxthelion.com",
|
|
"ip": "192.168.1.101",
|
|
"record_type": "A",
|
|
},
|
|
)
|
|
action = PlanAction(
|
|
endpoint_id="verizon://192.168.1.1",
|
|
operation=PlanOperation.CREATE,
|
|
desired=desired,
|
|
)
|
|
plan = EndpointPlan(action.endpoint_id, (action,))
|
|
|
|
self.assertEqual(plan.endpoint_id, "verizon://192.168.1.1")
|
|
self.assertEqual(plan.actions, (action,))
|