146 lines
5.0 KiB
Python
146 lines
5.0 KiB
Python
"""Tests for how level history is fetched out of Zabbix.
|
|
|
|
These are about the shape of the API call, not about the arithmetic done
|
|
afterwards, because that is where the forecast went wrong: the numbers were
|
|
computed correctly from the wrong rows.
|
|
|
|
Zabbix applies `limit` to the whole result set rather than to each item, so a
|
|
capped query sorted ascending returns the OLDEST rows in the window. A
|
|
four-cartridge printer polled every five minutes writes over 100k rows in 90
|
|
days; the report was reading the first few days of that and calling it current.
|
|
No arithmetic recovers from it, and nothing about the output looks wrong -
|
|
which is why it is pinned here.
|
|
"""
|
|
|
|
from plugins.printers.services.zabbix_service import ZabbixService
|
|
|
|
|
|
class RecordingService(ZabbixService):
|
|
"""A service whose API calls are recorded and answered from a script."""
|
|
|
|
def __init__(self, answers=None):
|
|
super().__init__()
|
|
self.calls = []
|
|
self.answers = answers or {}
|
|
|
|
def _apicall(self, method, params):
|
|
self.calls.append((method, params))
|
|
answer = self.answers.get(method)
|
|
if callable(answer):
|
|
return answer(params)
|
|
return answer or []
|
|
|
|
def paramsfor(self, method):
|
|
return [params for name, params in self.calls if name == method]
|
|
|
|
|
|
def rows(itemid, values, start=1780000000, step=3600):
|
|
"""history.get rows, newest first, as a DESC query returns them."""
|
|
return [{'itemid': str(itemid), 'clock': str(start + i * step),
|
|
'value': str(value)} for i, value in enumerate(values)][::-1]
|
|
|
|
|
|
def test_history_asks_for_the_newest_rows_not_the_oldest():
|
|
service = RecordingService({'history.get': rows(1, [90, 80, 70])})
|
|
|
|
service.gethistory(['1'], days=90)
|
|
|
|
params = service.paramsfor('history.get')[0]
|
|
assert params['sortorder'] == 'DESC'
|
|
assert params['sortfield'] == 'clock'
|
|
|
|
|
|
def test_history_budget_scales_with_the_number_of_items():
|
|
"""One cap shared by four cartridges is a quarter of a cartridge each."""
|
|
service = RecordingService({'history.get': rows(1, [90, 80])})
|
|
|
|
service.gethistory(['1', '2', '3', '4'], days=90)
|
|
|
|
params = service.paramsfor('history.get')[0]
|
|
assert params['limit'] == ZabbixService.HISTORY_PER_ITEM * 4
|
|
|
|
|
|
def test_history_comes_back_oldest_first():
|
|
"""The analysis reads a series forwards; DESC is a fetch detail."""
|
|
service = RecordingService({'history.get': rows(1, [90, 80, 70])})
|
|
|
|
result = service.gethistory(['1'], days=90)
|
|
|
|
clocks = [int(clock) for clock, _ in result['1']]
|
|
assert clocks == sorted(clocks)
|
|
assert [value for _, value in result['1']] == ['90', '80', '70']
|
|
|
|
|
|
def test_items_split_across_both_history_tables_are_all_returned():
|
|
"""A printer can carry an integer-typed supply and a float-typed one.
|
|
|
|
Stopping at the first table that answered dropped every item in the other,
|
|
and a missing cartridge looks exactly like a cartridge with no history.
|
|
"""
|
|
def answer(params):
|
|
if params['history'] == 3:
|
|
return rows(1, [90, 80])
|
|
return rows(2, [70, 60])
|
|
|
|
service = RecordingService({'history.get': answer})
|
|
|
|
result = service.gethistory(['1', '2'], days=90)
|
|
|
|
assert set(result) == {'1', '2'}
|
|
|
|
|
|
def test_the_second_table_is_only_asked_about_what_is_still_missing():
|
|
def answer(params):
|
|
return rows(1, [90, 80]) if params['history'] == 3 else []
|
|
|
|
service = RecordingService({'history.get': answer})
|
|
|
|
service.gethistory(['1', '2'], days=90)
|
|
|
|
assert service.paramsfor('history.get')[1]['itemids'] == ['2']
|
|
|
|
|
|
def test_a_long_window_reads_trends_rather_than_every_poll():
|
|
"""90 days of raw readings is 26k rows an item and hours of transfer for a
|
|
number that is a slope."""
|
|
service = RecordingService({'trend.get': [
|
|
{'itemid': '1', 'clock': '1780000000', 'value_avg': '80'},
|
|
{'itemid': '1', 'clock': '1779996400', 'value_avg': '90'},
|
|
]})
|
|
|
|
result = service.getlevelhistory(['1'], days=90)
|
|
|
|
assert service.paramsfor('trend.get')
|
|
assert not service.paramsfor('history.get')
|
|
assert [value for _, value in result['1']] == ['90', '80']
|
|
|
|
|
|
def test_a_short_window_reads_raw_history():
|
|
"""Over a few days the hourly average has too few points to fit."""
|
|
service = RecordingService({'history.get': rows(1, [90, 80])})
|
|
|
|
service.getlevelhistory(['1'], days=2)
|
|
|
|
assert service.paramsfor('history.get')
|
|
assert not service.paramsfor('trend.get')
|
|
|
|
|
|
def test_an_item_with_no_trends_falls_back_to_its_raw_history():
|
|
"""A site can keep trends for 0 days. That is a setting, not a fault."""
|
|
service = RecordingService({
|
|
'trend.get': [{'itemid': '1', 'clock': '1780000000', 'value_avg': '80'}],
|
|
'history.get': rows(2, [70, 60]),
|
|
})
|
|
|
|
result = service.getlevelhistory(['1', '2'], days=90)
|
|
|
|
assert service.paramsfor('history.get')[0]['itemids'] == ['2']
|
|
assert set(result) == {'1', '2'}
|
|
|
|
|
|
def test_no_items_asks_nothing():
|
|
service = RecordingService()
|
|
|
|
assert service.getlevelhistory([], days=90) == {}
|
|
assert service.calls == []
|