From c4e13ccd9326a8552898d73bb71776bced1b022b Mon Sep 17 00:00:00 2001 From: Zio Gabber <78922322+Gabrymi93@users.noreply.github.com> Date: Mon, 10 Aug 2026 13:46:46 +0100 Subject: [PATCH 1/2] chore(registry): load_catalog/load_signals dal registry fusion (registry.json) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Sostituisce clean_catalog.json/pipeline_signals.json con registry.json: - load_catalog: legge registry.json (sezione datasets, struttura identica) - load_signals: legge registry.json e normalizza il blocco run → sample_run (status/checked_at/run_url derivati da run_id/started_at) per compat con le pagine - test: campionato su registry fusion, verifica normalizzazione sample_run --- sources.py | 34 +++++++++++++++++++++++++++++++--- tests/test_sources.py | 32 ++++++++++++++++++++++++-------- 2 files changed, 55 insertions(+), 11 deletions(-) diff --git a/sources.py b/sources.py index d61786b..35431e0 100644 --- a/sources.py +++ b/sources.py @@ -67,8 +67,10 @@ def _fetch_yaml(url: str) -> dict: # ── Caricatori con cache — errori mostrati nella UI ────────────────────────────── @st.cache_data(ttl=300, show_spinner=False) def load_catalog(): + """Catalogo dataset dal registry fusion (registry.json).""" try: - return _fetch_json(f"{REGISTRY_BASE}/clean_catalog.json") + reg = _fetch_json(f"{REGISTRY_BASE}/registry.json") + return reg except Exception as e: st.error(f"❌ Catalogo non disponibile: {e}") return {} @@ -76,11 +78,37 @@ def load_catalog(): @st.cache_data(ttl=300, show_spinner=False) def load_signals(): + """Segnali pipeline dal registry fusion (registry.json). + + Restituisce un payload compatibile col vecchio pipeline_signals.json: + il blocco ``run`` del registry viene esposto come ``sample_run`` (i campi + ``checked_at``/``run_url`` non esistono più; si derivano da ``started_at`` + e ``run_id``) così i consumer non cambiano. + """ try: - return _fetch_json(f"{REGISTRY_BASE}/pipeline_signals.json") + reg = _fetch_json(f"{REGISTRY_BASE}/registry.json") + signals = [] + for s in reg.get("signals", []): + sig = dict(s) + run = s.get("run") or {} + if run: + sr = { + "status": "passed" + if run.get("status") == "SUCCESS" + else run.get("status", "").lower(), + "run_id": run.get("run_id", ""), + "checked_at": (run.get("started_at") or "")[:10], + "run_url": f"https://github.com/dataciviclab/dataset-incubator/actions/runs/{run.get('run_id', '')}" + if run.get("run_id") + else "", + "year": run.get("year"), + } + sig["sample_run"] = sr + signals.append(sig) + return {"schema_version": reg.get("schema_version", "1"), "signals": signals} except Exception as e: st.error(f"❌ Segnali pipeline non disponibili: {e}") - return {} + return {"signals": []} @st.cache_data(ttl=300, show_spinner=False) diff --git a/tests/test_sources.py b/tests/test_sources.py index 1913da7..90c6d5d 100644 --- a/tests/test_sources.py +++ b/tests/test_sources.py @@ -121,7 +121,7 @@ def test_http_error(self): ("load_sources_dashboard", load_sources_dashboard, {"sources": []}), ("load_inventory_report", load_inventory_report, {}), ("load_catalog", load_catalog, {}), - ("load_signals", load_signals, {}), + ("load_signals", load_signals, {"signals": []}), ] @@ -214,11 +214,25 @@ def test_loader_fallback_on_http_error(name, loader, expected_fallback): "sources": {"istat_sdmx": {"status": "ok", "rows": 4849, "method": "dataflow_count"}} } -CATALOG_SAMPLE = {"datasets": [{"slug": "test", "stage": "published"}]} - -PIPELINE_SAMPLE = {"signals": [{"id": "test", "status": "ok"}]} +# Registry fusion (registry.json) — fonte unica per load_catalog/load_signals +REGISTRY_SAMPLE = { + "schema_version": 1, + "datasets": [{"slug": "test", "name": "Test", "stage": "published", "period": {}}], + "signals": [ + { + "id": "test", + "status": "ok", + "run": { + "run_id": "20260101T000000Z_abc", + "year": 2025, + "status": "SUCCESS", + "started_at": "2026-01-01T00:00:00+00:00", + }, + } + ], +} -REGISTRY_SAMPLE = """istat_sdmx: +REGISTRY_SAMPLE_YAML = """istat_sdmx: protocol: sdmx verdict: go observation_mode: catalog-watch @@ -263,17 +277,19 @@ def test_load_inventory_report(self, mock_get): result = load_inventory_report() assert result["sources"]["istat_sdmx"]["rows"] == 4849 - @patch("sources._HTTP.get", return_value=_resp(CATALOG_SAMPLE)) + @patch("sources._HTTP.get", return_value=_resp(REGISTRY_SAMPLE)) def test_load_catalog(self, mock_get): result = load_catalog() assert len(result["datasets"]) == 1 - @patch("sources._HTTP.get", return_value=_resp(PIPELINE_SAMPLE)) + @patch("sources._HTTP.get", return_value=_resp(REGISTRY_SAMPLE)) def test_load_signals(self, mock_get): result = load_signals() assert len(result["signals"]) == 1 + # il blocco run del registry viene normalizzato in sample_run + assert result["signals"][0]["sample_run"]["status"] == "passed" - @patch("sources._HTTP.get", return_value=_yaml_resp(REGISTRY_SAMPLE)) + @patch("sources._HTTP.get", return_value=_yaml_resp(REGISTRY_SAMPLE_YAML)) def test_load_sources_registry(self, mock_get): result = load_sources_registry() assert result["istat_sdmx"]["verdict"] == "go" From 5ff6cc26ae3da72bcd64aaf1b93f774b65e0f436 Mon Sep 17 00:00:00 2001 From: Zio Gabber <78922322+Gabrymi93@users.noreply.github.com> Date: Mon, 10 Aug 2026 13:55:19 +0100 Subject: [PATCH 2/2] fix(dashboard): usa 'run' invece di 'sample_run' nei signals MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Il vecchio pipeline_signals.json esponeva il run come 'latest_run'; le pagine usavano 'sample_run' (campo inesistente → bug latente: i run non venivano mai mostrati). Con il registry, il blocco è 'run': normalizzazione loader + pagine allineate a 'run', fix del bug latente. --- pages/00_Vista_Insieme.py | 2 +- pages/02_Pipeline_Health.py | 10 +++++----- sources.py | 10 ++++------ tests/test_sources.py | 4 ++-- 4 files changed, 12 insertions(+), 14 deletions(-) diff --git a/pages/00_Vista_Insieme.py b/pages/00_Vista_Insieme.py index 518a6d0..b8f8ddf 100644 --- a/pages/00_Vista_Insieme.py +++ b/pages/00_Vista_Insieme.py @@ -196,7 +196,7 @@ st.dataframe(col_df, hide_index=True, width="stretch") # -- Alert run falliti -- -run_failed = [s for s in sigs if s.get("sample_run", {}).get("status") == "failed"] +run_failed = [s for s in sigs if s.get("run", {}).get("status") == "failed"] if run_failed: st.warning( f"⚠️ **{len(run_failed)} candidate con run CI fallito** " diff --git a/pages/02_Pipeline_Health.py b/pages/02_Pipeline_Health.py index a8bf262..4c313d8 100644 --- a/pages/02_Pipeline_Health.py +++ b/pages/02_Pipeline_Health.py @@ -45,7 +45,7 @@ for sig in sigs: slug = sig["id"].replace("-", "_") if slug not in catalog_slugs: - sr = sig.get("sample_run", {}) or {} + sr = sig.get("run", {}) or {} candidati.append( { "slug": slug, @@ -67,7 +67,7 @@ stage = ds.get("stage", "") slug = ds["slug"] sig_data = signals_by_slug.get(slug, {}) - sr = sig_data.get("sample_run", {}) or {} + sr = sig_data.get("run", {}) or {} item = { "slug": slug, "name": ds.get("name", ""), @@ -87,7 +87,7 @@ # Elenco completo di tutti i segnali con run falliti (candidate + catalogo) all_failed = [] for sig in sigs: - sr = sig.get("sample_run", {}) or {} + sr = sig.get("run", {}) or {} if sr.get("status") == "failed": slug = sig["id"].replace("-", "_") all_failed.append( @@ -117,8 +117,8 @@ ok_count = sum(1 for s in sigs if s.get("status") == "ok") warn_count = sum(1 for s in sigs if s.get("status") == "warn") err_count = sum(1 for s in sigs if s.get("status") == "error") -run_passed = sum(1 for s in sigs if s.get("sample_run", {}).get("status") == "passed") -run_failed = sum(1 for s in sigs if s.get("sample_run", {}).get("status") == "failed") +run_passed = sum(1 for s in sigs if s.get("run", {}).get("status") == "passed") +run_failed = sum(1 for s in sigs if s.get("run", {}).get("status") == "failed") run_none = len(sigs) - run_passed - run_failed max_n = max(n_intake, n_validation, n_published, 1) diff --git a/sources.py b/sources.py index 35431e0..208ec39 100644 --- a/sources.py +++ b/sources.py @@ -80,10 +80,9 @@ def load_catalog(): def load_signals(): """Segnali pipeline dal registry fusion (registry.json). - Restituisce un payload compatibile col vecchio pipeline_signals.json: - il blocco ``run`` del registry viene esposto come ``sample_run`` (i campi - ``checked_at``/``run_url`` non esistono più; si derivano da ``started_at`` - e ``run_id``) così i consumer non cambiano. + Il blocco ``run`` del registry viene esposto direttamente (chiave ``run``): + i campi ``checked_at``/``run_url`` non esistono nel registry, si derivano + da ``started_at`` e ``run_id``. """ try: reg = _fetch_json(f"{REGISTRY_BASE}/registry.json") @@ -92,7 +91,7 @@ def load_signals(): sig = dict(s) run = s.get("run") or {} if run: - sr = { + sig["run"] = { "status": "passed" if run.get("status") == "SUCCESS" else run.get("status", "").lower(), @@ -103,7 +102,6 @@ def load_signals(): else "", "year": run.get("year"), } - sig["sample_run"] = sr signals.append(sig) return {"schema_version": reg.get("schema_version", "1"), "signals": signals} except Exception as e: diff --git a/tests/test_sources.py b/tests/test_sources.py index 90c6d5d..7cd5250 100644 --- a/tests/test_sources.py +++ b/tests/test_sources.py @@ -286,8 +286,8 @@ def test_load_catalog(self, mock_get): def test_load_signals(self, mock_get): result = load_signals() assert len(result["signals"]) == 1 - # il blocco run del registry viene normalizzato in sample_run - assert result["signals"][0]["sample_run"]["status"] == "passed" + # il blocco run del registry viene esposto come run (status normalizzato) + assert result["signals"][0]["run"]["status"] == "passed" @patch("sources._HTTP.get", return_value=_yaml_resp(REGISTRY_SAMPLE_YAML)) def test_load_sources_registry(self, mock_get):