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
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
|
"""Test scripts/zfs-replicate, the nightly ZFS replication to TrueNAS.
Two defects the script had, both of which make a failing backup look like a
working one. It runs as a systemd oneshot (zfs-replicate.service) on a nightly
timer, so its exit code and its journal output are the only signals anyone ever
sees.
1. The full-replication loop caught each syncoid failure, warned, and carried
on -- then printed "Replication complete." and exited 0 regardless. With
every dataset failing, systemd recorded a clean success. A backup that had
not run for months was indistinguishable from one that had.
2. determine_host runs inside a command substitution, and its error() wrote to
stdout. So on an unreachable TrueNAS the message was captured into
TRUENAS_HOST and discarded, and `set -e` killed the script with exit 1 and
no output whatsoever -- a nightly service failing silently with nothing in
the journal to say why.
Both are fixed by sending every diagnostic to stderr (which also keeps the
command substitution's stdout clean, so only the hostname can land in
TRUENAS_HOST) and by exiting non-zero when any dataset failed.
Method: run the REAL script with a fake syncoid and a fake ping on PATH, in the
style of tests/zfs-pre-snapshot's fake-zfs.
Run from repo root:
python3 -m unittest tests.zfs-replicate.test_zfs_replicate
"""
import os
import shutil
import stat
import subprocess
import tempfile
import unittest
REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
SCRIPT = os.path.join(REPO_ROOT, "scripts", "zfs-replicate")
# The four datasets the script replicates, in order.
DATASETS = ["zroot/ROOT/default", "zroot/home", "zroot/media", "zroot/vms"]
class ZfsReplicate(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.mkdtemp(prefix="zfs-replicate-test-")
self.bin = os.path.join(self.tmp, "bin")
os.makedirs(self.bin)
def tearDown(self):
shutil.rmtree(self.tmp, ignore_errors=True)
def fake(self, name, body):
path = os.path.join(self.bin, name)
with open(path, "w") as f:
f.write("#!/bin/bash\n" + body)
os.chmod(path, os.stat(path).st_mode | stat.S_IEXEC | stat.S_IXGRP | stat.S_IXOTH)
def install_fakes(self, reachable="local", fail_datasets=(), no_syncoid=False):
"""ping answers for the named host only; syncoid fails for the listed
datasets (the last argument-but-one is the source dataset)."""
if reachable == "local":
ping = 'case "$3" in truenas.local) exit 0 ;; *) exit 1 ;; esac\n'
elif reachable == "tailscale":
ping = 'case "$3" in truenas) exit 0 ;; *) exit 1 ;; esac\n'
else:
ping = "exit 1\n"
# `ping -c 1 -W 2 <host>` puts the host in $5, but be permissive and
# scan every argument so an option-order change does not silently make
# every host unreachable (which would make these tests pass for the
# wrong reason).
ping = 'for a in "$@"; do case "$a" in\n' + \
(" truenas.local) exit 0 ;;\n" if reachable == "local" else "") + \
(" truenas) exit 0 ;;\n" if reachable == "tailscale" else "") + \
"esac; done\nexit 1\n"
self.fake("ping", ping)
if not no_syncoid:
arms = "".join(
' %s) echo "fake syncoid failing for %s" >&2; exit 1 ;;\n' % (d, d)
for d in fail_datasets
)
self.fake("syncoid", (
'for a in "$@"; do case "$a" in\n' + arms + "esac; done\nexit 0\n"
))
def run_script(self, *args):
env = dict(os.environ)
# Front of PATH so the fakes win; keep the rest so bash/coreutils work.
env["PATH"] = self.bin + os.pathsep + env["PATH"]
return subprocess.run(
["bash", SCRIPT, *args],
capture_output=True, text=True, env=env, timeout=20,
)
# ---------------------------------------------------------- normal ----
def test_all_datasets_succeed_exits_zero(self):
self.install_fakes()
r = self.run_script()
self.assertEqual(r.returncode, 0, r.stdout + r.stderr)
self.assertIn("Replication complete", r.stdout + r.stderr)
def test_single_dataset_mode_succeeds(self):
self.install_fakes()
r = self.run_script("zroot/home")
self.assertEqual(r.returncode, 0, r.stdout + r.stderr)
def test_falls_back_to_tailscale_when_local_is_unreachable(self):
self.install_fakes(reachable="tailscale")
r = self.run_script()
self.assertEqual(r.returncode, 0, r.stdout + r.stderr)
out = r.stdout + r.stderr
self.assertIn("Using TrueNAS host: truenas", out)
self.assertNotIn("truenas.local", out)
# -------------------------------------------------------- boundary ----
def test_every_dataset_failing_exits_non_zero(self):
# THE BUG: this used to print "Replication complete." and exit 0, so a
# nightly oneshot recorded success while backing nothing up.
self.install_fakes(fail_datasets=DATASETS)
r = self.run_script()
self.assertNotEqual(r.returncode, 0,
"a run where every dataset failed must not report success")
def test_one_dataset_failing_exits_non_zero_and_still_tries_the_rest(self):
self.install_fakes(fail_datasets=["zroot/media"])
r = self.run_script()
self.assertNotEqual(r.returncode, 0)
out = r.stdout + r.stderr
# The failure must not abort the run -- the dataset after it still goes.
self.assertIn("zroot/vms", out,
"a mid-loop failure must not stop the remaining datasets")
def test_failure_count_is_reported(self):
self.install_fakes(fail_datasets=["zroot/home", "zroot/vms"])
r = self.run_script()
self.assertIn("2", (r.stdout + r.stderr).split("complete")[-1] or "",
"the summary must say how many datasets failed")
# ----------------------------------------------------------- error ----
def test_unreachable_truenas_reports_why(self):
# THE OTHER BUG: error() wrote to stdout inside a command substitution,
# so this exited 1 with completely empty output.
self.install_fakes(reachable="none")
r = self.run_script()
self.assertNotEqual(r.returncode, 0)
self.assertIn("Cannot reach TrueNAS", r.stdout + r.stderr,
"the reason must reach the journal, not the command substitution")
def test_missing_syncoid_reports_why(self):
self.install_fakes(no_syncoid=True)
# PATH is the fake bin only, so `command -v syncoid` fails whether or
# not this machine has a real syncoid. bash must then be invoked by
# absolute path -- a PATH that cannot find syncoid cannot find bash
# either, and subprocess would fail to launch instead of testing
# anything. The script's syncoid check runs before any PATH lookup it
# actually needs, so nothing else breaks.
env = dict(os.environ)
env["PATH"] = self.bin
r = subprocess.run([shutil.which("bash") or "/usr/bin/bash", SCRIPT],
capture_output=True, text=True, env=env, timeout=20)
self.assertNotEqual(r.returncode, 0)
self.assertIn("syncoid not found", r.stdout + r.stderr)
def test_diagnostics_do_not_land_on_stdout_of_the_host_probe(self):
# Whatever determine_host prints on stdout becomes TRUENAS_HOST, so a
# stray diagnostic there corrupts every destination path.
self.install_fakes()
r = self.run_script()
self.assertIn("Using TrueNAS host: truenas.local", r.stdout + r.stderr)
self.assertNotIn("[INFO] Using TrueNAS host: [", r.stdout + r.stderr)
if __name__ == "__main__":
unittest.main()
|
