388 lines
12 KiB
Python
388 lines
12 KiB
Python
import io
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
from contextlib import redirect_stderr, redirect_stdout
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
from dns_sync.adapters import (
|
|
DNS_RECORDS,
|
|
PROXY_HOSTS,
|
|
Capability,
|
|
CapabilityOperation,
|
|
DiscoveredObject,
|
|
Endpoint,
|
|
)
|
|
from dns_sync.cli import build_parser, run_reconcile
|
|
|
|
|
|
class FakeAdapter:
|
|
def __init__(
|
|
self,
|
|
endpoint,
|
|
capabilities,
|
|
discovered,
|
|
):
|
|
self.endpoint = endpoint
|
|
self.capabilities = capabilities
|
|
self.discovered = discovered
|
|
self.apply_count = 0
|
|
self.verify_count = 0
|
|
|
|
def connect(self):
|
|
pass
|
|
|
|
def discover(self, object_type):
|
|
return list(self.discovered.get(object_type, []))
|
|
|
|
def apply(self, action):
|
|
self.apply_count += 1
|
|
raise AssertionError("read-only CLI called apply")
|
|
|
|
def verify(self, object):
|
|
self.verify_count += 1
|
|
raise AssertionError("read-only CLI called verify")
|
|
|
|
|
|
class ReconcileCliTests(unittest.TestCase):
|
|
def test_parser_exposes_read_only_reconcile_command(self):
|
|
args = build_parser().parse_args(
|
|
[
|
|
"reconcile",
|
|
"--state-db",
|
|
"/tmp/test-state.sqlite3",
|
|
"--include-disabled",
|
|
]
|
|
)
|
|
|
|
self.assertEqual(args.command, "reconcile")
|
|
self.assertEqual(
|
|
args.state_db,
|
|
"/tmp/test-state.sqlite3",
|
|
)
|
|
self.assertTrue(args.include_disabled)
|
|
self.assertFalse(args.apply)
|
|
self.assertIsNone(args.max_changes)
|
|
|
|
apply_args = build_parser().parse_args(
|
|
[
|
|
"reconcile",
|
|
"--apply",
|
|
"--max-changes",
|
|
"0",
|
|
]
|
|
)
|
|
self.assertTrue(apply_args.apply)
|
|
self.assertEqual(apply_args.max_changes, 0)
|
|
|
|
with redirect_stderr(io.StringIO()):
|
|
with self.assertRaises(SystemExit):
|
|
build_parser().parse_args(
|
|
["reconcile", "--max-changes", "-1"]
|
|
)
|
|
|
|
def test_reconcile_builds_plan_without_applying_it(self):
|
|
npm_endpoint = Endpoint("npm", "192.168.1.3:81")
|
|
verizon_endpoint = Endpoint(
|
|
"verizon",
|
|
"192.168.1.1:443",
|
|
)
|
|
npm = FakeAdapter(
|
|
npm_endpoint,
|
|
frozenset(
|
|
{
|
|
Capability(
|
|
PROXY_HOSTS,
|
|
CapabilityOperation.DISCOVER,
|
|
)
|
|
}
|
|
),
|
|
{
|
|
PROXY_HOSTS: [
|
|
DiscoveredObject(
|
|
object_type=PROXY_HOSTS,
|
|
local_key="12",
|
|
payload={
|
|
"domain_names": [
|
|
"app.example.com"
|
|
],
|
|
"forward_scheme": "http",
|
|
"forward_host": "192.168.1.50",
|
|
"forward_port": 8080,
|
|
"enabled": True,
|
|
},
|
|
)
|
|
]
|
|
},
|
|
)
|
|
verizon = FakeAdapter(
|
|
verizon_endpoint,
|
|
frozenset(
|
|
{
|
|
Capability(
|
|
DNS_RECORDS,
|
|
CapabilityOperation.DISCOVER,
|
|
),
|
|
Capability(
|
|
DNS_RECORDS,
|
|
CapabilityOperation.APPLY,
|
|
),
|
|
}
|
|
),
|
|
{
|
|
DNS_RECORDS: [
|
|
DiscoveredObject(
|
|
object_type=DNS_RECORDS,
|
|
local_key="41",
|
|
payload={
|
|
"hostname": "app.example.com",
|
|
"ip": "192.168.1.50",
|
|
"record_type": "A",
|
|
},
|
|
)
|
|
]
|
|
},
|
|
)
|
|
adapters = {
|
|
npm_endpoint.identity: npm,
|
|
verizon_endpoint.identity: verizon,
|
|
}
|
|
|
|
def loader(name, config, endpoint):
|
|
return adapters[endpoint.identity]
|
|
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
state_path = Path(directory) / "state.sqlite3"
|
|
stdout = io.StringIO()
|
|
stderr = io.StringIO()
|
|
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"NPM_ENDPOINTS": "192.168.1.3",
|
|
"VERIZON_ENDPOINTS": "192.168.1.1",
|
|
"DNS_TARGET_IP": "192.168.1.101",
|
|
},
|
|
clear=True,
|
|
), patch(
|
|
"dns_sync.cli.load_adapter",
|
|
side_effect=loader,
|
|
), redirect_stdout(stdout), redirect_stderr(stderr):
|
|
result = run_reconcile(
|
|
SimpleNamespace(
|
|
state_db=str(state_path),
|
|
include_disabled=False,
|
|
apply=False,
|
|
max_changes=None,
|
|
)
|
|
)
|
|
|
|
self.assertTrue(state_path.exists())
|
|
|
|
output = stdout.getvalue()
|
|
self.assertEqual(result, 0)
|
|
self.assertEqual(stderr.getvalue(), "")
|
|
self.assertIn(
|
|
"Generic reconciliation mode: READ ONLY",
|
|
output,
|
|
)
|
|
self.assertIn(
|
|
"UPDATE dns_records app.example.com "
|
|
"-> 192.168.1.101",
|
|
output,
|
|
)
|
|
self.assertIn(
|
|
"READ ONLY: no endpoint changes were applied.",
|
|
output,
|
|
)
|
|
self.assertEqual(npm.apply_count, 0)
|
|
self.assertEqual(verizon.apply_count, 0)
|
|
self.assertEqual(npm.verify_count, 0)
|
|
self.assertEqual(verizon.verify_count, 0)
|
|
|
|
|
|
def test_apply_flag_invokes_verified_executor(self):
|
|
config = SimpleNamespace(
|
|
target_ip="192.168.1.101",
|
|
include_disabled=False,
|
|
domain_matching=False,
|
|
domain=None,
|
|
)
|
|
adapter = object()
|
|
reconciliation = SimpleNamespace(
|
|
observations=(),
|
|
desired=(),
|
|
plans=(),
|
|
failures=(),
|
|
)
|
|
execution = SimpleNamespace(
|
|
applied=(),
|
|
skipped=(),
|
|
retired=(),
|
|
)
|
|
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
state_path = Path(directory) / "state.sqlite3"
|
|
stdout = io.StringIO()
|
|
|
|
with patch(
|
|
"dns_sync.cli.AppConfig.from_env",
|
|
return_value=config,
|
|
), patch(
|
|
"dns_sync.cli._configured_adapters",
|
|
return_value=[adapter],
|
|
), patch(
|
|
"dns_sync.cli.discover_reconcile_plan",
|
|
return_value=reconciliation,
|
|
), patch(
|
|
"dns_sync.cli.execute_endpoint_plans",
|
|
return_value=execution,
|
|
) as execute, redirect_stdout(stdout):
|
|
result = run_reconcile(
|
|
SimpleNamespace(
|
|
state_db=str(state_path),
|
|
include_disabled=False,
|
|
apply=True,
|
|
max_changes=None,
|
|
)
|
|
)
|
|
|
|
self.assertEqual(result, 0)
|
|
execute.assert_called_once()
|
|
self.assertIn(
|
|
"Generic reconciliation mode: APPLY",
|
|
stdout.getvalue(),
|
|
)
|
|
self.assertIn(
|
|
"APPLY COMPLETE: 0 action(s) applied and verified; "
|
|
"0 retirement(s) completed",
|
|
stdout.getvalue(),
|
|
)
|
|
|
|
def test_apply_budget_counts_planned_retirement_before_mutation(
|
|
self,
|
|
):
|
|
config = SimpleNamespace(
|
|
target_ip="192.168.1.101",
|
|
include_disabled=False,
|
|
domain_matching=False,
|
|
domain=None,
|
|
)
|
|
adapter = object()
|
|
|
|
retirement = SimpleNamespace(
|
|
object=SimpleNamespace(
|
|
object_type="certificates",
|
|
logical_key="app.example.com",
|
|
local_key="63",
|
|
)
|
|
)
|
|
plan = SimpleNamespace(
|
|
actions=(),
|
|
retirements=(retirement,),
|
|
)
|
|
reconciliation = SimpleNamespace(
|
|
observations=(),
|
|
desired=(),
|
|
plans=(plan,),
|
|
failures=(),
|
|
)
|
|
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
state_path = Path(directory) / "state.sqlite3"
|
|
|
|
with patch(
|
|
"dns_sync.cli.AppConfig.from_env",
|
|
return_value=config,
|
|
), patch(
|
|
"dns_sync.cli._configured_adapters",
|
|
return_value=[adapter],
|
|
), patch(
|
|
"dns_sync.cli.discover_reconcile_plan",
|
|
return_value=reconciliation,
|
|
), patch(
|
|
"dns_sync.cli.print_endpoint_plans",
|
|
), patch(
|
|
"dns_sync.cli.execute_endpoint_plans",
|
|
) as execute, redirect_stdout(io.StringIO()):
|
|
with self.assertRaisesRegex(
|
|
RuntimeError,
|
|
(
|
|
r"Refusing to apply 1 mutation\(s\); "
|
|
r"--max-changes allows 0"
|
|
),
|
|
):
|
|
run_reconcile(
|
|
SimpleNamespace(
|
|
state_db=str(state_path),
|
|
include_disabled=False,
|
|
apply=True,
|
|
max_changes=0,
|
|
)
|
|
)
|
|
|
|
execute.assert_not_called()
|
|
|
|
def test_apply_budget_blocks_executor_before_mutation(self):
|
|
config = SimpleNamespace(
|
|
target_ip="192.168.1.101",
|
|
include_disabled=False,
|
|
domain_matching=False,
|
|
domain=None,
|
|
)
|
|
adapter = object()
|
|
action = SimpleNamespace(
|
|
operation=SimpleNamespace(value="create"),
|
|
)
|
|
plan = SimpleNamespace(
|
|
actions=(action,),
|
|
retirements=(),
|
|
)
|
|
reconciliation = SimpleNamespace(
|
|
observations=(),
|
|
desired=(),
|
|
plans=(plan,),
|
|
failures=(),
|
|
)
|
|
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
state_path = Path(directory) / "state.sqlite3"
|
|
|
|
with patch(
|
|
"dns_sync.cli.AppConfig.from_env",
|
|
return_value=config,
|
|
), patch(
|
|
"dns_sync.cli._configured_adapters",
|
|
return_value=[adapter],
|
|
), patch(
|
|
"dns_sync.cli.discover_reconcile_plan",
|
|
return_value=reconciliation,
|
|
), patch(
|
|
"dns_sync.cli.print_endpoint_plans",
|
|
), patch(
|
|
"dns_sync.cli.execute_endpoint_plans",
|
|
) as execute, redirect_stdout(io.StringIO()):
|
|
with self.assertRaisesRegex(
|
|
RuntimeError,
|
|
(
|
|
r"Refusing to apply 1 mutation\(s\); "
|
|
r"--max-changes allows 0"
|
|
),
|
|
):
|
|
run_reconcile(
|
|
SimpleNamespace(
|
|
state_db=str(state_path),
|
|
include_disabled=False,
|
|
apply=True,
|
|
max_changes=0,
|
|
)
|
|
)
|
|
|
|
execute.assert_not_called()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|