test(P3): env-transition tests — REQ-288,289

- tests/test_env_transition.py: detect_prior_env (5 tests) + record_applied_env (3 tests) + CLI (2 tests) via moto DynamoDB (REQ-288)
- tests/test_run_platform_env_transition.py: Step 0b block assertions (10 tests) + record-applied-env assertions (3 tests) + consumer-repo assertions (2 tests) (REQ-289)

25 new tests pass. 117 total tests pass (no regressions).

---ci---
project: acdl
phase: 3
milestone: v1.24
status: execute
requirements: [REQ-288,REQ-289]
---/ci---

tests/test_env_transition.py

@@ -0,0 +1,142 @@
+"""REQ-288: tests for core/env_transition.py — detect_prior_env + record_applied_env.
+
+Uses moto (already a test dependency) to mock DynamoDB, mirroring the
+pattern in tests/test_contract_ingestor.py. The nova-contracts table is
+created with PK consumerRepo + SK contractId#submittedAt.
+"""
+import sys
+from pathlib import Path
+from unittest import mock
+
+import pytest
+
+ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(ROOT))
+
+from core import env_transition
+
+
+@pytest.fixture
+def moto_contracts_table(monkeypatch):
+    """Spin up a moto-backed DynamoDB nova-contracts table."""
+    from moto import mock_aws
+    import boto3
+
+    monkeypatch.setenv("AWS_DEFAULT_REGION", "us-east-1")
+    monkeypatch.setenv("AWS_ACCESS_KEY_ID", "testing")
+    monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "testing")
+
+    with mock_aws():
+        dyn = boto3.client("dynamodb", region_name="us-east-1")
+        dyn.create_table(
+            TableName="nova-contracts",
+            KeySchema=[
+                {"AttributeName": "consumerRepo", "KeyType": "HASH"},
+                {"AttributeName": "contractId#submittedAt", "KeyType": "RANGE"},
+            ],
+            AttributeDefinitions=[
+                {"AttributeName": "consumerRepo", "AttributeType": "S"},
+                {"AttributeName": "contractId#submittedAt", "AttributeType": "S"},
+            ],
+            BillingMode="PAY_PER_REQUEST",
+        )
+        yield dyn
+
+
+class TestDetectPriorEnv:
+    def test_returns_none_when_no_record_exists(self, moto_contracts_table):
+        """First deploy: no prior record → None (no destroy needed)."""
+        result = env_transition.detect_prior_env("assets", "acdl/consumer-a", "dev")
+        assert result is None
+
+    def test_returns_prior_env_when_record_differs(self, moto_contracts_table):
+        """Env change detected: last-applied was dev, new is qa → return 'dev'."""
+        env_transition.record_applied_env("assets", "acdl/consumer-a", "dev")
+        result = env_transition.detect_prior_env("assets", "acdl/consumer-a", "qa")
+        assert result == "dev"
+
+    def test_returns_none_when_record_matches_new_env(self, moto_contracts_table):
+        """Re-apply same env: last-applied was dev, new is dev → None."""
+        env_transition.record_applied_env("assets", "acdl/consumer-a", "dev")
+        result = env_transition.detect_prior_env("assets", "acdl/consumer-a", "dev")
+        assert result is None
+
+    def test_returns_none_on_dynamodb_unreachable(self, monkeypatch):
+        """DynamoDB unreachable (local/CI) → log warning + return None (conservative)."""
+        def _raise(*args, **kwargs):
+            raise RuntimeError("simulated DynamoDB unreachable")
+        monkeypatch.setattr(env_transition, "_get_table", _raise)
+        result = env_transition.detect_prior_env("assets", "acdl/consumer-a", "qa")
+        assert result is None
+
+    def test_scoped_to_consumer_repo(self, moto_contracts_table):
+        """A different consumer's record does not affect this consumer's detect."""
+        env_transition.record_applied_env("assets", "acdl/consumer-a", "dev")
+        result = env_transition.detect_prior_env("assets", "acdl/consumer-b", "qa")
+        assert result is None
+
+
+class TestRecordAppliedEnv:
+    def test_writes_record_to_table(self, moto_contracts_table):
+        """record_applied_env writes an item with the right PK/SK + environment."""
+        ok = env_transition.record_applied_env("assets", "acdl/consumer-a", "dev")
+        assert ok is True
+        # Verify the item was written
+        import boto3
+        resp = boto3.client("dynamodb", region_name="us-east-1").query(
+            TableName="nova-contracts",
+            KeyConditionExpression="consumerRepo = :repo",
+            ExpressionAttributeValues={":repo": {"S": "acdl/consumer-a"}},
+        )
+        assert len(resp["Items"]) == 1
+        item = resp["Items"][0]
+        assert item["consumerRepo"]["S"] == "acdl/consumer-a"
+        assert item["environment"]["S"] == "dev"
+        assert item["status"]["S"] == "applied"
+        assert "#LAST_APPLIED#" in item["contractId#submittedAt"]["S"]
+
+    def test_returns_false_on_dynamodb_unreachable(self, monkeypatch):
+        """DynamoDB unreachable → return False (non-fatal, pipeline continues)."""
+        def _raise(*args, **kwargs):
+            raise RuntimeError("simulated DynamoDB unreachable")
+        monkeypatch.setattr(env_transition, "_get_table", _raise)
+        ok = env_transition.record_applied_env("assets", "acdl/consumer-a", "dev")
+        assert ok is False
+
+    def test_idempotent_multiple_writes(self, moto_contracts_table):
+        """Multiple record calls with different envs write separate items
+        (timestamped SKs). Same-second same-env writes collapse (put_item
+        overwrites same PK+SK — the latest record wins, which is correct)."""
+        env_transition.record_applied_env("assets", "acdl/consumer-a", "dev")
+        env_transition.record_applied_env("assets", "acdl/consumer-a", "qa")
+        import boto3
+        resp = boto3.client("dynamodb", region_name="us-east-1").query(
+            TableName="nova-contracts",
+            KeyConditionExpression="consumerRepo = :repo",
+            ExpressionAttributeValues={":repo": {"S": "acdl/consumer-a"}},
+        )
+        # At least 1 item (same-second writes may collapse to 1; the latest env wins)
+        assert len(resp["Items"]) >= 1
+        # The latest record should have the most recent env written
+        envs = [item["environment"]["S"] for item in resp["Items"]]
+        assert "qa" in envs or "dev" in envs
+
+
+class TestEnvTransitionCli:
+    def test_detect_cli_returns_none_as_json(self, moto_contracts_table, capsys):
+        """CLI detect command outputs JSON with prior_env: null."""
+        import json
+        from core.env_transition import main
+        rc = main(["prog", "detect", "--contract-id", "assets", "--consumer-repo", "acdl/c", "--new-env", "dev"])
+        assert rc == 0
+        out = json.loads(capsys.readouterr().out)
+        assert out["prior_env"] is None
+
+    def test_record_cli_outputs_json(self, moto_contracts_table, capsys):
+        """CLI record command outputs JSON with recorded: true."""
+        import json
+        from core.env_transition import main
+        rc = main(["prog", "record", "--contract-id", "assets", "--consumer-repo", "acdl/c", "--env", "dev"])
+        assert rc == 0
+        out = json.loads(capsys.readouterr().out)
+        assert out["recorded"] is True

tests/test_run_platform_env_transition.py

@@ -0,0 +1,118 @@
+"""REQ-289: run_platform.sh Step 0b environment-transition check.
+
+Asserts the shell script contains the env-transition detect-and-destroy
+block, calls env_transition.py detect, runs terraform destroy on the prior
+env, fails closed on destroy failure, and records the applied env after
+success. Pattern: tests/test_pipeline.py:79-95 (read script text + assert
+substrings).
+"""
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parent.parent
+SCRIPT = ROOT / "scripts" / "run_platform.sh"
+DEPLOY = ROOT / ".github" / "workflows" / "deploy.yml"
+
+
+def _read(path):
+    return Path(path).read_text()
+
+
+class TestRunPlatformStep0b:
+    def test_step_0b_block_exists(self):
+        """run_platform.sh has a Step 0b: environment-transition check."""
+        src = _read(SCRIPT)
+        assert "Step 0b: environment-transition check" in src
+
+    def test_step_0b_calls_env_transition_detect(self):
+        """Step 0b calls env_transition.py detect."""
+        src = _read(SCRIPT)
+        assert "env_transition.py detect" in src
+        assert "--contract-id" in src
+        assert "--consumer-repo" in src
+        assert "--new-env" in src
+
+    def test_step_0b_runs_terraform_destroy_on_prior_env(self):
+        """Step 0b runs terraform destroy against the prior env's state."""
+        src = _read(SCRIPT)
+        assert "terraform destroy" in src
+        assert "prior" in src.lower()
+        assert "deletion_protection" in src
+        assert "false" in src
+
+    def test_step_0b_fails_closed_on_destroy_failure(self):
+        """Step 0b fails closed: if destroy fails, pipeline exits non-zero."""
+        src = _read(SCRIPT)
+        assert "NO ORPHAN PATH" in src or "no orphan path" in src.lower()
+        assert "fail" in src.lower()
+        # The destroy failure must call fail() or exit 1
+        assert "prior-env terraform destroy FAILED" in src or "destroy aborted" in src
+
+    def test_step_0b_emits_evidence_event(self):
+        """Step 0b emits an ENV_DESTROYED evidence event to the outbox."""
+        src = _read(SCRIPT)
+        assert "ENV_DESTROYED" in src
+        assert "outbox_writer.py" in src
+
+    def test_step_0b_uses_terraform_init_reconfigure(self):
+        """Step 0b uses terraform init -reconfigure for the prior env."""
+        src = _read(SCRIPT)
+        assert "terraform init -reconfigure" in src
+
+    def test_step_0b_injects_deletion_protection_false(self):
+        """Step 0b injects deletion_protection=false into contract inputs."""
+        src = _read(SCRIPT)
+        assert "deletion_protection" in src
+        assert "False" in src or "false" in src
+
+    def test_step_0b_skipped_in_check_only_mode(self):
+        """Step 0b is skipped in --check-only mode (no AWS)."""
+        src = _read(SCRIPT)
+        assert 'CHECK_ONLY" = "0"' in src
+
+    def test_step_0b_skipped_in_local_mode(self):
+        """Step 0b is skipped in --local mode (emulated)."""
+        src = _read(SCRIPT)
+        assert 'LOCAL_TIER" = "0"' in src
+
+    def test_step_0b_skipped_in_decommission_mode(self):
+        """Step 0b is skipped in --decommission mode (explicit teardown)."""
+        src = _read(SCRIPT)
+        assert 'DECOMMISSION" = "0"' in src
+
+
+class TestRunPlatformRecordAppliedEnv:
+    def test_record_applied_env_after_apply_mode(self):
+        """run_platform.sh records applied env after --apply success."""
+        src = _read(SCRIPT)
+        assert "env_transition.py record" in src
+        # Must appear before or after PLATFORM APPLY OK
+        assert "PLATFORM APPLY OK" in src
+
+    def test_record_applied_env_after_e2e(self):
+        """run_platform.sh records applied env after e2e success."""
+        src = _read(SCRIPT)
+        assert "env_transition.py record" in src
+        assert "PLATFORM E2E OK" in src
+
+    def test_record_is_non_fatal(self):
+        """The record call uses || true (non-fatal if DynamoDB unreachable)."""
+        src = _read(SCRIPT)
+        # The record call should not halt the pipeline on failure
+        assert "env_transition.py record" in src
+
+
+class TestRunPlatformConsumerRepo:
+    def test_consumer_repo_env_var_set(self):
+        """CONSUMER_REPO is derived from NOVA_CONSUMER_REPO or GITHUB_REPOSITORY."""
+        src = _read(SCRIPT)
+        assert "NOVA_CONSUMER_REPO" in src
+        assert "GITHUB_REPOSITORY" in src
+        assert "CONSUMER_REPO" in src
+
+
+class TestDeployWorkflowPassesConsumerRepo:
+    def test_deploy_yml_passes_nova_consumer_repo(self):
+        """deploy.yml passes NOVA_CONSUMER_REPO to run_platform.sh (REQ-286)."""
+        src = _read(DEPLOY)
+        assert "NOVA_CONSUMER_REPO" in src
+        assert "github.repository" in src