96 lines
3.1 KiB
Python
96 lines
3.1 KiB
Python
"""Tests for real-time monitor."""
|
|
import pytest
|
|
import time
|
|
from ayn_antivirus.monitor.realtime import RealtimeMonitor
|
|
from ayn_antivirus.core.engine import ScanEngine
|
|
from ayn_antivirus.config import Config
|
|
|
|
|
|
@pytest.fixture
|
|
def monitor(tmp_path):
|
|
config = Config()
|
|
engine = ScanEngine(config)
|
|
m = RealtimeMonitor(config, engine)
|
|
yield m
|
|
if m.is_running:
|
|
m.stop()
|
|
|
|
|
|
def test_monitor_init(monitor):
|
|
assert monitor is not None
|
|
assert monitor.is_running is False
|
|
|
|
|
|
def test_monitor_should_skip():
|
|
"""Temporary / lock / editor files should be skipped."""
|
|
config = Config()
|
|
engine = ScanEngine(config)
|
|
m = RealtimeMonitor(config, engine)
|
|
|
|
assert m._should_skip("/tmp/test.tmp") is True
|
|
assert m._should_skip("/tmp/test.swp") is True
|
|
assert m._should_skip("/tmp/test.lock") is True
|
|
assert m._should_skip("/tmp/.#backup") is True
|
|
assert m._should_skip("/tmp/test.part") is True
|
|
|
|
assert m._should_skip("/tmp/test.txt") is False
|
|
assert m._should_skip("/tmp/test.py") is False
|
|
assert m._should_skip("/var/www/index.html") is False
|
|
|
|
|
|
def test_monitor_debounce(monitor):
|
|
"""After the first call records the path, an immediate repeat is debounced."""
|
|
import time as _time
|
|
|
|
# Prime the path so it's recorded with the current monotonic time.
|
|
# On fresh processes, monotonic() can be close to 0.0 which is the
|
|
# default in _recent, so we explicitly set a realistic timestamp.
|
|
monitor._recent["/tmp/test.txt"] = _time.monotonic() - 10
|
|
assert monitor._is_debounced("/tmp/test.txt") is False
|
|
# Immediate second call should be debounced (within 2s window)
|
|
assert monitor._is_debounced("/tmp/test.txt") is True
|
|
|
|
|
|
def test_monitor_debounce_different_paths(monitor):
|
|
"""Different paths should not debounce each other."""
|
|
import time as _time
|
|
|
|
# Prime both paths far enough in the past to avoid the initial-value edge case
|
|
past = _time.monotonic() - 10
|
|
monitor._recent["/tmp/a.txt"] = past
|
|
monitor._recent["/tmp/b.txt"] = past
|
|
assert monitor._is_debounced("/tmp/a.txt") is False
|
|
assert monitor._is_debounced("/tmp/b.txt") is False
|
|
|
|
|
|
def test_monitor_start_stop(tmp_path, monitor):
|
|
monitor.start(paths=[str(tmp_path)], recursive=True)
|
|
assert monitor.is_running is True
|
|
time.sleep(0.3)
|
|
monitor.stop()
|
|
assert monitor.is_running is False
|
|
|
|
|
|
def test_monitor_double_start(tmp_path, monitor):
|
|
"""Starting twice should be harmless."""
|
|
monitor.start(paths=[str(tmp_path)])
|
|
assert monitor.is_running is True
|
|
monitor.start(paths=[str(tmp_path)]) # Should log warning, not crash
|
|
assert monitor.is_running is True
|
|
monitor.stop()
|
|
|
|
|
|
def test_monitor_stop_when_not_running(monitor):
|
|
"""Stopping when not running should be harmless."""
|
|
assert monitor.is_running is False
|
|
monitor.stop()
|
|
assert monitor.is_running is False
|
|
|
|
|
|
def test_monitor_nonexistent_path(monitor):
|
|
"""Non-existent paths should be skipped without crash."""
|
|
monitor.start(paths=["/nonexistent/path/xyz123"])
|
|
# Should still be running (observer started, just no schedules)
|
|
assert monitor.is_running is True
|
|
monitor.stop()
|