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

Add NPM proxy host apply paths

Browse Source
This commit is contained in:
Lyncoln
2026-09-20 01:00:42 -04:00
parent 6ec4f2932e
commit c4cdc92865
4 changed files with 748 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_npm_adapter.py
+438
View File
@@ -1161,6 +1161,444 @@ class NpmAdapterTests(unittest.TestCase):
target_client.upload_certificate.assert_not_called()
target_client.delete_certificate.assert_not_called()
def test_proxy_host_create_uses_resolved_target_certificate_id(self):
client = Mock()
client.create_proxy_host.return_value = {
"id": 120,
}
client.proxy_host.return_value = {
"id": 120,
"domain_names": ["app.example.com"],
"forward_scheme": "http",
"forward_host": "backend.internal",
"forward_port": 8080,
"certificate_id": 84,
"ssl_forced": True,
"caching_enabled": False,
"block_exploits": True,
"allow_websocket_upgrade": True,
"http2_support": True,
"hsts_enabled": True,
"hsts_subdomains": False,
"enabled": True,
}
adapter = self.adapter(client)
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,
"ssl_forced": True,
"caching_enabled": False,
"block_exploits": True,
"allow_websocket_upgrade": True,
"http2_support": True,
"hsts_enabled": True,
"hsts_subdomains": False,
"enabled": True,
},
references=(
ObjectReference(
relation="certificate",
object_type=CERTIFICATES,
local_key="84",
),
),
)
result = adapter.apply(
PlanAction(
endpoint_id=adapter.endpoint.identity,
operation=PlanOperation.CREATE,
desired=desired,
)
)
client.create_proxy_host.assert_called_once_with(
{
"domain_names": ["app.example.com"],
"forward_scheme": "http",
"forward_host": "backend.internal",
"forward_port": 8080,
"certificate_id": 84,
"ssl_forced": True,
"caching_enabled": False,
"block_exploits": True,
"allow_websocket_upgrade": True,
"http2_support": True,
"hsts_enabled": True,
"hsts_subdomains": False,
"enabled": True,
}
)
self.assertEqual(
result.object.local_key,
"120",
)
self.assertEqual(
result.object.references,
desired.references,
)
client.delete_proxy_host.assert_not_called()
def test_proxy_host_create_without_certificate_uses_zero(self):
client = Mock()
client.create_proxy_host.return_value = {
"id": 120,
}
client.proxy_host.return_value = {
"id": 120,
"domain_names": ["plain.example.com"],
"forward_scheme": "http",
"forward_host": "backend.internal",
"forward_port": 8080,
"certificate_id": 0,
"enabled": True,
}
adapter = self.adapter(client)
desired = normalize_discovered_object(
adapter._proxy_host_from_row(
client.proxy_host.return_value
)
)
desired = NormalizedObject(
object_type=desired.object_type,
logical_key=desired.logical_key,
payload=desired.payload,
references=desired.references,
)
result = adapter.apply(
PlanAction(
endpoint_id=adapter.endpoint.identity,
operation=PlanOperation.CREATE,
desired=desired,
)
)
payload = client.create_proxy_host.call_args.kwargs
if not payload:
payload = client.create_proxy_host.call_args.args[0]
self.assertEqual(
payload["certificate_id"],
0,
)
self.assertEqual(
result.object.references,
(),
)
def test_proxy_host_create_rolls_back_when_refetch_mismatches(self):
client = Mock()
client.create_proxy_host.return_value = {
"id": 120,
}
client.proxy_host.return_value = {
"id": 120,
"domain_names": ["wrong.example.com"],
"forward_scheme": "http",
"forward_host": "backend.internal",
"forward_port": 8080,
"certificate_id": 0,
"enabled": True,
}
adapter = self.adapter(client)
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,
"ssl_forced": False,
"caching_enabled": False,
"block_exploits": False,
"allow_websocket_upgrade": False,
"http2_support": False,
"hsts_enabled": False,
"hsts_subdomains": False,
"enabled": True,
},
)
with self.assertRaisesRegex(
RuntimeError,
"does not match desired",
):
adapter.apply(
PlanAction(
endpoint_id=adapter.endpoint.identity,
operation=PlanOperation.CREATE,
desired=desired,
)
)
client.delete_proxy_host.assert_called_once_with(
"120"
)
def test_proxy_host_create_reports_rollback_failure(self):
client = Mock()
client.create_proxy_host.return_value = {
"id": 120,
}
client.proxy_host.side_effect = RuntimeError(
"refetch failed"
)
client.delete_proxy_host.side_effect = RuntimeError(
"delete failed"
)
adapter = self.adapter(client)
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,
},
)
with self.assertRaisesRegex(
RuntimeError,
"rollback also failed",
):
adapter.apply(
PlanAction(
endpoint_id=adapter.endpoint.identity,
operation=PlanOperation.CREATE,
desired=desired,
)
)
def test_proxy_host_update_uses_current_local_id_and_resolved_reference(self):
client = Mock()
client.proxy_host.return_value = {
"id": 12,
"domain_names": ["app.example.com"],
"forward_scheme": "https",
"forward_host": "new-backend.internal",
"forward_port": 8443,
"certificate_id": 84,
"ssl_forced": True,
"enabled": True,
}
adapter = self.adapter(client)
current = NormalizedObject(
object_type=PROXY_HOSTS,
logical_key="app.example.com",
local_key="12",
payload={
"domain_names": ["app.example.com"],
"forward_scheme": "http",
"forward_host": "old-backend.internal",
"forward_port": 8080,
"ssl_forced": False,
"caching_enabled": False,
"block_exploits": False,
"allow_websocket_upgrade": False,
"http2_support": False,
"hsts_enabled": False,
"hsts_subdomains": False,
"enabled": True,
},
)
actual = normalize_discovered_object(
adapter._proxy_host_from_row(
client.proxy_host.return_value
)
)
desired = NormalizedObject(
object_type=actual.object_type,
logical_key=actual.logical_key,
payload=actual.payload,
references=actual.references,
)
result = adapter.apply(
PlanAction(
endpoint_id=adapter.endpoint.identity,
operation=PlanOperation.UPDATE,
desired=desired,
current=current,
)
)
client.update_proxy_host.assert_called_once()
args = client.update_proxy_host.call_args.args
self.assertEqual(args[0], "12")
self.assertEqual(
args[1]["certificate_id"],
84,
)
self.assertEqual(
result.object.local_key,
"12",
)
self.assertEqual(
result.object.references,
desired.references,
)
client.delete_proxy_host.assert_not_called()
def test_proxy_host_update_fails_when_target_does_not_converge(self):
client = Mock()
client.proxy_host.return_value = {
"id": 12,
"domain_names": ["app.example.com"],
"forward_scheme": "http",
"forward_host": "still-old.internal",
"forward_port": 8080,
"certificate_id": 0,
"enabled": True,
}
adapter = self.adapter(client)
current = normalize_discovered_object(
adapter._proxy_host_from_row(
client.proxy_host.return_value
)
)
desired = NormalizedObject(
object_type=PROXY_HOSTS,
logical_key="app.example.com",
payload={
"domain_names": ["app.example.com"],
"forward_scheme": "http",
"forward_host": "new.internal",
"forward_port": 8080,
"ssl_forced": False,
"caching_enabled": False,
"block_exploits": False,
"allow_websocket_upgrade": False,
"http2_support": False,
"hsts_enabled": False,
"hsts_subdomains": False,
"enabled": True,
},
)
with self.assertRaisesRegex(
RuntimeError,
"does not match desired",
):
adapter.apply(
PlanAction(
endpoint_id=adapter.endpoint.identity,
operation=PlanOperation.UPDATE,
desired=desired,
current=current,
)
)
client.delete_proxy_host.assert_not_called()
def test_proxy_host_apply_rejects_unresolved_certificate_reference(self):
client = Mock()
adapter = self.adapter(client)
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",
),
),
)
with self.assertRaisesRegex(
RuntimeError,
"not resolved",
):
adapter.apply(
PlanAction(
endpoint_id=adapter.endpoint.identity,
operation=PlanOperation.CREATE,
desired=desired,
)
)
client.create_proxy_host.assert_not_called()
client.update_proxy_host.assert_not_called()
def test_proxy_host_apply_rejects_unsupported_reference(self):
client = Mock()
adapter = self.adapter(client)
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="other",
object_type=CERTIFICATES,
local_key="84",
),
),
)
with self.assertRaisesRegex(
RuntimeError,
"unsupported object reference",
):
adapter.apply(
PlanAction(
endpoint_id=adapter.endpoint.identity,
operation=PlanOperation.CREATE,
desired=desired,
)
)
client.create_proxy_host.assert_not_called()
def test_unsupported_operations_fail_explicitly(self):
adapter = self.adapter(Mock())
tests/test_npm_client.py
+13
View File
@@ -186,6 +186,19 @@ class NpmClientMutationTests(unittest.TestCase):
)
self.assertEqual(result["id"], 12)
def test_deletes_proxy_host_by_endpoint_local_id(self):
client = self.client()
client.request = Mock()
result = client.delete_proxy_host("12")
self.assertIsNone(result)
client.request.assert_called_once_with(
"DELETE",
"/nginx/proxy-hosts/12",
expected_status=200,
)
def test_gets_one_certificate_by_endpoint_local_id(self):
client = self.client()
client.get_json = Mock(