diff --git a/setup.py b/setup.py index c3893a4..b17d0ce 100644 --- a/setup.py +++ b/setup.py @@ -67,6 +67,10 @@ def _read_requirements(filename): # schema/ingest.v1.json, and dispatches to the right ingestor. "console_scripts": [ "tracebloc-ingest = tracebloc_ingestor.cli.run:main", + # One-time pre-cutover metadata backfill (backend#1166/#1198): reads + # DB + backend creds from the same env as the ingestor (no + # ingest.yaml) and sweeps this client's registered datasets. + "tracebloc-backfill = tracebloc_ingestor.metadata_backfill_runner:main", ], }, classifiers=[ diff --git a/tests/test_api_client_methods.py b/tests/test_api_client_methods.py index b7ca4b6..2830d2a 100644 --- a/tests/test_api_client_methods.py +++ b/tests/test_api_client_methods.py @@ -263,3 +263,85 @@ def test_refresh_token_false_when_reauth_raises(): client.token = "old_token" with patch.object(client, "authenticate", side_effect=RuntimeError("auth down")): assert client._refresh_token() is False + + +# --------------------------------------------------------------------------- +# get_dataset_metadata() — read-by-ingestor_id (backend#1198) +# --------------------------------------------------------------------------- + + +def test_get_dataset_metadata_success(): + client = _client() + body = {"table_name": "tb", "category": "tabular_classification", "meta_data": {}} + with patch.object(client.session, "get", return_value=_resp(200, body)): + result = client.get_dataset_metadata("ing-1") + assert result == body + + +def test_get_dataset_metadata_404_returns_none(): + """A 404 (no dataset for this ingestor_id owned by this edge) is a skip + signal for the sweep, not an error — return None rather than raise.""" + client = _client() + with patch.object(client.session, "get", return_value=_resp(404, text="nope")): + assert client.get_dataset_metadata("missing") is None + + +def test_get_dataset_metadata_http_error_raises(): + client = _client() + with patch.object(client.session, "get", return_value=_resp(500, text="boom")): + with pytest.raises(requests.exceptions.RequestException): + client.get_dataset_metadata("ing-1") + + +def test_get_dataset_metadata_local_mode(): + client = _client(EDGE_ENV="local") + with patch.object(client.session, "get") as get: + assert client.get_dataset_metadata("ing-1") is None + get.assert_not_called() + + +# --------------------------------------------------------------------------- +# send_metadata_backfill() — upsert recomputed metadata (backend#1166/#1198) +# --------------------------------------------------------------------------- + + +def test_send_metadata_backfill_success(): + client = _client() + body = {"table_name": "tb", "created": True, "competitions_refolded": 2} + with patch.object(client.session, "post", return_value=_resp(201, body)): + result = client.send_metadata_backfill("tb", {"x": {"dtype": "int"}}, {}) + assert result == body + + +def test_send_metadata_backfill_http_error_raises(): + client = _client() + with patch.object(client.session, "post", return_value=_resp(404, text="no table")): + with pytest.raises(requests.exceptions.RequestException): + client.send_metadata_backfill("tb", {"x": {"dtype": "int"}}, {}) + + +def test_send_metadata_backfill_local_mode(): + client = _client(EDGE_ENV="local") + with patch.object(client.session, "post") as post: + result = client.send_metadata_backfill("tb", {"x": {"dtype": "int"}}, {}) + post.assert_not_called() + assert result["table_name"] == "tb" + + +def test_send_metadata_backfill_payload_shape(): + import json as _json + + client = _client() + captured = {} + + def _capture(url, **kwargs): + captured["url"] = url + captured["data"] = kwargs.get("data") + return _resp(201, {"table_name": "tb", "created": False, "competitions_refolded": 0}) + + with patch.object(client.session, "post", side_effect=_capture): + client.send_metadata_backfill("tb", {"x": {"dtype": "int"}}, {"attributes": {}}) + + assert captured["url"].endswith("/global_meta/metadata_backfill/tb/") + sent = _json.loads(captured["data"]) + assert sent == {"schema": {"x": {"dtype": "int"}}, "meta_data": {"attributes": {}}} diff --git a/tests/test_data_validator.py b/tests/test_data_validator.py index c39360a..4df3f0c 100644 --- a/tests/test_data_validator.py +++ b/tests/test_data_validator.py @@ -8,11 +8,11 @@ from tracebloc_ingestor.validators.data_validator import DataValidator - # --------------------------------------------------------------------------- # validate() top-level flow # --------------------------------------------------------------------------- + def test_no_schema_passes_with_warning(): result = DataValidator().validate(pd.DataFrame({"a": [1]})) assert result.is_valid @@ -73,7 +73,7 @@ def test_loads_from_json(tmp_path): Surfaced by an end-to-end cluster ingestion: a 20-record JSON file with explicit schema {id INT, age INT, score FLOAT, active BOOL, label - VARCHAR(20)} failed with "Data Validator Validator failed: No data found + VARCHAR(20)} failed with "Data Validator failed: No data found to validate" before any record was read. """ p = tmp_path / "d.json" @@ -186,9 +186,7 @@ def test_loads_from_json_scalar_array_fails_not_silently_passes(tmp_path): """ p = tmp_path / "scalars.json" p.write_text("[1, 2, 3]") - result = DataValidator(schema={"id": "INT", "label": "VARCHAR(8)"}).validate( - str(p) - ) + result = DataValidator(schema={"id": "INT", "label": "VARCHAR(8)"}).validate(str(p)) assert not result.is_valid assert "No data found" in result.errors[0] @@ -199,9 +197,7 @@ def test_loads_from_json_mixed_array_keeps_only_objects(tmp_path): the dict records (bugbot #233).""" p = tmp_path / "mixed.json" p.write_text('[{"id": 1, "label": "A"}, 42, {"id": 2, "label": "B"}]') - result = DataValidator(schema={"id": "INT", "label": "VARCHAR(8)"}).validate( - str(p) - ) + result = DataValidator(schema={"id": "INT", "label": "VARCHAR(8)"}).validate(str(p)) assert result.is_valid, f"expected valid; errors={result.errors}" assert result.metadata["rows_checked"] == 2 @@ -218,13 +214,9 @@ def test_loads_from_json_jsonl_surfaces_clear_error(tmp_path): """ p = tmp_path / "events.json" p.write_text( - '{"id":1,"label":"A"}\n' - '{"id":2,"label":"B"}\n' - '{"id":3,"label":"A"}\n' - ) - result = DataValidator(schema={"id": "INT", "label": "VARCHAR(8)"}).validate( - str(p) + '{"id":1,"label":"A"}\n' '{"id":2,"label":"B"}\n' '{"id":3,"label":"A"}\n' ) + result = DataValidator(schema={"id": "INT", "label": "VARCHAR(8)"}).validate(str(p)) assert not result.is_valid err = result.errors[0] assert "JSONL" in err or "newline-delimited" in err @@ -392,6 +384,7 @@ def test_dataframe_sparse_schema_field_still_passes(): # INT / BIGINT # --------------------------------------------------------------------------- + def test_int_valid(): df = pd.DataFrame({"n": [1, 2, 3]}) assert DataValidator(schema={"n": "INT"}).validate(df).is_valid @@ -420,7 +413,10 @@ def test_bigint_delegates_to_int(): # FLOAT / DOUBLE / DECIMAL # --------------------------------------------------------------------------- -@pytest.mark.parametrize("dtype", ["FLOAT", "DOUBLE", "DECIMAL(10,2)", "NUMERIC(8,3)", "NUMERIC"]) + +@pytest.mark.parametrize( + "dtype", ["FLOAT", "DOUBLE", "DECIMAL(10,2)", "NUMERIC(8,3)", "NUMERIC"] +) def test_float_family_valid(dtype): # NUMERIC is a MySQL alias for DECIMAL; #190 bugbot caught that the DDL # and ingestor type-cast layers accepted NUMERIC but the validator's @@ -440,6 +436,7 @@ def test_float_non_numeric_fails(): # Missing values are NULL, not "non-numeric" (regression) # --------------------------------------------------------------------------- + def test_int_with_missing_values_is_valid(): # NaN/empty in an INT column is a missing value (stored as NULL), not a # non-numeric one — and not a "non-integer" one either. @@ -480,6 +477,7 @@ def test_non_numeric_error_points_at_rows_without_leaking_values(): # Non-finite (inf) + realistic messy-data scenarios # --------------------------------------------------------------------------- + def test_float_infinity_is_rejected(): # inf survives pd.to_numeric (it is "numeric") but flows through the training # scaler unchanged and produces a NaN loss — so the gate must reject it. @@ -530,6 +528,7 @@ def test_scientific_notation_is_valid(): # VARCHAR / CHAR / TEXT # --------------------------------------------------------------------------- + def test_varchar_valid(): df = pd.DataFrame({"s": ["abc", "de"]}) assert DataValidator(schema={"s": "VARCHAR(255)"}).validate(df).is_valid @@ -614,8 +613,10 @@ def test_issue_188_full_repro_csv(make_csv): ) result = DataValidator( schema={ - "id": "INT", "feat": "FLOAT", - "code": "VARCHAR(10)", "label": "VARCHAR(8)", + "id": "INT", + "feat": "FLOAT", + "code": "VARCHAR(10)", + "label": "VARCHAR(8)", } ).validate(str(path)) assert result.is_valid, f"expected valid; errors={result.errors}" @@ -626,7 +627,7 @@ def test_issue_188_numeric_scalars_pass_string_family(dtype): # All three string-family validators share the same astype(str)!=value # over-rejection; the fix is applied uniformly. df = pd.DataFrame({"s": ["abc", 100, 200]}) # ints mixed in (CHAR case - # is special — see below) + # is special — see below) if "CHAR(" in dtype: # CHAR enforces fixed length, so use values that all stringify to 3. df = pd.DataFrame({"s": ["abc", 100, 200]}) @@ -650,6 +651,7 @@ def test_text_valid(): # BOOLEAN across dtypes # --------------------------------------------------------------------------- + def test_boolean_bool_dtype_valid(): df = pd.DataFrame({"b": [True, False]}) assert DataValidator(schema={"b": "BOOLEAN"}).validate(df).is_valid @@ -698,6 +700,7 @@ def test_boolean_float_invalid_fails(): # DATE / DATETIME / TIMESTAMP / TIME # --------------------------------------------------------------------------- + @pytest.mark.parametrize("dtype", ["DATE", "DATETIME", "TIMESTAMP", "TIME"]) def test_date_family_valid(dtype): df = pd.DataFrame({"d": ["2024-01-01", "2024-02-02"]}) @@ -737,6 +740,7 @@ def test_date_still_flags_real_bad_value_with_nulls_present(): # type parsing + auto-detect helper # --------------------------------------------------------------------------- + def test_constraints_are_stripped_from_type(): df = pd.DataFrame({"n": [1, 2]}) # "INT NOT NULL" should resolve to the INT validator. @@ -761,6 +765,7 @@ def test_detect_column_type(series, expected): # Streaming / large-file validation (bounded memory, full coverage) # --------------------------------------------------------------------------- + def test_validate_csv_streams_and_catches_late_chunk(tmp_path): # A bad value PAST the first chunk must still be caught — the validator # scans the whole file chunk by chunk (memory-bounded), not just a sample diff --git a/tests/test_database.py b/tests/test_database.py index ee502d9..60a2cbe 100644 --- a/tests/test_database.py +++ b/tests/test_database.py @@ -1004,3 +1004,90 @@ def test_create_table_rejects_reserved_bookkeeping_tables(reserved): db = Database.__new__(Database) with pytest.raises(ValueError, match="reserved"): db.create_table(reserved, {"feature_0": "FLOAT"}) + + +# --------------------------------------------------------------------------- +# Run journal: enumeration for the metadata backfill sweep +# --------------------------------------------------------------------------- + + +def test_list_registered_runs_returns_registered_only(db, mock_engine_factory): + """The backfill sweep enumerates only REGISTERED runs, one row each, mapped + to {ingestor_id, table_name, task}. A NULL task (pre-task-column run) is + preserved as None.""" + _, _, conn = mock_engine_factory + conn.execute.return_value.fetchall.return_value = [ + ("run-a", "ta", "tabular_classification"), + ("run-b", "tb", None), + ] + result = db.list_registered_runs() + assert result == [ + {"ingestor_id": "run-a", "table_name": "ta", "task": "tabular_classification"}, + {"ingestor_id": "run-b", "table_name": "tb", "task": None}, + ] + select = next( + s for s in _executed_sql(conn) if "SELECT ingestor_id, table_name, task" in s + ) + assert "WHERE registered = 1" in select + + +def test_list_dataset_ingestor_ids_scans_tables_excluding_framework( + db, mock_engine_factory +): + """Discovers dataset tables via information_schema (tables carrying an + ingestor_id column), skips the framework tables (run journal + salt store), + and returns each table's distinct non-null ingestor_id.""" + _, _, conn = mock_engine_factory + + def _result(rows): + r = MagicMock() + r.fetchall.return_value = rows + return r + + # 1st execute: information_schema table list (includes a framework table). + # then one execute per NON-framework table, in returned order. + conn.execute.side_effect = [ + _result([("ds_one",), ("tracebloc_ingest_runs",), ("ds_two",)]), + _result([("i1",), ("i2",)]), # ds_one distinct ids + _result([("i3",)]), # ds_two distinct ids + ] + + result = db.list_dataset_ingestor_ids() + assert result == [ + {"ingestor_id": "i1", "table_name": "ds_one"}, + {"ingestor_id": "i2", "table_name": "ds_one"}, + {"ingestor_id": "i3", "table_name": "ds_two"}, + ] + + executed = _executed_sql(conn) + # Discovery query hit information_schema for the ingestor_id column. + assert any("information_schema.columns" in s and "ingestor_id" in s for s in executed) + # The framework run-journal table was skipped — never SELECTed as a dataset. + assert not any("`tracebloc_ingest_runs`" in s for s in executed) + # Per-dataset-table id scans excluded null/empty ids. + assert any("FROM `ds_one`" in s and "IS NOT NULL" in s for s in executed) + + +def test_list_dataset_ingestor_ids_skips_a_table_that_errors(db, mock_engine_factory): + """A single unreadable table (scan timeout / dropped mid-sweep / perms) is + skipped and discovery continues — the sweep's 'one bad table can't abort the + rollout' guarantee must hold at enumeration time too. (bugbot)""" + _, _, conn = mock_engine_factory + + def _result(rows): + r = MagicMock() + r.fetchall.return_value = rows + return r + + conn.execute.side_effect = [ + _result([("ds_one",), ("ds_bad",), ("ds_two",)]), # information_schema + _result([("i1",)]), # ds_one + RuntimeError("scan timeout"), # ds_bad — per-table query fails + _result([("i2",)]), # ds_two still scanned + ] + + result = db.list_dataset_ingestor_ids() + assert result == [ + {"ingestor_id": "i1", "table_name": "ds_one"}, + {"ingestor_id": "i2", "table_name": "ds_two"}, + ] diff --git a/tests/test_image_validator_min_size.py b/tests/test_image_validator_min_size.py index 67d9df9..3427a74 100644 --- a/tests/test_image_validator_min_size.py +++ b/tests/test_image_validator_min_size.py @@ -196,6 +196,37 @@ def test_too_small_result_still_reports_corrupt_files(clean_env, images_dir): assert any("broken.jpg" in f for f in result.metadata["invalid_files"]) +def test_floor_wins_when_corrupt_file_is_first(clean_env, images_dir, monkeypatch): + """Order-independence guard: auto-detection must skip an unreadable first + file and detect from the first *readable* image. Before the fix, when the + corrupt file happened to come first in glob order, auto-detection gave up + and the result was 'auto-detection failed' instead of the floor verdict — + so the outcome depended on filesystem ordering (green on macOS, red on the + Linux CI). This forces the corrupt-first order explicitly.""" + src, add = images_dir + add("tiny.jpg", (8, 8)) + (src / "images" / "broken.jpg").write_bytes(b"not a real jpeg") + clean_env.setenv("SRC_PATH", str(src)) + + validator = ImageResolutionValidator() + real_get_files = validator._get_image_files + # Deterministically place the corrupt file first, whatever glob returns. + monkeypatch.setattr( + validator, + "_get_image_files", + lambda *a, **k: sorted( + real_get_files(*a, **k), + key=lambda p: 0 if p.name == "broken.jpg" else 1, + ), + ) + + result = validator.validate(None) + assert not result.is_valid + assert "below the minimum size" in result.errors[0] + assert "tiny.jpg" in result.errors[0] + assert any("broken.jpg" in f for f in result.metadata["invalid_files"]) + + # --- _meets_min_size unit ---------------------------------------------------- diff --git a/tests/test_ingestor_base.py b/tests/test_ingestor_base.py index cac2254..6b79df5 100644 --- a/tests/test_ingestor_base.py +++ b/tests/test_ingestor_base.py @@ -610,6 +610,70 @@ def test_validate_data_validator_exception_raises(): ing.validate_data("src") +# --------------------------------------------------------------------------- +# validator label normalization (#396 Bugbot: log labels drop suffix) +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "name, expected", + [ + # Already ends in "Validator" — must NOT double it. + ("Data Validator", "Data Validator"), + # Case-insensitive on the existing suffix. + ("Pascal VOC XML validator", "Pascal VOC XML validator"), + # Missing the suffix — must gain it. + ("Ingestable Records", "Ingestable Records Validator"), + ("Text Content", "Text Content Validator"), + ("BIO Label", "BIO Label Validator"), + ("Keypoint Annotation", "Keypoint Annotation Validator"), + # Surrounding whitespace is stripped before deciding. + (" Data Validator ", "Data Validator"), + ], +) +def test_validator_label_normalizes(name, expected): + assert base_mod._validator_label(name) == expected + + +def test_validate_data_success_line_keeps_suffix_when_missing(capsys): + # A name that does NOT end in "Validator" (e.g. "Ingestable Records") must + # keep the suffix in its "successfully passed" line. + ing = make_ingestor(category=None) + v = MagicMock() + v.name = "Ingestable Records" + v.validate.return_value = ValidationResult(True, [], [], {}) + with patch.object(base_mod, "map_validators", return_value=[v]): + assert ing.validate_data("src") is True + out = capsys.readouterr().out + assert "Ingestable Records Validator successfully passed" in out + + +def test_validate_data_success_line_does_not_double_suffix(capsys): + # A name that already ends in "Validator" (e.g. "Data Validator") must not + # double the word. + ing = make_ingestor(category=None) + v = MagicMock() + v.name = "Data Validator" + v.validate.return_value = ValidationResult(True, [], [], {}) + with patch.object(base_mod, "map_validators", return_value=[v]): + assert ing.validate_data("src") is True + out = capsys.readouterr().out + assert "Data Validator successfully passed" in out + assert "Data Validator Validator" not in out + + +def test_validate_data_failure_message_keeps_suffix_when_missing(): + # The error-append line normalizes too, so a suffix-less name isn't stripped. + ing = make_ingestor(category=None) + v = MagicMock() + v.name = "Ingestable Records" + v.validate.return_value = ValidationResult(False, ["nope"], [], {}) + with patch.object(base_mod, "map_validators", return_value=[v]): + with pytest.raises(ValueError) as exc: + ing.validate_data("src") + assert "Ingestable Records Validator failed" in str(exc.value) + + # --------------------------------------------------------------------------- # ingest (full flow, Session patched) # --------------------------------------------------------------------------- diff --git a/tests/test_metadata_backfill_runner.py b/tests/test_metadata_backfill_runner.py new file mode 100644 index 0000000..d797fef --- /dev/null +++ b/tests/test_metadata_backfill_runner.py @@ -0,0 +1,231 @@ +"""Tests for the metadata backfill runner — the GET → recompute → POST sweep +that wires build_dataset_metadata to the backend metadata-backfill endpoint. + +build_dataset_metadata and the APIClient are the units-under-test's collaborators +and are exercised elsewhere; here they are mocked so the runner's own +orchestration/skip/guard logic is what's covered. +""" + +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +from tracebloc_ingestor import metadata_backfill_runner as runner +from tracebloc_ingestor.metadata_backfill_runner import ( + STATUS_ERROR, + STATUS_NOT_FOUND, + STATUS_OK, + STATUS_SKIPPED_COMPETITION, + STATUS_SKIPPED_CURRENT, + backfill_dataset, + backfill_datasets, +) + +# A pre-cutover plain dataset record as the GET returns it — no new-shape +# attributes yet, not a competition. +_PLAIN_RECORD = { + "id": 1, + "table_name": "tb", + "category": "tabular_classification", + "data_format": "tabular", + "intent": "train", + "ingestor_id": "ing-1", + "is_competition": False, + "source_dataset_ids": [], + "schema": {"x": "INT"}, + "meta_data": {}, +} + +_PAYLOAD = { + "schema": {"x": {"dtype": "int"}}, + "meta_data": {"attributes": {"feature_stats": {"x": {"count": 5}}}}, +} + + +def _api(record=None, send_result=None): + api = MagicMock() + api.get_dataset_metadata.return_value = record + api.send_metadata_backfill.return_value = send_result or { + "table_name": "tb", + "created": True, + "competitions_refolded": 2, + } + return api + + +# --------------------------------------------------------------------------- +# backfill_dataset — single dataset +# --------------------------------------------------------------------------- + + +def test_backfill_dataset_happy_path_gets_builds_and_posts(): + api = _api(record=dict(_PLAIN_RECORD)) + db = MagicMock() + with patch.object(runner, "build_dataset_metadata", return_value=_PAYLOAD) as build: + result = backfill_dataset(db, api, "ing-1") + + # GET by ingestor_id, build keyed on the backend's category/data_format, + # then POST the recomputed payload back. + api.get_dataset_metadata.assert_called_once_with("ing-1") + build.assert_called_once() + assert build.call_args.kwargs["category"] == "tabular_classification" + assert build.call_args.kwargs["data_format"] == "tabular" + api.send_metadata_backfill.assert_called_once_with( + "tb", _PAYLOAD["schema"], _PAYLOAD["meta_data"] + ) + assert result.status == STATUS_OK + assert result.created is True + assert result.competitions_refolded == 2 + + +def test_backfill_dataset_not_found_returns_status_and_skips_build(): + api = _api(record=None) # GET 404 → None + db = MagicMock() + with patch.object(runner, "build_dataset_metadata") as build: + result = backfill_dataset(db, api, "missing") + assert result.status == STATUS_NOT_FOUND + build.assert_not_called() + api.send_metadata_backfill.assert_not_called() + + +def test_backfill_dataset_skips_competition(): + record = dict(_PLAIN_RECORD, is_competition=True) + api = _api(record=record) + db = MagicMock() + with patch.object(runner, "build_dataset_metadata") as build: + result = backfill_dataset(db, api, "ing-1") + assert result.status == STATUS_SKIPPED_COMPETITION + build.assert_not_called() + api.send_metadata_backfill.assert_not_called() + + +def test_backfill_dataset_skips_merged_with_source_ids(): + record = dict(_PLAIN_RECORD, source_dataset_ids=[7, 8]) + api = _api(record=record) + with patch.object(runner, "build_dataset_metadata") as build: + result = backfill_dataset(MagicMock(), api, "ing-1") + assert result.status == STATUS_SKIPPED_COMPETITION + build.assert_not_called() + + +def test_backfill_dataset_skips_when_already_new_shape(): + record = dict( + _PLAIN_RECORD, + meta_data={"attributes": {"feature_stats": {"x": {"count": 1}}}}, + ) + api = _api(record=record) + with patch.object(runner, "build_dataset_metadata") as build: + result = backfill_dataset(MagicMock(), api, "ing-1") + assert result.status == STATUS_SKIPPED_CURRENT + build.assert_not_called() + api.send_metadata_backfill.assert_not_called() + + +def test_backfill_dataset_reprocesses_current_when_skip_disabled(): + record = dict( + _PLAIN_RECORD, + meta_data={"attributes": {"feature_stats": {"x": {"count": 1}}}}, + ) + api = _api(record=record) + with patch.object(runner, "build_dataset_metadata", return_value=_PAYLOAD): + result = backfill_dataset(MagicMock(), api, "ing-1", skip_if_current=False) + assert result.status == STATUS_OK + api.send_metadata_backfill.assert_called_once() + + +def test_backfill_dataset_error_when_record_missing_category(): + record = dict(_PLAIN_RECORD, category=None) + api = _api(record=record) + with patch.object(runner, "build_dataset_metadata") as build: + result = backfill_dataset(MagicMock(), api, "ing-1") + assert result.status == STATUS_ERROR + build.assert_not_called() + + +# --------------------------------------------------------------------------- +# backfill_datasets — the sweep +# --------------------------------------------------------------------------- + + +def test_backfill_datasets_discovers_tables_when_ids_omitted(): + # Default enumeration scans dataset tables (covers the pre-cutover backlog), + # NOT the runs journal. + db = MagicMock() + db.list_dataset_ingestor_ids.return_value = [ + {"ingestor_id": "a", "table_name": "ta"}, + {"ingestor_id": "b", "table_name": "tb"}, + ] + api = MagicMock() + api.get_dataset_metadata.side_effect = lambda iid: dict( + _PLAIN_RECORD, ingestor_id=iid, table_name={"a": "ta", "b": "tb"}[iid] + ) + api.send_metadata_backfill.side_effect = lambda table, schema, meta=None: { + "table_name": table, + "created": True, + "competitions_refolded": 0, + } + with patch.object(runner, "build_dataset_metadata", return_value=_PAYLOAD): + results = backfill_datasets(db, api) + db.list_dataset_ingestor_ids.assert_called_once() + db.list_registered_runs.assert_not_called() + assert [r.ingestor_id for r in results] == ["a", "b"] + assert all(r.status == STATUS_OK for r in results) + + +def test_backfill_datasets_backfills_a_multi_id_table_once(): + # A table appended by several ingest runs has multiple ingestor_ids; its + # metadata is table-level, so it is backfilled ONCE and the sibling id is + # skipped (before any backend call). + db = MagicMock() + db.list_dataset_ingestor_ids.return_value = [ + {"ingestor_id": "run1", "table_name": "shared"}, + {"ingestor_id": "run2", "table_name": "shared"}, + ] + api = _api(record=dict(_PLAIN_RECORD, table_name="shared")) + with patch.object(runner, "build_dataset_metadata", return_value=_PAYLOAD): + results = backfill_datasets(db, api) + assert len(results) == 1 # second id skipped by per-table dedup + assert results[0].table_name == "shared" + api.send_metadata_backfill.assert_called_once() + + +def test_backfill_datasets_retries_sibling_id_when_first_not_found(): + # If a table's first ingestor_id 404s, a sibling id is still tried (the table + # isn't marked done on not_found). + db = MagicMock() + db.list_dataset_ingestor_ids.return_value = [ + {"ingestor_id": "missing", "table_name": "shared"}, + {"ingestor_id": "good", "table_name": "shared"}, + ] + api = MagicMock() + api.get_dataset_metadata.side_effect = lambda iid: ( + None if iid == "missing" else dict(_PLAIN_RECORD, table_name="shared") + ) + api.send_metadata_backfill.return_value = { + "table_name": "shared", + "created": True, + "competitions_refolded": 0, + } + with patch.object(runner, "build_dataset_metadata", return_value=_PAYLOAD): + results = backfill_datasets(db, api) + assert [r.status for r in results] == [STATUS_NOT_FOUND, STATUS_OK] + + +def test_backfill_datasets_continues_past_a_failing_dataset(): + api = MagicMock() + # First GET raises (e.g. transient), second succeeds. + api.get_dataset_metadata.side_effect = [ + RuntimeError("boom"), + dict(_PLAIN_RECORD, ingestor_id="b"), + ] + api.send_metadata_backfill.return_value = { + "table_name": "tb", + "created": False, + "competitions_refolded": 0, + } + with patch.object(runner, "build_dataset_metadata", return_value=_PAYLOAD): + results = backfill_datasets(MagicMock(), api, ["a", "b"]) + assert results[0].status == STATUS_ERROR + # Exception TYPE only — never the raw message (may embed customer cell values). + assert results[0].error == "RuntimeError" + assert results[1].status == STATUS_OK # sweep continued diff --git a/tracebloc_ingestor/__init__.py b/tracebloc_ingestor/__init__.py index 232a199..7308003 100644 --- a/tracebloc_ingestor/__init__.py +++ b/tracebloc_ingestor/__init__.py @@ -28,7 +28,7 @@ # Single source of truth for the package version. setup.py parses this literal # (see _read_version in setup.py) so the two can't drift again (#175). Bump here # only — setup.py picks it up automatically. -__version__ = "0.7.5" +__version__ = "0.7.7" __all__ = [ "Config", diff --git a/tracebloc_ingestor/api/client.py b/tracebloc_ingestor/api/client.py index d4ebbe7..8df074e 100644 --- a/tracebloc_ingestor/api/client.py +++ b/tracebloc_ingestor/api/client.py @@ -322,6 +322,105 @@ def send_ingest_summary( logger.error(f"{RED}Error sending ingest summary: {str(e)[:500]}{RESET}") raise + def get_dataset_metadata(self, ingestor_id: str) -> Optional[Dict[str, Any]]: + """Fetch a dataset's backend record by ``ingestor_id`` (backend#1198). + + Returns the core fields — ``table_name`` / ``category`` / ``data_format`` + / ``intent`` / ``is_competition`` / ``source_dataset_ids`` — plus the + currently-stored ``schema`` / ``meta_data``. Backs the pre-cutover + backfill runner: ``category`` and ``data_format`` drive the recompute and + are not recoverable from the table alone, so the runner reads them here + first. + + Returns ``None`` when the backend has no dataset for this ``ingestor_id`` + owned by this edge (404) so a sweep can skip it rather than crash. + + Raises: + requests.exceptions.HTTPError: on any non-404 error status. + """ + if self.config.EDGE_ENV == "local": + logger.info( + f"Mock: would fetch dataset metadata for ingestor_id={ingestor_id}" + ) + return None + + url = ( + f"{self.config.API_ENDPOINT}" + f"/global_meta/metadata_backfill/by-ingestor/{ingestor_id}/" + ) + response = self._authed_request("GET", url, timeout=API_TIMEOUT) + if response.status_code == 404: + logger.warning( + f"{YELLOW}No backend dataset for ingestor_id={ingestor_id} " + f"owned by this edge (404); skipping.{RESET}" + ) + return None + if response.status_code >= 400: + # Status only in the message — NOT response.text. A backend error + # body can echo the dataset's metadata (categorical vocab = customer + # cell values), and this exception surfaces in the backfill runner's + # install-log output. The response object is attached for a caller + # that needs to inspect it deliberately. (bugbot) + raise requests.exceptions.HTTPError( + f"HTTP {response.status_code}", response=response + ) + return self._parse_json(response, required=True) + + def send_metadata_backfill( + self, + table_name: str, + schema: Dict[str, Any], + meta_data: Optional[Dict[str, Any]] = None, + ) -> Dict[str, Any]: + """Upsert recomputed ``{schema, meta_data}`` for a PRE-EXISTING table via + the metadata-backfill endpoint (backend#1166/#1198). + + Metadata-only: never creates a dataset (the table must already exist + backend-side — 404 otherwise). The POST also propagates the refresh into + any competition built from the table (re-fold). Idempotent, so a re-run + is a safe overwrite. + + Returns: + The backend body — ``{"table_name", "created", "competitions_refolded"}``. + + Raises: + requests.exceptions.HTTPError: If the API call fails after retries. + """ + if self.config.EDGE_ENV == "local": + logger.info( + f"Mock: would backfill metadata for {table_name} " + f"({len(schema)} schema column(s))" + ) + return { + "table_name": table_name, + "created": False, + "competitions_refolded": 0, + } + + payload = json.dumps({"schema": schema, "meta_data": meta_data or {}}) + response = self._authed_request( + "POST", + f"{self.config.API_ENDPOINT}/global_meta/metadata_backfill/{table_name}/", + extra_headers={"Content-Type": "application/json"}, + data=payload, + timeout=API_TIMEOUT, + ) + if response.status_code >= 400: + # Status only — NOT response.text (see get_dataset_metadata): the + # POST body carries this table's categorical vocab, so a backend + # error echoing it back would embed customer cell values in the + # runner's install-log output. (bugbot) + raise requests.exceptions.HTTPError( + f"HTTP {response.status_code}", response=response + ) + result = self._parse_json(response, required=True) + logger.info( + f"{GREEN}Backfilled metadata for {table_name}: " + f"created={result.get('created')}, " + f"competitions_refolded={result.get('competitions_refolded')}{RESET}" + ) + return result + def __del__(self): """Cleanup when the client is destroyed""" if hasattr(self, "session"): diff --git a/tracebloc_ingestor/database.py b/tracebloc_ingestor/database.py index b850b83..14f6462 100644 --- a/tracebloc_ingestor/database.py +++ b/tracebloc_ingestor/database.py @@ -862,6 +862,96 @@ def reclaim_dead_run_rows( ) return reclaimed + def list_registered_runs(self) -> List[Dict[str, str]]: + """Return the REGISTERED ingest runs journalled on this client, one per + dataset: ``[{"ingestor_id", "table_name", "task"}, ...]``. + + The run journal (``record_ingest_started`` / ``mark_ingest_registered``) + is the authoritative local list of datasets this edge has ingested and + successfully registered with the backend — exactly the set the + pre-cutover metadata backfill sweeps. Only ``registered = 1`` rows are + returned, so a started-but-never-registered (dead) run is never + backfilled. ``task`` is the recorded category (may be NULL on runs + journalled before the column existed); the backfill runner re-reads the + authoritative category from the backend anyway. + """ + with self.engine.connect() as connection: + self._ensure_runs_table(connection) + rows = _execute_with_retry( + connection, + text( + f"SELECT ingestor_id, table_name, task " + f"FROM `{self.RUNS_TABLE}` WHERE registered = 1 " + f"ORDER BY started_at" + ), + ).fetchall() + return [ + {"ingestor_id": iid, "table_name": tname, "task": task} + for (iid, tname, task) in rows + ] + + def list_dataset_ingestor_ids(self) -> List[Dict[str, str]]: + """Discover every dataset table directly and return its distinct + ``ingestor_id``\\ s: ``[{"ingestor_id", "table_name"}, ...]``. + + This is the enumeration the metadata backfill sweeps by default, in + preference to :meth:`list_registered_runs`. The pre-cutover backlog the + backfill exists for was ingested BEFORE the run journal existed, so those + datasets have no journal row — but their tables still carry the framework + ``ingestor_id`` column. Scanning for that column finds them; the journal + would miss them entirely. + + A dataset table is any table in this schema with an ``ingestor_id`` + column, minus the framework bookkeeping tables (the run journal — which + also has the column — and the salt store). Only non-null, non-empty + ``ingestor_id`` values are returned; a table appended by several runs + yields one pair per distinct id (the caller backfills the table once). + """ + framework_tables = {self.SALT_TABLE, self.RUNS_TABLE} + with self.engine.connect() as connection: + table_rows = _execute_with_retry( + connection, + text( + "SELECT DISTINCT table_name FROM information_schema.columns " + "WHERE table_schema = DATABASE() AND column_name = 'ingestor_id'" + ), + ).fetchall() + + pairs: List[Dict[str, str]] = [] + for (table_name,) in table_rows: + if table_name in framework_tables: + continue + # Identifier interpolated (bound params can't name a table); the name + # comes from information_schema on our own DB, and backticks are + # escaped, so it is not attacker-controlled free text. + safe_table = table_name.replace("`", "``") + # Per-table guard: a single unreadable table (timeout on a large + # unindexed scan, a table dropped mid-sweep, a permissions gap) must + # NOT abort discovery — otherwise every other table is skipped and the + # sweep's "one bad table can't stop the rollout" guarantee is lost. + # Log the table name + exception TYPE only (never str(exc): a driver + # message can embed cell values, and this feeds install logs). + try: + with self.engine.connect() as connection: + id_rows = _execute_with_retry( + connection, + text( + f"SELECT DISTINCT ingestor_id FROM `{safe_table}` " + f"WHERE ingestor_id IS NOT NULL AND ingestor_id <> ''" + ), + ).fetchall() + except Exception as exc: # noqa: BLE001 — skip the table, continue the sweep + logger.warning( + "list_dataset_ingestor_ids: skipping table %s — could not read " + "ingestor_ids (%s)", + table_name, + type(exc).__name__, + ) + continue + for (ingestor_id,) in id_rows: + pairs.append({"ingestor_id": ingestor_id, "table_name": table_name}) + return pairs + def get_label_counts(self, table_name: str, ingestor_id: str) -> Dict[str, int]: """ Return ``{label: row_count}`` for every label inserted by *ingestor_id*. diff --git a/tracebloc_ingestor/ingestors/base.py b/tracebloc_ingestor/ingestors/base.py index 241b32c..8555de3 100644 --- a/tracebloc_ingestor/ingestors/base.py +++ b/tracebloc_ingestor/ingestors/base.py @@ -137,6 +137,20 @@ def _rows_state_clause(inserted_records: int) -> str: return "no rows were ingested, so nothing was left in the database" +def _validator_label(name: str) -> str: + """Human-readable label for a validator's preflight log/error lines. + + Validator ``.name`` values are inconsistent: some already end in + "Validator" (e.g. "Data Validator") while many do not (e.g. "Ingestable + Records", "Text Content", "BIO Label", "Keypoint Annotation"). Normalize at + the log site so every label reads " Validator" exactly once — + appending the word only when it's missing, so suffixed names don't double + ("Data Validator Validator") and unsuffixed names don't drop it. + """ + name = (name or "").strip() + return name if name.lower().endswith("validator") else f"{name} Validator" + + # NOTE: _TABULAR_FAMILY_CATEGORIES and _FILE_BEARING_CATEGORIES are imported # above from modalities.registry (derived from the per-category ModalitySpec # flags — backend#796 P3a). The ``self.category in `` checks throughout @@ -545,14 +559,15 @@ def validate_data(self, source: Any) -> bool: validation_errors = [] for validator in validators: + label = _validator_label(validator.name) try: - logger.info(f"{CYAN}Running validator: {validator.name}{RESET}") + logger.info(f"{CYAN}Running validator: {label}{RESET}") result = validator.validate(source) if not result.is_valid: all_valid = False validation_errors.append( - f"{BOLD}{validator.name} Validator failed: {RESET} \n {RED}" + f"{BOLD}{label} failed: {RESET} \n {RED}" ) validation_errors.extend(result.errors) validation_errors.append(f"{RESET}") @@ -560,15 +575,17 @@ def validate_data(self, source: Any) -> bool: # Log warnings if any for warning in result.warnings: logger.warning( - f"{YELLOW}Validation warning - {validator.name}: {warning}{RESET}" + f"{YELLOW}Validation warning - {label}: {warning}{RESET}" ) if result.is_valid: - print( - f"{GREEN}{validator.name} Validator successfully passed{RESET}" - ) + # Normalize the label so it always reads " Validator" + # exactly once: names that already end in "Validator" (e.g. + # "Data Validator") don't double, and names that don't (e.g. + # "Ingestable Records") keep the suffix. + print(f"{GREEN}{label} successfully passed{RESET}") except Exception as e: all_valid = False - validation_errors.append(f"Validator {validator.name} error: {str(e)}") + validation_errors.append(f"{label} error: {str(e)}") if not all_valid: error_summary = "\n".join(validation_errors) @@ -864,9 +881,7 @@ def _ingest_with_lock( and not self.unique_id_column and self._table_salt is None ): - self._table_salt = self.database.get_or_create_table_salt( - self.table_name - ) + self._table_salt = self.database.get_or_create_table_salt(self.table_name) if self.table is None: # Grouped categories get a composite (group, time) secondary @@ -1130,8 +1145,7 @@ def _ingest_with_lock( elif not label_counts: counts_helper = ( "get_label_sequence_counts" - if grouping is not None - and grouping.count_unit == "sequences" + if grouping is not None and grouping.count_unit == "sequences" else "get_label_counts" ) raise RuntimeError( diff --git a/tracebloc_ingestor/metadata_backfill_runner.py b/tracebloc_ingestor/metadata_backfill_runner.py new file mode 100644 index 0000000..a809558 --- /dev/null +++ b/tracebloc_ingestor/metadata_backfill_runner.py @@ -0,0 +1,266 @@ +"""Pre-cutover metadata backfill runner (backend#1166 / #1198 — the send side +of #378). + +Ties the backfill pieces together into a one-time, per-client sweep of datasets +ingested BEFORE the federated-alignment cutover. For each dataset it: + + 1. **GETs the backend record** by ``ingestor_id`` + (:meth:`APIClient.get_dataset_metadata`). The authoritative ``category`` and + ``data_format`` drive the recompute and are not recoverable from the table + alone, so they are read from the backend first. + 2. **Recomputes** the enriched ``{schema, meta_data}`` from the persisted rows + via :func:`build_dataset_metadata` — SQL aggregates, no row re-ingest — + keyed on that category. + 3. **POSTs** it back (:meth:`APIClient.send_metadata_backfill`), which upserts + ``GlobalMetaData`` and re-folds any competition built from the table. + +The set of datasets to sweep is the client's REGISTERED ingest journal +(:meth:`Database.list_registered_runs`); explicit ``ingestor_id``\\ s can be +passed instead. + +**Rollout, not always-on.** New ingests already emit the new-shape metadata +(#361), so this runs once per client during rollout. It is idempotent: a dataset +already carrying new-shape metadata is skipped, and the backend upsert is itself +a safe overwrite, so a re-run converges. + +**Scope.** Only plain/source datasets are recomputed. Competition/merged +datasets have no client-side raw table to read — their fold is rebuilt +backend-side by the re-fold this POST triggers — so they are skipped here +(detected via the backend record's ``is_competition`` / ``source_dataset_ids``). +``label_column`` and the uploader-declared scalar attributes are not persisted +backend-side, so they are not passed; the SQL-derivable enriched schema + +``feature_stats`` (the backfill's primary value) are recomputed regardless. See +:func:`build_dataset_metadata` for the per-fact limitations. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass +from typing import Any, Dict, List, Optional + +from .api.client import APIClient +from .config import Config +from .database import Database +from .metadata_backfill import build_dataset_metadata +from .utils.logging import setup_logging + +logger = logging.getLogger(__name__) + + +# Outcome statuses for a single dataset's backfill attempt. +STATUS_OK = "ok" +STATUS_NOT_FOUND = "not_found" +STATUS_SKIPPED_COMPETITION = "skipped_competition" +STATUS_SKIPPED_CURRENT = "skipped_current" +STATUS_ERROR = "error" + + +@dataclass +class BackfillResult: + """The outcome of backfilling one dataset (see the ``STATUS_*`` constants).""" + + ingestor_id: str + table_name: Optional[str] = None + status: str = STATUS_OK + created: Optional[bool] = None + competitions_refolded: Optional[int] = None + error: Optional[str] = None + + +def _already_new_shape(record: Dict[str, Any]) -> bool: + """Whether the backend record already carries new-shape metadata, so the + recompute + POST can be skipped. + + Marker: a non-empty ``meta_data.attributes`` — the typed alignment contract + that only the post-cutover ingest (#361) / a prior backfill writes. A + pre-cutover row has no ``attributes``, so it is not skipped. Conservative: if + a category legitimately has empty attributes we simply re-POST (idempotent). + """ + meta_data = record.get("meta_data") or {} + return bool(meta_data.get("attributes")) + + +def backfill_dataset( + database: Database, + api_client: APIClient, + ingestor_id: str, + *, + skip_if_current: bool = True, +) -> BackfillResult: + """GET → recompute → POST for a single dataset. Never raises for an expected + skip/not-found; those are returned as a :class:`BackfillResult` status. + Unexpected failures (SQL / network) propagate to the caller. + """ + record = api_client.get_dataset_metadata(ingestor_id) + if record is None: + return BackfillResult(ingestor_id, status=STATUS_NOT_FOUND) + + table_name = record.get("table_name") + + # Competition/merged datasets have no client-side raw table for the SQL + # recompute; their reconciled fold is rebuilt backend-side (the re-fold this + # backfill triggers on source POSTs), so never recompute them here. + if record.get("is_competition") or record.get("source_dataset_ids"): + return BackfillResult( + ingestor_id, table_name, status=STATUS_SKIPPED_COMPETITION + ) + + if skip_if_current and _already_new_shape(record): + return BackfillResult(ingestor_id, table_name, status=STATUS_SKIPPED_CURRENT) + + category = record.get("category") + if not category or not table_name: + return BackfillResult( + ingestor_id, + table_name, + status=STATUS_ERROR, + error="backend record missing category/table_name", + ) + + payload = build_dataset_metadata( + database, + table_name, + category=category, + data_format=record.get("data_format"), + ) + result = api_client.send_metadata_backfill( + table_name, payload["schema"], payload.get("meta_data") + ) + return BackfillResult( + ingestor_id, + table_name, + status=STATUS_OK, + created=result.get("created"), + competitions_refolded=result.get("competitions_refolded"), + ) + + +def backfill_datasets( + database: Database, + api_client: APIClient, + ingestor_ids: Optional[List[str]] = None, + *, + skip_if_current: bool = True, +) -> List[BackfillResult]: + """Backfill datasets. With an explicit ``ingestor_ids`` list, one attempt per + id; otherwise **discover every dataset table** in the client DB and backfill + each — see :meth:`Database.list_dataset_ingestor_ids`. + + Table discovery (not the runs journal) is the default because the whole point + of the backfill is the PRE-CUTOVER backlog — datasets ingested before the + runs journal existed, so they have no journal entry — yet their tables still + carry the framework ``ingestor_id`` column. Enumerating tables therefore + covers both journalled and pre-journal datasets; the journal is a strict + subset. + + A table can carry several ingestor_ids (multiple ingest runs into one table); + its metadata is table-level, so a table is backfilled once — the first of its + ids the backend resolves wins and the rest are skipped. + + Per-dataset guarded: an unexpected failure on one dataset is logged (type + only) and recorded as an ``error`` result; the sweep continues (a single bad + table can't abort the rollout). Returns one :class:`BackfillResult` per + attempt made. + """ + if ingestor_ids is not None: + # Caller-supplied ids: one attempt each; the table is learned from the + # backend GET inside backfill_dataset. + items = [(ingestor_id, None) for ingestor_id in ingestor_ids] + else: + items = [ + (row["ingestor_id"], row["table_name"]) + for row in database.list_dataset_ingestor_ids() + ] + + logger.info("Metadata backfill: %d ingestor_id(s) to process.", len(items)) + + results: List[BackfillResult] = [] + done_tables: set = set() + for ingestor_id, table_name in items: + # A table's metadata is backfilled once. Skip the remaining ids of a + # table already resolved — checked by the ENUMERATED name (no backend + # call) so multi-run tables don't re-POST. + if table_name is not None and table_name in done_tables: + continue + try: + result = backfill_dataset( + database, api_client, ingestor_id, skip_if_current=skip_if_current + ) + except Exception as exc: # noqa: BLE001 — one bad table must not abort the sweep + # Record/log the exception TYPE only — never str(exc) or a traceback. + # A SQL driver error or backend response body can embed customer cell + # values (e.g. categorical vocab), and this runs in a Helm-hook Job + # whose output lands in install logs. (bugbot) + error_type = type(exc).__name__ + logger.error( + "Metadata backfill failed for ingestor_id=%s (%s)", + ingestor_id, + error_type, + ) + result = BackfillResult( + ingestor_id, status=STATUS_ERROR, error=error_type + ) + results.append(result) + _log_result(result) + # Mark the table done only on a DEFINITIVE backend resolution (backfilled, + # already-current, or a competition we intentionally skip). not_found / + # error leave it open so a sibling ingestor_id can still resolve it. + if ( + result.status + in (STATUS_OK, STATUS_SKIPPED_CURRENT, STATUS_SKIPPED_COMPETITION) + and result.table_name + ): + done_tables.add(result.table_name) + + _log_summary(results) + return results + + +def _log_result(result: BackfillResult) -> None: + if result.status == STATUS_OK: + logger.info( + " ✓ %s (%s): created=%s, competitions_refolded=%s", + result.table_name, + result.ingestor_id, + result.created, + result.competitions_refolded, + ) + elif result.status == STATUS_ERROR: + logger.error( + " ✗ %s (%s): %s", result.table_name, result.ingestor_id, result.error + ) + else: + logger.info( + " – %s (%s): %s", result.table_name, result.ingestor_id, result.status + ) + + +def _log_summary(results: List[BackfillResult]) -> None: + counts: Dict[str, int] = {} + for result in results: + counts[result.status] = counts.get(result.status, 0) + 1 + logger.info( + "Metadata backfill complete: %s", + ", ".join(f"{status}={count}" for status, count in sorted(counts.items())) + or "nothing to do", + ) + + +def main(argv: Optional[List[str]] = None) -> int: # pragma: no cover - thin shell + """Console-script entrypoint (``tracebloc-backfill``). + + Reads DB + backend credentials from the environment via :class:`Config` (the + same env the ingestor uses — ``MYSQL_HOST``, ``BACKEND_TOKEN``, ``EDGE_ENV``, + …); no ``ingest.yaml`` is needed. Returns a non-zero exit code iff any + dataset errored, so a rollout job surfaces failures. + """ + config = Config() + setup_logging(config) + + database = Database(config) + api_client = APIClient(config) # triggers config.validate() + + results = backfill_datasets(database, api_client) + errored = sum(1 for result in results if result.status == STATUS_ERROR) + return 1 if errored else 0 diff --git a/tracebloc_ingestor/validators/image_validator.py b/tracebloc_ingestor/validators/image_validator.py index 2017a75..567e49d 100644 --- a/tracebloc_ingestor/validators/image_validator.py +++ b/tracebloc_ingestor/validators/image_validator.py @@ -172,14 +172,24 @@ def validate(self, path: Any, **kwargs) -> ValidationResult: metadata={"files_checked": 0}, ) - # Auto-detect resolution from first image if not specified + # Auto-detect resolution from the first *readable* image if not + # specified. Using image_files[0] blindly made auto-detection fail + # whenever the first file in (arbitrary) glob order happened to be + # corrupt/unreadable — which masked the real floor/uniformity + # verdict behind "auto-detection failed" and made the outcome + # depend on filesystem ordering. Skip unreadable files and detect + # from the first one that yields a resolution; if none do, leave + # expected_resolution unset and let _validate_image_resolutions + # report the honest "auto-detection failed". if auto_detect_resolution and not self.expected_resolution: - first_image_resolution = self._get_image_resolution(image_files[0]) - if first_image_resolution: - self.expected_resolution = first_image_resolution - logger.info( - f"Auto-detected expected resolution: {self.expected_resolution}" - ) + for candidate in image_files: + candidate_resolution = self._get_image_resolution(candidate) + if candidate_resolution: + self.expected_resolution = candidate_resolution + logger.info( + f"Auto-detected expected resolution: {self.expected_resolution}" + ) + break # Validate image resolutions return self._validate_image_resolutions(image_files)