Rename application to Samsara and add assessment summary, detailed tabs and local Linux inventory
This commit is contained in:
@@ -2,10 +2,10 @@ import json
|
||||
from pathlib import Path
|
||||
import tempfile
|
||||
import unittest
|
||||
from server_buddy.core.validation import validate_hostname
|
||||
from server_buddy.collectors.demo import collect_demo, DEMO_HOST
|
||||
from server_buddy.analyzers.assessment import assess
|
||||
from server_buddy.reporting.export import export_html, export_json
|
||||
from samsara.core.validation import validate_hostname
|
||||
from samsara.collectors.demo import collect_demo, DEMO_HOST
|
||||
from samsara.analyzers.assessment import assess
|
||||
from samsara.reporting.export import export_html, export_json
|
||||
|
||||
class CoreTests(unittest.TestCase):
|
||||
def test_hostname(self):
|
||||
@@ -39,6 +39,8 @@ class CoreTests(unittest.TestCase):
|
||||
text = report.read_text()
|
||||
self.assertNotIn('<script>', text)
|
||||
self.assertIn('SYNTHETIC', text)
|
||||
self.assertIn('<title>Samsara</title>', text)
|
||||
self.assertIn('Samsara assessment', text)
|
||||
self.assertIn('missing evidence', text)
|
||||
self.assertIn('manual verification', text)
|
||||
output = Path(directory) / 'report.json'
|
||||
|
||||
34
tests/test_detection.py
Normal file
34
tests/test_detection.py
Normal file
@@ -0,0 +1,34 @@
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
from samsara.core.detection import detect_local_os, check_profile, PROFILES
|
||||
|
||||
class DetectionTests(unittest.TestCase):
|
||||
@patch('samsara.core.detection.platform.system', return_value='Linux')
|
||||
def test_rhel_and_compatible_release(self, system):
|
||||
for release in ['ID=rhel\nVERSION_ID="9.4"\nPRETTY_NAME="Red Hat Enterprise Linux 9.4"',
|
||||
'ID=rocky\nID_LIKE="rhel centos fedora"\nVERSION_ID=9.4']:
|
||||
with patch('samsara.core.detection.Path.read_text', return_value=release):
|
||||
info = detect_local_os()
|
||||
self.assertEqual(info['family'], 'rhel')
|
||||
self.assertEqual(info['version'], '9')
|
||||
check_profile(info, 'rhel:9')
|
||||
check_profile(info, 'rhel:any')
|
||||
with self.assertRaises(ValueError):
|
||||
check_profile(info, 'rhel:8')
|
||||
|
||||
@patch('samsara.core.detection.platform.system', return_value='Linux')
|
||||
@patch('samsara.core.detection.Path.read_text', return_value='ID=linuxmint\nID_LIKE="ubuntu debian"\nVERSION_ID=22\nPRETTY_NAME="Linux Mint"')
|
||||
def test_generic_linux_not_misidentified_as_rhel(self, release, system):
|
||||
info = detect_local_os()
|
||||
self.assertEqual(info['family'], 'linux')
|
||||
check_profile(info, 'auto')
|
||||
with self.assertRaises(ValueError):
|
||||
check_profile(info, 'rhel:any')
|
||||
|
||||
@patch('samsara.core.detection.platform.system', return_value='Windows')
|
||||
@patch('samsara.core.detection.platform.win32_ver', return_value=('11', '', '', ''))
|
||||
def test_windows_desktop_not_invented_as_server(self, version, system):
|
||||
info = detect_local_os()
|
||||
self.assertEqual(info['platform'], 'windows')
|
||||
self.assertEqual(info['version'], '')
|
||||
self.assertIn(('Windows Server 2025', 'windows:2025'), PROFILES)
|
||||
@@ -3,10 +3,10 @@ from pathlib import Path
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
from server_buddy.collectors.demo import collect_demo, LINUX_DEMO_HOST
|
||||
from server_buddy.analyzers.assessment import assess
|
||||
from samsara.collectors.demo import collect_demo, LINUX_DEMO_HOST
|
||||
from samsara.analyzers.assessment import assess
|
||||
try:
|
||||
from server_buddy.collectors.importer import load_inventory
|
||||
from samsara.collectors.importer import load_inventory
|
||||
except ImportError:
|
||||
load_inventory = None
|
||||
|
||||
|
||||
@@ -4,11 +4,11 @@ import tempfile
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
from server_buddy.collectors.linux import (collect_local_linux, parse_connections,
|
||||
from samsara.collectors.linux import (collect_local_linux, parse_connections,
|
||||
parse_units, run_command, CollectionCancelled)
|
||||
from server_buddy.collectors.demo import collect_demo, LINUX_DEMO_HOST
|
||||
from server_buddy.analyzers.assessment import assess
|
||||
from server_buddy.reporting.export import export_html
|
||||
from samsara.collectors.demo import collect_demo, LINUX_DEMO_HOST
|
||||
from samsara.analyzers.assessment import assess
|
||||
from samsara.reporting.export import export_html
|
||||
|
||||
|
||||
class LinuxTests(unittest.TestCase):
|
||||
@@ -28,12 +28,12 @@ class LinuxTests(unittest.TestCase):
|
||||
self.assertEqual(rows[1]['process'], 'nginx')
|
||||
self.assertEqual(rows[1]['direction'], 'Unknown')
|
||||
|
||||
@patch('server_buddy.collectors.linux.platform.system', return_value='Windows')
|
||||
@patch('samsara.collectors.linux.platform.system', return_value='Windows')
|
||||
def test_wrong_execution_platform(self, system):
|
||||
with self.assertRaises(ValueError):
|
||||
collect_local_linux()
|
||||
|
||||
@patch('server_buddy.collectors.linux.platform.system', return_value='Linux')
|
||||
@patch('samsara.collectors.linux.platform.system', return_value='Linux')
|
||||
def test_partial_failures_never_use_demo(self, system):
|
||||
def unavailable(args, **kwargs):
|
||||
raise RuntimeError('SECRET diagnostics must not be reported')
|
||||
@@ -46,7 +46,7 @@ class LinuxTests(unittest.TestCase):
|
||||
self.assertNotIn('SECRET', json.dumps(data))
|
||||
self.assertEqual(data['checks']['certificates']['status'], 'not_collected')
|
||||
|
||||
@patch('server_buddy.collectors.linux.platform.system', return_value='Linux')
|
||||
@patch('samsara.collectors.linux.platform.system', return_value='Linux')
|
||||
def test_cancel(self, system):
|
||||
with self.assertRaises(CollectionCancelled):
|
||||
collect_local_linux(cancelled=lambda: True)
|
||||
@@ -61,7 +61,7 @@ class LinuxTests(unittest.TestCase):
|
||||
if args[0] == 'dpkg-query':
|
||||
return 'installed\tnginx:amd64\t1.2\nconfig-files\tremoved\t1.0\n'
|
||||
return ''
|
||||
with self.subTest(rpm=rpm), patch('server_buddy.collectors.linux.platform.system', return_value='Linux'), patch('server_buddy.collectors.linux.Path.is_file', return_value=rpm):
|
||||
with self.subTest(rpm=rpm), patch('samsara.collectors.linux.platform.system', return_value='Linux'), patch('samsara.collectors.linux.Path.is_file', return_value=rpm):
|
||||
data = collect_local_linux(runner=command)
|
||||
self.assertEqual(len(data['inventory']['software']), 1)
|
||||
self.assertIn('nginx', data['inventory']['software'][0]['name'])
|
||||
|
||||
29
tests/test_schedules.py
Normal file
29
tests/test_schedules.py
Normal file
@@ -0,0 +1,29 @@
|
||||
import unittest
|
||||
from samsara.collectors.linux_schedules import cron_rows, enrich_timers
|
||||
|
||||
class ScheduleTests(unittest.TestCase):
|
||||
def test_cron_schedules_and_scripts_without_sensitive_arguments(self):
|
||||
text = '# ignored\nMAILTO=secret@example.test\n0 2 * * * app /usr/bin/python3 /opt/jobs/report.py --password=SECRET\n@daily root /opt/jobs/backup.sh --token=SECRET\n'
|
||||
rows = cron_rows(text, '/etc/cron.d/example', system_file=True)
|
||||
self.assertEqual(len(rows), 2)
|
||||
self.assertIn('report.py', rows[0]['name'])
|
||||
self.assertIn('0 2 * * *', rows[0]['triggers'])
|
||||
self.assertEqual(rows[0]['account'], 'app')
|
||||
self.assertIn('@daily', rows[1]['triggers'])
|
||||
self.assertNotIn('SECRET', str(rows))
|
||||
self.assertNotIn('secret@example.test', str(rows))
|
||||
self.assertNotIn('arguments', rows[0])
|
||||
|
||||
def test_timer_metadata_and_missing_next_run(self):
|
||||
rows = [{'name': 'report.timer'}, {'name': 'other.timer'}]
|
||||
def command(args):
|
||||
self.assertIn('--', args)
|
||||
return 'Id=report.timer\nTimersCalendar={ OnCalendar=*-*-* 02:00:00 ; next_elapse=... }\nTriggers=report.service\nNextElapseUSecRealtime=Mon 2026-11-02 02:00:00 UTC\n\nId=other.timer\nTimersCalendar=\n'
|
||||
enrich_timers(rows, command)
|
||||
self.assertIn('02:00:00', rows[0]['triggers'])
|
||||
self.assertEqual(rows[0]['associated_unit'], 'report.service')
|
||||
self.assertNotIn('triggers', rows[1])
|
||||
|
||||
def test_inline_shell_not_exposed(self):
|
||||
rows = cron_rows("@hourly bash -c 'echo SECRET'", 'current-user')
|
||||
self.assertNotIn('SECRET', str(rows))
|
||||
@@ -1,17 +1,17 @@
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
from server_buddy.core.startup import desktop_error
|
||||
from samsara.core.startup import desktop_error
|
||||
|
||||
class StartupTests(unittest.TestCase):
|
||||
@patch('server_buddy.core.startup.sys.platform', 'linux')
|
||||
@patch('samsara.core.startup.sys.platform', 'linux')
|
||||
@patch.dict('os.environ', {'DISPLAY': ':0', 'QT_QPA_PLATFORM': 'xcb'}, clear=True)
|
||||
@patch('server_buddy.core.startup.ctypes.CDLL', side_effect=OSError)
|
||||
@patch('samsara.core.startup.ctypes.CDLL', side_effect=OSError)
|
||||
def test_missing_xcb_returns_install_instruction(self, loader):
|
||||
self.assertIn('sudo apt install libxcb-cursor0', desktop_error())
|
||||
|
||||
@patch('server_buddy.core.startup.sys.platform', 'linux')
|
||||
@patch('samsara.core.startup.sys.platform', 'linux')
|
||||
@patch.dict('os.environ', {'QT_QPA_PLATFORM': 'offscreen'}, clear=True)
|
||||
@patch('server_buddy.core.startup.ctypes.CDLL')
|
||||
@patch('samsara.core.startup.ctypes.CDLL')
|
||||
def test_offscreen_does_not_require_xcb(self, loader):
|
||||
self.assertIsNone(desktop_error())
|
||||
loader.assert_not_called()
|
||||
|
||||
35
tests/test_summary.py
Normal file
35
tests/test_summary.py
Normal file
@@ -0,0 +1,35 @@
|
||||
import unittest
|
||||
from samsara.analyzers.summary import summarize
|
||||
from samsara.analyzers.assessment import assess
|
||||
from samsara.collectors.demo import collect_demo, LINUX_DEMO_HOST, DEMO_HOST
|
||||
|
||||
class SummaryTests(unittest.TestCase):
|
||||
def test_os_version_workload_and_schedule_evidence(self):
|
||||
for host, version in [(LINUX_DEMO_HOST, '9.4'), (DEMO_HOST, '2022')]:
|
||||
data = collect_demo(host)
|
||||
summary = summarize(data, assess(data))
|
||||
self.assertEqual(summary['version'], version)
|
||||
self.assertTrue(summary['software'])
|
||||
self.assertTrue(summary['jobs'])
|
||||
self.assertEqual(summary['risk_score'], assess(data)['score'])
|
||||
self.assertTrue(all('evidence' in row for row in summary['software']))
|
||||
|
||||
def test_unknown_version_not_kernel_and_disabled_jobs_not_active(self):
|
||||
data = collect_demo(LINUX_DEMO_HOST)
|
||||
data['server']['operating_system'] = 'Enterprise Linux'
|
||||
data['server']['os_build'] = '5.14.0'
|
||||
data['inventory']['tasks'] = [{'name': 'disabled', 'enabled': False}]
|
||||
data['inventory']['software'] = [{'name': 'libsqlite3'}, {'name': 'nginx-doc'}]
|
||||
data['inventory']['services'] = []
|
||||
summary = summarize(data, assess(data))
|
||||
self.assertEqual(summary['version'], 'Not collected')
|
||||
self.assertEqual(summary['jobs'], [])
|
||||
self.assertEqual(summary['software'], [])
|
||||
|
||||
def test_explicit_version_and_missing_schedule(self):
|
||||
data = collect_demo(LINUX_DEMO_HOST)
|
||||
data['server']['os_version'] = '9.5'
|
||||
data['inventory']['tasks'] = [{'name': 'backup', 'enabled': True}]
|
||||
summary = summarize(data, assess(data))
|
||||
self.assertEqual(summary['version'], '9.5')
|
||||
self.assertEqual(summary['jobs'][0]['schedule'], 'Schedule not collected')
|
||||
139
tests/test_ui.py
139
tests/test_ui.py
@@ -1,12 +1,17 @@
|
||||
import os
|
||||
os.environ.setdefault('QT_QPA_PLATFORM', 'offscreen')
|
||||
import time
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
try:
|
||||
from PySide6.QtWidgets import QApplication
|
||||
from server_buddy.ui.window import MainWindow
|
||||
from samsara.ui.window import MainWindow
|
||||
except ImportError:
|
||||
QApplication = None
|
||||
from samsara.collectors.demo import collect_demo, LINUX_DEMO_HOST
|
||||
|
||||
|
||||
@unittest.skipIf(QApplication is None, 'PySide6 not installed')
|
||||
class UITests(unittest.TestCase):
|
||||
@@ -14,74 +19,98 @@ class UITests(unittest.TestCase):
|
||||
def setUpClass(cls):
|
||||
cls.app = QApplication.instance() or QApplication([])
|
||||
|
||||
def test_navigation_collection_and_environment_errors(self):
|
||||
def finish(self, window):
|
||||
deadline = time.monotonic() + 5
|
||||
while window.job is not None and time.monotonic() < deadline:
|
||||
self.app.processEvents()
|
||||
time.sleep(.01)
|
||||
self.assertIsNone(window.job)
|
||||
|
||||
def test_default_workflow_navigation_and_no_fabricated_history(self):
|
||||
window = MainWindow()
|
||||
window.show()
|
||||
self.assertEqual(window.tabs.count(), 10)
|
||||
for index in range(6):
|
||||
self.assertEqual(window.windowTitle(), 'Samsara')
|
||||
self.assertEqual(window.source.currentData(), 'local')
|
||||
self.assertEqual(window.target.currentData(), 'auto')
|
||||
self.assertIsNone(window.data)
|
||||
self.assertEqual(window.history, [])
|
||||
self.assertFalse(window.report_buttons[0].isEnabled())
|
||||
self.assertEqual(window.nav.item(0).text(), 'Summary')
|
||||
for index in range(5):
|
||||
window.nav.setCurrentRow(index)
|
||||
self.assertEqual(window.pages.currentIndex(), index)
|
||||
window.hostname.setText('host;bad')
|
||||
window.start_collection()
|
||||
self.assertIn('Enter a hostname', window.status.text())
|
||||
window.hostname.setText('demo-web-01.example.test')
|
||||
window.environment.setCurrentIndex(1)
|
||||
window.start_collection()
|
||||
self.assertIn('not implemented', window.status.text())
|
||||
window.environment.setCurrentIndex(0)
|
||||
window.start_collection()
|
||||
deadline = time.monotonic() + 5
|
||||
while window.job is not None and time.monotonic() < deadline:
|
||||
self.app.processEvents()
|
||||
time.sleep(.01)
|
||||
self.assertIsNone(window.job)
|
||||
self.assertEqual(len(window.history), 2)
|
||||
self.assertTrue(window.analyze_button.isEnabled())
|
||||
self.assertIn('Complete', window.status.text())
|
||||
self.assertEqual(window.pages.currentIndex(), {0: 1, 1: 0, 2: 2, 3: 3, 4: 4}[index])
|
||||
window.close()
|
||||
|
||||
def test_linux_demo_collection(self):
|
||||
def test_samples_details_and_history(self):
|
||||
window = MainWindow()
|
||||
window.target.setCurrentIndex(1)
|
||||
self.assertEqual(window.hostname.text(), 'demo-rhel-01.example.test')
|
||||
window.start_collection()
|
||||
deadline = time.monotonic() + 5
|
||||
while window.job is not None and time.monotonic() < deadline:
|
||||
self.app.processEvents()
|
||||
time.sleep(.01)
|
||||
self.assertIsNone(window.job)
|
||||
for source, expected in [('linux_demo', 'linux'), ('windows_demo', 'windows')]:
|
||||
window.source.setCurrentIndex(window.source.findData(source))
|
||||
self.assertFalse(window.target.isEnabled())
|
||||
window.start_collection()
|
||||
self.finish(window)
|
||||
self.assertEqual(window.data['server']['platform'], expected)
|
||||
self.assertIn('Synthetic', window.banner.text())
|
||||
self.assertEqual(window.nav.currentRow(), 0)
|
||||
self.assertEqual(window.result_tabs.tabText(0), 'Summary')
|
||||
self.assertIn('/100', window.retirement_score.text())
|
||||
self.assertIn('Version:', window.os_version.text())
|
||||
for index in range(1, window.result_tabs.count()):
|
||||
window.result_tabs.setCurrentIndex(index)
|
||||
self.assertGreaterEqual(list(window.detail_areas.values())[index - 1].count(), 1)
|
||||
window.open_history(window.history_list.item(0))
|
||||
self.assertEqual(window.data['server']['platform'], 'linux')
|
||||
self.assertIn('SYNTHETIC', window.banner.text())
|
||||
self.assertEqual(window.tabs.count(), 10)
|
||||
self.assertEqual(len(window.records), 2)
|
||||
window.close()
|
||||
|
||||
def test_local_collection_source_and_report(self):
|
||||
from unittest.mock import patch
|
||||
from server_buddy.collectors.demo import collect_demo, LINUX_DEMO_HOST
|
||||
def test_auto_detect_runs_local_collector_and_saves_report(self):
|
||||
data = collect_demo(LINUX_DEMO_HOST)
|
||||
data['synthetic'] = False
|
||||
data['server']['collection_source'] = 'Local Linux read-only collection'
|
||||
with patch('server_buddy.ui.window.platform.system', return_value='Linux'), patch('server_buddy.ui.window.collect_local_linux', return_value=data):
|
||||
detected = {'platform': 'linux', 'family': 'linux', 'version': '22', 'name': 'Local Linux'}
|
||||
with patch('samsara.ui.window.detect_local_os', return_value=detected) as detection, patch('samsara.ui.window.collect_local_linux', return_value=data) as collector:
|
||||
window = MainWindow()
|
||||
# Local collection does not validate or contact the hostname field.
|
||||
window.hostname.setText('not-a-remote-target;')
|
||||
window.local_button.click()
|
||||
window.analyze_button.click()
|
||||
self.assertFalse(window.analyze_button.isEnabled())
|
||||
self.assertTrue(window.cancel_button.isEnabled())
|
||||
deadline = time.monotonic() + 5
|
||||
while window.job is not None and time.monotonic() < deadline:
|
||||
self.app.processEvents()
|
||||
time.sleep(.01)
|
||||
self.assertIsNone(window.job)
|
||||
self.assertIn('LOCAL LINUX', window.banner.text())
|
||||
self.assertEqual(window.history[-1]['source'], 'Local Linux read-only collection')
|
||||
self.assertTrue(window.local_button.isEnabled())
|
||||
self.assertFalse(window.cancel_button.isEnabled())
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
self.finish(window)
|
||||
detection.assert_called_once()
|
||||
collector.assert_called_once()
|
||||
self.assertIn('Local Linux', window.banner.text())
|
||||
self.assertTrue(window.analyze_button.isEnabled())
|
||||
self.assertTrue(window.report_buttons[0].isEnabled())
|
||||
with tempfile.TemporaryDirectory() as folder:
|
||||
report = Path(folder) / 'local.html'
|
||||
with patch('server_buddy.ui.window.QFileDialog.getSaveFileName', return_value=(str(report), '')), patch('server_buddy.ui.window.QMessageBox.information'):
|
||||
window.save_report('html')
|
||||
with patch('samsara.ui.window.QFileDialog.getSaveFileName', return_value=(str(report), '')), patch('samsara.ui.window.QMessageBox.information'):
|
||||
window.quick_save.click()
|
||||
self.assertIn('LOCAL LINUX INVENTORY', report.read_text())
|
||||
window.close()
|
||||
|
||||
def test_mismatched_profile_and_windows_unavailable_keep_prior_results(self):
|
||||
window = MainWindow()
|
||||
window.collection_done(collect_demo(LINUX_DEMO_HOST))
|
||||
previous = window.data
|
||||
detected = {'platform': 'linux', 'family': 'linux', 'version': '22', 'name': 'Local Linux'}
|
||||
with patch('samsara.ui.window.detect_local_os', return_value=detected), patch('samsara.ui.window.collect_local_linux') as collector:
|
||||
window.target.setCurrentIndex(window.target.findData('rhel:9'))
|
||||
window.start_collection()
|
||||
self.finish(window)
|
||||
collector.assert_not_called()
|
||||
self.assertIn('does not match', window.status.text())
|
||||
self.assertIs(window.data, previous)
|
||||
with patch('samsara.ui.window.detect_local_os', return_value={'platform': 'windows', 'family': 'windows', 'version': '', 'name': 'Windows'}):
|
||||
window.start_collection()
|
||||
self.finish(window)
|
||||
self.assertIn('not available yet', window.status.text())
|
||||
window.close()
|
||||
|
||||
def test_import_requires_file_and_reads_recorded_os(self):
|
||||
window = MainWindow()
|
||||
window.source.setCurrentIndex(window.source.findData('file'))
|
||||
self.assertFalse(window.target.isEnabled())
|
||||
window.start_collection()
|
||||
self.assertIn('Choose an inventory', window.status.text())
|
||||
with patch('samsara.ui.window.load_inventory', return_value=collect_demo(LINUX_DEMO_HOST)) as loader, patch('samsara.ui.window.detect_local_os') as detect:
|
||||
window.inventory_path.setText('/tmp/saved-inventory.json')
|
||||
window.start_collection()
|
||||
self.finish(window)
|
||||
loader.assert_called_once_with('/tmp/saved-inventory.json')
|
||||
detect.assert_not_called()
|
||||
window.close()
|
||||
|
||||
Reference in New Issue
Block a user