mirror of
https://github.com/MacRimi/ProxMenux.git
synced 2026-11-02 02:42:08 +00:00
set apart host backups of other nodes and of former jobs
- Each host backup carries the node it belongs to, read from its source host or its name (`hostcfg-<node>`, a job of this node or its PBS backup ID); both backup lists also return the node name. - The backup list offers This node / All nodes when some backup belongs to another node, with This node as the default, a count that follows the choice and a note with the backups not shown. - Backups whose node cannot be read from their name are listed under All nodes. - A local backup made on this node by a job that no longer exists is marked as Former job, read from its metadata file; backups without it, manual ones and those of other nodes are not marked. - New texts in English and Spanish, modelled in the translation steady-state test. - Tests for the node and the former job read from each backup, and for the cluster node membership notifications.
This commit is contained in:
@@ -0,0 +1,91 @@
|
||||
"""Each host backup in the list carries the node it belongs to, read from its
|
||||
name, so a cluster node can show its own backups apart from the others, and
|
||||
a local backup says when the job that made it no longer exists."""
|
||||
import ast
|
||||
import os
|
||||
from pathlib import Path
|
||||
import re
|
||||
import shlex
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
SERVER = Path(__file__).resolve().parents[1] / 'flask_server.py'
|
||||
HELPERS = ('_parse_job_env', '_host_backup_own_names', '_host_backup_node',
|
||||
'_host_backup_from_former_job')
|
||||
|
||||
|
||||
class HostBackupNodeTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
text = SERVER.read_text(encoding='utf-8')
|
||||
tree = ast.parse(text)
|
||||
nodes = [n for n in tree.body if isinstance(n, ast.FunctionDef) and n.name in HELPERS]
|
||||
self.assertEqual(sorted(n.name for n in nodes), sorted(HELPERS))
|
||||
folder = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(folder.cleanup)
|
||||
self.jobs = folder.name
|
||||
self.scope = {'os': os, 're': re, 'shlex': shlex,
|
||||
'_BACKUP_JOBS_DIR': self.jobs,
|
||||
'_BACKUP_TAR_SUFFIXES': ('.tar', '.tar.zst', '.tar.gz'),
|
||||
'_HOST_BACKUP_STAMP_RE': re.compile(r'-\d{8}_\d{6}$')}
|
||||
exec(compile(ast.Module(body=nodes, type_ignores=[]), str(SERVER), 'exec'), self.scope)
|
||||
self.own_names = self.scope['_host_backup_own_names']
|
||||
self.node = self.scope['_host_backup_node']
|
||||
self.former = self.scope['_host_backup_from_former_job']
|
||||
|
||||
def job(self, job_id, backup_id=None):
|
||||
lines = [f'JOB_ID={job_id}']
|
||||
if backup_id:
|
||||
lines.append(f'PBS_BACKUP_ID={backup_id}')
|
||||
Path(self.jobs, f'{job_id}.env').write_text('\n'.join(lines) + '\n')
|
||||
|
||||
def test_the_default_name_says_the_node(self):
|
||||
own = self.own_names('pve1')
|
||||
self.assertEqual(self.node('hostcfg-pve1', 'pve1', own), 'pve1')
|
||||
self.assertEqual(self.node('hostcfg-pve2', 'pve1', own), 'pve2')
|
||||
|
||||
def test_a_local_archive_name_says_the_node(self):
|
||||
own = self.own_names('pve1')
|
||||
self.assertEqual(self.node('hostcfg-pve2-20261014_031500.tar.zst', 'pve1', own), 'pve2')
|
||||
|
||||
def test_a_job_of_this_node_is_this_node(self):
|
||||
self.job('nightly-pbs', backup_id='my-host-config')
|
||||
own = self.own_names('pve1')
|
||||
self.assertEqual(self.node('nightly-pbs-20261014_031500.tar.zst', 'pve1', own), 'pve1')
|
||||
self.assertEqual(self.node('my-host-config', 'pve1', own), 'pve1')
|
||||
|
||||
def test_a_name_that_does_not_say_is_unknown(self):
|
||||
own = self.own_names('pve1')
|
||||
self.assertIsNone(self.node('custom-backup', 'pve1', own))
|
||||
self.assertIsNone(self.node('hostcfg-', 'pve1', own))
|
||||
self.assertIsNone(self.node(None, 'pve1', own))
|
||||
|
||||
def test_a_missing_jobs_folder_leaves_the_default_name(self):
|
||||
self.scope['_BACKUP_JOBS_DIR'] = os.path.join(self.jobs, 'absent')
|
||||
self.assertEqual(self.own_names('pve1'), {'hostcfg-pve1'})
|
||||
|
||||
def sidecar(self, job_id, hostname='pve1'):
|
||||
return {'kind': 'scheduled', 'job_id': job_id, 'source_hostname': hostname,
|
||||
'detected_via': 'sidecar'}
|
||||
|
||||
def test_a_backup_of_a_job_that_no_longer_exists_is_from_a_former_job(self):
|
||||
self.assertTrue(self.former(self.sidecar('PVE1-local-backuo'), 'pve1', {'PVE1-local-backup'}))
|
||||
|
||||
def test_a_backup_of_an_existing_job_is_not(self):
|
||||
self.assertFalse(self.former(self.sidecar('PVE1-local-backup'), 'pve1', {'PVE1-local-backup'}))
|
||||
|
||||
def test_a_backup_made_on_another_node_is_left_unknown(self):
|
||||
self.assertFalse(self.former(self.sidecar('pve2-nightly', hostname='pve2'), 'pve1', set()))
|
||||
|
||||
def test_a_backup_without_its_sidecar_is_left_unknown(self):
|
||||
for info in ({'kind': 'legacy', 'job_id': None, 'source_hostname': None, 'detected_via': 'tar_peek'},
|
||||
{'kind': 'manual', 'job_id': None, 'source_hostname': None, 'detected_via': 'hostcfg_prefix'}):
|
||||
self.assertFalse(self.former(info, 'pve1', set()))
|
||||
|
||||
def test_a_manual_backup_has_no_job(self):
|
||||
info = self.sidecar(None)
|
||||
info['kind'] = 'manual'
|
||||
self.assertFalse(self.former(info, 'pve1', set()))
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user