david/dns-sync
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add portable NPM reconciliation policies

Browse Source
This commit is contained in:
Lyncoln
2026-09-20 01:05:34 -04:00
parent c4cdc92865
commit 3022b76876
3 changed files with 813 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_planning.py
+211
View File
@@ -289,6 +289,217 @@ class EndpointPlanningTests(unittest.TestCase):
)
def test_certificate_provider_difference_is_noop_when_portable_state_matches(self):
endpoint = Endpoint("npm", "node-two:81")
npm_adapter = FakeAdapter(
endpoint=endpoint,
capabilities=frozenset(
{
Capability(
CERTIFICATES,
CapabilityOperation.APPLY,
),
}
),
)
desired = NormalizedObject(
object_type=CERTIFICATES,
logical_key="app.example.com",
payload={
"domain_names": ["app.example.com"],
"provider": "letsencrypt",
"nice_name": "source",
"valid_from": "2026-09-01T00:00:00+00:00",
"expires_on": "2026-12-01T00:00:00+00:00",
},
)
current = Observation(
endpoint_id=endpoint.identity,
object=NormalizedObject(
object_type=CERTIFICATES,
logical_key="app.example.com",
local_key="84",
payload={
"domain_names": ["app.example.com"],
"provider": "other",
"nice_name": "imported copy",
"valid_from": "2026-09-01T00:00:00+00:00",
"expires_on": "2026-12-01T00:00:00+00:00",
},
),
observed_at=NOW,
)
plans = build_endpoint_plans(
[npm_adapter],
[desired],
[current],
)
self.assertEqual(
plans[0].actions[0].operation,
PlanOperation.NOOP,
)
def test_proxy_reference_change_is_update_even_when_payload_matches(self):
endpoint = Endpoint("npm", "node-two:81")
npm_adapter = FakeAdapter(
endpoint=endpoint,
capabilities=frozenset(
{
Capability(
PROXY_HOSTS,
CapabilityOperation.APPLY,
),
}
),
)
desired = NormalizedObject(
object_type=PROXY_HOSTS,
logical_key="app.example.com",
payload={
"domain_names": ["app.example.com"],
"forward_scheme": "http",
"forward_host": "backend.internal",
"forward_port": 8080,
},
references=(
ObjectReference(
relation="certificate",
object_type=CERTIFICATES,
logical_key="new.example.com",
),
),
)
observations = [
Observation(
endpoint_id=endpoint.identity,
object=NormalizedObject(
object_type=CERTIFICATES,
logical_key="old.example.com",
local_key="67",
payload={
"domain_names": ["old.example.com"],
},
),
observed_at=NOW,
),
Observation(
endpoint_id=endpoint.identity,
object=NormalizedObject(
object_type=PROXY_HOSTS,
logical_key="app.example.com",
local_key="12",
payload=dict(desired.payload),
references=(
ObjectReference(
relation="certificate",
object_type=CERTIFICATES,
local_key="67",
),
),
),
observed_at=NOW,
),
]
plans = build_endpoint_plans(
[npm_adapter],
[desired],
observations,
)
self.assertEqual(
plans[0].actions[0].operation,
PlanOperation.UPDATE,
)
def test_planner_logicalizes_matching_local_reference_to_noop(self):
endpoint = Endpoint("npm", "node-two:81")
npm_adapter = FakeAdapter(
endpoint=endpoint,
capabilities=frozenset(
{
Capability(
PROXY_HOSTS,
CapabilityOperation.APPLY,
),
}
),
)
desired = NormalizedObject(
object_type=PROXY_HOSTS,
logical_key="app.example.com",
payload={
"domain_names": ["app.example.com"],
"forward_scheme": "http",
"forward_host": "backend.internal",
"forward_port": 8080,
},
references=(
ObjectReference(
relation="certificate",
object_type=CERTIFICATES,
logical_key="app.example.com",
),
),
)
observations = [
Observation(
endpoint_id=endpoint.identity,
object=NormalizedObject(
object_type=CERTIFICATES,
logical_key="app.example.com",
local_key="84",
payload={
"domain_names": ["app.example.com"],
},
),
observed_at=NOW,
),
Observation(
endpoint_id=endpoint.identity,
object=NormalizedObject(
object_type=PROXY_HOSTS,
logical_key="app.example.com",
local_key="12",
payload=dict(desired.payload),
references=(
ObjectReference(
relation="certificate",
object_type=CERTIFICATES,
local_key="84",
),
),
),
observed_at=NOW,
),
]
plans = build_endpoint_plans(
[npm_adapter],
[desired],
observations,
)
self.assertEqual(
plans[0].actions[0].operation,
PlanOperation.NOOP,
)
self.assertEqual(
plans[0].actions[0].current.references,
desired.references,
)
def test_duplicate_endpoint_objects_fail_without_payload_output(self):
desired = [
dns_object("app.example.com", "192.168.1.101")