"""Synthetic public fixtures: no site inventory or production measurements.""" import json from pathlib import Path import subprocess import sys import time import unittest from unittest.mock import Mock, patch sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from agent_control import metrics as m NODE = 'MemTotal: 1000 kB\nMemAvailable: 400 kB\n__LOAD__\n1.20 2.30 3.40 1/20 123\n' NFS = 'Filesystem Type 1B-blocks Used Avail Use% Mounted on\nstorage.example.test:/export nfs4 1000 300 700 30% /mnt/shared\n' HOSTS = 'HOST_NAME STATUS JL/U MAX NJOBS RUN SSUSP USUSP RSV\nnode-a.example.test ok - 8 6 6 0 0 0\nnode-b.example.test unavail - 1 0 0 0 0 0\n' QUEUES = 'QUEUE_NAME PRIO STATUS MAX JL/U JL/P JL/H NJOBS PEND RUN SUSP RSV\nbatch 30 Open:Active - - - - 6 0 6 0 0\n' CONFIG = {'entry_hostname': 'entry.example.test', 'hosts': ['entry.example.test', 'node-a.example.test'], 'authorized_hosts': ['node-a.example.test'], 'ssh_user': 'observer', 'ssh_identity_file': '/home/observer/.ssh/metrics', 'storage_hosts': ['entry.example.test'], 'wrapper_paths': {'slurm': '/opt/site/bin/scheduler-exec', 'lsf': '/opt/site/bin/scheduler-exec'}} class MetricsTests(unittest.TestCase): def test_memory_load_units(self): row = m.parse_node(NODE) self.assertEqual(row['memory_total_bytes'], 1024000) self.assertEqual(row['memory_available_bytes'], 409600) self.assertEqual(row['memory_used_percent'], 60) self.assertEqual(row['load_15m'], 3.4) def test_missing_memory_and_nan_are_not_zero(self): for data in (NODE.replace('MemAvailable:', 'Other:'), NODE.replace('1.20', 'nan')): with self.assertRaises((ValueError, KeyError)): m.parse_node(data) def test_nfs_actual_mount(self): row = m.parse_nfs(NFS)[0] self.assertEqual(row['mount_source'], 'storage.example.test:/export') self.assertEqual(row['available_bytes'], 700) self.assertEqual(row['used_percent'], 30) with self.assertRaises(ValueError): m.parse_nfs('Permission denied') def test_slurm_dedup_and_empty_jobs(self): line = 'node-a.example.test|idle|8|16000\n' self.assertEqual(m.parse_slurm_nodes(line * 2)['count'], 1) self.assertEqual(m.parse_jobs('')['count'], 0) self.assertEqual(m.parse_jobs('RUNNING\nPENDING\nRUNNING')['by_state'], {'RUNNING': 2, 'PENDING': 1}) with self.assertRaises(ValueError): m.parse_jobs('Permission denied') def test_lsf_slots_are_not_job_records(self): hosts = m.parse_lsf_table(HOSTS, 'hosts') self.assertEqual(hosts['hosts'][1]['state'], 'unavail') self.assertEqual(hosts['hosts'][0]['run'], 6) self.assertEqual(m.parse_lsf_table(QUEUES, 'queues')['queues'][0]['run'], 6) self.assertEqual(m.parse_jobs('RUN\nRUN\n')['count'], 2) with self.assertRaises(ValueError): m.parse_lsf_table(HOSTS.replace('6', '-'), 'hosts') def test_default_only_executes_local_node(self): with patch.object(m, '_run', return_value=NODE) as run: result = m.collect({}) self.assertEqual(run.call_count, 1) self.assertEqual(run.call_args.args[:2], ('localhost', 'node')) self.assertEqual(result['storage'], []) self.assertEqual(result['schedulers']['slurm']['jobs']['status'], 'not_configured') self.assertIsNone(result['schedulers']['lsf']['jobs']['data']) self.assertEqual(result['status'], 'partial') def test_collect_partial_no_fabricated_data(self): def run(host, collector, timeout, settings): if host != settings['entry']: raise RuntimeError('permission denied (exit 255)') return {'node': NODE, 'nfs': NFS, 'slurm_nodes': 'node-a.example.test|idle|8|16000', 'slurm_jobs': '', 'lsf_hosts': HOSTS, 'lsf_queues': QUEUES, 'lsf_jobs': 'RUN\nRUN'}[collector] with patch.object(m, '_run', side_effect=run): result = m.collect(CONFIG) self.assertEqual(result['status'], 'partial') self.assertIsNone(result['nodes'][1]['data']) self.assertEqual(len(result['errors']), 1) self.assertEqual(result['schedulers']['lsf']['jobs']['data']['count'], 2) self.assertTrue(result['collected_at'].endswith('Z')) json.dumps(result, allow_nan=False) def test_config_rejects_shell_and_unapproved_targets(self): invalid = [None, {'command': 'id'}, {'hosts': ['node-a.example.test']}, {'hosts': ['localhost', 'localhost']}, {'hosts': ['localhost'], 'nodes': []}, {'schedulers': ['lsf; id']}, {'total_timeout_seconds': float('nan')}, {'hosts': 'all'}, {'wrapper_paths': {'slurm': '/bin/sh -c id'}}, {'wrapper_paths': {'slurm': '/opt/../bin/sh'}}, {'ssh_user': '-oProxyCommand=id'}, {'entry_hostname': 'localhost;id'}, {'authorized_hosts': [['bad']]}] with patch.object(m, '_run') as run: for config in invalid: with self.subTest(config=config): self.assertEqual(m.collect(config)['status'], 'error') run.assert_not_called() def test_timeout_no_zero(self): with patch.object(m, '_run', side_effect=TimeoutError('collector timed out')): result = m.collect({'schedulers': []}) self.assertEqual(result['status'], 'error') self.assertIsNone(result['nodes'][0]['data']) def test_real_subprocess_timeout_kills_group(self): with patch.dict(m.COMMANDS, {'node': [sys.executable, '-c', 'import time; time.sleep(10)']}): started = time.monotonic() with self.assertRaises(TimeoutError): m._run('localhost', 'node', 0.15, m._settings({})) self.assertLess(time.monotonic() - started, 2) def test_remote_ssh_strict_fixed_and_identity_not_source(self): process = Mock(returncode=0) process.communicate.return_value = (NODE, '') settings = m._settings(CONFIG) with patch.object(subprocess, 'Popen', return_value=process) as popen: record = m._sample('node-a.example.test', 'node', settings, time.monotonic()+5) command = popen.call_args.args[0] self.assertIn('StrictHostKeyChecking=yes', command) self.assertIn('BatchMode=yes', command) self.assertIn('observer@node-a.example.test', command) self.assertIn(CONFIG['ssh_identity_file'], command) self.assertNotIn(CONFIG['ssh_identity_file'], record['source']) self.assertNotIn('shell', popen.call_args.kwargs) def test_scheduler_argv_and_empty_lsf(self): command = m._command('slurm_jobs', m._settings(CONFIG)) self.assertEqual(command, ['/opt/site/bin/scheduler-exec', 'slurm', 'squeue', '-h', '-o', '%T']) process = Mock(returncode=255) process.communicate.return_value = ('', 'No unfinished job found.\n') with patch.object(subprocess, 'Popen', return_value=process): self.assertEqual(m._run('entry.example.test', 'lsf_jobs', 1, m._settings(CONFIG)), '') def test_deadline_does_not_start_command(self): with patch.object(m, '_run') as run: result = m._sample('localhost', 'node', m._settings({}), time.monotonic()-1) run.assert_not_called() self.assertEqual(result['status'], 'error') self.assertIsNone(result['data']) def test_missing_wrapper_is_error_not_empty_jobs(self): with patch.object(subprocess, 'Popen', side_effect=FileNotFoundError): result = m.collect(dict(CONFIG, hosts=[], storage_hosts=[])) self.assertEqual(result['status'], 'error') self.assertIsNone(result['schedulers']['lsf']['jobs']['data']) if __name__ == '__main__': unittest.main()