107 lines
3.9 KiB
Python
107 lines
3.9 KiB
Python
from datetime import datetime
|
|
from pathlib import Path
|
|
import sys
|
|
import time
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
|
|
from pytorrent.services import download_planner, poller_control
|
|
|
|
|
|
def test_planner_evaluate_network_caps():
|
|
download_planner._override_until = lambda profile_id: ""
|
|
settings = download_planner.normalize({
|
|
"enabled": True,
|
|
"profile_name": "low power mode",
|
|
"network_protection_enabled": True,
|
|
"network_max_down": 100,
|
|
"network_max_up": 50,
|
|
"weekday_down": 1000,
|
|
"weekday_up": 500,
|
|
})
|
|
decision = download_planner.evaluate({"id": 1}, settings, datetime(2026, 5, 13, 12, 0))
|
|
assert decision["profile_name"] == "low power mode"
|
|
assert decision["down"] == 100
|
|
assert decision["up"] == 50
|
|
assert "network_limit_down" in decision["reasons"]
|
|
|
|
|
|
def test_poller_metrics_and_fallback():
|
|
settings = poller_control.normalize_settings({
|
|
"active_interval_seconds": -1,
|
|
"safe_fallback_enabled": True,
|
|
"slow_response_threshold_ms": 200,
|
|
})
|
|
assert settings["active_interval_seconds"] > 0
|
|
state = poller_control.ProfilePollState(profile_id=1)
|
|
runtime = poller_control.mark_tick(
|
|
state,
|
|
time.monotonic() - 0.5,
|
|
active=True,
|
|
ok=True,
|
|
emitted_payload_size=1234,
|
|
rtorrent_call_count=2,
|
|
skipped_emissions=1,
|
|
settings=settings,
|
|
)
|
|
assert runtime["emitted_payload_size"] == 1234
|
|
assert runtime["rtorrent_call_count"] == 2
|
|
assert runtime["adaptive_mode"] in {"normal", "idle", "slowdown", "recovery"}
|
|
|
|
fixed_state = poller_control.ProfilePollState(profile_id=2, adaptive_mode="slowdown", slow_count=5)
|
|
fixed_runtime = poller_control.mark_tick(
|
|
fixed_state,
|
|
time.monotonic() - 1.0,
|
|
active=True,
|
|
ok=True,
|
|
settings={**settings, "adaptive_enabled": False},
|
|
)
|
|
assert fixed_runtime["adaptive_enabled"] is False
|
|
assert fixed_runtime["adaptive_mode"] == "fixed"
|
|
assert fixed_runtime["slow_count"] == 0
|
|
|
|
|
|
def test_poller_background_slow_task_state():
|
|
state = poller_control.ProfilePollState(profile_id=3)
|
|
assert state.slow_task_running is False
|
|
state.slow_task_running = True
|
|
runtime = poller_control.mark_tick(
|
|
state,
|
|
time.monotonic() - 0.05,
|
|
active=True,
|
|
ok=True,
|
|
settings={"adaptive_enabled": False, "slow_response_threshold_ms": 200},
|
|
skipped_emissions=1,
|
|
)
|
|
assert runtime["adaptive_mode"] == "fixed"
|
|
assert runtime["skipped_emissions"] >= 1
|
|
assert state.slow_task_running is True
|
|
|
|
|
|
def test_poller_requested_fast_defaults():
|
|
settings = poller_control.normalize_settings({})
|
|
assert settings["active_interval_seconds"] == 0.5
|
|
assert settings["torrent_list_interval_seconds"] == 0.5
|
|
assert settings["idle_interval_seconds"] == 3.0
|
|
assert settings["error_interval_seconds"] == 2.0
|
|
assert settings["system_stats_interval_seconds"] == 1.0
|
|
assert settings["tracker_stats_interval_seconds"] == 30.0
|
|
assert settings["disk_stats_interval_seconds"] == 30.0
|
|
assert settings["queue_stats_interval_seconds"] == 5.0
|
|
assert settings["heartbeat_interval_seconds"] == 5.0
|
|
assert settings["slow_response_threshold_ms"] == 10000.0
|
|
assert settings["slowdown_multiplier"] == 1.0
|
|
state = poller_control.ProfilePollState(profile_id=4)
|
|
runtime = poller_control.mark_tick(state, time.monotonic() - 0.01, active=True, ok=True, settings=settings)
|
|
assert runtime["effective_interval_seconds"] == 0.5
|
|
assert runtime["configured_min_interval_seconds"] == 0.5
|
|
assert "last_tick_gap_ms" in runtime
|
|
|
|
|
|
if __name__ == "__main__":
|
|
test_planner_evaluate_network_caps()
|
|
test_poller_metrics_and_fallback()
|
|
test_poller_background_slow_task_state()
|
|
test_poller_requested_fast_defaults()
|
|
print("planner/poller service smoke tests passed")
|