Remove obsolete K3s Ansible tests
这个提交包含在：
admin 于
1 父节点
0d00596402
当前提交
1029f26163
2 个文件已更改
-143
没有与搜索条件相匹配的文件
@@ -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)
|
||||
@@ -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()
|
||||
在新议题中引用
屏蔽一个用户