133 lines
4.2 KiB
Python
133 lines
4.2 KiB
Python
import time
|
|
|
|
from exporter.metrics import TransceiverCollector
|
|
from exporter.models import (
|
|
DeviceHealthState,
|
|
DeviceMetricsSnapshot,
|
|
TransceiverChannelRecord,
|
|
TransceiverRecord,
|
|
)
|
|
|
|
|
|
def _collect_metrics(collector: TransceiverCollector):
|
|
return list(collector.collect())
|
|
|
|
|
|
def test_collector_exports_transceiver_and_info_with_component_name():
|
|
snapshot = DeviceMetricsSnapshot(
|
|
device="dev1",
|
|
collected_at=time.time(),
|
|
transceivers=(
|
|
TransceiverRecord(
|
|
device="dev1",
|
|
component_name="comp1",
|
|
logical_port="1/0/1",
|
|
present="PRESENT",
|
|
oper_status="ACTIVE",
|
|
temperature_c=40.0,
|
|
supply_voltage_v=3.3,
|
|
vendor="H3C",
|
|
serial="SN1",
|
|
part_number="PN",
|
|
hardware_rev="1.0",
|
|
),
|
|
),
|
|
channels=(),
|
|
)
|
|
cache = {"dev1": snapshot}
|
|
health = {
|
|
"dev1": DeviceHealthState(
|
|
last_scrape_success=True,
|
|
last_scrape_duration=0.1,
|
|
last_scrape_timestamp=time.time(),
|
|
last_error_type=None,
|
|
)
|
|
}
|
|
|
|
collector = TransceiverCollector(cache, health)
|
|
metrics = _collect_metrics(collector)
|
|
|
|
temp_metric = [m for m in metrics if m.name == "transceiver_temperature_celsius"][0]
|
|
sample = temp_metric.samples[0]
|
|
assert sample.value == 40.0
|
|
assert sample.labels["device"] == "dev1"
|
|
assert sample.labels["port"] == "1/0/1"
|
|
assert sample.labels["component_name"] == "comp1"
|
|
|
|
info_metric = [m for m in metrics if m.name == "transceiver_info"][0]
|
|
info_labels = info_metric.samples[0].labels
|
|
assert info_labels["device"] == "dev1"
|
|
assert info_labels["component_name"] == "comp1"
|
|
assert info_labels["vendor"] == "H3C"
|
|
assert info_labels["serial"] == "SN1"
|
|
|
|
|
|
def test_collector_channel_metrics_include_component_name():
|
|
snapshot = DeviceMetricsSnapshot(
|
|
device="dev1",
|
|
collected_at=time.time(),
|
|
transceivers=(),
|
|
channels=(
|
|
TransceiverChannelRecord(
|
|
device="dev1",
|
|
component_name="comp1",
|
|
logical_port="1/0/1",
|
|
channel_index=0,
|
|
logical_channel="1/0/1:1",
|
|
rx_power_dbm=-3.0,
|
|
tx_power_dbm=-1.0,
|
|
bias_current_ma=10.0,
|
|
laser_temperature_c=25.0,
|
|
),
|
|
),
|
|
)
|
|
cache = {"dev1": snapshot}
|
|
health = {
|
|
"dev1": DeviceHealthState(
|
|
last_scrape_success=True,
|
|
last_scrape_duration=0.1,
|
|
last_scrape_timestamp=time.time(),
|
|
last_error_type=None,
|
|
)
|
|
}
|
|
|
|
collector = TransceiverCollector(cache, health)
|
|
metrics = _collect_metrics(collector)
|
|
|
|
tx_metric = [m for m in metrics if m.name == "transceiver_channel_tx_power_dbm"][0]
|
|
sample = tx_metric.samples[0]
|
|
assert sample.labels["device"] == "dev1"
|
|
assert sample.labels["port"] == "1/0/1"
|
|
assert sample.labels["channel"] == "1/0/1:1"
|
|
assert sample.labels["component_name"] == "comp1"
|
|
|
|
info_metric = [m for m in metrics if m.name == "transceiver_channel_info"][0]
|
|
info_labels = info_metric.samples[0].labels
|
|
assert info_labels["device"] == "dev1"
|
|
assert info_labels["channel"] == "1/0/1:1"
|
|
assert info_labels["component_name"] == "comp1"
|
|
|
|
|
|
def test_collector_no_business_metrics_on_failure():
|
|
cache = {} # 没有成功快照
|
|
health = {
|
|
"dev1": DeviceHealthState(
|
|
last_scrape_success=False,
|
|
last_scrape_duration=0.2,
|
|
last_scrape_timestamp=time.time(),
|
|
last_error_type="TimeoutError",
|
|
)
|
|
}
|
|
|
|
collector = TransceiverCollector(cache, health)
|
|
metrics = _collect_metrics(collector)
|
|
|
|
# 有健康指标
|
|
success_metric = [m for m in metrics if m.name == "netconf_scrape_success"][0]
|
|
assert any(s.labels["device"] == "dev1" and s.value == 0 for s in success_metric.samples)
|
|
|
|
# 不应有 transceiver_temperature_celsius
|
|
temp_metrics = [m for m in metrics if m.name == "transceiver_temperature_celsius"]
|
|
assert not temp_metrics or not temp_metrics[0].samples
|
|
|