diff --git a/.coveragerc b/.coveragerc new file mode 100644 index 0000000..c1e43ee --- /dev/null +++ b/.coveragerc @@ -0,0 +1,7 @@ +[run] +omit = + tests/* + */site-packages/* + +[report] +show_missing = True diff --git a/.gitea/workflows/lint.yml b/.gitea/workflows/lint.yml index ecb4021..942f6e3 100644 --- a/.gitea/workflows/lint.yml +++ b/.gitea/workflows/lint.yml @@ -21,3 +21,21 @@ jobs: - name: Run flake8 run: flake8 . --exclude=__pycache__,.git + + notify-failure: + name: Notify on failure + runs-on: ubuntu-latest + needs: [python-lint] + if: failure() && github.event_name == 'push' + steps: + - name: Send Matrix alert + env: + MATRIX_WEBHOOK_URL: ${{ secrets.MATRIX_WEBHOOK_URL }} + REPO: ${{ github.repository }} + BRANCH: ${{ github.ref_name }} + RUN_URL: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }} + run: | + if [ -z "$MATRIX_WEBHOOK_URL" ] || [ "$MATRIX_WEBHOOK_URL" = "CONFIGURE_ME" ]; then exit 0; fi + curl -sf -X POST "$MATRIX_WEBHOOK_URL" \ + -H "Content-Type: application/json" \ + -d "{\"text\":\"CI FAILED: ${REPO} @ ${BRANCH} — ${RUN_URL}\"}" diff --git a/.gitea/workflows/test.yml b/.gitea/workflows/test.yml new file mode 100644 index 0000000..b335c68 --- /dev/null +++ b/.gitea/workflows/test.yml @@ -0,0 +1,23 @@ +name: Test + +on: + push: + branches: ["**"] + pull_request: + branches: ["**"] + +jobs: + pytest: + name: Python Tests (pytest) + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v3 + + - name: Install Python and dependencies + run: | + apt-get update -qq + apt-get install -y -qq python3 python3-pip python3-pytest python3-psutil python3-requests + pip3 install pytest-cov + + - name: Run pytest with coverage + run: python3 -m pytest tests/ -v --cov=. --cov-report=term-missing --cov-config=.coveragerc diff --git a/.gitignore b/.gitignore index 9c68ef2..c86df22 100644 --- a/.gitignore +++ b/.gitignore @@ -1,2 +1,4 @@ .claude -settings.local.json \ No newline at end of file +settings.local.json +__pycache__/ +*.pyc \ No newline at end of file diff --git a/tests/test_hwmon.py b/tests/test_hwmon.py new file mode 100644 index 0000000..327495f --- /dev/null +++ b/tests/test_hwmon.py @@ -0,0 +1,222 @@ +"""Tests for SystemHealthMonitor pure methods — no external processes or filesystem.""" +import sys +import os +from unittest.mock import patch, MagicMock +sys.path.insert(0, os.path.dirname(os.path.dirname(__file__))) + +import pytest +from hwmonDaemon import SystemHealthMonitor + + +@pytest.fixture(scope='module') +def monitor(): + """Create a minimal monitor instance with all external side-effects patched out.""" + with patch.object(SystemHealthMonitor, 'load_env_config'), \ + patch.object(SystemHealthMonitor, '_check_tool_availability', return_value={}), \ + patch('os.makedirs'): + return SystemHealthMonitor(dry_run=True) + + +# ── _format_bytes_human ────────────────────────────────────────────────────── + +class TestFormatBytesHuman: + def test_bytes(self, monitor): + assert monitor._format_bytes_human(512) == '512.0 B' + + def test_kilobytes(self, monitor): + assert monitor._format_bytes_human(1024) == '1.0 KB' + + def test_megabytes(self, monitor): + assert monitor._format_bytes_human(1024 ** 2) == '1.0 MB' + + def test_gigabytes(self, monitor): + assert monitor._format_bytes_human(1024 ** 3) == '1.0 GB' + + def test_terabytes(self, monitor): + assert monitor._format_bytes_human(1024 ** 4) == '1.0 TB' + + def test_fractional(self, monitor): + assert monitor._format_bytes_human(1536) == '1.5 KB' + + def test_zero(self, monitor): + assert monitor._format_bytes_human(0) == '0.0 B' + + +# ── _parse_size ─────────────────────────────────────────────────────────────── + +class TestParseSize: + def test_gigabytes(self, monitor): + result = monitor._parse_size('15.7G') + assert abs(result - 15.7 * 1024**3) < 1 + + def test_terabytes(self, monitor): + result = monitor._parse_size('21.8T') + assert abs(result - 21.8 * 1024**4) < 1 + + def test_megabytes(self, monitor): + result = monitor._parse_size('512M') + assert result == 512 * 1024**2 + + def test_kilobytes(self, monitor): + result = monitor._parse_size('100K') + assert result == 100 * 1024 + + def test_bytes(self, monitor): + result = monitor._parse_size('100B') + assert result == 100 + + def test_invalid_returns_zero(self, monitor): + assert monitor._parse_size('notasize') == 0.0 + + def test_non_string_returns_zero(self, monitor): + assert monitor._parse_size(None) == 0.0 + assert monitor._parse_size(42) == 0.0 + + +# ── _parse_smart_value ──────────────────────────────────────────────────────── + +class TestParseSmartValue: + def test_plain_integer(self, monitor): + assert monitor._parse_smart_value('42') == 42 + + def test_temperature_with_celsius(self, monitor): + assert monitor._parse_smart_value('38 °C') == 38 + + def test_time_format(self, monitor): + assert monitor._parse_smart_value('15589h+17m+33.939s') == 15589 + + def test_hex_value(self, monitor): + assert monitor._parse_smart_value('0x0a') == 10 + + def test_invalid_returns_zero(self, monitor): + assert monitor._parse_smart_value('not_a_number') == 0 + + +# ── _detect_manufacturer ────────────────────────────────────────────────────── + +class TestDetectManufacturer: + def test_western_digital(self, monitor): + assert monitor._detect_manufacturer('WDC WD40EFRX') == 'Western Digital' + + def test_hgst(self, monitor): + assert monitor._detect_manufacturer('HGST HUH728080ALE604') == 'Western Digital' + + def test_seagate(self, monitor): + assert monitor._detect_manufacturer('ST4000DM004') == 'Seagate' + + def test_samsung(self, monitor): + assert monitor._detect_manufacturer('Samsung SSD 870 EVO') == 'Samsung' + + def test_intel(self, monitor): + assert monitor._detect_manufacturer('INTEL SSDSC2KB480G8') == 'Intel' + + def test_micron(self, monitor): + assert monitor._detect_manufacturer('Crucial CT500MX500SSD1') == 'Micron' + + def test_toshiba(self, monitor): + assert monitor._detect_manufacturer('TOSHIBA MG06ACA10TE') == 'Toshiba' + + def test_unknown(self, monitor): + assert monitor._detect_manufacturer('GENERICDRIVE XYZ') == 'Unknown' + + def test_empty_model(self, monitor): + assert monitor._detect_manufacturer('') == 'Unknown' + + def test_none_model(self, monitor): + assert monitor._detect_manufacturer(None) == 'Unknown' + + +# ── _check_thermal_health ───────────────────────────────────────────────────── + +class TestCheckThermalHealth: + def test_hdd_ok_temperature(self, monitor): + issues = monitor._check_thermal_health('sda', 45, 'HDD') + assert issues == [] + + def test_hdd_info_temperature(self, monitor): + issues = monitor._check_thermal_health('sda', 62, 'HDD') + assert len(issues) == 1 + assert 'INFO' in issues[0] + + def test_hdd_warning_temperature(self, monitor): + issues = monitor._check_thermal_health('sda', 66, 'HDD') + assert len(issues) == 1 + assert 'WARNING' in issues[0] + + def test_hdd_critical_temperature(self, monitor): + issues = monitor._check_thermal_health('sda', 76, 'HDD') + assert len(issues) == 1 + assert 'CRITICAL' in issues[0] + + def test_ssd_has_higher_warning_threshold(self, monitor): + # HDD warning=65°C, SSD warning=70°C; at 67°C: + # HDD → WARNING, SSD → INFO (above optimal_max=65 but below warning=70) + issues_hdd = monitor._check_thermal_health('sda', 67, 'HDD') + issues_ssd = monitor._check_thermal_health('sda', 67, 'SSD') + assert any('WARNING' in i for i in issues_hdd) + assert not any('WARNING' in i for i in issues_ssd) + assert any('INFO' in i for i in issues_ssd) + + def test_none_temperature_returns_empty(self, monitor): + issues = monitor._check_thermal_health('sda', None, 'HDD') + assert issues == [] + + +# ── _is_excluded_mount ──────────────────────────────────────────────────────── + +class TestIsExcludedMount: + def test_exact_excluded_mount(self, monitor): + assert monitor._is_excluded_mount('/media') is True + + def test_pattern_excluded(self, monitor): + assert monitor._is_excluded_mount('/media/external') is True + + def test_downloads_excluded(self, monitor): + assert monitor._is_excluded_mount('/mnt/data/downloads') is True + + def test_normal_mount_not_excluded(self, monitor): + assert monitor._is_excluded_mount('/') is False + assert monitor._is_excluded_mount('/var') is False + assert monitor._is_excluded_mount('/mnt/ceph') is False + + +# ── _is_new_drive ───────────────────────────────────────────────────────────── + +class TestIsNewDrive: + def test_brand_new_drive(self, monitor): + assert monitor._is_new_drive(0) is True + + def test_one_hour_drive(self, monitor): + assert monitor._is_new_drive(1) is True + + def test_under_threshold(self, monitor): + assert monitor._is_new_drive(719) is True + + def test_at_threshold_is_not_new(self, monitor): + assert monitor._is_new_drive(720) is False + + def test_old_drive(self, monitor): + assert monitor._is_new_drive(50000) is False + + +# ── _is_physical_disk ──────────────────────────────────────────────────────── + +class TestIsPhysicalDisk: + def test_real_sata_disk(self, monitor): + # /dev/sda should pass (no exclusion pattern matches) + # Note: _is_physical_disk also checks os.path.exists and reads sysfs, + # but the exclusion logic runs first and can return False early. + # We test the exclusion cases which are pure. + assert monitor._is_physical_disk('/dev/mapper/data') is False + + def test_device_mapper_excluded(self, monitor): + assert monitor._is_physical_disk('/dev/dm-0') is False + + def test_loop_device_excluded(self, monitor): + assert monitor._is_physical_disk('/dev/loop0') is False + + def test_partition_excluded(self, monitor): + assert monitor._is_physical_disk('/dev/sda1') is False + + def test_rbd_excluded(self, monitor): + assert monitor._is_physical_disk('/dev/rbd0') is False