Commit bc753574 authored by Christoph Alt's avatar Christoph Alt
Browse files

started with an dependeny inversion

parent 0af971b8
Pipeline #47133 failed with stages
in 1 minute and 41 seconds
import grafanalib.core
from grafanalib.influxdb import InfluxDBTarget
from .panels import get_arguments
def get_target_class_name(name, *, prefix="Wrapped"):
if not name.startswith(prefix):
raise ValueError(f"{name} does not start with {prefix}")
return name.replace(prefix, "")
def get_grafanalib_object(wrapped_object):
cls_name = get_target_class_name(wrapped_object.__class__.__name__)
if cls_name == "InfluxDBTarget":
target_class = InfluxDBTarget
else:
target_class = getattr(grafanalib.core, cls_name)
kwargs = get_arguments(wrapped_object)
for kwarg, value in kwargs.items():
if value.__class__.__name__.startswith("Wrapped"):
kwargs[kwarg] = get_grafanalib_object(value)
elif isinstance(value, list):
if value and value[0].__class__.__name__.startswith("Wrapped"):
kwargs[kwarg] = [get_grafanalib_object(o) for o in value]
return target_class(**kwargs)
from dataclasses import dataclass, field, fields
from typing import List
@dataclass
class WrappedTarget:
query: str
alias: str = None
class WrappedInfluxDBTarget(WrappedTarget):
pass
@dataclass
class WrappedRepeat:
direction: str
variable: str
maxPerRow: int = None
@dataclass
class WrappedPanel:
title: str
dataSource: str
targets: List[WrappedTarget] = field(default_factory=list)
thresholdType: str = 'absolute'
thresholds: List = field(default_factory=list)
extra_kwargs: dict = field(default_factory=dict)
@dataclass
class WrappedTimeSeries(WrappedPanel):
unit: str = 'none'
axisLabel: str = ''
pointSize: int = 1
repeat: WrappedRepeat = None
@dataclass
class WrappedStat(WrappedPanel):
alignment: str = 'center'
colorMode: str = 'value'
graphMode: str = 'area'
reduceCalc: str = 'last'
orientation: str = 'auto'
transparent: bool = True
@dataclass
class WrappedRowPanel:
title: str
collapsed: bool = True
panels: List[WrappedPanel] = field(default_factory=list)
@dataclass
class WrappedDashboardList(WrappedPanel):
searchTags: List[str] = field(default_factory=list)
showHeadings: bool = False
showSearch: bool = True
showStarred: bool = False
@dataclass
class WrappedTime:
start: str
end: str
@dataclass
class WrappedDashboard:
title: str
description: str
tags: List[str] = field(default_factory=list)
timezone: str = "browser"
rows: List = field(default_factory=list)
panels: List = field(default_factory=list)
templating = None
annotation = None
time: WrappedTime = field(default_factory=lambda: WrappedTime(start='now-30d', end='now'))
def __post_init__(self):
if len(self.rows) != 0 and len(self.panels) != 0:
raise ValueError("Either rows or panels")
@dataclass
class WrappedTemplate:
name: str
query: str
type: str = "query"
includeAll: bool = True
multi: bool = True
regex: str = None
def get_arguments(wrapped) -> dict:
ret = dict()
for cur_field in fields(wrapped):
if cur_field.type == dict:
ret.update(getattr(wrapped, cur_field.name))
elif value := getattr(wrapped, cur_field.name):
ret.setdefault(cur_field.name, value)
return ret
......@@ -8,7 +8,8 @@ setup(name="cb-util",
author_email="Christoph.alt@fau.de",
packages=find_packages(include=["cbutil",
"cbutil.postprocessing",
"dashboards"]),
"dashboards",
"common"]),
install_requires=[
"python-dotenv",
"influxdb",
......
from common.queries import Query, ShowTagQuery
def test_showtagquery_from():
q = ShowTagQuery('host', 'UniformGridCPU')
assert str(q) == 'SHOW TAG VALUES FROM \"UniformGridCPU\" WITH key = \"host\"'
def test_showtagquery_without_from():
q = ShowTagQuery('host')
assert str(q) == 'SHOW TAG VALUES WITH key = \"host\"'
def test_big_query():
q = Query(
_select='mlupsPerProcess',
_from='UniformGridGPU',
_where='"host" =~ /^$host$/ AND "collisionSetup" =~ /^$collisionSetup$/',
_group_by=[
"blocks_0",
"blocks_1",
"blocks_2",
"cellsPerBlock_0",
"cellsPerBlock_1",
"cellsPerBlock_2",
"gpuBlockSize_0",
"gpuBlockSize_1",
"gpuBlockSize_2",
"collisionSetup",
"stencil",
"streamingPattern"
]
)
q1 = ('SELECT "mlupsPerProcess" '
'FROM "UniformGridGPU" '
'WHERE ("host" =~ /^$host$/ AND "collisionSetup" =~ /^$collisionSetup$/) AND $timeFilter '
'GROUP BY "blocks_0", "blocks_1", "blocks_2", '
'"cellsPerBlock_0", "cellsPerBlock_1", "cellsPerBlock_2", '
'"gpuBlockSize_0", "gpuBlockSize_1", "gpuBlockSize_2", '
'"collisionSetup", "stencil", "streamingPattern"')
assert q1 == str(q)
def test_min_query():
q = Query("UniformGridGPU")
assert str(q) == 'SELECT "*" FROM "UniformGridGPU" WHERE $timeFilter'
def test_query_with_select():
q = Query('UniformGridGPU', 'mlupsPerProcess')
assert str(q) == 'SELECT "mlupsPerProcess" FROM "UniformGridGPU" WHERE $timeFilter'
from unittest.mock import patch
from dashboards.grafanalib_backend import (get_grafanalib_object,
get_target_class_name)
from dashboards.panels import (WrappedDashboard, WrappedInfluxDBTarget,
WrappedPanel, WrappedTime, WrappedTimeSeries,
get_arguments)
def test_get_target_class_Panel():
assert get_target_class_name("WrappedPanel") == "Panel"
assert get_target_class_name("PrefixPanel", prefix="Prefix") == "Panel"
panel = WrappedPanel("test", "InfluxDB-1")
assert get_target_class_name(panel.__class__.__name__)
@patch('dashboards.grafanalib_backend.grafanalib.core')
def test_get_grafanalib_panel(mock_grafanalib):
panel = WrappedPanel("test", "InfluxDB-1")
kwargs = get_arguments(panel)
get_grafanalib_object(panel)
mock_grafanalib.Panel.assert_called_with(**kwargs)
@patch('grafanalib.core')
def test_get_grafanalib_ts(mock_grafanalib):
panel = WrappedTimeSeries("test", "InfluxDB-1")
kwargs = get_arguments(panel)
get_grafanalib_object(panel)
mock_grafanalib.TimeSeries.assert_called_with(**kwargs)
@patch('dashboards.grafanalib_backend.InfluxDBTarget')
@patch('grafanalib.core')
def test_get_grafanalib_ts_with_target(mock_grafanalib_core, mock_grafanalib_influx):
target = WrappedInfluxDBTarget('SELECT "mlupsperprocess" FROM "UniformGridCPU"')
panel = WrappedTimeSeries("test", "InfluxDB-1", targets=[target])
get_grafanalib_object(panel)
mock_grafanalib_core.TimeSeries.assert_called()
mock_grafanalib_influx.assert_called_with(**get_arguments(target))
@patch('grafanalib.core')
def test_time(mock_grafanalib):
time = WrappedTime('now-30d', 'now')
get_grafanalib_object(time)
mock_grafanalib.Time.assert_called_with(start='now-30d', end='now')
@patch('grafanalib.core')
def test_dashboard(mock_grafanalib):
dashboard = WrappedDashboard("title", "description")
get_grafanalib_object(dashboard)
mock_grafanalib.Time.assert_called()
mock_grafanalib.Dashboard.assert_called()
from dashboards.panels import WrappedPanel, WrappedTimeSeries, get_arguments
def test_get_arguments_panel():
panel = WrappedPanel("test", "InfluxDB-1")
kwargs = get_arguments(panel)
assert kwargs['title'] == "test"
assert kwargs['dataSource'] == "InfluxDB-1"
def test_get_arguments_ts():
panel = WrappedTimeSeries("test", "InfluxDB-1", unit='W')
kwargs = get_arguments(panel)
assert kwargs['title'] == "test"
assert kwargs['dataSource'] == "InfluxDB-1"
assert kwargs['unit'] == 'W'
def test_extra_kwargs():
panel = WrappedPanel("test", "InfluxDB-1", extra_kwargs={'color': 'red'})
kwargs = get_arguments(panel)
assert kwargs['title'] == "test"
assert kwargs['dataSource'] == "InfluxDB-1"
assert kwargs['color'] == 'red'
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment