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

[IceS2]

Summary

First vertical slice of the v2 CLI E2E test framework — replaces the v1 test_cli_<connector>.py per-source approach with a declarative shape-then-assert framework. Ships the framework core plus MySQL as the pilot connector. Other connectors land in follow-up PRs.

What's new

Framework (tests/cli_e2e_v2/core/)

• Declarative source baselines via SQLAlchemy Core (MetaData + TableSeed + ViewDefinition + StoredProcedureDefinition).
• Dialect-agnostic SqlBaselineEnforcer (Inspector-based introspect, dialect-specific apply). Adding a new SQL dialect is ~30 lines.
• WorkflowConfig builder with .pipeline(options) transition + filter overlay; renders ${E2E_*} env refs (no secrets in tmp_path).
• Fluent assertion catalog: OmClient → TableAssert / StoredProcedureAssert / ServiceAssert + descend into ColumnAssert / LineageAssert / ProfileAssert. One-shot .eventually(timeout) arms the next terminal for polling.
• Structural differ that walks Expected* against OM, raises StructuralMismatch with category-summary + path-grouped diffs.
• Expected* derived from the SQLAlchemy MetaData + dialect-specific type map — no hand-rolled column lists.

MySQL pilot (tests/cli_e2e_v2/mysql/)

• 12 tests covering metadata structural, profiler metrics, lineage from view DDL, FK constraint, auto-classification PII, mark-deleted, error containment, and a parametrized 4-scenario filter matrix.
• Self-contained: a session-scoped mysql_container testcontainers fixture boots mysql:8.0, creates a scoped om_user with the documented OM-MySQL minimum GRANTs (SELECT, SHOW VIEW, EXECUTE on the schema; PROCESS + SHOW_ROUTINE globally for SP body introspection), and populates E2E_MYSQL_* env vars. Teammates run pytest with no env plumbing.

Meta-tests (tests/cli_e2e_v2/meta/)

• 48 synthetic unit tests proving the framework catches each documented failure mode. Run in <2s, no testcontainers / network.
• Each fluent assertion paired with positive + negative cases; structural differ exercises every DiffKind via parametrize.

CI

• New workflow .github/workflows/py-cli-e2e-tests-v2.yml runs the suite on push and PR. Drops the v1 GRANT bootstrap step — the testcontainers fixture owns container lifecycle, schema bootstrap, and grants.

What's NOT in this PR

• v1 mysql stays for now. Per the strangler pattern, v1 test_cli_mysql.py is removed in a follow-up PR after v2 has run cleanly in CI for one week. This avoids the "delete-the-old-path-in-the-same-PR-as-the-new-one" trap.
• Other connectors. Postgres, Snowflake, BigQuery, etc. land in subsequent slice PRs, each ~6 files mirroring the MySQL scaffold.
• DQ / usage / query-log lineage pipelines — Bucket D, post-MVP per the spec.

Test plan

• pytest tests/cli_e2e_v2/mysql -v → 12 passed
• pytest tests/cli_e2e_v2/meta -v → 48 passed
• make py_format_check clean
• nox -s static-checks no new errors vs baseline
• CI workflow green on the PR

Notable workflow fixes bundled in

fix(workflow): basedpyright cleanup for write_status_file — three pyright-ignore comments on BaseWorkflow.config access in the new write_status_file method (added earlier in this branch). The Union[Any, Dict] config type is load-bearing for the ApplicationWorkflow subclass which uses OpenMetadataApplicationConfig, so tightening the base type isn't safe; targeted ignores keep the annotation correct without breaking polymorphism. Also tightens build_ingestion_status return type from Optional[IngestionStatus] to IngestionStatus (body unconditionally returns non-None).

Docs

• tests/cli_e2e_v2/README.md — how to run + debug
• tests/cli_e2e_v2/CONNECTORS.md — six-file scaffold for adding a new connector, mirrors mysql/

[gitar-bot]

⚠️ Bug: run_app lost print_status() call — silent CLI output regression

The old run_app explicitly called workflow.print_status() between stop() and raise_from_status(), giving operators a human-readable summary on stdout. The new unified execute_workflow in common.py does not call print_status(), so the app command now runs silently on success. Other run_* functions (ingest, profile, etc.) may have had their own print_status() calls that were similarly dropped during unification.

Suggested fix:

Add `workflow.print_status()` inside `execute_workflow` (after `workflow.stop()` in the finally block, or after the try/finally before `raise_from_status`), so all workflow types get consistent output:

    try:
        workflow.execute()
    finally:
        workflow.stop()
        if status_file is not None:
            workflow.write_status_file(status_file)
    workflow.print_status()
    if config_dict.get("workflowConfig", {}).get("raiseOnError", True):
        workflow.raise_from_status()

_{Was this helpful? React with 👍 / 👎 | Reply gitar fix to apply this suggestion}

[gitar-bot]

⚠️ Edge Case: write_status_file assumes config.source.type — crashes for ApplicationWorkflow

write_status_file accesses self.config.source.type (line 432) and self.config.ingestionPipelineFQN (line 433). For ApplicationWorkflow, config is OpenMetadataApplicationConfig which uses a different schema and may not expose .source.type. If a user passes --status-file to the app subcommand, this will raise AttributeError inside the finally block, masking any original workflow exception.

Suggested fix:

Guard the attribute access with hasattr/getattr, or use a polymorphic method:

    payload = {
        "pipeline_type": getattr(getattr(self.config, "source", None), "type", None),
        "ingestion_pipeline_fqn": getattr(self.config, "ingestionPipelineFQN", None),
        "success": success,
        "steps": ingestion_status.model_dump(),
    }

_{Was this helpful? React with 👍 / 👎 | Reply gitar fix to apply this suggestion}

[gitar-bot]

💡 Edge Case: write_status_file in finally can suppress the original exception

In execute_workflow, write_status_file runs inside the finally block. If workflow.execute() raises and write_status_file also raises (e.g., because steps aren't populated after an early crash), the original exception is replaced by the status-file error, making debugging harder. This only triggers when --status-file is explicitly provided.

Suggested fix:

Wrap the write_status_file call in a try/except to log-and-swallow:

    finally:
        workflow.stop()
        if status_file is not None:
            try:
                workflow.write_status_file(status_file)
            except Exception:
                logger.warning("Failed to write status file", exc_info=True)

_{Was this helpful? React with 👍 / 👎 | Reply gitar fix to apply this suggestion}

[gitar-bot]

Code Review ⚠️ Changes requested 0 resolved / 3 findings

Introduces a declarative v2 E2E CLI framework with a MySQL pilot and meta-testing suite. Changes are requested to address silent CLI output regressions in run_app, TypeErrors in write_status_file for ApplicationWorkflow, and exception masking in the finally block.

⚠️ Bug: run_app lost print_status() call — silent CLI output regression

📄 ingestion/src/metadata/cli/app.py:45 📄 ingestion/src/metadata/cli/common.py:28-35

The old run_app explicitly called workflow.print_status() between stop() and raise_from_status(), giving operators a human-readable summary on stdout. The new unified execute_workflow in common.py does not call print_status(), so the app command now runs silently on success. Other run_* functions (ingest, profile, etc.) may have had their own print_status() calls that were similarly dropped during unification.

Suggested fix

Add `workflow.print_status()` inside `execute_workflow` (after `workflow.stop()` in the finally block, or after the try/finally before `raise_from_status`), so all workflow types get consistent output:

    try:
        workflow.execute()
    finally:
        workflow.stop()
        if status_file is not None:
            workflow.write_status_file(status_file)
    workflow.print_status()
    if config_dict.get("workflowConfig", {}).get("raiseOnError", True):
        workflow.raise_from_status()

⚠️ Edge Case: write_status_file assumes config.source.type — crashes for ApplicationWorkflow

📄 ingestion/src/metadata/workflow/base.py:432-433 📄 ingestion/src/metadata/cli/app.py:28 📄 ingestion/src/metadata/cli/app.py:45

write_status_file accesses self.config.source.type (line 432) and self.config.ingestionPipelineFQN (line 433). For ApplicationWorkflow, config is OpenMetadataApplicationConfig which uses a different schema and may not expose .source.type. If a user passes --status-file to the app subcommand, this will raise AttributeError inside the finally block, masking any original workflow exception.

Suggested fix

Guard the attribute access with hasattr/getattr, or use a polymorphic method:

    payload = {
        "pipeline_type": getattr(getattr(self.config, "source", None), "type", None),
        "ingestion_pipeline_fqn": getattr(self.config, "ingestionPipelineFQN", None),
        "success": success,
        "steps": ingestion_status.model_dump(),
    }

💡 Edge Case: write_status_file in finally can suppress the original exception

📄 ingestion/src/metadata/cli/common.py:30-33

In execute_workflow, write_status_file runs inside the finally block. If workflow.execute() raises and write_status_file also raises (e.g., because steps aren't populated after an early crash), the original exception is replaced by the status-file error, making debugging harder. This only triggers when --status-file is explicitly provided.

Suggested fix

Wrap the write_status_file call in a try/except to log-and-swallow:

    finally:
        workflow.stop()
        if status_file is not None:
            try:
                workflow.write_status_file(status_file)
            except Exception:
                logger.warning("Failed to write status file", exc_info=True)

🤖 Prompt for agents

Code Review: Introduces a declarative v2 E2E CLI framework with a MySQL pilot and meta-testing suite. Changes are requested to address silent CLI output regressions in `run_app`, TypeErrors in `write_status_file` for `ApplicationWorkflow`, and exception masking in the `finally` block.

1. ⚠️ Bug: run_app lost print_status() call — silent CLI output regression
   Files: ingestion/src/metadata/cli/app.py:45, ingestion/src/metadata/cli/common.py:28-35

   The old `run_app` explicitly called `workflow.print_status()` between `stop()` and `raise_from_status()`, giving operators a human-readable summary on stdout. The new unified `execute_workflow` in `common.py` does not call `print_status()`, so the `app` command now runs silently on success. Other `run_*` functions (ingest, profile, etc.) may have had their own `print_status()` calls that were similarly dropped during unification.

   Suggested fix:
   Add `workflow.print_status()` inside `execute_workflow` (after `workflow.stop()` in the finally block, or after the try/finally before `raise_from_status`), so all workflow types get consistent output:
   
       try:
           workflow.execute()
       finally:
           workflow.stop()
           if status_file is not None:
               workflow.write_status_file(status_file)
       workflow.print_status()
       if config_dict.get("workflowConfig", {}).get("raiseOnError", True):
           workflow.raise_from_status()

2. ⚠️ Edge Case: write_status_file assumes config.source.type — crashes for ApplicationWorkflow
   Files: ingestion/src/metadata/workflow/base.py:432-433, ingestion/src/metadata/cli/app.py:28, ingestion/src/metadata/cli/app.py:45

   `write_status_file` accesses `self.config.source.type` (line 432) and `self.config.ingestionPipelineFQN` (line 433). For `ApplicationWorkflow`, config is `OpenMetadataApplicationConfig` which uses a different schema and may not expose `.source.type`. If a user passes `--status-file` to the `app` subcommand, this will raise `AttributeError` inside the `finally` block, masking any original workflow exception.

   Suggested fix:
   Guard the attribute access with hasattr/getattr, or use a polymorphic method:
   
       payload = {
           "pipeline_type": getattr(getattr(self.config, "source", None), "type", None),
           "ingestion_pipeline_fqn": getattr(self.config, "ingestionPipelineFQN", None),
           "success": success,
           "steps": ingestion_status.model_dump(),
       }

3. 💡 Edge Case: write_status_file in finally can suppress the original exception
   Files: ingestion/src/metadata/cli/common.py:30-33

   In `execute_workflow`, `write_status_file` runs inside the `finally` block. If `workflow.execute()` raises and `write_status_file` also raises (e.g., because steps aren't populated after an early crash), the original exception is replaced by the status-file error, making debugging harder. This only triggers when `--status-file` is explicitly provided.

   Suggested fix:
   Wrap the write_status_file call in a try/except to log-and-swallow:
   
       finally:
           workflow.stop()
           if status_file is not None:
               try:
                   workflow.write_status_file(status_file)
               except Exception:
                   logger.warning("Failed to write status file", exc_info=True)

Options

Display: compact → Showing less information.

Comment with these commands to change:

Compact

gitar display:verbose

_{Was this helpful? React with 👍 / 👎 | Gitar}

[NiltonDuarte]

Great typing communicates intent across time. Right now, a signature like Env[True] is a bit opaque. We can make it entirely self-documenting by introducing semantic aliases:

# Constrained — callers can only parameterize Env with Required (True)  or NotRequired (False), matching
# the two concrete `required` states. Anything else is a type error.
Required: TypeAlias = Literal[True]
NotRequired: TypeAlias = Literal[False]
_Req = TypeVar("_Req", Required, NotRequired)

This allows us to write/hint Env[Required] or Env[NotRequired], whic tells the developer what is expected. While it might be a minor detail for this specific, simple class, adopting this pattern sets a great standard for readability across the codebase.

[NiltonDuarte]

Overriding new with a custom signature breaks standard object instantiation conventions and feels hacky here.

We can achieve the exact same type resolution much more cleanly by using @overload on init. This respects standard interfaces and keeps the logic straightforward without having to touch new. Here is how that looks:

    # Two __init__ overloads — one per Literal[required] value — let the type
    # checker pick the right `Env[NotRequired|Required]` specialization at the call
    # site.
    @overload
    def __init__(
        self: Env[Required],
        key: str,
        default: str | None = None,
        *,
        required: Literal[True],
    ) -> None: ...

    @overload
    def __init__(
        self: Env[NotRequired],
        key: str,
        default: str | None = None,
        *,
        required: Literal[False] = False,
    ) -> None: ...

    def __init__(
        self,
        key: str,
        default: str | None = None,
        *,
        required: bool = True,
    ) -> None:
        self.key = key
        if default is not None:
            os.environ.setdefault(key, default)
        if required and not os.environ.get(key):
            raise EnvLoadError(
                f"required env var {key} not set. Set it in your shell or GitHub Actions secrets."
            )

    # Two .get() overloads narrow by the specialization of Env:
    #   Env[Required].get()  -> str          (construction validated)
    #   Env[NotRequired].get() -> str | None   (caller must handle None)
    @overload
    def get(self: Env[Required]) -> str: ...

    @overload
    def get(self: Env[NotRequired]) -> str | None: ...

    def get(self) -> str | None:
        return os.environ.get(self.key)