235 lines
12 KiB
Python
235 lines
12 KiB
Python
import fcntl
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
spec = importlib.util.spec_from_file_location("ec_update_integration", Path(__file__).resolve().parents[1] / "ec_agent.py")
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
|
|
|
|
class UpdateIntegrationTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temporary = tempfile.TemporaryDirectory()
|
|
self.base = Path(self.temporary.name)
|
|
self.config_dir = self.base / "config"
|
|
self.config_dir.mkdir()
|
|
replacements = {
|
|
"AGENT_CONFIG": self.config_dir / "agent.json",
|
|
"UPDATE_JOURNAL": self.config_dir / "update" / "journal.json",
|
|
"UPDATE_STATUS": self.config_dir / "update-status.json",
|
|
"AGENT_HEALTH": self.config_dir / "agent-health.json",
|
|
}
|
|
self.patches = [patch.object(module, key, value) for key, value in replacements.items()]
|
|
self.patches.append(patch.object(module, "protected_file", side_effect=lambda path, **_: path))
|
|
for entry in self.patches:
|
|
entry.start()
|
|
self.agent = module.Agent({"root": str(self.base / "data"), "configPath": str(module.AGENT_CONFIG), "controller": "https://controller.example", "hostId": "one", "credential": "secret"})
|
|
self.agent.can_update = lambda: True
|
|
self.agent.inventory = lambda: {"agentVersion": module.VERSION}
|
|
|
|
def tearDown(self):
|
|
self.agent.journal.close()
|
|
for entry in reversed(self.patches):
|
|
entry.stop()
|
|
self.temporary.cleanup()
|
|
|
|
def assert_activity_held(self):
|
|
with (self.agent.root / "agent" / "activity.lock").open("a") as stream:
|
|
with self.assertRaises(BlockingIOError):
|
|
fcntl.flock(stream, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
|
|
def test_execution_and_reports_exclude_updates_but_long_poll_does_not(self):
|
|
operation = {"id": "one", "kind": "runtime.pull", "payload": {}}
|
|
calls = []
|
|
def request(method, path, body=None):
|
|
if method != "GET":
|
|
self.assert_activity_held()
|
|
else:
|
|
with (self.agent.root / "agent" / "activity.lock").open("a") as stream:
|
|
fcntl.flock(stream, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
calls.append((method, path))
|
|
return {"operations": [operation]} if method == "GET" else {}
|
|
def dispatch(value):
|
|
self.assert_activity_held()
|
|
return {"ok": True}
|
|
self.agent.request = request
|
|
self.agent.dispatch = dispatch
|
|
self.agent.poll_operations()
|
|
self.assertEqual(self.agent.journal.get("one")["state"], "completed")
|
|
self.assertEqual(len(calls), 4)
|
|
self.assertIsNone(self.agent.current_operation)
|
|
with self.agent.activity() as acquired:
|
|
self.assertTrue(acquired)
|
|
|
|
def test_updater_lock_defers_delivery_without_blocking_health(self):
|
|
self.agent.request = lambda method, path, body=None: {"host": {}}
|
|
with self.agent.activity() as acquired:
|
|
self.assertTrue(acquired)
|
|
with patch.object(self.agent, "request", wraps=self.agent.request) as request:
|
|
self.agent.poll_operations()
|
|
request.assert_not_called()
|
|
self.agent.heartbeat()
|
|
self.assertTrue(module.AGENT_HEALTH.exists())
|
|
|
|
def test_journal_blocks_operations_during_verification_and_recovery(self):
|
|
module.UPDATE_JOURNAL.parent.mkdir()
|
|
for content in ('{"id":"candidate","phase":"verifying"}', '{"id":"candidate","phase":"rollback"}', '{broken'):
|
|
module.UPDATE_JOURNAL.write_text(content)
|
|
with patch.object(self.agent, "request") as request:
|
|
self.agent.poll_operations()
|
|
request.assert_not_called()
|
|
module.UPDATE_JOURNAL.unlink()
|
|
with patch.object(self.agent, "request", return_value={"operations": []}) as request:
|
|
self.agent.poll_operations()
|
|
request.assert_called_once_with("GET", "/api/agent/operations?lane=write&wait=20")
|
|
|
|
def test_read_lane_completes_while_mutation_is_busy_and_excludes_updater(self):
|
|
read_done = threading.Event()
|
|
operation = {"id": "read-one", "kind": "host.inspect", "payload": {}}
|
|
self.agent.request = lambda method, path, body=None: {"operations": [operation]} if method == "GET" else {}
|
|
def dispatch(value):
|
|
self.assert_activity_held()
|
|
read_done.set()
|
|
self.agent.stopping.set()
|
|
return {"ok": True}
|
|
self.agent.dispatch = dispatch
|
|
with self.agent.activity() as acquired:
|
|
self.assertTrue(acquired)
|
|
reader = threading.Thread(target=self.agent.read_loop)
|
|
reader.start()
|
|
self.assertTrue(read_done.wait(2), "Read waited behind unrelated mutation activity")
|
|
reader.join(2)
|
|
self.assertFalse(reader.is_alive())
|
|
self.assertEqual(self.agent.journal.get("read-one")["state"], "completed")
|
|
self.assertIsNone(self.agent.executing_id)
|
|
|
|
def test_read_lane_never_dispatches_mutations(self):
|
|
def request(method, path, body=None):
|
|
self.agent.stopping.set()
|
|
return {"operations": [{"id": "bad-read", "kind": "instance.remove", "payload": {}}]}
|
|
self.agent.request = request
|
|
with patch.object(self.agent, "dispatch") as dispatch:
|
|
self.agent.read_loop()
|
|
dispatch.assert_not_called()
|
|
|
|
def test_health_lane_runs_while_file_receipt_for_same_instance_is_slow(self):
|
|
receipt_waiting, release_receipt, health_done = threading.Event(), threading.Event(), threading.Event()
|
|
file_op = {"id": "file-one", "kind": "files.read", "instanceId": "same", "payload": {}}
|
|
health_op = {"id": "health-one", "kind": "instance.inspect", "instanceId": "same", "payload": {}}
|
|
errors = []
|
|
def request(method, path, body=None):
|
|
if method == "GET":
|
|
return {"operations": [health_op if "lane=health" in path else file_op]}
|
|
if path.endswith("file-one") and body["state"] == "completed":
|
|
receipt_waiting.set()
|
|
release_receipt.wait(3)
|
|
return {}
|
|
def dispatch(operation):
|
|
if operation["id"] == "health-one":
|
|
health_done.set()
|
|
self.agent.stopping.set()
|
|
return {"ok": True}
|
|
def run(lane):
|
|
try: self.agent.read_loop(lane)
|
|
except Exception as error: errors.append(error)
|
|
self.agent.request, self.agent.dispatch = request, dispatch
|
|
files = threading.Thread(target=run, args=("read",))
|
|
health = threading.Thread(target=run, args=("health",))
|
|
files.start()
|
|
try:
|
|
self.assertTrue(receipt_waiting.wait(2))
|
|
health.start()
|
|
self.assertTrue(health_done.wait(2), "Health waited for a file receipt")
|
|
finally:
|
|
self.agent.stopping.set()
|
|
release_receipt.set()
|
|
files.join(3)
|
|
if health.ident is not None: health.join(3)
|
|
self.assertFalse(files.is_alive())
|
|
self.assertFalse(health.is_alive())
|
|
self.assertEqual(errors, [])
|
|
self.assertEqual(self.agent.journal.get("health-one")["state"], "completed")
|
|
|
|
def test_instance_mutations_and_global_mutations_exclude_observations(self):
|
|
operation = {"instanceId": "one"}
|
|
entered = threading.Event()
|
|
release = threading.Event()
|
|
def read():
|
|
with self.agent.read_gate.read(), self.agent.instance_lock(operation):
|
|
entered.set()
|
|
release.wait(2)
|
|
with self.agent.instance_lock(operation):
|
|
reader = threading.Thread(target=read)
|
|
reader.start()
|
|
self.assertFalse(entered.wait(0.05))
|
|
self.assertTrue(entered.wait(2))
|
|
writer_entered = threading.Event()
|
|
def write():
|
|
with self.agent.read_gate.write():
|
|
writer_entered.set()
|
|
writer = threading.Thread(target=write)
|
|
writer.start()
|
|
self.assertFalse(writer_entered.wait(0.05))
|
|
release.set()
|
|
reader.join(2)
|
|
writer.join(2)
|
|
self.assertTrue(writer_entered.is_set())
|
|
|
|
def test_failed_controller_contact_never_writes_healthy_receipt(self):
|
|
with patch.object(self.agent, "request", side_effect=module.AgentError("offline")):
|
|
with self.assertRaises(module.AgentError):
|
|
self.agent.heartbeat()
|
|
self.assertFalse(module.AGENT_HEALTH.exists())
|
|
|
|
def test_health_proves_current_process_version_and_transaction(self):
|
|
module.atomic_json(module.UPDATE_JOURNAL, {"id": "candidate-id", "phase": "verifying"})
|
|
with patch.object(self.agent, "request", return_value={"host": {}}), patch.object(module.time, "time", return_value=12345.5):
|
|
self.agent.heartbeat()
|
|
self.assertEqual(json.loads(module.AGENT_HEALTH.read_text()), {"version": module.VERSION, "pid": os.getpid(), "checkedAt": 12345.5, "updateId": "candidate-id"})
|
|
module.UPDATE_JOURNAL.unlink()
|
|
self.agent.record_health()
|
|
self.assertNotIn("updateId", json.loads(module.AGENT_HEALTH.read_text()))
|
|
|
|
def test_real_agent_health_receipt_is_accepted_by_the_independent_updater(self):
|
|
updater_spec = importlib.util.spec_from_file_location("ec_update_contract", Path(__file__).resolve().parents[1] / "ec_agent_update.py")
|
|
updater_module = importlib.util.module_from_spec(updater_spec)
|
|
updater_spec.loader.exec_module(updater_module)
|
|
journal = {"id": "candidate-id", "phase": "verifying", "targetVersion": module.VERSION, "switchedAt": module.time.time() - 1}
|
|
module.atomic_json(module.UPDATE_JOURNAL, journal)
|
|
with patch.object(self.agent, "request", return_value={"host": {}}):
|
|
self.agent.heartbeat()
|
|
updater = updater_module.Updater()
|
|
with patch.object(updater, "check_uninstall"), patch.object(updater_module, "AGENT_HEALTH", module.AGENT_HEALTH), patch.object(updater_module, "trusted", side_effect=lambda path, **_: path), patch.object(updater_module, "service_state", return_value=(True, os.getpid())):
|
|
self.assertTrue(updater.wait_healthy(journal))
|
|
|
|
def test_operation_exception_releases_update_lock(self):
|
|
with patch.object(self.agent, "request", side_effect=module.AgentError("offline")):
|
|
with self.assertRaises(module.AgentError):
|
|
self.agent.poll_operations()
|
|
with self.agent.activity() as acquired:
|
|
self.assertTrue(acquired)
|
|
|
|
def test_update_inventory_excludes_local_recovery_and_secret_fields(self):
|
|
module.atomic_json(module.UPDATE_STATUS, {"state": "failed", "targetVersion": "1.3.0", "currentVersion": "untrusted", "error": "Restored previous agent", "checkedAt": 1700000000, "credential": "secret", "journal": {"private": True}, "failedSha256": "digest"})
|
|
self.assertEqual(self.agent.update_inventory(), {"enabled": True, "state": "failed", "currentVersion": module.VERSION, "targetVersion": "1.3.0", "error": "Restored previous agent", "checkedAt": "2023-11-14T22:13:20Z"})
|
|
|
|
def test_malformed_update_status_is_visible_without_breaking_inventory(self):
|
|
module.UPDATE_STATUS.write_text("[]")
|
|
self.assertEqual(self.agent.update_inventory()["state"], "failed")
|
|
|
|
def test_shutdown_never_polls_work(self):
|
|
self.agent.stopping.set()
|
|
with patch.object(self.agent, "request") as request:
|
|
self.agent.poll_operations()
|
|
request.assert_not_called()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|