diff options
Diffstat (limited to 'tests/test_upstream.py')
| -rw-r--r-- | tests/test_upstream.py | 112 |
1 files changed, 112 insertions, 0 deletions
diff --git a/tests/test_upstream.py b/tests/test_upstream.py new file mode 100644 index 0000000..38cba54 --- /dev/null +++ b/tests/test_upstream.py @@ -0,0 +1,112 @@ +# sb-builder - self-checks for the nvchecker glue +# Copyright (C) 2026 Danilo M. <danix@danix.xyz> +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License version 2 as +# published by the Free Software Foundation. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# 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. + +"""Run: python3 -m unittest discover -s tests (no network: nvchecker is faked)""" + +import os +os.environ["QT_QPA_PLATFORM"] = "offscreen" + +import shutil # noqa: E402 +import sys # noqa: E402 +import tempfile # noqa: E402 +import time # noqa: E402 +import tomllib # noqa: E402 +import unittest # noqa: E402 +from unittest import mock # noqa: E402 + +ROOT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..") +sys.path.insert(0, ROOT) + +from PyQt6.QtWidgets import QApplication # noqa: E402 + +from sb_builder import upstream # noqa: E402 + +app = QApplication.instance() or QApplication([]) +LOG = "\n".join([ + '{"name": "borgbackup", "version": "1.4.5", "event": "updated", "level": "info"}', + '{"name": "bogus", "event": "unexpected error happened", "error": "HTTPError(404)", "level": "error"}', + '{"name": "bogus", "event": "no-result", "error": "later", "level": "error"}', + 'not json', +]) + + +def row(name, hint=None, info="1.0"): + return {"name": name, "hint_version": hint, "info_version": info} + + +class UpstreamTest(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, self.tmp) + self.stanzas = os.path.join(self.tmp, "stanzas.toml") + with open(self.stanzas, "w") as f: + f.write('[x]\nsource = "manual"\nmanual = "2.0"\n') + + def test_repo_file(self): + with open(upstream.STANZAS, "rb") as f: + self.assertNotIn("__config__", tomllib.load(f)) + self.assertIn("borgbackup", upstream.stanzas()) + self.assertEqual(upstream.stanzas("/nonexistent"), set()) + + def test_write_config(self): + keys = os.path.join(self.tmp, "keys.toml") + out = upstream.write_config(self.tmp, keys, self.stanzas) + with open(out, "rb") as f: + self.assertEqual(tomllib.load(f), {"__config__": {}, "x": {"source": "manual", "manual": "2.0"}}) + open(keys, "w").close() + with open(upstream.write_config(self.tmp, keys, self.stanzas), "rb") as f: + self.assertEqual(tomllib.load(f)["__config__"], {"keyfile": keys}) + + def test_parse_log(self): + self.assertEqual(upstream.parse_log(LOG), ({"borgbackup": "1.4.5"}, {"bogus": "HTTPError(404)"})) + + def test_state(self): + names = {"a", "b", "c"} + v = {"a": "2.0", "b": "1.0"} + self.assertEqual(upstream.state(row("z"), names, v, {}), "no stanza") + self.assertEqual(upstream.state(row("a"), names, None, {}), "not checked") + self.assertEqual(upstream.state(row("a"), names, v, {}), "outdated") + self.assertEqual(upstream.state(row("a", hint="2.0"), names, v, {}), "current") + self.assertEqual(upstream.state(row("b"), names, v, {}), "current") # .info version fallback + self.assertEqual(upstream.state(row("c"), names, v, {}), "nvchecker error") + + def test_checker_runs_nvchecker(self): + bindir = os.path.join(self.tmp, "bin") + os.mkdir(bindir) + with open(os.path.join(bindir, "nvchecker"), "w") as f: + f.write('#!/bin/sh\n[ "$1" = -c ] && [ -f "$2" ] && [ "$3 $4" = "--logger json" ] || exit 9\n' + 'echo \'{"name": "x", "version": "2.0", "level": "info"}\'\n') + os.chmod(os.path.join(bindir, "nvchecker"), 0o755) + with mock.patch.dict(os.environ, {"PATH": bindir + os.pathsep + os.environ["PATH"]}): + got = self.run_checker() + self.assertEqual(got, [({"x": "2.0"}, {}, "")]) + + def test_checker_missing_program(self): + with mock.patch.dict(os.environ, {"PATH": self.tmp}): + got = self.run_checker() + self.assertEqual(len(got), 1) + self.assertTrue(got[0][2]) + + def run_checker(self): + c = upstream.Checker() + self.addCleanup(c.deleteLater) + got = [] + c.done.connect(lambda *a: got.append(a)) + c.run(os.path.join(self.tmp, "nokeys.toml"), self.stanzas) + end = time.monotonic() + 5 + while not got and time.monotonic() < end: + app.processEvents() + return got |