From 1dc9498fbc783634873ea486d1a3538c009bf1c1 Mon Sep 17 00:00:00 2001 From: fas89 <50082482+fas89@users.noreply.github.com> Date: Mon, 12 Oct 2026 12:41:35 +0200 Subject: [PATCH 1/4] fix(iac): apply refuses an MWAA environment it cannot create instead of reporting success An AWS contract with orchestration.engine airflow (or mwaa) and an orchestration.schedule made the AWS planner plan mwaa.ensure_environment, and the OpenTofu emitter's _emit_from_actions had no branch for it. The generated main.tf.json held no aws_mwaa_environment, fluid apply reported success, and no Airflow environment existed. OpenTofu is the only AWS apply engine, so nothing else created it. fluid apply and fluid generate iac now stop with unsupported_binding, kind mwaa-environment-not-emitted, naming the environment and two ways out: engine none keeps the schedule on EventBridge Scheduler, or push the DAG to an MWAA environment you own with fluid schedule-sync. The OpenTofu engine cannot create one itself: aws_mwaa_environment needs a VPC network_configuration with private subnets, which the contract does not carry (aws-ia/terraform-aws-mwaa requires private_subnet_ids with no default; the CloudFormation NetworkConfiguration reference says the same). Every other planned op without an emitter is accounted for too. An op no list knows is refused by name (kind planned-op-not-emitted), the way dbt raises MissingMaterializationError ("No materialization ... was found for adapter", dbt-adapters exceptions/compilation.py) rather than skipping the model. Data-plane ops emit() derives from exposes[] and build steps pass quietly; known planner ops nothing provisions yet (IAM bindings from metadata.policies, Athena views, Redshift schemas and tables, execution.trigger schedules) and orchestration.tasks steps are named in a warning. --shadow now maps an environment op to a logical resource, so a dropped one would show as a native-only gap. --- AUTOGEN_SPIKE.md | 5 +- CHANGELOG.md | 13 ++ fluid_build/iac/providers/aws.py | 115 +++++++++++- fluid_build/iac/shadow.py | 5 + tests/iac/test_iac_aws.py | 4 +- tests/iac/test_iac_aws_unemitted_ops.py | 228 ++++++++++++++++++++++++ tests/iac/test_iac_shadow.py | 5 +- 7 files changed, 369 insertions(+), 6 deletions(-) create mode 100644 tests/iac/test_iac_aws_unemitted_ops.py diff --git a/AUTOGEN_SPIKE.md b/AUTOGEN_SPIKE.md index dd7687f7..a1c80cf3 100644 --- a/AUTOGEN_SPIKE.md +++ b/AUTOGEN_SPIKE.md @@ -246,7 +246,10 @@ self-skipping when `tofu` or the gcloud emulator is absent. What is *not* mapped is genuinely imperative — an R8 boundary, not an emitter gap: - **AWS MWAA** (`aws_mwaa_environment`) — needs VPC `network_configuration` - (subnets, security groups) the contract does not carry. + (subnets, security groups) the contract does not carry. A planned + `mwaa.ensure_environment` is refused (`unsupported_binding`, kind + `mwaa-environment-not-emitted`) rather than skipped, as is any AWS op the + emitter does not know (`planned-op-not-emitted`). - **Non-SQL Snowflake procedures / UDFs** (Python / Java / Scala) — need runtime / handler / package config the contract does not carry. - **`composer.trigger_dag`**, **Athena query execution**, **Redshift table diff --git a/CHANGELOG.md b/CHANGELOG.md index ab11a72c..61647a9a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -28,6 +28,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 longer does (`rest`, `iceberg_rest`, `polaris`, `unity`, `nessie` and `glue` never had one). `fluid apply` stops with `iceberg_catalog_move_blocked` and the `tofu state rm` command for the volume instead of planning to drop it. +- An AWS contract with `orchestration.engine: airflow` or `mwaa` and an + `orchestration.schedule` now fails `fluid apply` and `fluid generate iac` with + `unsupported_binding` (kind `mwaa-environment-not-emitted`). Its plan promised an MWAA + environment that the OpenTofu engine never created, since `aws_mwaa_environment` needs a + VPC network the contract does not carry; apply reported success and no environment + existed. To schedule without Airflow, set `orchestration.engine: none` and keep the + schedule (an EventBridge Scheduler schedule is created). To keep Airflow, create the + MWAA environment yourself, push the DAG with `fluid generate schedule` and + `fluid schedule-sync --scheduler mwaa`, and apply a copy of the contract without + `orchestration.schedule`. Any other planned AWS op the engine does not know now fails + the same way (kind `planned-op-not-emitted`), and the known ops it does not provision + yet (IAM bindings from `metadata.policies`, Athena views, Redshift schemas and tables, + `execution.trigger` schedules, `orchestration.tasks`) are named in a warning. - A GCP Iceberg expose that a streaming sink writes to must now name its catalog in `binding.location.catalog`. Without one, the sink wrote through whatever catalog reached the worker (REST by default, Glue for `sink.catalog: glue` or a hand-written diff --git a/fluid_build/iac/providers/aws.py b/fluid_build/iac/providers/aws.py index 62082502..9a276352 100644 --- a/fluid_build/iac/providers/aws.py +++ b/fluid_build/iac/providers/aws.py @@ -1100,6 +1100,65 @@ def _emit_kinesis( ) +#: Planned ops ``_emit_from_actions`` leaves alone because something else owns +#: them: the data plane :meth:`AwsIacPlugin.emit` derives from ``exposes[]`` +#: (see :class:`~fluid_build.iac.base.IacProviderPlugin`), and the build steps +#: the build runners execute rather than provision. +_OPS_OWNED_ELSEWHERE = frozenset( + { + "s3.ensure_bucket", + "glue.ensure_database", + "glue.ensure_table", + "glue.ensure_iceberg_table", + "dbt.prepare_profile", + "athena.execute_query", + } +) + +#: Ops the AWS planner emits that nothing in the OpenTofu engine provisions +#: yet. Each is named in a warning on every emit rather than refused: refusing +#: them would fail contracts the MWAA fix is not about, and dropping them +#: without a word is the defect this list exists to end. +_OPS_NOT_YET_EMITTED = frozenset( + { + "iam.bind_glue_database", + "iam.bind_s3_bucket", + "athena.create_view", + "redshift.ensure_schema", + "redshift.ensure_table", + "events.ensure_schedule", + } +) + + +def _refuse_mwaa_environment(action: Mapping[str, Any]) -> UnsupportedBindingError: + """``mwaa.ensure_environment`` — planned, but OpenTofu cannot create it. + + ``aws_mwaa_environment`` requires ``network_configuration`` (two private + ``subnet_ids`` and ``security_group_ids``), and the contract carries no + network. Skipping the op let ``fluid apply`` report success with no + Airflow environment, so it is refused instead. + """ + name = action.get("environment_name") or "?" + return UnsupportedBindingError( + "mwaa-environment-not-emitted", + f"The plan asks for the MWAA environment {name!r} (orchestration.engine airflow or " + "mwaa with orchestration.schedule), but the OpenTofu engine cannot create an " + "aws_mwaa_environment: it needs a VPC network_configuration (subnets and security " + "groups) that the contract does not carry. Applying would report success with no " + "Airflow environment, so nothing was applied.", + ( + "To schedule without Airflow, set orchestration.engine to 'none' and keep " + "orchestration.schedule: fluid apply then creates an EventBridge Scheduler " + "schedule.", + "To keep Airflow, create the MWAA environment yourself (it needs your VPC), push " + "this contract's DAG to it with `fluid generate schedule` and `fluid schedule-sync " + "--scheduler mwaa --environment-name `, and run fluid apply on a copy of the " + "contract without orchestration.schedule.", + ), + ) + + def _emit_from_actions( resources: Dict[str, Any], actions: Iterable[Mapping[str, Any]], cid: str ) -> None: @@ -1107,8 +1166,8 @@ def _emit_from_actions( Covers Glue ETL jobs, Step Functions, and the Lambda schedule / event path — inline Lambda source is zipped by ``data.archive_file`` (see - :meth:`AwsIacPlugin.emit_data`). MWAA is still skipped: ``aws_mwaa_environment`` - needs VPC ``network_configuration`` the contract does not carry. + :meth:`AwsIacPlugin.emit_data`). An op with no emitter goes to + :func:`_account_for_unemitted_op`, so no planned op is dropped without a word. """ for action in actions or []: if not isinstance(action, Mapping): @@ -1130,6 +1189,58 @@ def _emit_from_actions( _emit_event_rule(resources, action, cid) elif op == "s3.ensure_notification": _emit_s3_notification(resources, action, cid) + else: + _account_for_unemitted_op(action) + + +def _account_for_unemitted_op(action: Mapping[str, Any]) -> None: + """Pass, refuse, or name in a warning a planned op the emitter has no translation for. + + A data-plane or build op owned elsewhere (``_OPS_OWNED_ELSEWHERE``) passes. + ``mwaa.ensure_environment`` is refused (:func:`_refuse_mwaa_environment`); + an op no list here knows is refused by name, the way dbt fails with + ``MissingMaterializationError`` ("No materialization ... was found for + adapter", dbt-adapters ``exceptions/compilation.py``) rather than skipping + the model; a known op nothing provisions yet, and an orchestration task, + is named in a warning. + """ + op = action.get("op") + known = op if isinstance(op, str) else "" + # An orchestration.tasks[] step (providers/aws/plan/orchestration.py) is a + # task the scheduler runs, in whatever vocabulary the contract wrote: its + # ``s3.ensure_bucket`` is not the bucket exposes[] declares. + task = isinstance(action.get("orchestration"), Mapping) + if known in _OPS_OWNED_ELSEWHERE and not task: + return + if known == "mwaa.ensure_environment": + raise _refuse_mwaa_environment(action) + if known in _OPS_NOT_YET_EMITTED: + LOG.warning( + "aws_planned_op_not_emitted op=%s id=%s: the plan lists it, but the OpenTofu " + "engine provisions nothing for it, so fluid apply will not create it", + known, + action.get("id"), + ) + return + if task: + LOG.warning( + "aws_orchestration_task_not_emitted op=%s id=%s: orchestration tasks run in " + "the scheduler, and fluid apply provisions nothing for them", + op, + action.get("id"), + ) + return + where = f" (action {action['id']!r})" if action.get("id") else "" + raise UnsupportedBindingError( + "planned-op-not-emitted", + f"The plan contains the AWS op {op!r}{where}, and the OpenTofu engine " + "has no translation for it. Applying would skip it and report success, so nothing " + "was applied.", + ( + "Remove the contract field that plans this op, or report it as a fluid " + "bug if the planner produced it from a valid contract.", + ), + ) def _emit_glue_job(resources: Dict[str, Any], action: Mapping[str, Any], cid: str) -> None: diff --git a/fluid_build/iac/shadow.py b/fluid_build/iac/shadow.py index e6ba1b45..6b1788e9 100644 --- a/fluid_build/iac/shadow.py +++ b/fluid_build/iac/shadow.py @@ -51,6 +51,7 @@ class LogicalResource: "aws_glue_catalog_table": "table", "aws_s3_bucket": "bucket", "aws_kinesis_stream": "stream", + "aws_mwaa_environment": "environment", "google_bigquery_dataset": "dataset", "google_bigquery_table": "table", "google_storage_bucket": "bucket", @@ -73,6 +74,9 @@ class LogicalResource: ("bucket", "bucket"), ("topic", "topic"), ("stream", "stream"), + # `mwaa.ensure_environment`: without it the op resolved to no kind, so an + # environment the emitter dropped never showed up as a native-only gap. + ("environment", "environment"), ) _TF_NAME_FIELDS = ("name", "bucket", "table_id", "dataset_id", "topic", "stream") @@ -85,6 +89,7 @@ class LogicalResource: "database", "topic", "stream", + "environment_name", "name", ) diff --git a/tests/iac/test_iac_aws.py b/tests/iac/test_iac_aws.py index 63d3fdf0..0a375aae 100644 --- a/tests/iac/test_iac_aws.py +++ b/tests/iac/test_iac_aws.py @@ -314,9 +314,11 @@ def test_lambda_ops_are_skipped(self): [ {"op": "lambda.ensure_function", "function_name": "f", "code": {}}, {"op": "eventbridge.ensure_schedule", "schedule_name": "s"}, - {"op": "mwaa.ensure_environment", "environment_name": "e"}, + {"op": "lambda.add_permission", "function_name": "f"}, ], ) + # mwaa.ensure_environment used to be skipped here too; it is now + # refused (tests/iac/test_iac_aws_unemitted_ops.py). assert res == {} def test_no_actions_emits_no_planned_resources(self): diff --git a/tests/iac/test_iac_aws_unemitted_ops.py b/tests/iac/test_iac_aws_unemitted_ops.py new file mode 100644 index 00000000..8b0ddb55 --- /dev/null +++ b/tests/iac/test_iac_aws_unemitted_ops.py @@ -0,0 +1,228 @@ +# Copyright 2024-2026 Agentics Transformation Ltd +# +# 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. + +"""A planned AWS op the OpenTofu emitter does not translate is never dropped without a word. + +``orchestration.engine: airflow`` (or ``mwaa``) with a schedule made the AWS +planner promise an ``mwaa.ensure_environment`` in ``plan.json``, and the +emitter's ``_emit_from_actions`` had no branch for it: ``main.tf.json`` held no +``aws_mwaa_environment``, ``fluid apply`` reported success, and no Airflow +environment existed. OpenTofu is the only AWS apply engine, so nothing else +created it either. +""" + +from __future__ import annotations + +import json +import logging +from pathlib import Path +from typing import Any, Dict + +import pytest + +from fluid_build.cli import generate_iac +from fluid_build.cli._common import CLIError +from fluid_build.iac import build_module, get_iac_plugin, shadow_compare +from fluid_build.iac.base import UnsupportedBindingError +from fluid_build.iac.shadow import native_logical_resources +from fluid_build.providers.aws.provider import AwsProvider +from tests.iac.test_iac_generate_offline import _args, _write, no_aws # noqa: F401 (fixture) +from tests.iac.test_iac_generate_offline import _orders as _contract + +pytestmark = [pytest.mark.unit, pytest.mark.provider] + +ACCOUNT = "123456789012" +_AWS_LOGGER = "fluid_build.iac.providers.aws" + + +def _aws(): + return get_iac_plugin("aws") + + +def _planned(contract: Dict[str, Any]) -> list: + """The real AWS planner's actions — the same call ``fluid apply`` makes.""" + provider = AwsProvider(account_id=ACCOUNT, region="us-east-1", logger=logging.getLogger("t")) + return list(provider.plan(contract)) + + +class TestMwaaEnvironmentIsRefused: + def test_the_op_is_refused_with_its_environment_and_a_way_out(self): + with pytest.raises(UnsupportedBindingError) as excinfo: + _aws().emit( + _contract(), + [{"op": "mwaa.ensure_environment", "environment_name": "mwaa-orders"}], + ) + err = excinfo.value + assert err.kind == "mwaa-environment-not-emitted" + assert "mwaa-orders" in str(err) + assert "network_configuration" in str(err) + assert err.remediation, "the CLI renders these as next steps" + assert any("schedule-sync" in step for step in err.remediation) + + @pytest.mark.parametrize("engine", ["airflow", "mwaa"]) + def test_the_planned_path_fails_before_a_module_is_built(self, engine): + contract = _contract(orchestration={"engine": engine, "schedule": "0 * * * *"}) + actions = _planned(contract) + # The plan promises the environment: provider.py calls _plan_schedule + # unconditionally, so this is the path `fluid apply` takes. + assert any(a.get("op") == "mwaa.ensure_environment" for a in actions) + with pytest.raises(UnsupportedBindingError) as excinfo: + build_module(_aws(), contract, actions=actions) + assert excinfo.value.kind == "mwaa-environment-not-emitted" + assert "mwaa-offline_orders" in str(excinfo.value) + + def test_a_named_existing_environment_is_still_refused(self): + # orchestration.mwaa.environmentName names the environment, but the + # planned op still asks for it to exist — OpenTofu cannot ensure that. + contract = _contract( + orchestration={ + "engine": "airflow", + "schedule": "0 * * * *", + "mwaa": {"environmentName": "shared-airflow"}, + } + ) + with pytest.raises(UnsupportedBindingError) as excinfo: + build_module(_aws(), contract, actions=_planned(contract)) + assert "shared-airflow" in str(excinfo.value) + + def test_a_schedule_without_airflow_still_emits(self): + contract = _contract(orchestration={"schedule": "rate(1 hour)"}) + actions = _planned(contract) + assert not any(a.get("op") == "mwaa.ensure_environment" for a in actions) + module = json.loads(build_module(_aws(), contract, actions=actions)) + assert "aws_scheduler_schedule" in module["resource"] + + def test_no_orchestration_is_unaffected(self): + contract = _contract() + module = json.loads(build_module(_aws(), contract, actions=_planned(contract))) + assert set(module["resource"]) == { + "aws_glue_catalog_database", + "aws_glue_catalog_table", + "aws_s3_bucket", + } + + +class TestGenerateIac: + @pytest.fixture(autouse=True) + def _account(self, no_aws: None, monkeypatch: pytest.MonkeyPatch) -> None: + # After no_aws, which clears it: the EventBridge path names the account. + monkeypatch.setenv("AWS_ACCOUNT_ID", ACCOUNT) + + def _run(self, tmp_path: Path, contract: Dict[str, Any]) -> Path: + out = tmp_path / "out" + generate_iac.run(_args(_write(tmp_path, contract), out), logging.getLogger("t")) + return out + + def test_airflow_with_a_schedule_is_a_typed_cli_error(self, tmp_path): + contract = _contract(orchestration={"engine": "airflow", "schedule": "0 * * * *"}) + with pytest.raises(CLIError) as excinfo: + self._run(tmp_path, contract) + error = excinfo.value + assert (error.event, error.context["kind"]) == ( + "unsupported_binding", + "mwaa-environment-not-emitted", + ) + assert any("schedule-sync" in step for step in error.context["remediation"]) + # Refused before the output directory is even created. + assert not (tmp_path / "out").exists() + + def test_a_schedule_without_airflow_writes_the_module(self, tmp_path): + out = self._run(tmp_path, _contract(orchestration={"schedule": "rate(1 hour)"})) + module = json.loads((out / "main.tf.json").read_text(encoding="utf-8")) + assert "aws_scheduler_schedule" in module["resource"] + + +class TestEveryOtherOp: + # A non-string op must be the typed refusal too, not a TypeError from hashing it. + @pytest.mark.parametrize( + ("action", "named"), + [ + ({"op": "glue.ensure_crawler", "name": "c"}, "'glue.ensure_crawler'"), + ({"id": "x"}, "None (action 'x')"), + ({"id": "x", "op": ["glue", "ensure_job"]}, "['glue', 'ensure_job']"), + ], + ) + def test_an_op_with_no_translation_is_refused_by_name(self, action, named): + with pytest.raises(UnsupportedBindingError) as excinfo: + _aws().emit(_contract(), [action]) + err = excinfo.value + assert (err.kind, named in str(err)) == ("planned-op-not-emitted", True) + assert err.remediation + + @pytest.mark.parametrize( + "op", + [ + # The data plane: emit() derives these from exposes[]. + "s3.ensure_bucket", + "glue.ensure_database", + "glue.ensure_table", + "glue.ensure_iceberg_table", + # Build steps the build runners execute, never provisioned. + "dbt.prepare_profile", + "athena.execute_query", + ], + ) + def test_an_op_provisioned_or_run_elsewhere_passes_quietly(self, op, caplog): + caplog.set_level(logging.WARNING, logger=_AWS_LOGGER) + _aws().emit(_contract(), [{"op": op, "id": "a"}]) + assert not [r for r in caplog.records if op in r.getMessage()] + + @pytest.mark.parametrize( + "op", + [ + "iam.bind_glue_database", + "iam.bind_s3_bucket", + "athena.create_view", + "redshift.ensure_schema", + "redshift.ensure_table", + "events.ensure_schedule", + ], + ) + def test_a_known_op_with_no_emitter_is_named_in_a_warning(self, op, caplog): + caplog.set_level(logging.WARNING, logger=_AWS_LOGGER) + _aws().emit(_contract(), [{"op": op, "id": "a"}]) + (record,) = [r for r in caplog.records if op in r.getMessage()] + assert record.levelno == logging.WARNING + assert "aws_planned_op_not_emitted" in record.getMessage() + + # A task's ``s3.ensure_bucket`` names a bucket exposes[] may not declare. + @pytest.mark.parametrize("op", ["glue.run_job", "s3.ensure_bucket"]) + def test_an_orchestration_task_is_named_in_a_warning(self, op, caplog): + caplog.set_level(logging.WARNING, logger=_AWS_LOGGER) + task = { + "id": "task_1", + "op": op, + "orchestration": {"task_id": "task_1", "depends_on": [], "engine": "airflow"}, + } + _aws().emit(_contract(), [task]) + (record,) = [r for r in caplog.records if op in r.getMessage()] + assert "aws_orchestration_task_not_emitted" in record.getMessage() + + +class TestShadow: + def test_an_environment_op_is_a_logical_resource(self): + # Were the refusal ever removed, --shadow would report the dropped + # environment as a native-only gap instead of leaving it out. + found = native_logical_resources( + [{"op": "mwaa.ensure_environment", "environment_name": "mwaa-orders"}] + ) + assert {(r.kind, r.identity) for r in found} == {("environment", "mwaa-orders")} + + def test_shadow_compare_surfaces_the_refusal(self): + with pytest.raises(UnsupportedBindingError): + shadow_compare( + _contract(), + plugin=_aws(), + native_actions=[{"op": "mwaa.ensure_environment", "environment_name": "e"}], + ) diff --git a/tests/iac/test_iac_shadow.py b/tests/iac/test_iac_shadow.py index deaa4a6e..bbe38cb7 100644 --- a/tests/iac/test_iac_shadow.py +++ b/tests/iac/test_iac_shadow.py @@ -49,10 +49,11 @@ def _aws_contract(): # Native plan that fully matches the AWS contract above. +# The AWS planner's own op names: the emitter refuses an op it does not know. _AWS_NATIVE_MATCH = [ {"op": "glue.ensure_database", "database": "sales"}, - {"op": "glue.create_table", "table": "orders"}, - {"op": "s3.create_bucket", "bucket": "lake"}, + {"op": "glue.ensure_table", "table": "orders"}, + {"op": "s3.ensure_bucket", "bucket": "lake"}, ] From 0bbd3b120cd344150c2715ef205b4616f8c18215 Mon Sep 17 00:00:00 2001 From: fas89 <50082482+fas89@users.noreply.github.com> Date: Mon, 12 Oct 2026 15:05:19 +0200 Subject: [PATCH 2/4] fix(iac): --provider aws applies a multi-cloud contract again, and an MWAA contract is refused even when the planner fails The previous commit refused any planned op the AWS emitter did not know. native_actions plans with the first expose's provider, so on a contract whose first expose is on GCP or Snowflake, `--provider aws` (what the ambiguous-provider error tells the user to run) fed bq.* / sf.* ops to the AWS emitter and failed the AWS half with planned-op-not-emitted. An unknown op is now named in a warning (op and id rendered with repr, so a control character from a task's op never reaches the terminal) instead of refused. What kept the refusal honest is now a test: every "op" literal under fluid_build/providers/aws, and a mode="replace" plan, must be emitted (_ACTION_EMITTERS, which replaces the if/elif chain) or classified. It found aws.s3.copy_prefix, the pre-replace backup the rollback writer records, which is now owned elsewhere. dbt-core asserts adapter materialization coverage the same way. native_actions returns [] when the planner raises (a Glue table name its validator refuses, a cyclic orchestration.tasks), and the emitter then saw no mwaa.ensure_environment and wrote the data plane, exit 0. An empty plan for a contract that asks for MWAA (the schedule planner's own test: a schedule and engine airflow or mwaa, lower-cased, unstripped) is refused the same way. The remediation named `schedule-sync --environment-name `, which the MWAA dispatch never reads: it syncs to --destination. It now gives the full generate-schedule / schedule-sync pair with --destination filled from orchestration.mwaa.dagS3Path, and tells a contract that names its environment to push to it rather than create one. --- AUTOGEN_SPIKE.md | 5 +- CHANGELOG.md | 14 +- fluid_build/iac/providers/aws.py | 212 ++++++++++++++------ tests/iac/test_iac_aws_unemitted_ops.py | 252 ++++++++++++++++++++++-- 4 files changed, 403 insertions(+), 80 deletions(-) diff --git a/AUTOGEN_SPIKE.md b/AUTOGEN_SPIKE.md index a1c80cf3..64824406 100644 --- a/AUTOGEN_SPIKE.md +++ b/AUTOGEN_SPIKE.md @@ -248,8 +248,9 @@ self-skipping when `tofu` or the gcloud emulator is absent. - **AWS MWAA** (`aws_mwaa_environment`) — needs VPC `network_configuration` (subnets, security groups) the contract does not carry. A planned `mwaa.ensure_environment` is refused (`unsupported_binding`, kind - `mwaa-environment-not-emitted`) rather than skipped, as is any AWS op the - emitter does not know (`planned-op-not-emitted`). + `mwaa-environment-not-emitted`) rather than skipped, and so is a contract + that asks for one when the planner produced no plan; an op the emitter + does not translate is named in a warning. - **Non-SQL Snowflake procedures / UDFs** (Python / Java / Scala) — need runtime / handler / package config the contract does not carry. - **`composer.trigger_dag`**, **Athena query execution**, **Redshift table diff --git a/CHANGELOG.md b/CHANGELOG.md index 61647a9a..f9405ff5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -35,12 +35,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 VPC network the contract does not carry; apply reported success and no environment existed. To schedule without Airflow, set `orchestration.engine: none` and keep the schedule (an EventBridge Scheduler schedule is created). To keep Airflow, create the - MWAA environment yourself, push the DAG with `fluid generate schedule` and - `fluid schedule-sync --scheduler mwaa`, and apply a copy of the contract without - `orchestration.schedule`. Any other planned AWS op the engine does not know now fails - the same way (kind `planned-op-not-emitted`), and the known ops it does not provision - yet (IAM bindings from `metadata.policies`, Athena views, Redshift schemas and tables, - `execution.trigger` schedules, `orchestration.tasks`) are named in a warning. + MWAA environment yourself (or name your existing one in + `orchestration.mwaa.environmentName`), push the DAG with `fluid generate schedule` and + `fluid schedule-sync --scheduler mwaa --destination s3:///dags/`, and apply a + copy of the contract without `orchestration.schedule`. The refusal holds when the AWS + planner fails on another field and apply falls back to the data plane. A planned op the + engine does not provision (IAM bindings from `metadata.policies`, Athena views, Redshift + schemas and tables, `execution.trigger` schedules, `orchestration.tasks`, another + cloud's op under `--provider aws`) is named in a warning. - A GCP Iceberg expose that a streaming sink writes to must now name its catalog in `binding.location.catalog`. Without one, the sink wrote through whatever catalog reached the worker (REST by default, Glue for `sink.catalog: glue` or a hand-written diff --git a/fluid_build/iac/providers/aws.py b/fluid_build/iac/providers/aws.py index 9a276352..72a24266 100644 --- a/fluid_build/iac/providers/aws.py +++ b/fluid_build/iac/providers/aws.py @@ -54,7 +54,18 @@ import os import re from dataclasses import dataclass, field -from typing import Any, Dict, Iterable, List, Mapping, Optional, Sequence, Tuple +from typing import ( + Any, + Callable, + Dict, + Iterable, + List, + Mapping, + NamedTuple, + Optional, + Sequence, + Tuple, +) import yaml @@ -480,7 +491,7 @@ def emit( _emit_lf_bucket_policy(resources, bucket_policy) # Glue ETL jobs / Step Functions / the Lambda schedule path — # the planner's build & orchestration ops. - _emit_from_actions(resources, actions, cid) + _emit_from_actions(resources, actions, cid, contract) # Second pass — wire ordering edges that the literal-string fields # on Redshift external schemas / planned-action resources don't # carry by value. See :func:`_wire_aws_deps`. @@ -1102,8 +1113,10 @@ def _emit_kinesis( #: Planned ops ``_emit_from_actions`` leaves alone because something else owns #: them: the data plane :meth:`AwsIacPlugin.emit` derives from ``exposes[]`` -#: (see :class:`~fluid_build.iac.base.IacProviderPlugin`), and the build steps -#: the build runners execute rather than provision. +#: (see :class:`~fluid_build.iac.base.IacProviderPlugin`), the build steps the +#: build runners execute rather than provision, and the pre-replace backup copy +#: (``aws.s3.copy_prefix``, planned only for a destructive apply mode) whose +#: ``rollback_snapshot`` marker the apply's rollback writer records. _OPS_OWNED_ELSEWHERE = frozenset( { "s3.ensure_bucket", @@ -1112,6 +1125,7 @@ def _emit_kinesis( "glue.ensure_iceberg_table", "dbt.prepare_profile", "athena.execute_query", + "aws.s3.copy_prefix", } ) @@ -1119,6 +1133,10 @@ def _emit_kinesis( #: yet. Each is named in a warning on every emit rather than refused: refusing #: them would fail contracts the MWAA fix is not about, and dropping them #: without a word is the defect this list exists to end. +#: +#: Every op the AWS planner can produce is in :data:`_ACTION_EMITTERS`, in +#: :data:`_OPS_OWNED_ELSEWHERE`, here, or is ``mwaa.ensure_environment``; +#: ``tests/iac/test_iac_aws_unemitted_ops.py`` fails when one is not. _OPS_NOT_YET_EMITTED = frozenset( { "iam.bind_glue_database", @@ -1130,20 +1148,88 @@ def _emit_kinesis( } ) +#: The op the AWS schedule planner (``providers/aws/plan/schedule.py``) plans +#: for ``orchestration.engine`` airflow or mwaa with an ``orchestration.schedule``. +_MWAA_OP = "mwaa.ensure_environment" + + +class _MwaaRequest(NamedTuple): + """What a contract that asks for MWAA says about the environment.""" + + #: ``orchestration.mwaa.environmentName``, else ``mwaa-``. + name: str + #: True when the contract names it: the user most likely runs it already. + named: bool + #: ``orchestration.mwaa.dagS3Path``: where the environment reads DAGs. + dags: Optional[str] + -def _refuse_mwaa_environment(action: Mapping[str, Any]) -> UnsupportedBindingError: - """``mwaa.ensure_environment`` — planned, but OpenTofu cannot create it. +def _mwaa_requested(contract: Mapping[str, Any]) -> Optional[_MwaaRequest]: + """The MWAA environment the contract asks for, or ``None``. + + The same test the AWS schedule planner applies before it plans + :data:`_MWAA_OP` (``_plan_scheduled_execution``: a schedule, and the + engine lower-cased, unstripped, is ``airflow`` or ``mwaa``), and the same + name it gives the environment. + """ + orchestration = contract.get("orchestration") + if not isinstance(orchestration, Mapping) or not orchestration.get("schedule"): + return None + engine = orchestration.get("engine", "") + if not isinstance(engine, str) or engine.lower() not in ("airflow", "mwaa"): + return None + mwaa = orchestration.get("mwaa") + mwaa = mwaa if isinstance(mwaa, Mapping) else {} + named = mwaa.get("environmentName") + dags = mwaa.get("dagS3Path") + return _MwaaRequest( + name=str(named) if named else f"mwaa-{contract.get('id', 'unknown')}", + named=bool(named), + dags=str(dags) if dags else None, + ) + + +def _refuse_mwaa_environment(name: Any, request: Optional[_MwaaRequest]) -> UnsupportedBindingError: + """The contract asks for an MWAA environment, and OpenTofu cannot create it. ``aws_mwaa_environment`` requires ``network_configuration`` (two private ``subnet_ids`` and ``security_group_ids``), and the contract carries no network. Skipping the op let ``fluid apply`` report success with no Airflow environment, so it is refused instead. + + The way to keep Airflow is the Path-A push: ``fluid schedule-sync + --scheduler mwaa`` copies the DAGs to the S3 folder the environment reads + (``--destination``; its MWAA dispatch reads no ``--environment-name``), + filled in from ``orchestration.mwaa.dagS3Path`` when the contract gives it. """ - name = action.get("environment_name") or "?" + shown = name if name else "?" + named = bool(request and request.named and request.name == name) + dags = request.dags if request else None + # Printed as-is in a command line, so a control character in the + # contract's path falls back to the placeholder rather than reach the terminal. + dest = dags if dags and dags.isprintable() else "s3:///dags/" + push = ( + "`fluid generate schedule --scheduler airflow --output dags` and " + f"`fluid schedule-sync --scheduler mwaa --dags-dir dags --destination {dest}`" + ) + without = ( + "fluid apply still plans the environment while the contract has " + "orchestration.schedule, so apply a copy of the contract without it." + ) + if named: + keep_airflow = ( + f"Your contract names the existing MWAA environment {shown!r}: push this " + f"contract's DAG to it with {push}. {without}" + ) + else: + keep_airflow = ( + f"To keep Airflow, create the MWAA environment {shown!r} yourself (it needs your " + f"VPC), then push this contract's DAG to it with {push}. {without}" + ) return UnsupportedBindingError( "mwaa-environment-not-emitted", - f"The plan asks for the MWAA environment {name!r} (orchestration.engine airflow or " - "mwaa with orchestration.schedule), but the OpenTofu engine cannot create an " + f"The contract asks for the MWAA environment {shown!r} (orchestration.engine airflow " + "or mwaa with orchestration.schedule), but the OpenTofu engine cannot create an " "aws_mwaa_environment: it needs a VPC network_configuration (subnets and security " "groups) that the contract does not carry. Applying would report success with no " "Airflow environment, so nothing was applied.", @@ -1151,58 +1237,58 @@ def _refuse_mwaa_environment(action: Mapping[str, Any]) -> UnsupportedBindingErr "To schedule without Airflow, set orchestration.engine to 'none' and keep " "orchestration.schedule: fluid apply then creates an EventBridge Scheduler " "schedule.", - "To keep Airflow, create the MWAA environment yourself (it needs your VPC), push " - "this contract's DAG to it with `fluid generate schedule` and `fluid schedule-sync " - "--scheduler mwaa --environment-name `, and run fluid apply on a copy of the " - "contract without orchestration.schedule.", + keep_airflow, ), ) def _emit_from_actions( - resources: Dict[str, Any], actions: Iterable[Mapping[str, Any]], cid: str + resources: Dict[str, Any], + actions: Iterable[Mapping[str, Any]], + cid: str, + contract: Mapping[str, Any], ) -> None: """Translate the planner's build / orchestration ops into ``hashicorp/aws`` resources. Covers Glue ETL jobs, Step Functions, and the Lambda schedule / event - path — inline Lambda source is zipped by ``data.archive_file`` (see - :meth:`AwsIacPlugin.emit_data`). An op with no emitter goes to - :func:`_account_for_unemitted_op`, so no planned op is dropped without a word. + path (:data:`_ACTION_EMITTERS`) — inline Lambda source is zipped by + ``data.archive_file`` (see :meth:`AwsIacPlugin.emit_data`). An op with no + emitter goes to :func:`_account_for_unemitted_op`, so no planned op is + dropped without a word. + + A contract that asks for MWAA is refused even when ``actions`` is empty: + ``fluid apply`` and ``fluid generate iac`` pass ``[]`` when the native + planner could not run or failed on an unrelated field (``native_actions`` + in ``cli/generate_iac.py`` logs that at DEBUG), and the module would then + lack the environment as silently as before. A non-empty ``actions`` came + from a planner that ran; that the AWS one plans the environment it is + asked for is :func:`_account_for_unemitted_op`'s to refuse. """ - for action in actions or []: - if not isinstance(action, Mapping): - continue + planned = [action for action in actions or [] if isinstance(action, Mapping)] + requested = _mwaa_requested(contract) + if not planned and requested is not None: + raise _refuse_mwaa_environment(requested.name, requested) + for action in planned: op = action.get("op") - if op == "glue.ensure_job": - _emit_glue_job(resources, action, cid) - elif op == "stepfunctions.ensure_state_machine": - _emit_state_machine(resources, action, cid) - elif op == "lambda.ensure_function": - _emit_lambda_function(resources, action, cid) - elif op == "lambda.add_permission": - _emit_lambda_permission(resources, action, cid) - elif op == "lambda.create_event_source_mapping": - _emit_event_source_mapping(resources, action, cid) - elif op == "eventbridge.ensure_schedule": - _emit_scheduler_schedule(resources, action, cid) - elif op == "eventbridge.ensure_rule": - _emit_event_rule(resources, action, cid) - elif op == "s3.ensure_notification": - _emit_s3_notification(resources, action, cid) + emitter = _ACTION_EMITTERS.get(op) if isinstance(op, str) else None + if emitter is not None: + emitter(resources, action, cid) else: - _account_for_unemitted_op(action) + _account_for_unemitted_op(action, requested) -def _account_for_unemitted_op(action: Mapping[str, Any]) -> None: +def _account_for_unemitted_op(action: Mapping[str, Any], requested: Optional[_MwaaRequest]) -> None: """Pass, refuse, or name in a warning a planned op the emitter has no translation for. A data-plane or build op owned elsewhere (``_OPS_OWNED_ELSEWHERE``) passes. - ``mwaa.ensure_environment`` is refused (:func:`_refuse_mwaa_environment`); - an op no list here knows is refused by name, the way dbt fails with - ``MissingMaterializationError`` ("No materialization ... was found for - adapter", dbt-adapters ``exceptions/compilation.py``) rather than skipping - the model; a known op nothing provisions yet, and an orchestration task, - is named in a warning. + :data:`_MWAA_OP` is refused (:func:`_refuse_mwaa_environment`). Anything + else is named in a warning: a known op nothing provisions yet, an + orchestration task, and an op this module does not know. The last is not + refused: an AWS planner op cannot reach it unclassified (the planner- + vocabulary test sees to that), so what does is another cloud's op, which + ``native_actions`` passes when a multi-cloud contract's first expose is not + on AWS and the user chose ``--provider aws``, as the ambiguous-provider + error tells them to. Refusing it failed the AWS half of every such contract. """ op = action.get("op") known = op if isinstance(op, str) else "" @@ -1212,11 +1298,13 @@ def _account_for_unemitted_op(action: Mapping[str, Any]) -> None: task = isinstance(action.get("orchestration"), Mapping) if known in _OPS_OWNED_ELSEWHERE and not task: return - if known == "mwaa.ensure_environment": - raise _refuse_mwaa_environment(action) + if known == _MWAA_OP: + raise _refuse_mwaa_environment(action.get("environment_name"), requested) + # Op and id are %r: a task's op and every id come from the contract, and + # repr keeps a control character in either out of the terminal. if known in _OPS_NOT_YET_EMITTED: LOG.warning( - "aws_planned_op_not_emitted op=%s id=%s: the plan lists it, but the OpenTofu " + "aws_planned_op_not_emitted op=%r id=%r: the plan lists it, but the OpenTofu " "engine provisions nothing for it, so fluid apply will not create it", known, action.get("id"), @@ -1224,22 +1312,17 @@ def _account_for_unemitted_op(action: Mapping[str, Any]) -> None: return if task: LOG.warning( - "aws_orchestration_task_not_emitted op=%s id=%s: orchestration tasks run in " + "aws_orchestration_task_not_emitted op=%r id=%r: orchestration tasks run in " "the scheduler, and fluid apply provisions nothing for them", op, action.get("id"), ) return - where = f" (action {action['id']!r})" if action.get("id") else "" - raise UnsupportedBindingError( - "planned-op-not-emitted", - f"The plan contains the AWS op {op!r}{where}, and the OpenTofu engine " - "has no translation for it. Applying would skip it and report success, so nothing " - "was applied.", - ( - "Remove the contract field that plans this op, or report it as a fluid " - "bug if the planner produced it from a valid contract.", - ), + LOG.warning( + "planned_op_not_emitted op=%r id=%r: the plan lists it, and the AWS OpenTofu " + "module has no translation for it, so fluid apply on AWS will not create it", + op, + action.get("id"), ) @@ -1505,6 +1588,19 @@ def _emit_s3_notification(resources: Dict[str, Any], action: Mapping[str, Any], ) +#: The planned ops this module translates, op -> emitter. +_ACTION_EMITTERS: Mapping[str, Callable[[Dict[str, Any], Mapping[str, Any], str], None]] = { + "glue.ensure_job": _emit_glue_job, + "stepfunctions.ensure_state_machine": _emit_state_machine, + "lambda.ensure_function": _emit_lambda_function, + "lambda.add_permission": _emit_lambda_permission, + "lambda.create_event_source_mapping": _emit_event_source_mapping, + "eventbridge.ensure_schedule": _emit_scheduler_schedule, + "eventbridge.ensure_rule": _emit_event_rule, + "s3.ensure_notification": _emit_s3_notification, +} + + # --------------------------------------------------------------------------- # Redshift Serverless — namespace + workgroup + external schema bridge # diff --git a/tests/iac/test_iac_aws_unemitted_ops.py b/tests/iac/test_iac_aws_unemitted_ops.py index 8b0ddb55..3f8a3092 100644 --- a/tests/iac/test_iac_aws_unemitted_ops.py +++ b/tests/iac/test_iac_aws_unemitted_ops.py @@ -26,6 +26,7 @@ import json import logging +import re from pathlib import Path from typing import Any, Dict @@ -35,8 +36,15 @@ from fluid_build.cli._common import CLIError from fluid_build.iac import build_module, get_iac_plugin, shadow_compare from fluid_build.iac.base import UnsupportedBindingError +from fluid_build.iac.providers.aws import ( + _ACTION_EMITTERS, + _MWAA_OP, + _OPS_NOT_YET_EMITTED, + _OPS_OWNED_ELSEWHERE, +) from fluid_build.iac.shadow import native_logical_resources from fluid_build.providers.aws.provider import AwsProvider +from fluid_build.providers.base import ProviderError from tests.iac.test_iac_generate_offline import _args, _write, no_aws # noqa: F401 (fixture) from tests.iac.test_iac_generate_offline import _orders as _contract @@ -68,7 +76,11 @@ def test_the_op_is_refused_with_its_environment_and_a_way_out(self): assert "mwaa-orders" in str(err) assert "network_configuration" in str(err) assert err.remediation, "the CLI renders these as next steps" - assert any("schedule-sync" in step for step in err.remediation) + (keep_airflow,) = [step for step in err.remediation if "schedule-sync" in step] + assert "create the MWAA environment 'mwaa-orders' yourself" in keep_airflow + # MWAA's schedule-sync dispatch reads --destination, never --environment-name. + assert "--scheduler mwaa --dags-dir dags --destination s3://" in keep_airflow + assert "--environment-name" not in keep_airflow @pytest.mark.parametrize("engine", ["airflow", "mwaa"]) def test_the_planned_path_fails_before_a_module_is_built(self, engine): @@ -89,12 +101,32 @@ def test_a_named_existing_environment_is_still_refused(self): orchestration={ "engine": "airflow", "schedule": "0 * * * *", - "mwaa": {"environmentName": "shared-airflow"}, + "mwaa": { + "environmentName": "shared-airflow", + "dagS3Path": "s3://shared-airflow-dags/dags/", + }, } ) + # Both paths: the plan's op, and no plan at all (the planner failed). + for actions in (_planned(contract), []): + with pytest.raises(UnsupportedBindingError) as excinfo: + build_module(_aws(), contract, actions=actions) + err = excinfo.value + assert "shared-airflow" in str(err) + # The user already runs it: the way out pushes the DAG to the + # folder the contract says it reads, and does not say to create one. + (keep_airflow,) = [step for step in err.remediation if "schedule-sync" in step] + assert "names the existing MWAA environment 'shared-airflow'" in keep_airflow + assert "--destination s3://shared-airflow-dags/dags/" in keep_airflow + assert "yourself" not in keep_airflow + + def test_a_control_character_in_the_dag_path_is_not_printed(self): + contract = _contract( + orchestration={**_AIRFLOW, "mwaa": {"dagS3Path": "s3://b/dags/\x1b[2J"}} + ) with pytest.raises(UnsupportedBindingError) as excinfo: - build_module(_aws(), contract, actions=_planned(contract)) - assert "shared-airflow" in str(excinfo.value) + _aws().emit(contract, []) + assert not any("\x1b" in step for step in excinfo.value.remediation) def test_a_schedule_without_airflow_still_emits(self): contract = _contract(orchestration={"schedule": "rate(1 hour)"}) @@ -144,21 +176,29 @@ def test_a_schedule_without_airflow_writes_the_module(self, tmp_path): class TestEveryOtherOp: - # A non-string op must be the typed refusal too, not a TypeError from hashing it. + # A non-string op is named too, not a TypeError from hashing it. @pytest.mark.parametrize( ("action", "named"), [ - ({"op": "glue.ensure_crawler", "name": "c"}, "'glue.ensure_crawler'"), - ({"id": "x"}, "None (action 'x')"), - ({"id": "x", "op": ["glue", "ensure_job"]}, "['glue', 'ensure_job']"), + ({"op": "glue.ensure_crawler", "id": "c"}, "op='glue.ensure_crawler' id='c'"), + ({"id": "x"}, "op=None id='x'"), + ({"id": "x", "op": ["glue", "ensure_job"]}, "op=['glue', 'ensure_job'] id='x'"), ], ) - def test_an_op_with_no_translation_is_refused_by_name(self, action, named): - with pytest.raises(UnsupportedBindingError) as excinfo: - _aws().emit(_contract(), [action]) - err = excinfo.value - assert (err.kind, named in str(err)) == ("planned-op-not-emitted", True) - assert err.remediation + def test_an_op_with_no_translation_is_named_in_a_warning(self, action, named, caplog): + caplog.set_level(logging.WARNING, logger=_AWS_LOGGER) + resources = _aws().emit(_contract(), [action]) + # The data plane is still emitted: the op is named, not fatal. + assert "aws_glue_catalog_table" in resources + (record,) = [r for r in caplog.records if "planned_op_not_emitted" in r.getMessage()] + assert named in record.getMessage() + + def test_a_control_character_in_an_op_does_not_reach_the_terminal(self, caplog): + caplog.set_level(logging.WARNING, logger=_AWS_LOGGER) + _aws().emit(_contract(), [{"op": "x\x1b[31m", "id": "i\x1b[0m"}]) + (record,) = [r for r in caplog.records if "planned_op_not_emitted" in r.getMessage()] + assert "\x1b" not in record.getMessage() + assert "x\\x1b[31m" in record.getMessage() @pytest.mark.parametrize( "op", @@ -171,6 +211,8 @@ def test_an_op_with_no_translation_is_refused_by_name(self, action, named): # Build steps the build runners execute, never provisioned. "dbt.prepare_profile", "athena.execute_query", + # The pre-replace backup the apply's rollback writer records. + "aws.s3.copy_prefix", ], ) def test_an_op_provisioned_or_run_elsewhere_passes_quietly(self, op, caplog): @@ -210,6 +252,188 @@ def test_an_orchestration_task_is_named_in_a_warning(self, op, caplog): assert "aws_orchestration_task_not_emitted" in record.getMessage() +_AIRFLOW = {"engine": "airflow", "schedule": "0 * * * *"} + + +def _upper_case_table() -> Dict[str, Any]: + """Valid to ``fluid validate``; the AWS planner's Glue name check refuses it.""" + contract = _contract(orchestration=dict(_AIRFLOW)) + contract["exposes"][0]["binding"]["location"]["table"] = "Orders" + return contract + + +def _cyclic_tasks() -> Dict[str, Any]: + """``orchestration.tasks`` that depend on each other: the planner raises.""" + task = {"type": "provider_action", "action": "aws.glue.run_job", "params": {}} + return _contract( + orchestration={ + **_AIRFLOW, + "tasks": [ + {**task, "taskId": "a", "dependsOn": ["b"]}, + {**task, "taskId": "b", "dependsOn": ["a"]}, + ], + } + ) + + +class TestThePlannerFailing: + """The planner failing on another field does not let the environment drop. + + ``native_actions`` returns ``[]`` when the native planner raises (it logs + that at DEBUG), so the emitter saw no ``mwaa.ensure_environment`` and wrote + the data plane alone, exit 0. + """ + + @pytest.mark.parametrize("make", [_upper_case_table, _cyclic_tasks], ids=["glue-name", "cycle"]) + def test_the_contract_is_refused_without_a_plan(self, make): + contract = make() + with pytest.raises(ProviderError): + _planned(contract) + with pytest.raises(UnsupportedBindingError) as excinfo: + _aws().emit(contract, []) + assert excinfo.value.kind == "mwaa-environment-not-emitted" + assert "mwaa-offline_orders" in str(excinfo.value) + + def test_generate_iac_is_refused(self, no_aws, monkeypatch, tmp_path): + monkeypatch.setenv("AWS_ACCOUNT_ID", ACCOUNT) + out = tmp_path / "out" + with pytest.raises(CLIError) as excinfo: + generate_iac.run( + _args(_write(tmp_path, _upper_case_table()), out), logging.getLogger("t") + ) + assert excinfo.value.context["kind"] == "mwaa-environment-not-emitted" + assert not out.exists() + + @pytest.mark.parametrize( + "orchestration", + [ + None, + {"schedule": "rate(1 hour)"}, + {"engine": "airflow"}, + # The planner does not strip the engine: this one is EventBridge. + {"engine": "airflow ", "schedule": "0 * * * *"}, + {"engine": None, "schedule": "0 * * * *"}, + ], + ) + def test_a_contract_that_asks_for_no_mwaa_still_emits_without_a_plan(self, orchestration): + contract = _contract() if orchestration is None else _contract(orchestration=orchestration) + assert "aws_glue_catalog_table" in _aws().emit(contract, []) + + @pytest.mark.parametrize("engine", ["AIRFLOW", "mwaa"]) + def test_the_engine_is_matched_as_the_planner_matches_it(self, engine): + contract = _contract(orchestration={"engine": engine, "schedule": "0 * * * *"}) + assert any(a.get("op") == _MWAA_OP for a in _planned(contract)) + with pytest.raises(UnsupportedBindingError): + _aws().emit(contract, []) + + +def _two_clouds(other: str) -> Dict[str, Any]: + """A contract whose first expose is on ``other`` and whose second is on AWS.""" + contract = _contract() + first = { + "gcp": { + "platform": "gcp", + "format": "bigquery_table", + "location": {"project": "p-1", "dataset": "ds", "table": "t", "region": "US"}, + }, + "snowflake": { + "platform": "snowflake", + "format": "snowflake_table", + "location": {"database": "DB", "schema": "S", "table": "T", "account": "acct"}, + }, + }[other] + contract["exposes"].insert( + 0, + { + "exposeId": f"{other}_orders", + "kind": "table", + "binding": first, + "contract": {"schema": [{"name": "id", "type": "string"}]}, + }, + ) + return contract + + +class TestAMultiCloudContract: + """``--provider aws`` on a contract whose first expose is on another cloud. + + ``native_actions`` plans with the first expose's provider, so that cloud's + ops reach the AWS emitter. Refusing them as unknown AWS ops failed the AWS + half of the contract, which the ambiguous-provider error tells the user to + apply with ``--provider aws``. + """ + + @pytest.mark.parametrize("other", ["gcp", "snowflake"]) + def test_generate_iac_writes_the_aws_module(self, other, no_aws, monkeypatch, tmp_path): + monkeypatch.setenv("AWS_ACCOUNT_ID", ACCOUNT) + out = tmp_path / "out" + args = _args(_write(tmp_path, _two_clouds(other)), out) + args.provider = "aws" + generate_iac.run(args, logging.getLogger("t")) + module = json.loads((out / "main.tf.json").read_text(encoding="utf-8")) + assert {"aws_glue_catalog_table", "aws_s3_bucket"} <= set(module["resource"]) + + def test_another_clouds_op_is_named_not_refused(self, caplog): + caplog.set_level(logging.WARNING, logger=_AWS_LOGGER) + resources = _aws().emit( + _two_clouds("gcp"), [{"op": "bq.ensure_dataset", "id": "dataset_ds"}] + ) + assert "aws_glue_catalog_table" in resources + (record,) = [r for r in caplog.records if "bq.ensure_dataset" in r.getMessage()] + assert record.levelno == logging.WARNING + + +_AWS_PLANNER = Path(generate_iac.__file__).resolve().parents[1] / "providers" / "aws" +_OP_LITERAL = re.compile(r"""["']op["']\s*:\s*["']([A-Za-z0-9_.]+)["']""") + + +def _classified() -> set: + return set(_ACTION_EMITTERS) | _OPS_OWNED_ELSEWHERE | _OPS_NOT_YET_EMITTED | {_MWAA_OP} + + +class TestThePlannerVocabulary: + """Every op the AWS planner can produce is emitted or classified. + + The two allowlists in ``iac/providers/aws.py`` are a second copy of the + planner's vocabulary; this is what ties them together, the way dbt-core's + adapter tests assert every materialization an adapter ships is handled. + An op this fails on is warned about at apply, never provisioned. + """ + + def test_every_op_literal_in_the_planner_is_classified(self): + found = {} + for path in sorted(_AWS_PLANNER.rglob("*.py")): + if "tests" in path.relative_to(_AWS_PLANNER).parts: + continue + for op in _OP_LITERAL.findall(path.read_text(encoding="utf-8")): + found.setdefault(op, path.relative_to(_AWS_PLANNER).as_posix()) + # The scan must see the planner, or it proves nothing. + assert {"glue.ensure_job", _MWAA_OP, "aws.s3.copy_prefix"} <= set(found) + unclassified = {op: where for op, where in found.items() if op not in _classified()} + assert unclassified == {}, ( + "planner ops the AWS OpenTofu emitter neither emits nor classifies: add each to " + "_ACTION_EMITTERS, _OPS_OWNED_ELSEWHERE or _OPS_NOT_YET_EMITTED" + ) + + def test_a_destructive_plan_is_classified(self): + # mode="replace" adds the phase-3 snapshot ops the default plan skips. + contract = _contract(orchestration={"schedule": "rate(1 hour)"}) + contract["exposes"][0]["binding"]["format"] = "glue_table" + actions = AwsProvider( + account_id=ACCOUNT, region="us-east-1", logger=logging.getLogger("t") + ).plan(contract, mode="replace") + ops = {a.get("op") for a in actions} + assert "aws.s3.copy_prefix" in ops + assert ops - _classified() == set() + # And the emitter takes the whole plan: nothing refused, nothing unknown. + json.loads(build_module(_aws(), contract, actions=actions)) + + def test_the_lists_do_not_overlap(self): + lists = [set(_ACTION_EMITTERS), set(_OPS_OWNED_ELSEWHERE), set(_OPS_NOT_YET_EMITTED)] + assert sum(len(x) for x in lists) == len(set().union(*lists)) + assert not any(_MWAA_OP in x for x in lists) + + class TestShadow: def test_an_environment_op_is_a_logical_resource(self): # Were the refusal ever removed, --shadow would report the dropped From 561f37757c72698983732b4a4a1d5d69b15fa24a Mon Sep 17 00:00:00 2001 From: fas89 <50082482+fas89@users.noreply.github.com> Date: Mon, 12 Oct 2026 18:03:52 +0200 Subject: [PATCH 3/4] fix(iac): --provider aws refuses an MWAA contract even when another cloud's planner ran A multi-cloud contract whose first expose is on GCP or Snowflake, with orchestration.engine airflow and a schedule, still wrote an AWS module with no MWAA environment and exited 0 under `fluid generate iac --provider aws` (and `fluid apply --provider aws`). native_actions plans with the first expose's provider, so the AWS planner that plans mwaa.ensure_environment never ran; the emitter only refused when the action list was empty or held that op, and the foreign ops made it non-empty. FLUID_PROVIDER=local hit the same gap with LocalProvider's ops. The refusal now follows the contract: when it asks for MWAA (the same predicate the AWS schedule planner applies) and no planned action is the environment op, the emitter refuses with unsupported_binding, kind mwaa-environment-not-emitted, whichever planner produced the actions. Regression tests drive generate_iac with --provider aws on both multi-cloud shapes and the emitter with GCP, Snowflake and local ops. --- AUTOGEN_SPIKE.md | 3 +- CHANGELOG.md | 5 ++-- fluid_build/iac/providers/aws.py | 23 +++++++++------ tests/iac/test_iac_aws_unemitted_ops.py | 39 +++++++++++++++++++++++++ 4 files changed, 58 insertions(+), 12 deletions(-) diff --git a/AUTOGEN_SPIKE.md b/AUTOGEN_SPIKE.md index 64824406..7b4648fa 100644 --- a/AUTOGEN_SPIKE.md +++ b/AUTOGEN_SPIKE.md @@ -249,7 +249,8 @@ self-skipping when `tofu` or the gcloud emulator is absent. (subnets, security groups) the contract does not carry. A planned `mwaa.ensure_environment` is refused (`unsupported_binding`, kind `mwaa-environment-not-emitted`) rather than skipped, and so is a contract - that asks for one when the planner produced no plan; an op the emitter + that asks for one when no AWS planner produced the op (the planner failed, + or another cloud's planner ran under `--provider aws`); an op the emitter does not translate is named in a warning. - **Non-SQL Snowflake procedures / UDFs** (Python / Java / Scala) — need runtime / handler / package config the contract does not carry. diff --git a/CHANGELOG.md b/CHANGELOG.md index f9405ff5..df8524f2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -38,8 +38,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 MWAA environment yourself (or name your existing one in `orchestration.mwaa.environmentName`), push the DAG with `fluid generate schedule` and `fluid schedule-sync --scheduler mwaa --destination s3:///dags/`, and apply a - copy of the contract without `orchestration.schedule`. The refusal holds when the AWS - planner fails on another field and apply falls back to the data plane. A planned op the + copy of the contract without `orchestration.schedule`. The refusal follows the contract, + so it holds when the AWS planner fails on another field and when a multi-cloud contract + whose first expose is on another cloud is applied with `--provider aws`. A planned op the engine does not provision (IAM bindings from `metadata.policies`, Athena views, Redshift schemas and tables, `execution.trigger` schedules, `orchestration.tasks`, another cloud's op under `--provider aws`) is named in a warning. diff --git a/fluid_build/iac/providers/aws.py b/fluid_build/iac/providers/aws.py index 72a24266..0bd4f3c1 100644 --- a/fluid_build/iac/providers/aws.py +++ b/fluid_build/iac/providers/aws.py @@ -1256,17 +1256,20 @@ def _emit_from_actions( emitter goes to :func:`_account_for_unemitted_op`, so no planned op is dropped without a word. - A contract that asks for MWAA is refused even when ``actions`` is empty: - ``fluid apply`` and ``fluid generate iac`` pass ``[]`` when the native - planner could not run or failed on an unrelated field (``native_actions`` - in ``cli/generate_iac.py`` logs that at DEBUG), and the module would then - lack the environment as silently as before. A non-empty ``actions`` came - from a planner that ran; that the AWS one plans the environment it is - asked for is :func:`_account_for_unemitted_op`'s to refuse. + A contract that asks for MWAA is refused whatever ``actions`` holds, so + the refusal follows the contract, not the planner that happened to run. + ``actions`` may lack the environment op because no AWS planner produced + it: ``native_actions`` in ``cli/generate_iac.py`` passes ``[]`` when the + planner could not run or failed on an unrelated field (it logs that at + DEBUG), and passes another cloud's ops when a multi-cloud contract's first + expose is not on AWS and the user chose ``--provider aws`` (or + ``FLUID_PROVIDER`` names another planner). Either way the module would + lack the environment as silently as before. When the op is present, its + ``environment_name`` is the one the refusal names. """ planned = [action for action in actions or [] if isinstance(action, Mapping)] requested = _mwaa_requested(contract) - if not planned and requested is not None: + if requested is not None and not any(action.get("op") == _MWAA_OP for action in planned): raise _refuse_mwaa_environment(requested.name, requested) for action in planned: op = action.get("op") @@ -1281,7 +1284,9 @@ def _account_for_unemitted_op(action: Mapping[str, Any], requested: Optional[_Mw """Pass, refuse, or name in a warning a planned op the emitter has no translation for. A data-plane or build op owned elsewhere (``_OPS_OWNED_ELSEWHERE``) passes. - :data:`_MWAA_OP` is refused (:func:`_refuse_mwaa_environment`). Anything + :data:`_MWAA_OP` is refused (:func:`_refuse_mwaa_environment`; a contract + that asks for it is refused before the loop, in :func:`_emit_from_actions`, + so this guards an op handed in without that request). Anything else is named in a warning: a known op nothing provisions yet, an orchestration task, and an op this module does not know. The last is not refused: an AWS planner op cannot reach it unclassified (the planner- diff --git a/tests/iac/test_iac_aws_unemitted_ops.py b/tests/iac/test_iac_aws_unemitted_ops.py index 3f8a3092..9e2969c8 100644 --- a/tests/iac/test_iac_aws_unemitted_ops.py +++ b/tests/iac/test_iac_aws_unemitted_ops.py @@ -382,6 +382,45 @@ def test_another_clouds_op_is_named_not_refused(self, caplog): (record,) = [r for r in caplog.records if "bq.ensure_dataset" in r.getMessage()] assert record.levelno == logging.WARNING + @pytest.mark.parametrize("other", ["gcp", "snowflake"]) + def test_airflow_is_refused_though_another_cloud_planned( + self, other, no_aws, monkeypatch, tmp_path + ): + # The first expose's planner runs and returns its own ops, so the AWS + # planner that plans the environment never does: the refusal must + # follow the contract's request, not the ops that came back. + monkeypatch.setenv("AWS_ACCOUNT_ID", ACCOUNT) + contract = _two_clouds(other) + contract["orchestration"] = dict(_AIRFLOW) + out = tmp_path / "out" + args = _args(_write(tmp_path, contract), out) + args.provider = "aws" + with pytest.raises(CLIError) as excinfo: + generate_iac.run(args, logging.getLogger("t")) + assert (excinfo.value.event, excinfo.value.context["kind"]) == ( + "unsupported_binding", + "mwaa-environment-not-emitted", + ) + assert not out.exists() + + @pytest.mark.parametrize( + "foreign", + [ + [{"op": "bq.ensure_dataset", "id": "dataset_ds"}], + [{"op": "sf.table.ensure", "id": "t"}], + # FLUID_PROVIDER=local plans with LocalProvider. + [{"op": "materialize", "id": "m"}], + ], + ids=["gcp", "snowflake", "local"], + ) + def test_another_clouds_ops_do_not_stand_in_for_the_environment(self, foreign): + contract = _two_clouds("gcp") + contract["orchestration"] = dict(_AIRFLOW) + with pytest.raises(UnsupportedBindingError) as excinfo: + _aws().emit(contract, foreign) + assert excinfo.value.kind == "mwaa-environment-not-emitted" + assert "mwaa-offline_orders" in str(excinfo.value) + _AWS_PLANNER = Path(generate_iac.__file__).resolve().parents[1] / "providers" / "aws" _OP_LITERAL = re.compile(r"""["']op["']\s*:\s*["']([A-Za-z0-9_.]+)["']""") From e6260e2af917d90465c0ac7e33aae0c87b6c57fc Mon Sep 17 00:00:00 2001 From: fas89 <50082482+fas89@users.noreply.github.com> Date: Mon, 12 Oct 2026 22:33:53 +0200 Subject: [PATCH 4/4] fix(iac): the MWAA refusal's keep-Airflow commands run as printed The refusal printed `fluid generate schedule --output dags` and then `fluid schedule-sync --dags-dir dags`. schedule-sync's default (product delete scope) mirrors each top-level directory of --dags-dir and refuses loose files, so the printed pair failed with schedule_sync_dags_dir_not_product_scoped. The DAG is now generated into dags//, and a contract id that is not a valid product directory name prints a placeholder instead. A test runs the printed output path through schedule-sync's own scope check. --- fluid_build/iac/providers/aws.py | 17 ++++++++++++++++- tests/iac/test_iac_aws_unemitted_ops.py | 21 +++++++++++++++++++++ 2 files changed, 37 insertions(+), 1 deletion(-) diff --git a/fluid_build/iac/providers/aws.py b/fluid_build/iac/providers/aws.py index 0bd4f3c1..da0f890c 100644 --- a/fluid_build/iac/providers/aws.py +++ b/fluid_build/iac/providers/aws.py @@ -1153,6 +1153,11 @@ def _emit_kinesis( _MWAA_OP = "mwaa.ensure_environment" +# The directory-name rule ``fluid schedule-sync`` applies to product scopes +# (cli/schedule_sync.py ``_SCOPE_NAME_RE``). +_PRODUCT_DIR_RE = re.compile(r"[A-Za-z0-9_][A-Za-z0-9_.\-]{0,127}") + + class _MwaaRequest(NamedTuple): """What a contract that asks for MWAA says about the environment.""" @@ -1162,6 +1167,10 @@ class _MwaaRequest(NamedTuple): named: bool #: ``orchestration.mwaa.dagS3Path``: where the environment reads DAGs. dags: Optional[str] + #: The contract id when it is a valid ``schedule-sync`` product directory + #: name, else ``None``. ``schedule-sync`` (product delete scope, its + #: default) refuses DAGs that are not inside a ``/`` directory. + product: Optional[str] = None def _mwaa_requested(contract: Mapping[str, Any]) -> Optional[_MwaaRequest]: @@ -1182,10 +1191,12 @@ def _mwaa_requested(contract: Mapping[str, Any]) -> Optional[_MwaaRequest]: mwaa = mwaa if isinstance(mwaa, Mapping) else {} named = mwaa.get("environmentName") dags = mwaa.get("dagS3Path") + contract_id = str(contract.get("id") or "") return _MwaaRequest( name=str(named) if named else f"mwaa-{contract.get('id', 'unknown')}", named=bool(named), dags=str(dags) if dags else None, + product=contract_id if _PRODUCT_DIR_RE.fullmatch(contract_id) else None, ) @@ -1208,8 +1219,12 @@ def _refuse_mwaa_environment(name: Any, request: Optional[_MwaaRequest]) -> Unsu # Printed as-is in a command line, so a control character in the # contract's path falls back to the placeholder rather than reach the terminal. dest = dags if dags and dags.isprintable() else "s3:///dags/" + # schedule-sync mirrors each top-level directory of --dags-dir into the + # same-named folder of the destination, and refuses loose files, so the DAG + # is generated into dags// for the pair to run as printed. + product = request.product if request and request.product else "" push = ( - "`fluid generate schedule --scheduler airflow --output dags` and " + f"`fluid generate schedule --scheduler airflow --output dags/{product}` and " f"`fluid schedule-sync --scheduler mwaa --dags-dir dags --destination {dest}`" ) without = ( diff --git a/tests/iac/test_iac_aws_unemitted_ops.py b/tests/iac/test_iac_aws_unemitted_ops.py index 9e2969c8..89343330 100644 --- a/tests/iac/test_iac_aws_unemitted_ops.py +++ b/tests/iac/test_iac_aws_unemitted_ops.py @@ -82,6 +82,27 @@ def test_the_op_is_refused_with_its_environment_and_a_way_out(self): assert "--scheduler mwaa --dags-dir dags --destination s3://" in keep_airflow assert "--environment-name" not in keep_airflow + def test_the_printed_keep_airflow_commands_run_as_printed(self, tmp_path): + """The DAG goes into dags//: schedule-sync's default + (product delete scope) refuses a loose file at the top of --dags-dir.""" + from fluid_build.cli.schedule_sync import _product_scopes + + with pytest.raises(UnsupportedBindingError) as excinfo: + _aws().emit(_contract(orchestration=_AIRFLOW), []) + (keep_airflow,) = [step for step in excinfo.value.remediation if "schedule-sync" in step] + output = re.search(r"--output (\S+?)`", keep_airflow).group(1) + assert output == "dags/offline_orders" + dag = tmp_path / output / "offline_orders_dag.py" + dag.parent.mkdir(parents=True) + dag.write_text("# generated\n") + assert _product_scopes(tmp_path / "dags") == ["offline_orders"] + + def test_an_id_schedule_sync_cannot_use_as_a_directory_prints_a_placeholder(self): + with pytest.raises(UnsupportedBindingError) as excinfo: + _aws().emit(_contract(id="sales orders/v1", orchestration=_AIRFLOW), []) + (keep_airflow,) = [step for step in excinfo.value.remediation if "schedule-sync" in step] + assert "--output dags/`" in keep_airflow + @pytest.mark.parametrize("engine", ["airflow", "mwaa"]) def test_the_planned_path_fails_before_a_module_is_built(self, engine): contract = _contract(orchestration={"engine": engine, "schedule": "0 * * * *"})