feat(P3): powerbi export — CSV/JSON views + 8 placeholder views + data dictionary (REQ-190,199,208,209)

P3 (Wave 2, feat) — REQ-190, REQ-199, REQ-208, REQ-209

New components:
- core/metrics/powerbi_export.py — exports fact/dim tables + 8 placeholder views to CSV/JSON
- tests/test_powerbi_export.py — 6 tests (all pass)
- docs/METRICS_VIEWS.md — column-level data dictionary (REQ-209)
- metrics/powerbi/NOVA_DASHBOARD_README.md — folder-connector import guide + starter visual model (REQ-208)

8 placeholder views (deferred metrics, headers only):
- placeholder_live_infra_health (D-096)
- placeholder_live_outbox_rate (D-096)
- placeholder_tamper_evident_checkpoints (D-083)
- placeholder_onboarding_funnel (D-113/D-114/D-119)
- placeholder_drift_detection (D-096 + no scheduler)
- placeholder_live_cur_reconciliation (D-096)
- placeholder_sla_downtime (D-096)
- placeholder_predictive_reactive (future emitter)

D-120: Nova-native (CSV/JSON files, no live connector)
D-129: PowerBI ingests via folder connector

---ci---
project: acdl
phase: 3
milestone: v1.17
status: execute
---/ci---
This commit is contained in:
Jon Chery
2026-08-04 20:03:12 +00:00
parent 814fea6c3c
commit 3a7604dec0
4 changed files with 498 additions and 0 deletions
+91
View File
@@ -0,0 +1,91 @@
"""Tests for Nova PowerBI export (P3, REQ-190)."""
import json
import os
import sqlite3
import sys
from pathlib import Path
import pytest
ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))
@pytest.fixture
def tmp_export(tmp_path, monkeypatch):
metrics_dir = tmp_path / "metrics"
metrics_dir.mkdir()
export_dir = metrics_dir / "powerbi"
store_db = metrics_dir / "nova_metrics.db"
monkeypatch.setattr("core.metrics.powerbi_export._METRICS_DIR", str(metrics_dir))
monkeypatch.setattr("core.metrics.powerbi_export._STORE_PATH", str(store_db))
monkeypatch.setattr("core.metrics.powerbi_export._EXPORT_DIR", str(export_dir))
return {"metrics_dir": metrics_dir, "store_db": store_db, "export_dir": export_dir}
def _init_store_with_data(db_path):
conn = sqlite3.connect(str(db_path))
conn.executescript("""
CREATE TABLE fact_run (run_id TEXT PRIMARY KEY, contract_id TEXT, environment TEXT, exit_code INTEGER);
CREATE TABLE dim_capability (capability_id TEXT PRIMARY KEY, name TEXT, tier TEXT);
INSERT INTO fact_run VALUES ('run-1', 'cid-1', 'dev', 0);
INSERT INTO dim_capability VALUES ('CAP-001', 'test cap', 'local');
""")
conn.commit()
conn.close()
def test_export_csv(tmp_export):
from core.metrics.powerbi_export import export_all
_init_store_with_data(tmp_export["store_db"])
result = export_all(fmt="csv")
assert (tmp_export["export_dir"] / "fact_run.csv").exists()
assert (tmp_export["export_dir"] / "dim_capability.csv").exists()
assert result["fact_tables"]["fact_run"] == 1
def test_export_json(tmp_export):
from core.metrics.powerbi_export import export_all
_init_store_with_data(tmp_export["store_db"])
result = export_all(fmt="json")
assert (tmp_export["export_dir"] / "fact_run.json").exists()
data = json.loads((tmp_export["export_dir"] / "fact_run.json").read_text())
assert len(data) == 1
assert data[0]["run_id"] == "run-1"
def test_export_placeholder_views(tmp_export):
from core.metrics.powerbi_export import export_all, PLACEHOLDER_VIEWS
result = export_all(fmt="both")
for view_name in PLACEHOLDER_VIEWS:
assert (tmp_export["export_dir"] / f"{view_name}.csv").exists()
assert (tmp_export["export_dir"] / f"{view_name}.json").exists()
assert len(PLACEHOLDER_VIEWS) == 8
def test_export_placeholder_csv_headers_only(tmp_export):
from core.metrics.powerbi_export import export_all
export_all(fmt="csv")
csv_path = tmp_export["export_dir"] / "placeholder_drift_detection.csv"
lines = csv_path.read_text().strip().split("\n")
assert len(lines) == 1 # headers only, no data
assert "timestamp" in lines[0]
def test_export_placeholder_json_schema(tmp_export):
from core.metrics.powerbi_export import export_all
export_all(fmt="json")
json_path = tmp_export["export_dir"] / "placeholder_sla_downtime.json"
data = json.loads(json_path.read_text())
assert "schema" in data
assert data["schema"]["blocking_decision"] == "D-096"
assert data["data"] == []
def test_export_no_store(tmp_export):
from core.metrics.powerbi_export import export_all
result = export_all(fmt="csv")
assert "error" in result
# Placeholders still exported
assert (tmp_export["export_dir"] / "placeholder_drift_detection.csv").exists()