"""Regression tests for the generated Grafana v2 dashboard, without Grafana I/O.""" import importlib.util import json from pathlib import Path import pytest ROOT = Path(__file__).resolve().parents[1] @pytest.fixture(scope="module") def generator(): spec = importlib.util.spec_from_file_location( "generate_grafana_ups", ROOT / "tools/generate_grafana_ups.py", ) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module @pytest.fixture def panels(generator): return { panel["spec"]["title"]: panel["spec"] for panel in generator.build_dashboard()["elements"].values() } def queries(panel): return { query["spec"]["refId"]: query["spec"]["query"]["spec"]["query"] for query in panel["data"]["spec"]["queries"] } def properties_for(panel, ref): overrides = panel["vizConfig"]["spec"]["fieldConfig"]["overrides"] return { prop["id"]: prop["value"] for override in overrides if override["matcher"] == {"id": "byFrameRefID", "options": ref} for prop in override["properties"] } @pytest.mark.parametrize("title,primary,label,unit,secondary,secondary_unit,display_name", [ ("Network · Cell voltage history", "bms_voltage_cells_volts", "cell", "volt", "bms_current_amps", "amp", "Current"), ("Network · Temperature history", "bms_temperature_celsius", "sensor", "celsius", "bms_capacity_remaining_ah", "amph", "Remaining Capacity"), ]) def test_network_histories_keep_original_series_and_separate_axes( panels, title, primary, label, unit, secondary, secondary_unit, display_name, ): panel = panels[title] query = queries(panel) assert set(query) == {"A", "B"} assert f'r["_measurement"] == "{primary}"' in query["A"] assert f'r["_measurement"] == "{secondary}"' in query["B"] # Cell and sensor identities survive grouping and column selection. keys = ["ups", "_measurement", "_field", label] assert f'group(columns: {json.dumps(keys)})' in query["A"] assert f'keep(columns: {json.dumps(["_time", "_value"] + keys)})' in query["A"] defaults = panel["vizConfig"]["spec"]["fieldConfig"]["defaults"] assert defaults["unit"] == unit assert "${__field.labels." + label + "}" in defaults["displayName"] secondary_properties = properties_for(panel, "B") assert secondary_properties["unit"] == secondary_unit assert secondary_properties["displayName"] == display_name assert secondary_properties["custom.axisPlacement"] == "right" assert secondary_properties["custom.axisLabel"] @pytest.mark.parametrize("title", [ "Network · Cell voltage history", "Network · Temperature history", ]) def test_network_histories_do_not_invent_empty_intervals(panels, title): panel = panels[title] for query in queries(panel).values(): assert 'from(bucket: "${bucket:raw}")' in query assert 'r["ups"] == "network"' in query assert 'r["_field"] == "value"' in query assert 'range(start: v.timeRangeStart, stop: v.timeRangeStop)' in query assert 'sort(columns: ["_time"])' in query assert 'aggregateWindow(' not in query assert 'last()' not in query assert 'fill(' not in query assert 'map(' not in query # In particular, do not change the current's sign. custom = panel["vizConfig"]["spec"]["fieldConfig"]["defaults"]["custom"] assert custom["insertNulls"] is False assert custom["spanNulls"] is False # Keep actual nulls, rather than hiding them. assert custom["drawStyle"] == "line" assert custom["lineInterpolation"] == "linear" assert panel["data"]["spec"]["transformations"] == [] @pytest.mark.parametrize("device,title", [("desktop", "Desktop"), ("nas", "NAS")]) def test_ecoflow_histories_average_each_display_interval(panels, device, title): for suffix, refs in (("Temperature history", {"A", "B"}), ("Input / output history", {"A", "B", "C"})): panel = panels[f"{title} · {suffix}"] assert set(queries(panel)) == refs for query in queries(panel).values(): assert f'r["ups"] == "{device}"' in query assert 'aggregateWindow(every: v.windowPeriod, fn: mean, createEmpty: true)' in query assert 'float(v:' not in query # An empty interval stays null. @pytest.mark.parametrize("title,ref,measurement,divisor,expected", [ # River 2 reports pack current and remaining capacity. River 3 reports # neither, so its panels use battery power and charge. ("Desktop · Temperature history", "B", "ef_river2_bms_remain_cap", "1000.0", { "displayName": "Remaining Capacity", "unit": "amph", "custom.axisPlacement": "right", "custom.axisLabel": "Remaining capacity (Ah)"}), ("NAS · Temperature history", "B", "bms_capacity_charge_ratio", None, { "displayName": "Charge", "unit": "percentunit", "custom.axisPlacement": "right", "custom.axisLabel": "Charge (%)"}), ("Desktop · Input / output history", "C", "ef_river2_bms_amp", None, { "displayName": "Current", "unit": "amp", "custom.axisPlacement": "right", "custom.axisLabel": "Current (A)"}), ("NAS · Input / output history", "C", "ef_river3_display_pow_get_bms", None, { "displayName": "Battery"}), ]) def test_ecoflow_histories_add_a_battery_series(panels, title, ref, measurement, divisor, expected): panel = panels[title] query = queries(panel)[ref] assert f'r["_measurement"] == "{measurement}"' in query assert ("r._value / " + divisor in query) if divisor else ("map(" not in query) properties = properties_for(panel, ref) assert expected.items() <= properties.items() # Battery power stays on the watt axis of the panel. if "unit" not in expected: assert "unit" not in properties and "custom.axisPlacement" not in properties @pytest.mark.parametrize("title,series", [ # The River 2 decoder omits the estimate of the idle direction. ("Desktop", {"A": ("ef_discharge_remaining_time_minutes", "To empty"), "B": ("ef_charge_remaining_time_minutes", "To full")}), # River 3 sends both estimates at all times. Use the one in use only. ("NAS", {"A": ("ef_remaining_time_minutes", "Time remaining")}), ]) def test_time_remaining_shows_the_direction_in_use_in_hours(panels, title, series): panel = panels[f"{title} · Time remaining"] assert set(queries(panel)) == set(series) for ref, (measurement, name) in series.items(): query = queries(panel)[ref] assert f'r["_measurement"] == "{measurement}"' in query assert 'range(start: -5m, stop: now())' in query assert query.index("last()") < query.index("float(v: r._value) / 60.0") assert properties_for(panel, ref) == {"displayName": name} defaults = panel["vizConfig"]["spec"]["fieldConfig"]["defaults"] # A fixed unit: Grafana's duration units round 76 hours to "3 days". assert defaults["unit"] == "suffix: h" assert defaults["decimals"] == 1 assert defaults["noValue"] == "No estimate" assert not any("estimate" in name for name in panels if name != f"{title} · Time remaining") def test_only_the_time_remaining_panels_change_the_no_value_text(panels): for title, panel in panels.items(): expected = "No estimate" if title.endswith("Time remaining") else "No recent data" assert panel["vizConfig"]["spec"]["fieldConfig"]["defaults"]["noValue"] == expected def test_latest_readings_still_use_recent_samples(generator): query = generator.flux("network", "bms_voltage_cells_volts", label="cell") assert 'range(start: -5m, stop: now())' in query assert 'sort(columns: ["_time"])' in query assert 'last()' in query assert 'aggregateWindow(' not in query def test_generator_writes_the_dashboard_file(generator, tmp_path, monkeypatch): output = tmp_path / "dashboard.json" monkeypatch.setattr("sys.argv", ["generate_grafana_ups.py", "--output", str(output)]) generator.main() assert output.read_text(encoding="utf-8") == generator.render() def test_local_dashboard_file_is_current(generator): """Git ignores the generated file, so a clone has none until the generator runs.""" dashboard = ROOT / "grafana-ups.json" if not dashboard.exists(): pytest.skip("grafana-ups.json is not generated") assert generator.render() == dashboard.read_text(encoding="utf-8") def test_api_wrapper_is_optional(generator): dashboard = json.loads(generator.render()) assert "apiVersion" not in dashboard resource = json.loads(generator.render(api_resource=True)) assert resource["spec"] == dashboard assert resource["apiVersion"] == "dashboard.grafana.app/v2" def test_top_row_has_charge_and_capacity_or_flow_without_reading_age(generator): dashboard = generator.build_dashboard() top = {} for item in dashboard["layout"]["spec"]["items"]: spec = item["spec"] if spec["y"] == 0: title = dashboard["elements"][spec["element"]["name"]]["spec"]["title"] top[title] = (spec["x"], spec["width"]) expected = {} for col, (_, title, model) in enumerate(generator.DEVICES): expected[f"{title} · Charge ({model})"] = (col * 8, 5) secondary = "Capacity" if model == "JBD" else "Battery power flow" expected[f"{title} · {secondary}"] = (col * 8 + 5, 3) assert top == expected assert all("Last export" not in element["spec"]["title"] for element in dashboard["elements"].values()) for element in dashboard["elements"].values(): assert all("int(v: now())" not in query for query in queries(element["spec"]).values()) def test_three_column_layout_stays_within_original_height(generator): dashboard = generator.build_dashboard() occupied = set() for item in dashboard["layout"]["spec"]["items"]: spec = item["spec"] assert spec["element"]["name"] in dashboard["elements"] assert 0 <= spec["x"] < spec["x"] + spec["width"] <= 24 assert 0 <= spec["y"] < spec["y"] + spec["height"] <= 24 cells = {(x, y) for x in range(spec["x"], spec["x"] + spec["width"]) for y in range(spec["y"], spec["y"] + spec["height"])} assert not occupied.intersection(cells) occupied.update(cells)