mirror of
https://github.com/openhop-dev/openhop_repeater.git
synced 2026-10-23 13:32:50 +00:00
293 lines
12 KiB
Python
293 lines
12 KiB
Python
import copy
|
|
import ctypes
|
|
import ctypes.util
|
|
import errno
|
|
import os
|
|
import stat
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from threading import Event
|
|
from unittest.mock import Mock
|
|
|
|
import pytest
|
|
import yaml
|
|
|
|
from repeater.config_manager import ConfigManager
|
|
|
|
|
|
@pytest.mark.parametrize("failure", ["dump", "fsync", "replace"])
|
|
def test_failed_update_preserves_file_and_shared_config(tmp_path, monkeypatch, failure):
|
|
path = tmp_path / "config.yaml"
|
|
config = {"repeater": {"node_name": "before"}, "radio_type": "pymc_tcp"}
|
|
before = copy.deepcopy(config)
|
|
path.write_text(yaml.safe_dump(config))
|
|
original = path.read_bytes()
|
|
section = config["repeater"]
|
|
manager = ConfigManager(str(path), config)
|
|
live = Mock()
|
|
monkeypatch.setattr(manager, "live_update_daemon", live)
|
|
|
|
def fail(*args, **kwargs):
|
|
if failure == "dump":
|
|
args[1].write("partial: ")
|
|
assert config == before
|
|
raise OSError("synthetic write failure")
|
|
|
|
target = "yaml.safe_dump" if failure == "dump" else f"os.{failure}"
|
|
monkeypatch.setattr(f"repeater.config_manager.{target}", fail)
|
|
result = manager.update_and_save({"repeater": {"node_name": "after"}})
|
|
assert result["success"] is False
|
|
assert result["saved"] is False
|
|
assert result["live_updated"] is False
|
|
assert config == before
|
|
assert config["repeater"] is section
|
|
assert path.read_bytes() == original
|
|
assert list(tmp_path.iterdir()) == [path]
|
|
live.assert_not_called()
|
|
|
|
|
|
def test_failed_direct_save_does_not_normalize_shared_config(tmp_path, monkeypatch):
|
|
config = {"radio_type": "pymc_tcp", "pymc_tcp": {"host": "synthetic"}}
|
|
before = copy.deepcopy(config)
|
|
path = tmp_path / "config.yaml"
|
|
path.write_text(yaml.safe_dump(config))
|
|
original = path.read_bytes()
|
|
monkeypatch.setattr("repeater.config_manager.yaml.safe_dump", Mock(side_effect=OSError()))
|
|
assert not ConfigManager(str(path), config).save_to_file()
|
|
assert config == before
|
|
assert path.read_bytes() == original
|
|
assert list(tmp_path.iterdir()) == [path]
|
|
|
|
|
|
@pytest.mark.parametrize("mode", [None, 0o600, 0o640])
|
|
def test_atomic_save_permissions_and_publication(tmp_path, monkeypatch, mode):
|
|
path = tmp_path / "config.yaml"
|
|
config = {"repeater": {"node_name": "before", "other": 1}}
|
|
section = config["repeater"]
|
|
if mode is not None:
|
|
path.write_text(yaml.safe_dump(config))
|
|
path.chmod(mode)
|
|
replace = os.replace
|
|
observed = []
|
|
|
|
def checked_replace(source, destination):
|
|
assert config["repeater"]["node_name"] == "before"
|
|
assert stat.S_IMODE(os.stat(source).st_mode) == (mode or 0o600)
|
|
assert os.stat(source).st_uid == os.getuid()
|
|
observed.append(True)
|
|
replace(source, destination)
|
|
|
|
monkeypatch.setattr("repeater.config_manager.os.replace", checked_replace)
|
|
assert ConfigManager(str(path), config).update_and_save(
|
|
{"repeater": {"node_name": "after"}}, live_update=False
|
|
)["saved"]
|
|
assert observed == [True]
|
|
assert config["repeater"] is section
|
|
assert yaml.safe_load(path.read_text()) == config
|
|
assert stat.S_IMODE(path.stat().st_mode) == (mode or 0o600)
|
|
assert list(tmp_path.iterdir()) == [path]
|
|
|
|
|
|
def test_atomic_save_preserves_posix_access_acl(tmp_path):
|
|
if not hasattr(os, "getxattr") or not hasattr(os, "listxattr"):
|
|
pytest.skip("extended attributes unavailable")
|
|
library = ctypes.util.find_library("acl")
|
|
if not library:
|
|
pytest.skip("libacl unavailable")
|
|
acl = ctypes.CDLL(library, use_errno=True)
|
|
acl.acl_from_text.argtypes = [ctypes.c_char_p]
|
|
acl.acl_from_text.restype = ctypes.c_void_p
|
|
acl.acl_set_file.argtypes = [ctypes.c_char_p, ctypes.c_int, ctypes.c_void_p]
|
|
acl.acl_set_file.restype = ctypes.c_int
|
|
acl.acl_free.argtypes = [ctypes.c_void_p]
|
|
acl.acl_free.restype = ctypes.c_int
|
|
path = tmp_path / "config.yaml"
|
|
path.write_text("repeater: {node_name: before}\n")
|
|
path.chmod(0o640)
|
|
# An extra named user makes this a real extended POSIX ACL, not mode bits.
|
|
acl_ptr = acl.acl_from_text(b"u::rw-,u:12345:r--,g::r--,m::r--,o::---")
|
|
assert acl_ptr
|
|
try:
|
|
if acl.acl_set_file(os.fsencode(path), 0x8000, acl_ptr) != 0:
|
|
error = ctypes.get_errno()
|
|
if error in (errno.ENOTSUP, errno.EOPNOTSUPP):
|
|
pytest.skip("filesystem does not support POSIX ACLs")
|
|
raise OSError(error, os.strerror(error))
|
|
finally:
|
|
acl.acl_free(acl_ptr)
|
|
before_acl = os.getxattr(path, "system.posix_acl_access")
|
|
before_mode = stat.S_IMODE(path.stat().st_mode)
|
|
config = {"repeater": {"node_name": "before"}}
|
|
|
|
assert ConfigManager(str(path), config).update_and_save(
|
|
{"repeater": {"node_name": "after"}}, live_update=False
|
|
)["saved"]
|
|
assert os.getxattr(path, "system.posix_acl_access") == before_acl
|
|
assert stat.S_IMODE(path.stat().st_mode) == before_mode
|
|
assert yaml.safe_load(path.read_text()) == config
|
|
|
|
|
|
def _setup_directory_default_acl(tmp_path):
|
|
if not all(hasattr(os, name) for name in ("listxattr", "getxattr", "setxattr")):
|
|
pytest.skip("extended attributes unavailable")
|
|
library = ctypes.util.find_library("acl")
|
|
if not library:
|
|
pytest.skip("libacl unavailable")
|
|
acl = ctypes.CDLL(library, use_errno=True)
|
|
acl.acl_from_text.argtypes = [ctypes.c_char_p]
|
|
acl.acl_from_text.restype = ctypes.c_void_p
|
|
acl.acl_set_file.argtypes = [ctypes.c_char_p, ctypes.c_int, ctypes.c_void_p]
|
|
acl.acl_set_file.restype = ctypes.c_int
|
|
acl.acl_free.argtypes = [ctypes.c_void_p]
|
|
acl.acl_free.restype = ctypes.c_int
|
|
|
|
path = tmp_path / "config.yaml"
|
|
config = {"repeater": {"node_name": "before"}}
|
|
path.write_text(yaml.safe_dump(config))
|
|
path.chmod(0o640)
|
|
original_acl_names = set(os.listxattr(path))
|
|
assert "system.posix_acl_access" not in original_acl_names
|
|
original_mode = stat.S_IMODE(path.stat().st_mode)
|
|
|
|
# A default ACL is installed only after the original file exists. A new
|
|
# staging inode inherits the named reader, but the original never had one.
|
|
acl_ptr = acl.acl_from_text(b"u::rwx,u:12345:r-x,g::r-x,m::r-x,o::---")
|
|
assert acl_ptr
|
|
try:
|
|
if acl.acl_set_file(os.fsencode(tmp_path), 0x4000, acl_ptr) != 0:
|
|
error = ctypes.get_errno()
|
|
if error in (errno.ENOTSUP, errno.EOPNOTSUPP):
|
|
pytest.skip("filesystem does not support POSIX default ACLs")
|
|
raise OSError(error, os.strerror(error))
|
|
finally:
|
|
acl.acl_free(acl_ptr)
|
|
return path, config, original_acl_names, original_mode
|
|
|
|
|
|
def test_atomic_save_does_not_inherit_directory_default_acl(tmp_path):
|
|
path, config, original_acl_names, original_mode = _setup_directory_default_acl(tmp_path)
|
|
assert ConfigManager(str(path), config).update_and_save(
|
|
{"repeater": {"node_name": "after"}}, live_update=False
|
|
)["saved"]
|
|
assert set(os.listxattr(path)) == original_acl_names
|
|
assert "system.posix_acl_access" not in os.listxattr(path)
|
|
assert stat.S_IMODE(path.stat().st_mode) == original_mode
|
|
assert yaml.safe_load(path.read_text()) == config
|
|
|
|
|
|
def test_inherited_acl_removal_failure_preserves_file_and_shared_config(tmp_path, monkeypatch):
|
|
path, config, original_acl_names, _ = _setup_directory_default_acl(tmp_path)
|
|
previous_bytes = path.read_bytes()
|
|
previous_inode = path.stat().st_ino
|
|
section = config["repeater"]
|
|
manager = ConfigManager(str(path), config)
|
|
live = Mock()
|
|
monkeypatch.setattr(manager, "live_update_daemon", live)
|
|
attempted = []
|
|
|
|
def fail_removal(target, name):
|
|
assert isinstance(target, int)
|
|
assert name == "system.posix_acl_access"
|
|
attempted.append(True)
|
|
raise PermissionError(errno.EPERM, "inherited ACL removal denied")
|
|
|
|
monkeypatch.setattr("repeater.config_manager.os.removexattr", fail_removal)
|
|
result = manager.update_and_save({"repeater": {"node_name": "after"}})
|
|
assert attempted == [True]
|
|
assert result == {
|
|
"success": False,
|
|
"saved": False,
|
|
"live_updated": False,
|
|
"error": "Failed to save config to file",
|
|
}
|
|
assert path.read_bytes() == previous_bytes
|
|
assert path.stat().st_ino == previous_inode
|
|
assert set(os.listxattr(path)) == original_acl_names
|
|
assert config == {"repeater": {"node_name": "before"}}
|
|
assert config["repeater"] is section
|
|
assert list(tmp_path.iterdir()) == [path]
|
|
live.assert_not_called()
|
|
|
|
|
|
def test_xattr_copy_failure_does_not_commit_or_publish(tmp_path, monkeypatch):
|
|
if not all(hasattr(os, name) for name in ("listxattr", "getxattr", "setxattr")):
|
|
pytest.skip("extended attributes unavailable")
|
|
path = tmp_path / "config.yaml"
|
|
config = {"repeater": {"node_name": "before"}}
|
|
path.write_text(yaml.safe_dump(config))
|
|
path.chmod(0o640)
|
|
try:
|
|
os.setxattr(path, "user.review_marker", b"original")
|
|
except OSError as exc:
|
|
if exc.errno in (errno.ENOTSUP, errno.EOPNOTSUPP):
|
|
pytest.skip("filesystem does not support user extended attributes")
|
|
raise
|
|
previous_bytes = path.read_bytes()
|
|
previous_inode = path.stat().st_ino
|
|
section = config["repeater"]
|
|
manager = ConfigManager(str(path), config)
|
|
live = Mock()
|
|
monkeypatch.setattr(manager, "live_update_daemon", live)
|
|
attempted = []
|
|
|
|
def fail_staged_copy(target, name, value, *args, **kwargs):
|
|
assert isinstance(target, int) # A staging inode, never the original path.
|
|
assert name == "user.review_marker"
|
|
assert value == b"original"
|
|
assert path.read_bytes() == previous_bytes
|
|
attempted.append(True)
|
|
raise PermissionError(errno.EPERM, "metadata copy denied")
|
|
|
|
monkeypatch.setattr("repeater.config_manager.os.setxattr", fail_staged_copy)
|
|
result = manager.update_and_save({"repeater": {"node_name": "after"}})
|
|
assert attempted == [True]
|
|
assert result == {
|
|
"success": False,
|
|
"saved": False,
|
|
"live_updated": False,
|
|
"error": "Failed to save config to file",
|
|
}
|
|
assert path.read_bytes() == previous_bytes
|
|
assert path.stat().st_ino == previous_inode
|
|
assert config == {"repeater": {"node_name": "before"}}
|
|
assert config["repeater"] is section
|
|
assert os.getxattr(path, "user.review_marker") == b"original"
|
|
assert list(tmp_path.iterdir()) == [path]
|
|
live.assert_not_called()
|
|
|
|
|
|
def test_concurrent_managers_serialize_staging_and_saving(tmp_path, monkeypatch):
|
|
path = tmp_path / "config.yaml"
|
|
config = {"repeater": {"node_name": "before"}}
|
|
first = ConfigManager(str(path), config)
|
|
second = ConfigManager(str(path), config)
|
|
entered, release, second_started, second_dump = Event(), Event(), Event(), Event()
|
|
dump = yaml.safe_dump
|
|
|
|
def paused_dump(data, *args, **kwargs):
|
|
if data["repeater"].get("first") and not data["repeater"].get("second"):
|
|
entered.set()
|
|
assert release.wait(5)
|
|
else:
|
|
second_dump.set()
|
|
return dump(data, *args, **kwargs)
|
|
|
|
def second_update():
|
|
second_started.set()
|
|
return second.update_and_save({"repeater": {"second": True}}, live_update=False)
|
|
|
|
monkeypatch.setattr("repeater.config_manager.yaml.safe_dump", paused_dump)
|
|
with ThreadPoolExecutor(max_workers=2) as pool:
|
|
a = pool.submit(first.update_and_save, {"repeater": {"first": True}}, False)
|
|
try:
|
|
assert entered.wait(5)
|
|
b = pool.submit(second_update)
|
|
assert second_started.wait(5)
|
|
assert not second_dump.wait(0.1)
|
|
assert config == {"repeater": {"node_name": "before"}}
|
|
finally:
|
|
release.set()
|
|
assert a.result()["saved"]
|
|
assert b.result()["saved"]
|
|
assert config["repeater"] == {"node_name": "before", "first": True, "second": True}
|
|
assert yaml.safe_load(path.read_text()) == config
|