import importlib.util import json from pathlib import Path def load_dashboard_module(): module_path = Path(__file__).resolve().parents[1] / "superset" / "create_dashboard.py" spec = importlib.util.spec_from_file_location("create_dashboard", module_path) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module def chart_config(module, slice_name): return next(chart for chart in module.CHARTS_CONFIG if chart["slice_name"] == slice_name) def exported_chart(exported_dashboard, slice_name): return next( chart["__Slice__"] for chart in exported_dashboard["charts"] if chart["__Slice__"]["slice_name"] == slice_name ) def load_exported_dashboard(): export_path = Path(__file__).resolve().parents[1] / "superset" / "dashboards" / "ecommerce_analytics.zip.json" return json.loads(export_path.read_text()) def expected_export_params(chart, actual_params): return { **chart["params"], "datasource": actual_params["datasource"], "viz_type": chart["viz_type"], } class FakeDashboard: def __init__(self, slug): self.slug = slug class FakeChart: def __init__(self, chart_id, slice_name, datasource_id=1, dashboard_slugs=None): self.id = chart_id self.slice_name = slice_name self.datasource_id = datasource_id self.dashboards = [ FakeDashboard(slug) for slug in (dashboard_slugs or ["ecommerce-analytics"]) ] def test_geo_chart_uses_readable_top_countries_bar_config(): module = load_dashboard_module() geo_chart = chart_config(module, "🌍 Top Countries by Events") assert geo_chart["viz_type"] == "echarts_timeseries_bar" assert geo_chart["previous_slice_names"] == ["🌍 Geography Map"] params = geo_chart["params"] assert params["x_axis"] == "geo_country" assert params["metrics"] == [ { "expressionType": "SQL", "sqlExpression": "COUNT(*)", "label": "Events, pcs", } ] assert params["row_limit"] == 15 assert params["order_desc"] is True assert params["sort_series_type"] == "sum" assert params["show_legend"] is True assert params["rich_tooltip"] is True assert params["y_axis_title"] == "Events, pcs" assert params["x_axis_title"] == "Country" assert params["y_axis_format"] == ",d" def test_exported_dashboard_uses_same_readable_geo_chart(): module = load_dashboard_module() exported_dashboard = load_exported_dashboard() expected_chart = chart_config(module, "🌍 Top Countries by Events") geo_chart = exported_chart(exported_dashboard, "🌍 Top Countries by Events") assert geo_chart["viz_type"] == "echarts_timeseries_bar" params = json.loads(geo_chart["params"]) assert params == expected_export_params(expected_chart, params) assert params["viz_type"] == "echarts_timeseries_bar" assert params["x_axis"] == "geo_country" assert params["metrics"][0]["label"] == "Events, pcs" assert params["x_axis_sort"] == "Events, pcs" assert params["x_axis_sort_asc"] is False assert params["show_legend"] is True assert params["rich_tooltip"] is True assert params["y_axis_title"] == "Events, pcs" position_json = json.loads(exported_dashboard["dashboards"][0]["__Dashboard__"]["position_json"]) assert position_json["CHART-7"]["meta"]["sliceName"] == "🌍 Top Countries by Events" def test_exported_dashboard_chart_params_match_current_config(): module = load_dashboard_module() exported_dashboard = load_exported_dashboard() expected_charts = { chart["slice_name"]: chart for chart in module.CHARTS_CONFIG } for exported in exported_dashboard["charts"]: actual_chart = exported["__Slice__"] actual_params = json.loads(actual_chart["params"]) expected_chart = expected_charts[actual_chart["slice_name"]] assert actual_chart["viz_type"] == expected_chart["viz_type"] assert actual_params == expected_export_params(expected_chart, actual_params) def test_exported_dashboard_chart_names_match_current_config(): module = load_dashboard_module() exported_dashboard = load_exported_dashboard() expected_names = [chart["slice_name"] for chart in module.CHARTS_CONFIG] actual_names = [chart["__Slice__"]["slice_name"] for chart in exported_dashboard["charts"]] assert actual_names == expected_names def test_exported_dashboard_layout_matches_current_rows(): module = load_dashboard_module() exported_dashboard = load_exported_dashboard() position_json = json.loads(exported_dashboard["dashboards"][0]["__Dashboard__"]["position_json"]) chart_titles_by_id = { component_id: component["meta"]["sliceName"] for component_id, component in position_json.items() if isinstance(component, dict) and component.get("type") == "CHART" } actual_rows = [] for row_id in position_json["GRID_ID"]["children"]: row = [] for chart_component_id in position_json[row_id]["children"]: chart_component = position_json[chart_component_id] row.append( ( chart_titles_by_id[chart_component_id], chart_component["meta"]["width"], ) ) actual_rows.append(row) assert actual_rows == module.DASHBOARD_ROWS def test_choose_chart_to_sync_prefers_current_name_and_marks_old_name_duplicate(): module = load_dashboard_module() old_chart = FakeChart(7, "🌍 Geography Map", datasource_id=42) current_chart = FakeChart(11, "🌍 Top Countries by Events", datasource_id=42) selected, duplicates = module.choose_chart_to_sync( [old_chart, current_chart], "🌍 Top Countries by Events", dataset_id=42, ) assert selected is current_chart assert duplicates == [old_chart] def test_choose_chart_to_sync_ignores_same_name_from_other_dataset(): module = load_dashboard_module() other_dataset_chart = FakeChart(7, "🌍 Top Countries by Events", datasource_id=7) target_dataset_chart = FakeChart(11, "🌍 Geography Map", datasource_id=42) selected, duplicates = module.choose_chart_to_sync( [other_dataset_chart, target_dataset_chart], "🌍 Top Countries by Events", dataset_id=42, ) assert selected is target_dataset_chart assert duplicates == [] def test_dashboard_sync_ignores_same_name_same_dataset_from_other_dashboard(): module = load_dashboard_module() other_dashboard_chart = FakeChart( 7, "🌍 Top Countries by Events", datasource_id=42, dashboard_slugs=["mentee-dashboard"], ) target_dashboard_chart = FakeChart( 11, "🌍 Geography Map", datasource_id=42, dashboard_slugs=["ecommerce-analytics"], ) owned_charts = module.dashboard_owned_charts( [other_dashboard_chart, target_dashboard_chart], dashboard_slug="ecommerce-analytics", ) selected, duplicates = module.choose_chart_to_sync( owned_charts, "🌍 Top Countries by Events", dataset_id=42, ) assert owned_charts == [target_dashboard_chart] assert selected is target_dashboard_chart assert duplicates == []