57 行
2.3 KiB
Python
57 行
2.3 KiB
Python
import importlib.util
|
|
from pathlib import Path
|
|
import socket
|
|
import subprocess
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
|
|
spec = importlib.util.spec_from_file_location(
|
|
"relay", Path(__file__).parents[1] / "roles/common/files/k3s-control-plane-relay.py"
|
|
)
|
|
relay = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(relay)
|
|
BASE = (Path(__file__).parents[1] / "roles/common/templates/control-plane-relay.nft.j2").read_text()
|
|
|
|
|
|
class ControlPlaneRelayTest(unittest.TestCase):
|
|
@patch.object(relay.subprocess, "run")
|
|
@patch.object(relay.socket, "getaddrinfo")
|
|
def test_all_dns_addresses_are_deduplicated_and_checked_before_apply(self, resolve, run):
|
|
resolve.return_value = [
|
|
(socket.AF_INET, 0, 0, "", ("192.0.2.1", 0)),
|
|
(socket.AF_INET6, 0, 0, "", ("2001:db8::1", 0, 0, 0)),
|
|
(socket.AF_INET, 0, 0, "", ("192.0.2.1", 0)),
|
|
]
|
|
relay.apply_rules(["tthk.node.dev.cm"], BASE)
|
|
rules = run.call_args_list[0].kwargs["input"]
|
|
self.assertEqual(rules.count("peers4 { 192.0.2.1 }"), 1)
|
|
self.assertIn("peers6 { 2001:db8::1 }", rules)
|
|
self.assertIn("--check", run.call_args_list[0].args[0])
|
|
self.assertEqual(run.call_args_list[1].kwargs["input"], rules)
|
|
|
|
@patch.object(relay.subprocess, "run")
|
|
@patch.object(relay.socket, "getaddrinfo")
|
|
def test_any_dns_failure_preserves_existing_rules(self, resolve, run):
|
|
resolve.side_effect = [
|
|
[(socket.AF_INET, 0, 0, "", ("192.0.2.1", 0))],
|
|
socket.gaierror("temporary failure"),
|
|
]
|
|
with self.assertRaises(socket.gaierror):
|
|
relay.apply_rules(["tthk.node.dev.cm", "other.node.dev.cm"], BASE)
|
|
run.assert_not_called()
|
|
|
|
@patch.object(relay.subprocess, "run", side_effect=subprocess.CalledProcessError(1, "nft"))
|
|
def test_invalid_rules_are_not_applied(self, run):
|
|
with self.assertRaises(subprocess.CalledProcessError):
|
|
relay.apply_rules([], BASE)
|
|
self.assertEqual(run.call_count, 1)
|
|
|
|
@patch.object(relay.subprocess, "run")
|
|
def test_empty_list_clears_only_the_owned_table(self, run):
|
|
relay.apply_rules([], BASE)
|
|
rules = run.call_args.kwargs["input"]
|
|
self.assertIn("delete table inet k3s_control_plane_relay", rules)
|
|
self.assertNotIn("add element", rules)
|
|
self.assertNotIn("flush ruleset", rules)
|