Jon Chery
3a7604dec0
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---
91 lines
3.3 KiB
Python
91 lines
3.3 KiB
Python
"""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() |