feat(validation): validate ES uploads with bronze dataio converters and Pandera

src/webapp/databricks.py

@@ -62,6 +62,9 @@
     re.IGNORECASE,
 )
 
+# Only these basenames may be read from bronze_volume/training_inputs/ (path traversal safe).
+ALLOWED_BRONZE_TRAINING_INPUTS_FILES = frozenset({"dataio.py", "config.toml"})
+
 
 def _create_databricks_workspace_client(operation: str) -> WorkspaceClient:
     """
@@ -640,6 +643,76 @@ def download_bronze_volume_file(self, inst_name: str, file_name: str) -> Any:
             raise ValueError("Databricks download returned no contents.")
         return stream
 
+    def download_bronze_training_inputs_file(
+        self, inst_name: str, relative_path: str = "dataio.py"
+    ) -> Any:
+        """
+        Download a file from ``bronze_volume/training_inputs/`` for an institution.
+
+        Only allowlisted basenames (currently ``dataio.py`` and ``config.toml``) are
+        permitted; nested paths, ``..``, and absolute paths are rejected.
+
+        Args:
+            inst_name: Institution display name (passed through ``databricksify_inst_name``).
+            relative_path: Basename under ``training_inputs/`` (default ``dataio.py``).
+
+        Returns:
+            Byte stream of file contents (same shape as ``download_bronze_volume_file``).
+
+        Raises:
+            ValueError: If path is unsafe, Databricks is not configured, or download fails.
+        """
+        if not relative_path or not isinstance(relative_path, str):
+            raise ValueError("relative_path is required.")
+        if relative_path != os.path.basename(relative_path):
+            raise ValueError(
+                "relative_path must be a basename only (no directories or separators)."
+            )
+        if relative_path in (".", "..") or ".." in relative_path:
+            raise ValueError("relative_path must not contain path traversal.")
+        if relative_path not in ALLOWED_BRONZE_TRAINING_INPUTS_FILES:
+            raise ValueError(
+                f"relative_path must be one of {sorted(ALLOWED_BRONZE_TRAINING_INPUTS_FILES)}."
+            )
+        if not databricks_vars.get("DATABRICKS_HOST_URL") or not databricks_vars.get(
+            "CATALOG_NAME"
+        ):
+            raise ValueError("Databricks integration not configured.")
+        if not gcs_vars.get("GCP_SERVICE_ACCOUNT_EMAIL"):
+            raise ValueError("GCP service account email not configured.")
+
+        try:
+            w = WorkspaceClient(
+                host=databricks_vars["DATABRICKS_HOST_URL"],
+                google_service_account=gcs_vars["GCP_SERVICE_ACCOUNT_EMAIL"],
+            )
+        except Exception as e:
+            LOGGER.exception(
+                "Failed to create Databricks WorkspaceClient with host: %s and service account: %s",
+                databricks_vars.get("DATABRICKS_HOST_URL"),
+                gcs_vars.get("GCP_SERVICE_ACCOUNT_EMAIL"),
+            )
+            raise ValueError(f"Workspace client creation failed: {e}") from e
+
+        db_inst_name = databricksify_inst_name(inst_name)
+        volume_path = (
+            f"/Volumes/{databricks_vars['CATALOG_NAME']}/"
+            f"{db_inst_name}_bronze/bronze_volume/training_inputs/{relative_path}"
+        )
+
+        try:
+            response = w.files.download(volume_path)
+        except Exception as e:
+            LOGGER.exception("Failed to download from %s", volume_path)
+            raise ValueError(
+                f"Failed to download bronze training_inputs file: {e}"
+            ) from e
+
+        stream = getattr(response, "contents", None)
+        if stream is None:
+            raise ValueError("Databricks download returned no contents.")
+        return stream
+
     # Note that for each unique PIPELINE, we'll need a new function, this is by nature of how unique pipelines
     # may have unique parameters and would have a unique name (i.e. the name field specified in w.jobs.list()). But any run of a given pipeline (even across institutions) can use the same function.
     # E.g. there is one PDP inference pipeline, so one PDP inference function here.

src/webapp/databricks_test.py

@@ -308,6 +308,71 @@ def test_build_shared_inference_job_parameters_shape() -> None:
     )
 
 
+def test_download_bronze_training_inputs_rejects_path_traversal(
+    ctrl: DatabricksControl,
+) -> None:
+    with pytest.raises(ValueError, match="basename"):
+        ctrl.download_bronze_training_inputs_file("School", "../dataio.py")
+    with pytest.raises(ValueError, match="basename"):
+        ctrl.download_bronze_training_inputs_file("School", "foo/dataio.py")
+    with pytest.raises(ValueError, match="one of"):
+        ctrl.download_bronze_training_inputs_file("School", "secrets.py")
+
+
+def test_download_bronze_training_inputs_file_builds_training_inputs_path(
+    monkeypatch: pytest.MonkeyPatch, ctrl: DatabricksControl
+) -> None:
+    import src.webapp.databricks as db_mod
+
+    monkeypatch.setitem(
+        db_mod.databricks_vars, "DATABRICKS_HOST_URL", "https://example.databricks.com"
+    )
+    monkeypatch.setitem(db_mod.databricks_vars, "CATALOG_NAME", "dev_catalog")
+    monkeypatch.setitem(db_mod.gcs_vars, "GCP_SERVICE_ACCOUNT_EMAIL", "sa@example.com")
+
+    mock_stream = MagicMock()
+    mock_response = MagicMock()
+    mock_response.contents = mock_stream
+    workspace = MagicMock()
+    workspace.files.download.return_value = mock_response
+
+    with mock.patch.object(db_mod, "WorkspaceClient", return_value=workspace):
+        result = ctrl.download_bronze_training_inputs_file("Edvise School", "dataio.py")
+
+    assert result is mock_stream
+    workspace.files.download.assert_called_once_with(
+        "/Volumes/dev_catalog/edvise_school_bronze/bronze_volume/training_inputs/dataio.py"
+    )
+
+
+def test_download_bronze_training_inputs_allows_config_toml(
+    monkeypatch: pytest.MonkeyPatch, ctrl: DatabricksControl
+) -> None:
+    import src.webapp.databricks as db_mod
+
+    monkeypatch.setitem(
+        db_mod.databricks_vars, "DATABRICKS_HOST_URL", "https://example.databricks.com"
+    )
+    monkeypatch.setitem(db_mod.databricks_vars, "CATALOG_NAME", "dev_catalog")
+    monkeypatch.setitem(db_mod.gcs_vars, "GCP_SERVICE_ACCOUNT_EMAIL", "sa@example.com")
+
+    mock_stream = MagicMock()
+    mock_response = MagicMock()
+    mock_response.contents = mock_stream
+    workspace = MagicMock()
+    workspace.files.download.return_value = mock_response
+
+    with mock.patch.object(db_mod, "WorkspaceClient", return_value=workspace):
+        result = ctrl.download_bronze_training_inputs_file(
+            "Edvise School", "config.toml"
+        )
+
+    assert result is mock_stream
+    workspace.files.download.assert_called_once_with(
+        "/Volumes/dev_catalog/edvise_school_bronze/bronze_volume/training_inputs/config.toml"
+    )
+
+
 def test_run_validated_gcs_to_bronze_sync_calls_run_now_with_bundle_params(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:

src/webapp/gcsutil.py

@@ -396,8 +396,8 @@ def validate_file(
             file_name: Blob name under unvalidated/.
             allowed_schemas: List of schema/model names allowed.
             institution_id: Validation namespace: "edvise", "pdp", or "legacy".
-            institution_identifier: Optional institution ID (e.g. UUID). Used by
-                Edvise upload validation for caller context.
+            institution_identifier: For ES, institution name used to fetch bronze
+                ``training_inputs/dataio.py`` converters. Unused for PDP/Legacy.
 
         Returns:
             List of inferred schema names (e.g. ["STUDENT"]).

src/webapp/routers/data.py

@@ -1613,6 +1613,7 @@ def _run_validation_and_upsert_file_record(
     current_user: BaseUser,
     storage_control: StorageControl,
     sess: Session,
+    institution_name: Optional[str] = None,
 ) -> Dict[str, Any]:
     """Run storage validate_file, then upsert file record and return response dict."""
     try:
@@ -1621,7 +1622,9 @@ def _run_validation_and_upsert_file_record(
             file_name,
             allowed_schemas,
             institution_id=schema_namespace,
-            institution_identifier=inst_id if schema_namespace == "edvise" else None,
+            institution_identifier=(
+                institution_name if schema_namespace == "edvise" else None
+            ),
         )
     except HardValidationError as e:
         logging.debug("Inferred Schemas FAILED (hard) %s", e)
@@ -1781,6 +1784,7 @@ def validation_helper(
         current_user,
         storage_control,
         sess,
+        institution_name=inst.name,
     )
     if batch_id is not None:
         _load_batch_for_bronze_sync(sess, inst_id, batch_id)

src/webapp/routers/data_test.py

@@ -1464,7 +1464,7 @@ def test_validate_file_with_edvise_schema(edvise_client: TestClient) -> None:
     assert MOCK_STORAGE.validate_file.called
     call_kwargs = MOCK_STORAGE.validate_file.call_args.kwargs
     assert call_kwargs.get("institution_id") == "edvise"
-    assert call_kwargs.get("institution_identifier") == uuid_to_str(EDVISE_INST_UUID)
+    assert call_kwargs.get("institution_identifier") == "edvise_school"
 
     # Non-PDP bronze sync runs only when batch_id is provided at validation time.
     MOCK_DATABRICKS.run_validated_gcs_to_bronze_sync.assert_not_called()
@@ -1854,9 +1854,10 @@ def test_validate_file_with_edvise_does_not_require_active_schema_doc(
 
     assert response.status_code == 200
     assert MOCK_STORAGE.validate_file.call_args.kwargs.get("institution_id") == "edvise"
-    assert MOCK_STORAGE.validate_file.call_args.kwargs.get(
-        "institution_identifier"
-    ) == uuid_to_str(EDVISE_INST_UUID)
+    assert (
+        MOCK_STORAGE.validate_file.call_args.kwargs.get("institution_identifier")
+        == "edvise_school"
+    )
 
 
 def test_validation_helper_pdp_and_edvise_mutual_exclusivity(
@@ -1995,7 +1996,7 @@ def capture_schema(*args, **kwargs):
 
     assert captured_arg_count == 3
     assert captured_institution_id == "edvise"
-    assert captured_institution_identifier == uuid_to_str(EDVISE_INST_UUID)
+    assert captured_institution_identifier == "edvise_school"
 
     # Reset mock
     MOCK_STORAGE.validate_file.side_effect = None