From 1029f26163acc3e299d4a996329163d419019c28 Mon Sep 17 00:00:00 2001 From: rohow Date: Fri, 16 Oct 2026 14:38:40 +0800 Subject: [PATCH] Remove obsolete K3s Ansible tests --- ansible/tests/test_control_plane_relay.py | 56 --------------- ansible/tests/test_readiness.py | 87 ----------------------- 2 files changed, 143 deletions(-) delete mode 100644 ansible/tests/test_control_plane_relay.py delete mode 100644 ansible/tests/test_readiness.py diff --git a/ansible/tests/test_control_plane_relay.py b/ansible/tests/test_control_plane_relay.py deleted file mode 100644 index 0a8219d..0000000 --- a/ansible/tests/test_control_plane_relay.py +++ /dev/null @@ -1,56 +0,0 @@ -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) diff --git a/ansible/tests/test_readiness.py b/ansible/tests/test_readiness.py deleted file mode 100644 index 64e4939..0000000 --- a/ansible/tests/test_readiness.py +++ /dev/null @@ -1,87 +0,0 @@ -import json -import unittest -from pathlib import Path - -import yaml -from ansible.plugins.filter.core import combine -from jinja2 import Environment, StrictUndefined - - -class ReadinessTest(unittest.TestCase): - @classmethod - def setUpClass(cls): - tasks = yaml.safe_load( - (Path(__file__).parents[1] / "roles/k3s/tasks/main.yml").read_text() - ) - env = Environment(undefined=StrictUndefined) - env.filters["from_json"] = json.loads - env.filters["combine"] = combine - cls.env = env - cls.tasks = tasks - cls.conditions = { - task["register"]: env.compile_expression(task["until"]) - for task in tasks - if "until" in task - } - - def test_node_must_be_ready_at_the_requested_version(self): - condition = self.conditions["k3s_node_status"] - for version, ready, expected in [ - ("v1.35.9+k3s1", "True", True), - ("v1.35.6+k3s1", "True", False), - ("v1.35.9+k3s1", "False", False), - ]: - with self.subTest(version=version, ready=ready): - status = {"status": { - "nodeInfo": {"kubeletVersion": version}, - "conditions": [{"type": "Ready", "status": ready}], - }} - self.assertEqual(condition( - k3s_node_status={"rc": 0, "stdout": json.dumps(status)}, - k3s_version="v1.35.9+k3s1", - ), expected) - self.assertFalse(condition( - k3s_node_status={"rc": 1, "stdout": ""}, - k3s_version="v1.35.9+k3s1", - )) - - def test_api_requires_a_successful_readyz_response(self): - condition = self.conditions["k3s_api_ready"] - for rc, stdout, expected in [(0, "ok", True), (1, "ok", False), (0, "", False)]: - with self.subTest(rc=rc, stdout=stdout): - self.assertEqual(condition(k3s_api_ready={"rc": rc, "stdout": stdout}), expected) - - def test_mirror_changes_restart_existing_nodes_without_a_version_upgrade(self): - task = next(t for t in self.tasks if t["name"] == "Restart K3s after updating registry mirrors") - condition = self.env.compile_expression(" and ".join(f"({c})" for c in task["when"])) - for changed, exists, current, expected in [ - (True, True, "v1.35.9+k3s1", True), - (False, True, "v1.35.9+k3s1", False), - (True, True, "v1.35.6+k3s1", False), - (True, False, "", False), - ]: - with self.subTest(changed=changed, exists=exists, current=current): - self.assertEqual(condition( - k3s_registries_config={"changed": changed}, - k3s_binary={"stat": {"exists": exists}}, - k3s_current_version={"stdout": current} if exists else {}, - k3s_version="v1.35.9+k3s1", - ), expected) - - def test_host_mirror_override_preserves_other_registries(self): - template = self.env.from_string( - (Path(__file__).parents[1] / "roles/k3s/templates/registries.yaml.j2").read_text() - ) - mirrors = {"docker.io": ["docker.1ms.run", "docker.m.daocloud.io"], "ghcr.io": ["ghcr.m.daocloud.io"]} - for override in [{}, {"docker.io": list(reversed(mirrors["docker.io"]))}]: - result = yaml.safe_load(template.render( - registry_mirrors=mirrors, registry_mirror_overrides=override, - ))["mirrors"] - self.assertEqual(result["ghcr.io"]["endpoint"], ["https://ghcr.m.daocloud.io"]) - self.assertEqual(result["docker.io"]["endpoint"], [ - "https://" + host for host in override.get("docker.io", mirrors["docker.io"]) - ]) - - -if __name__ == "__main__": - unittest.main()