feat(cli-e2e-v2): Slice 1 — framework + MySQL pilot connector

.github/workflows/py-cli-e2e-tests-v2.yml

@@ -0,0 +1,99 @@
+#  Copyright 2026 Collate
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#  http://www.apache.org/licenses/LICENSE-2.0
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+# CLI E2E v2 — strangler-fig replacement for py-cli-e2e-tests.yml.
+#
+# Each connector lives under ingestion/tests/cli_e2e_v2/<connector>/ and
+# is a self-contained pytest module (no inheritance). The matrix below
+# grows by one entry per connector migration PR; the connector's
+# corresponding entry is removed from py-cli-e2e-tests.yml in the same PR.
+#
+# Triggers: workflow_dispatch only during the stabilization window for
+# the MySQL pilot. The schedule cron will be added once the pilot is
+# consistently green (see spec §7.1).
+
+name: py-cli-e2e-tests-v2
+on:
+  workflow_dispatch:
+    inputs:
+      connectors:
+        description: "Connectors to run (JSON array)"
+        required: true
+        default: '["mysql"]'
+
+permissions:
+  id-token: write
+  contents: read
+
+jobs:
+  py-cli-e2e-tests-v2:
+    runs-on: ubuntu-latest
+    timeout-minutes: 60
+    strategy:
+      fail-fast: false
+      matrix:
+        connector: ${{ fromJSON(inputs.connectors || '["mysql"]') }}
+    environment: test
+
+    steps:
+      - name: Free Disk Space (Ubuntu)
+        uses: jlumbroso/free-disk-space@main
+        with:
+          tool-cache: false
+          android: true
+          dotnet: true
+          haskell: true
+          large-packages: false
+          swap-storage: true
+          docker-images: false
+
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Setup Openmetadata Test Environment
+        uses: ./.github/actions/setup-openmetadata-test-environment
+        with:
+          python-version: '3.10'
+
+      - name: Run CLI E2E v2 tests
+        id: e2e-v2-test
+        env:
+          # MySQL test data lives in a dedicated MySQL container that the
+          # session-scoped `mysql_container` pytest fixture (testcontainers)
+          # boots, bootstraps with the OM-doc minimum grants, and tears
+          # down. Teammates run the same way locally — no env plumbing.
+          # Only OM-server admin creds (used to mint the ingestion-bot
+          # JWT) need to be exported here; they come from the bundled
+          # docker-compose and are not secrets.
+          OM_ADMIN_EMAIL: admin@open-metadata.org
+          OM_ADMIN_PASSWORD: admin
+        run: |
+          source env/bin/activate
+          cd ingestion
+          mkdir -p junit
+          pytest -v \
+            --junitxml=junit/test-results-v2-${{ matrix.connector }}.xml \
+            tests/cli_e2e_v2/${{ matrix.connector }}
+        shell: bash
+
+      - name: Upload tests artifact
+        if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          name: tests-v2-${{ matrix.connector }}
+          path: ingestion/junit/test-results-v2-*.xml
+
+      - name: Clean Up
+        if: always()
+        run: |
+          cd ./docker/development
+          docker compose down --remove-orphans
+          sudo rm -rf ${PWD}/docker-volume

ingestion/pyproject.toml

@@ -222,6 +222,12 @@ ignore = [
 # lands in a later stage tests don't immediately error out.
 "tests/**/*.py" = ["S101", "PLR2004", "PLC0415"]
 "ingestion/tests/**/*.py" = ["S101", "PLR2004", "PLC0415"]
+# v2 CLI E2E framework uses relative imports by design (connector-centric
+# layout — connectors live in subdirs and import from `..core.*` / `.connector`).
+# `T201` (print) is allowed in the top-level conftest for the session-start
+# posture banner. Path listed twice for the dual-cwd pattern above.
+"tests/cli_e2e_v2/**/*.py" = ["S101", "PLR2004", "PLC0415", "TID252", "T201"]
+"ingestion/tests/cli_e2e_v2/**/*.py" = ["S101", "PLR2004", "PLC0415", "TID252", "T201"]
 # Auto-generated from JSON Schema — never edit, never lint.
 "src/metadata/generated/**" = ["ALL"]
 "ingestion/src/metadata/generated/**" = ["ALL"]

ingestion/src/metadata/cli/app.py

@@ -17,20 +17,22 @@
 import traceback
 from pathlib import Path
 
+from metadata.cli.common import execute_workflow
 from metadata.config.common import load_config_file
 from metadata.utils.logger import cli_logger
 from metadata.workflow.application import ApplicationWorkflow
 
 logger = cli_logger()
 
 
-def run_app(config_path: Path) -> None:
+def run_app(config_path: Path, status_file: Path | None = None) -> None:
     """
     Run the application workflow from a config path
     to a JSON or YAML file
     :param config_path: Path to load JSON config
     """
 
+    config_dict = None
     try:
         config_dict = load_config_file(config_path)
         # no logging for config because apps might have custom secrets
@@ -40,7 +42,4 @@ def run_app(config_path: Path) -> None:
         logger.debug(traceback.format_exc())
         sys.exit(1)
 
-    workflow.execute()
-    workflow.stop()
-    workflow.print_status()
-    workflow.raise_from_status()
+    execute_workflow(workflow=workflow, config_dict=config_dict, status_file=status_file)

ingestion/src/metadata/cli/classify.py

@@ -28,7 +28,7 @@
 logger = cli_logger()
 
 
-def run_classification(config_path: Path) -> None:
+def run_classification(config_path: Path, status_file: Path | None = None) -> None:
     """
     Run the sampler workflow from a config path
     to a JSON or YAML file
@@ -48,4 +48,4 @@ def run_classification(config_path: Path) -> None:
         WorkflowInitErrorHandler.print_init_error(exc, config_dict, PipelineType.metadata)
         sys.exit(1)
 
-    execute_workflow(workflow=workflow, config_dict=config_dict)
+    execute_workflow(workflow=workflow, config_dict=config_dict, status_file=status_file)

ingestion/src/metadata/cli/common.py

@@ -13,14 +13,23 @@
 Handle workflow execution
 """
 
+from pathlib import Path
 from typing import Any, Dict  # noqa: UP035
 
 from metadata.workflow.base import BaseWorkflow
 
 
-def execute_workflow(workflow: BaseWorkflow, config_dict: Dict[str, Any]) -> None:  # noqa: UP006
-    """Execute the workflow and raise if needed"""
-    workflow.execute()
-    workflow.stop()
+def execute_workflow(
+    workflow: BaseWorkflow,
+    config_dict: Dict[str, Any],  # noqa: UP006
+    status_file: Path | None = None,
+) -> None:
+    """Execute the workflow, write status file if requested, raise on failure if configured."""
+    try:
+        workflow.execute()
+    finally:
+        workflow.stop()
+        if status_file is not None:
+            workflow.write_status_file(status_file)
     if config_dict.get("workflowConfig", {}).get("raiseOnError", True):
         workflow.raise_from_status()

ingestion/src/metadata/cli/dataquality.py

@@ -28,7 +28,7 @@
 logger = cli_logger()
 
 
-def run_test(config_path: Path) -> None:
+def run_test(config_path: Path, status_file: Path | None = None) -> None:
     """
     Run the Data Quality Test Suites workflow from a config path
     to a JSON or YAML file
@@ -48,4 +48,4 @@ def run_test(config_path: Path) -> None:
         WorkflowInitErrorHandler.print_init_error(exc, workflow_config_dict, PipelineType.TestSuite)
         sys.exit(1)
 
-    execute_workflow(workflow=workflow, config_dict=workflow_config_dict)
+    execute_workflow(workflow=workflow, config_dict=workflow_config_dict, status_file=status_file)

ingestion/src/metadata/cli/ingest.py

@@ -29,7 +29,7 @@
 logger = cli_logger()
 
 
-def run_ingest(config_path: Path) -> None:
+def run_ingest(config_path: Path, status_file: Path | None = None) -> None:
     """
     Run the ingestion workflow from a config path
     to a JSON or YAML file
@@ -46,4 +46,4 @@ def run_ingest(config_path: Path) -> None:
         WorkflowInitErrorHandler.print_init_error(exc, config_dict, PipelineType.metadata)
         sys.exit(1)
 
-    execute_workflow(workflow=workflow, config_dict=config_dict)
+    execute_workflow(workflow=workflow, config_dict=config_dict, status_file=status_file)

ingestion/src/metadata/cli/ingest_dbt.py

@@ -25,6 +25,7 @@
 from dotenv import load_dotenv
 from pydantic import BaseModel, Field, field_validator
 
+from metadata.cli.common import execute_workflow
 from metadata.ingestion.ometa.credentials import URL
 from metadata.utils.logger import cli_logger
 from metadata.workflow.metadata import MetadataWorkflow
@@ -291,7 +292,7 @@ def create_dbt_workflow_config(dbt_project_path: Path, om_config: OpenMetadataDB
     return config  # noqa: RET504
 
 
-def run_ingest_dbt(dbt_project_path: Path) -> None:
+def run_ingest_dbt(dbt_project_path: Path, status_file: Path | None = None) -> None:
     """
     Run the dbt artifacts ingestion workflow from a dbt project path
 
@@ -321,13 +322,13 @@ def run_ingest_dbt(dbt_project_path: Path) -> None:
         logger.info("Creating workflow configuration...")
         workflow_config = create_dbt_workflow_config(dbt_project_path, om_config)
 
-        # Create and execute the MetadataWorkflow (reusing existing infrastructure)
         logger.info("Starting OpenMetadata ingestion workflow...")
         workflow = MetadataWorkflow.create(workflow_config)
-        workflow.execute()
-        workflow.raise_from_status()
-        workflow.print_status()
-        workflow.stop()
+        execute_workflow(
+            workflow=workflow,
+            config_dict=workflow_config,
+            status_file=status_file,
+        )
 
         logger.info("DBT artifacts ingestion completed successfully")
 

ingestion/src/metadata/cli/lineage.py

@@ -47,7 +47,7 @@ class LineageWorkflow(BaseModel):
     parserType: Optional[QueryParserType] = QueryParserType.Auto  # noqa: N815, UP045
 
 
-def run_lineage(config_path: Path) -> None:
+def run_lineage(config_path: Path, status_file: Path | None = None) -> None:
     """
     Run the ingestion workflow from a config path
     to a JSON or YAML file

ingestion/src/metadata/cli/profile.py

@@ -29,7 +29,7 @@
 logger = cli_logger()
 
 
-def run_profiler(config_path: Path) -> None:
+def run_profiler(config_path: Path, status_file: Path | None = None) -> None:
     """
     Run the Profiler workflow from a config path
     to a JSON or YAML file
@@ -46,4 +46,4 @@ def run_profiler(config_path: Path) -> None:
         WorkflowInitErrorHandler.print_init_error(exc, workflow_config_dict, PipelineType.profiler)
         sys.exit(1)
 
-    execute_workflow(workflow=workflow, config_dict=workflow_config_dict)
+    execute_workflow(workflow=workflow, config_dict=workflow_config_dict, status_file=status_file)

ingestion/src/metadata/cli/usage.py

@@ -29,7 +29,7 @@
 logger = cli_logger()
 
 
-def run_usage(config_path: Path) -> None:
+def run_usage(config_path: Path, status_file: Path | None = None) -> None:
     """
     Run the usage workflow from a config path
     to a JSON or YAML file
@@ -46,4 +46,4 @@ def run_usage(config_path: Path) -> None:
         WorkflowInitErrorHandler.print_init_error(exc, config_dict, PipelineType.usage)
         sys.exit(1)
 
-    execute_workflow(workflow=workflow, config_dict=config_dict)
+    execute_workflow(workflow=workflow, config_dict=config_dict, status_file=status_file)

ingestion/src/metadata/cmd.py

@@ -77,6 +77,13 @@ def create_common_config_parser_args(parser: argparse.ArgumentParser):
         type=Path,
         required=True,
     )
+    parser.add_argument(
+        "--status-file",
+        help="path to write structured JSON status output (optional)",
+        type=Path,
+        required=False,
+        default=None,
+    )
 
 
 def create_dbt_parser_args(parser: argparse.ArgumentParser):
@@ -220,6 +227,7 @@ def metadata(args: Optional[List[str]] = None):  # noqa: UP006, UP045
     metadata_workflow = contains_args.get("command")
     config_file: Optional[Path] = contains_args.get("config")  # noqa: UP045
     dbt_project_path: Optional[Path] = contains_args.get("dbt_project_path")  # noqa: UP045
+    status_file: Optional[Path] = contains_args.get("status_file")  # noqa: UP045
 
     path = None
     if config_file:
@@ -234,7 +242,7 @@ def metadata(args: Optional[List[str]] = None):  # noqa: UP006, UP045
         set_loggers_level(log_level)
 
     if path and metadata_workflow and metadata_workflow in RUN_PATH_METHODS:
-        RUN_PATH_METHODS[metadata_workflow](path)
+        RUN_PATH_METHODS[metadata_workflow](path, status_file)
 
     if metadata_workflow == MetadataCommands.SCAFFOLD_CONNECTOR.value:
         has_name = contains_args.get("name")

ingestion/src/metadata/workflow/base.py

@@ -12,10 +12,12 @@
 Base workflow definition.
 """
 
+import json
 import traceback
 import uuid
 from abc import ABC, abstractmethod
 from datetime import datetime
+from pathlib import Path
 from statistics import mean
 from typing import Any, Dict, List, Optional, TypeVar, Union  # noqa: UP035
 
@@ -212,13 +214,21 @@ def get_failures(self) -> List[StackTraceError]:  # noqa: UP006
     def workflow_steps(self) -> List[Step]:  # noqa: UP006
         """Steps to report status from"""
 
+    def _step_meets_success_threshold(self, step: Step) -> bool:
+        """True iff the step has no failures, or its success ratio meets the workflow's threshold.
+
+        Shared by `raise_from_status_internal` (which raises on failure) and
+        `write_status_file` (which reports the CLI's observable success/failure state).
+        """
+        status = step.get_status()
+        if not status.failures:
+            return True
+        return status.calculate_success() >= self.workflow_config.successThreshold  # pyright: ignore[reportOperatorIssue]
+
     def raise_from_status_internal(self, raise_warnings=False) -> None:
         """Based on the internal workflow status, raise a WorkflowExecutionError"""
         for step in self.workflow_steps():
-            if (
-                step.get_status().failures
-                and step.get_status().calculate_success() < self.workflow_config.successThreshold
-            ):
+            if not self._step_meets_success_threshold(step):
                 raise WorkflowExecutionError(f"{step.name} reported errors: {Summary.from_step(step)}")
 
             if raise_warnings and step.status.warnings:
@@ -400,3 +410,28 @@ def print_status(self):
             start_time,
             self._is_debug_enabled(),
         )
+
+    def write_status_file(self, path: Path) -> None:
+        """Serialize per-step status to JSON at the given path.
+
+        The `success` field mirrors the CLI's exit-code semantic: True iff every
+        step meets its success threshold (the same condition under which
+        `raise_from_status_internal` does NOT raise).
+
+        Shape:
+            {
+              "pipeline_type": str,
+              "ingestion_pipeline_fqn": str | None,
+              "success": bool,
+              "steps": [<StepSummary dicts>]
+            }
+        """
+        ingestion_status = self.build_ingestion_status()
+        success = all(self._step_meets_success_threshold(step) for step in self.workflow_steps())
+        payload = {
+            "pipeline_type": self.config.source.type,  # pyright: ignore[reportAttributeAccessIssue]
+            "ingestion_pipeline_fqn": self.config.ingestionPipelineFQN,  # pyright: ignore[reportAttributeAccessIssue]
+            "success": success,
+            "steps": ingestion_status.model_dump(),
+        }
+        path.write_text(json.dumps(payload, indent=2, default=str))