acdl/tests/test_env_transition.py

"""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