131 lines
3.1 KiB
Python
131 lines
3.1 KiB
Python
import time
|
|
|
|
import pytest
|
|
|
|
from exporter.config import DeviceConfig
|
|
from exporter.registry import DeviceRegistry
|
|
|
|
|
|
def test_register_static_and_runtime_device():
|
|
registry = DeviceRegistry(global_scrape_interval=60)
|
|
|
|
static = DeviceConfig(
|
|
name="dev1",
|
|
host="1.1.1.1",
|
|
port=830,
|
|
username="u",
|
|
password="p",
|
|
source="static",
|
|
)
|
|
runtime = DeviceConfig(
|
|
name="dev1",
|
|
host="10.0.0.1",
|
|
port=830,
|
|
username="u",
|
|
password="p2",
|
|
source="runtime",
|
|
)
|
|
|
|
registry.register_static_device(static)
|
|
registry.register_runtime_device(runtime)
|
|
|
|
devices = registry.list_devices()
|
|
assert len(devices) == 1
|
|
dev = devices[0]
|
|
assert dev.name == "dev1"
|
|
assert dev.host == "10.0.0.1"
|
|
assert dev.password == "p2"
|
|
assert dev.source == "runtime"
|
|
|
|
|
|
def test_get_enabled_devices_respects_next_scrape_at(monkeypatch):
|
|
registry = DeviceRegistry(global_scrape_interval=60)
|
|
now = time.time()
|
|
|
|
dev1 = DeviceConfig(
|
|
name="dev1",
|
|
host="h1",
|
|
port=830,
|
|
username="u",
|
|
password="p",
|
|
)
|
|
dev2 = DeviceConfig(
|
|
name="dev2",
|
|
host="h2",
|
|
port=830,
|
|
username="u",
|
|
password="p",
|
|
)
|
|
|
|
registry.register_static_device(dev1)
|
|
registry.register_static_device(dev2)
|
|
|
|
# 手动调整 next_scrape_at
|
|
registry._states["dev1"].next_scrape_at = now - 1
|
|
registry._states["dev2"].next_scrape_at = now + 100
|
|
|
|
ready = registry.get_enabled_devices(now)
|
|
assert len(ready) == 1
|
|
assert ready[0].cfg.name == "dev1"
|
|
|
|
|
|
def test_backoff_increases_on_failure():
|
|
registry = DeviceRegistry(global_scrape_interval=60)
|
|
now = time.time()
|
|
dev = DeviceConfig(
|
|
name="dev",
|
|
host="h",
|
|
port=830,
|
|
username="u",
|
|
password="p",
|
|
)
|
|
registry.register_static_device(dev)
|
|
state = registry._states["dev"]
|
|
state.next_scrape_at = now
|
|
|
|
# 连续失败三次
|
|
for _ in range(3):
|
|
registry.update_after_scrape(
|
|
device_name="dev",
|
|
success=False,
|
|
now=now,
|
|
scrape_interval=60,
|
|
failure_threshold=3,
|
|
max_backoff_factor=8,
|
|
)
|
|
|
|
state = registry._states["dev"]
|
|
assert state.consecutive_failures == 3
|
|
assert state.backoff_factor >= 2
|
|
assert state.next_scrape_at > now
|
|
|
|
|
|
def test_reset_backoff_on_success():
|
|
registry = DeviceRegistry(global_scrape_interval=60)
|
|
now = time.time()
|
|
dev = DeviceConfig(
|
|
name="dev",
|
|
host="h",
|
|
port=830,
|
|
username="u",
|
|
password="p",
|
|
)
|
|
registry.register_static_device(dev)
|
|
state = registry._states["dev"]
|
|
state.consecutive_failures = 5
|
|
state.backoff_factor = 4
|
|
|
|
registry.update_after_scrape(
|
|
device_name="dev",
|
|
success=True,
|
|
now=now,
|
|
scrape_interval=60,
|
|
failure_threshold=3,
|
|
max_backoff_factor=8,
|
|
)
|
|
|
|
state = registry._states["dev"]
|
|
assert state.consecutive_failures == 0
|
|
assert state.backoff_factor == 1
|
|
assert state.next_scrape_at == pytest.approx(now + 60, rel=0.01)
|