diff --git a/AUTOGEN_SPIKE.md b/AUTOGEN_SPIKE.md index dd7687f7..7b4648fa 100644 --- a/AUTOGEN_SPIKE.md +++ b/AUTOGEN_SPIKE.md @@ -246,7 +246,12 @@ 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, and so is a contract + 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. - **`composer.trigger_dag`**, **Athena query execution**, **Redshift table diff --git a/CHANGELOG.md b/CHANGELOG.md index ab11a72c..df8524f2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -28,6 +28,22 @@ 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 (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 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. - 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..da0f890c 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`. @@ -1100,36 +1111,239 @@ 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`), 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", + "glue.ensure_database", + "glue.ensure_table", + "glue.ensure_iceberg_table", + "dbt.prepare_profile", + "athena.execute_query", + "aws.s3.copy_prefix", + } +) + +#: 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. +#: +#: 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", + "iam.bind_s3_bucket", + "athena.create_view", + "redshift.ensure_schema", + "redshift.ensure_table", + "events.ensure_schedule", + } +) + +#: 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" + + +# 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.""" + + #: ``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] + #: 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]: + """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") + 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, + ) + + +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. + """ + 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/" + # 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 = ( + f"`fluid generate schedule --scheduler airflow --output dags/{product}` 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 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.", + ( + "To schedule without Airflow, set orchestration.engine to 'none' and keep " + "orchestration.schedule: fluid apply then creates an EventBridge Scheduler " + "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`). MWAA is still skipped: ``aws_mwaa_environment`` - needs VPC ``network_configuration`` the contract does not carry. + 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 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. """ - 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 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") - 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, requested) + + +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. + :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- + 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 "" + # 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_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=%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"), + ) + return + if task: + LOG.warning( + "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 + 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"), + ) def _emit_glue_job(resources: Dict[str, Any], action: Mapping[str, Any], cid: str) -> None: @@ -1394,6 +1608,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/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..89343330 --- /dev/null +++ b/tests/iac/test_iac_aws_unemitted_ops.py @@ -0,0 +1,512 @@ +# 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 +import re +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.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 + +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" + (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 + + 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 * * * *"}) + 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", + "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: + _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)"}) + 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 is named too, not a TypeError from hashing it. + @pytest.mark.parametrize( + ("action", "named"), + [ + ({"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_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", + [ + # 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", + # 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): + 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() + + +_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 + + @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_.]+)["']""") + + +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 + # 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"}, ]