1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
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
|
