commit
1ccdebdcb2
5 changed files with 175 additions and 117 deletions
8
.github/workflows/tests.yml
vendored
8
.github/workflows/tests.yml
vendored
|
|
@ -17,6 +17,8 @@ jobs:
|
|||
with:
|
||||
python-version: 3.9
|
||||
- name: Install dependencies
|
||||
run: python -m pip install --upgrade pip pytest
|
||||
- name: Run tests with pytest
|
||||
run: pytest -v tests.py
|
||||
run: python -m pip install --upgrade pip pytest flake8
|
||||
- name: Run linter
|
||||
run: flake8 --max-line-length=120 scripts/
|
||||
- name: Run tests
|
||||
run: pytest -v
|
||||
|
|
|
|||
4
Makefile
4
Makefile
|
|
@ -23,9 +23,9 @@ SHARE_DIR = $(DESTDIR)$(PREFIX)/share
|
|||
.PHONY: install
|
||||
|
||||
install:
|
||||
@install -Dm755 snap_pac.py $(SHARE_DIR)/libalpm/scripts/snap-pac
|
||||
@install -Dm755 scripts/snap_pac.py $(SHARE_DIR)/libalpm/scripts/snap-pac
|
||||
@install -Dm644 hooks/* -t $(SHARE_DIR)/libalpm/hooks/
|
||||
@install -Dm644 LICENSE -t $(SHARE_DIR)/licenses/$(PKGNAME)/
|
||||
@install -Dm644 man8/* -t $(SHARE_DIR)/man/man8/
|
||||
@install -Dm644 README.md -t $(SHARE_DIR)/doc/$(PKGNAME)/
|
||||
@install -Dm644 extra/snap-pac.ini -t $(DESTDIR)/etc/snap-pac.ini.example
|
||||
@install -Dm644 extra/snap-pac.ini $(DESTDIR)/etc/snap-pac.ini.example
|
||||
|
|
|
|||
|
|
@ -14,9 +14,7 @@
|
|||
# You should have received a copy of the GNU General Public License along
|
||||
# with this program; if not, write to the Free Software Foundation, Inc.,
|
||||
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
|
||||
"""
|
||||
Script for taking pre/post snapshots; run from pacman hooks.
|
||||
"""
|
||||
"""Script for taking pre/post snapshots; run from pacman hooks."""
|
||||
|
||||
from argparse import ArgumentParser
|
||||
from configparser import ConfigParser
|
||||
|
|
@ -27,39 +25,30 @@ import sys
|
|||
logging.basicConfig(format="%(message)s", level=logging.INFO)
|
||||
|
||||
|
||||
def create_snapper_cmd(preorpost, config, section, prefile):
|
||||
"""Populate the snapper command."""
|
||||
try:
|
||||
chroot = os.stat("/") != os.stat("/proc/1/root/.")
|
||||
except PermissionError:
|
||||
logging.warning("Unable to detect if in chroot. Run script as root.")
|
||||
chroot = False
|
||||
snapper_cmd = ["snapper"]
|
||||
if chroot:
|
||||
snapper_cmd.append("--no-dbus")
|
||||
snapper_cmd.append(f"--config {section} create")
|
||||
snapper_cmd.append(f"--type {preorpost}")
|
||||
snapper_cmd.append(f"--cleanup-algorithm {config.get(section, 'cleanup_algorithm')}")
|
||||
snapper_cmd.append("--print-number")
|
||||
desc_limit = config.getint(section, "desc_limit")
|
||||
if preorpost == "pre":
|
||||
snapper_cmd.append(f"--description {config.get(section, 'pre_description')[:desc_limit]}")
|
||||
else:
|
||||
snapper_cmd.append(f"--description {config.get(section, 'post_description')[:desc_limit]}")
|
||||
with open(prefile, "r") as f:
|
||||
pre_number = f.read().rstrip("\n")
|
||||
snapper_cmd.append(f"--pre-number {pre_number}")
|
||||
os.remove(prefile)
|
||||
return snapper_cmd
|
||||
class SnapperCmd:
|
||||
|
||||
def __init__(self, config, snapshot_type, cleanup_algorithm, description="", nodbus=False, pre_number=None):
|
||||
self.cmd = ["snapper"]
|
||||
if nodbus:
|
||||
self.cmd.append("--no-dbus")
|
||||
self.cmd.append(f"--config {config} create")
|
||||
self.cmd.append(f"--type {snapshot_type}")
|
||||
self.cmd.append(f"--cleanup-algorithm {cleanup_algorithm}")
|
||||
self.cmd.append("--print-number")
|
||||
if description:
|
||||
self.cmd.append(f"--description \"{description}\"")
|
||||
if snapshot_type == "post":
|
||||
if pre_number is not None:
|
||||
self.cmd.append(f"--pre-number {pre_number}")
|
||||
else:
|
||||
raise ValueError("snapshot type specified as 'post' but no pre snapshot number passed.")
|
||||
print(self.__str__())
|
||||
|
||||
def do_snapshot(preorpost, cmd, prefile):
|
||||
"""Run the actual snapper command and save snapshot number if pre snapshot."""
|
||||
num = os.popen(" ".join(cmd)).read().rstrip("\n")
|
||||
if preorpost == "pre":
|
||||
with open(prefile, "w") as f:
|
||||
f.write(num)
|
||||
return num
|
||||
def __call__(self):
|
||||
return os.popen(self.__str__()).read().rstrip("\n")
|
||||
|
||||
def __str__(self):
|
||||
return " ".join(self.cmd)
|
||||
|
||||
|
||||
def get_snapper_configs(conf_file):
|
||||
|
|
@ -78,8 +67,8 @@ def setup_config_parser(ini_file, parent_cmd, packages):
|
|||
config["DEFAULT"] = {
|
||||
"snapshot": False,
|
||||
"cleanup_algorithm": "number",
|
||||
"pre_description": "".join(["\"", parent_cmd, "\""]),
|
||||
"post_description": "".join(["\"", packages, "\""]),
|
||||
"pre_description": parent_cmd,
|
||||
"post_description": packages,
|
||||
"desc_limit": 72
|
||||
}
|
||||
config["root"] = {
|
||||
|
|
@ -89,6 +78,31 @@ def setup_config_parser(ini_file, parent_cmd, packages):
|
|||
return config
|
||||
|
||||
|
||||
def get_description(preorpost, config, section):
|
||||
desc_limit = config.getint(section, "desc_limit")
|
||||
if preorpost == "pre":
|
||||
return config.get(section, "pre_description")[:desc_limit]
|
||||
else:
|
||||
return config.get(section, "post_description")[:desc_limit]
|
||||
|
||||
|
||||
def get_pre_number(preorpost, prefile):
|
||||
if preorpost == "pre":
|
||||
pre_number = None
|
||||
else:
|
||||
try:
|
||||
with open(prefile, "r") as f:
|
||||
pre_number = f.read().rstrip("\n")
|
||||
except FileNotFoundError:
|
||||
logging.error("prefile not found")
|
||||
return pre_number
|
||||
|
||||
|
||||
def write_pre_number(num, prefile):
|
||||
with open(prefile, "w") as f:
|
||||
f.write(num)
|
||||
|
||||
|
||||
def main(snap_pac_ini, snapper_conf_file, args):
|
||||
|
||||
if os.getenv("SNAP_PAC_SKIP", "n") in ["y", "Y", "yes", "Yes", "YES"]:
|
||||
|
|
@ -98,6 +112,7 @@ def main(snap_pac_ini, snapper_conf_file, args):
|
|||
packages = " ".join([line.rstrip("\n") for line in sys.stdin])
|
||||
config = setup_config_parser(snap_pac_ini, parent_cmd, packages)
|
||||
snapper_configs = get_snapper_configs(snapper_conf_file)
|
||||
chroot = os.stat("/") != os.stat("/proc/1/root/.")
|
||||
|
||||
for c in snapper_configs:
|
||||
|
||||
|
|
@ -105,18 +120,21 @@ def main(snap_pac_ini, snapper_conf_file, args):
|
|||
config.add_section(c)
|
||||
|
||||
logging.debug(f"{c = }")
|
||||
|
||||
if config.getboolean(c, "snapshot"):
|
||||
prefile = f"/tmp/snap-pac-pre_{c}"
|
||||
logging.debug(f"{prefile = }")
|
||||
try:
|
||||
snapper_cmd = create_snapper_cmd(args.preorpost, config, c, prefile)
|
||||
logging.debug(f"{snapper_cmd = }")
|
||||
except FileNotFoundError:
|
||||
logging.error(f"Error: File containing pre snapshot number not found for \"{c}\" configuration. "
|
||||
"If you are installing snap-pac for the first time, this is normal and can be ignored.")
|
||||
else:
|
||||
num = do_snapshot(args.preorpost, snapper_cmd, prefile)
|
||||
logging.info(f"==> {c}: {num}")
|
||||
|
||||
cleanup_algorithm = config.get(c, "cleanup_algorithm")
|
||||
description = get_description(args.preorpost, config, c)
|
||||
pre_number = get_pre_number(args.preorpost, prefile)
|
||||
|
||||
snapper_cmd = SnapperCmd(c, args.preorpost, cleanup_algorithm, description, chroot, pre_number)
|
||||
num = snapper_cmd()
|
||||
logging.info(f"==> {c}: {num}")
|
||||
|
||||
if args.preorpost == "pre":
|
||||
write_pre_number(num, prefile)
|
||||
|
||||
return True
|
||||
|
||||
105
test_script.py
Normal file
105
test_script.py
Normal file
|
|
@ -0,0 +1,105 @@
|
|||
from configparser import ConfigParser
|
||||
import tempfile
|
||||
import os
|
||||
|
||||
import pytest
|
||||
|
||||
from scripts.snap_pac import (
|
||||
SnapperCmd, get_pre_number, get_snapper_configs, main, setup_config_parser,
|
||||
write_pre_number, get_description
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def config():
|
||||
config = ConfigParser()
|
||||
config["DEFAULT"] = {
|
||||
"snapshot": False,
|
||||
"cleanup_algorithm": "number",
|
||||
"pre_description": "foo",
|
||||
"post_description": "bar",
|
||||
"desc_limit": 72
|
||||
}
|
||||
config["root"] = {
|
||||
"snapshot": True
|
||||
}
|
||||
config["home"] = {
|
||||
"snapshot": True,
|
||||
"desc_limit": 3,
|
||||
"post_description": "a really long description"
|
||||
}
|
||||
return config
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def prefile():
|
||||
with tempfile.NamedTemporaryFile("w", delete=False) as f:
|
||||
f.write("1234")
|
||||
name = f.name
|
||||
return name
|
||||
|
||||
|
||||
@pytest.mark.parametrize("snapper_cmd, actual_cmd", [
|
||||
(
|
||||
SnapperCmd("root", "pre", "number", "foo"),
|
||||
"snapper --config root create --type pre --cleanup-algorithm number --print-number --description \"foo\""
|
||||
),
|
||||
(
|
||||
SnapperCmd("root", "post", "number", "bar", False, 1234),
|
||||
"snapper --config root create --type post --cleanup-algorithm number --print-number"
|
||||
" --description \"bar\" --pre-number 1234"
|
||||
),
|
||||
(
|
||||
SnapperCmd("root", "post", "number", "bar", True, 1234),
|
||||
"snapper --no-dbus --config root create --type post --cleanup-algorithm number --print-number"
|
||||
" --description \"bar\" --pre-number 1234"
|
||||
)
|
||||
])
|
||||
def test_snapper_cmd(snapper_cmd, actual_cmd):
|
||||
assert str(snapper_cmd) == actual_cmd
|
||||
|
||||
|
||||
def test_get_snapper_configs():
|
||||
with tempfile.NamedTemporaryFile("w", delete=False) as f:
|
||||
f.write("## Path: System/Snapper\n")
|
||||
f.write("\n")
|
||||
f.write("## Type: string\n")
|
||||
f.write("## Default: \"\"\n")
|
||||
f.write("# List of snapper configurations.\n")
|
||||
f.write("SNAPPER_CONFIGS=\"home root foo bar\"\n")
|
||||
name = f.name
|
||||
assert get_snapper_configs(name) == ["home", "root", "foo", "bar"]
|
||||
|
||||
|
||||
def test_skip_snap_pac():
|
||||
os.environ["SNAP_PAC_SKIP"] = "y"
|
||||
assert main("foo", "bar", "yep") is False
|
||||
|
||||
|
||||
def test_setup_config_parser(config):
|
||||
with tempfile.NamedTemporaryFile("w", delete=False) as f:
|
||||
f.write("[home]\n")
|
||||
f.write("snapshot = True\n")
|
||||
f.write("desc_limit = 3\n")
|
||||
f.write("post_description = a really long description\n")
|
||||
name = f.name
|
||||
config2 = setup_config_parser(name, "foo", "bar")
|
||||
assert config == config2
|
||||
|
||||
|
||||
def test_get_pre_number_pre(prefile):
|
||||
assert get_pre_number("pre", prefile) is None
|
||||
|
||||
|
||||
def test_get_pre_number_post(prefile):
|
||||
assert get_pre_number("post", prefile) == "1234"
|
||||
|
||||
|
||||
def test_write_pre_number(prefile):
|
||||
write_pre_number("5678", prefile)
|
||||
assert get_pre_number("post", prefile) == "5678"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("snapshot_type, description", [("pre", "foo"), ("post", "a r")])
|
||||
def test_get_description(snapshot_type, description, config):
|
||||
assert get_description(snapshot_type, config, "home") == description
|
||||
67
tests.py
67
tests.py
|
|
@ -1,67 +0,0 @@
|
|||
from configparser import ConfigParser
|
||||
import tempfile
|
||||
import os
|
||||
|
||||
import pytest
|
||||
|
||||
from snap_pac import create_snapper_cmd, get_snapper_configs, main, setup_config_parser
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def config():
|
||||
config = ConfigParser()
|
||||
config["DEFAULT"] = {
|
||||
"snapshot": False,
|
||||
"cleanup_algorithm": "number",
|
||||
"pre_description": "\"foo\"",
|
||||
"post_description": "\"bar\"",
|
||||
"desc_limit": 72
|
||||
}
|
||||
config["root"] = {
|
||||
"snapshot": True
|
||||
}
|
||||
config["home"] = {
|
||||
"snapshot": True
|
||||
}
|
||||
return config
|
||||
|
||||
|
||||
def test_create_snapper_pre_cmd(config):
|
||||
snapper_cmd = create_snapper_cmd("pre", config, "root", "/tmp/somefile")
|
||||
cmd = "snapper --config root create --type pre --cleanup-algorithm number --print-number --description \"foo\""
|
||||
assert " ".join(snapper_cmd) == cmd
|
||||
|
||||
|
||||
def test_create_snapper_post_cmd(config):
|
||||
with tempfile.NamedTemporaryFile("w", delete=False) as f:
|
||||
f.write("1234")
|
||||
snapper_cmd = create_snapper_cmd("post", config, "root", f.name)
|
||||
cmd = "snapper --config root create --type post --cleanup-algorithm number --print-number"
|
||||
cmd += " --description \"bar\" --pre-number 1234"
|
||||
assert " ".join(snapper_cmd) == cmd
|
||||
|
||||
|
||||
def test_get_snapper_configs():
|
||||
with tempfile.NamedTemporaryFile("w", delete=False) as f:
|
||||
f.write("## Path: System/Snapper\n")
|
||||
f.write("\n")
|
||||
f.write("## Type: string\n")
|
||||
f.write("## Default: \"\"\n")
|
||||
f.write("# List of snapper configurations.\n")
|
||||
f.write("SNAPPER_CONFIGS=\"home root foo bar\"\n")
|
||||
name = f.name
|
||||
assert get_snapper_configs(name) == ["home", "root", "foo", "bar"]
|
||||
|
||||
|
||||
def test_skip_snap_pac():
|
||||
os.environ["SNAP_PAC_SKIP"] = "y"
|
||||
assert main("foo", "bar", "yep") is False
|
||||
|
||||
|
||||
def test_setup_config_parser(config):
|
||||
with tempfile.NamedTemporaryFile("w", delete=False) as f:
|
||||
f.write("[home]\n")
|
||||
f.write("snapshot = True")
|
||||
name = f.name
|
||||
config2 = setup_config_parser(name, "foo", "bar")
|
||||
assert config == config2
|
||||
Loading…
Add table
Reference in a new issue