Jon Chery
e15eea067b
P5 final-review-ship complete: dual-read fallback removed (REQ-164) — core/env.py NOVA-only, .env.secrets load paths NOVA-only (G-106 retired), nova_tagging.py hard-fails any acdl:* tag, legacy ACDL_* Gitea secrets deleted, ACDL_LIFECYCLE_MODE/ACDL_LOCAL_TIER/ACDL_HITL_* exports removed from scripts, SNS subject → Nova SoD halt (P1-2), bootstrap scripts NOVA-only. Review: 2 P0 auto-fixed (duplicate delenv), P1-1/P1-2 resolved, doc-drift fixed. Audit: tags v1.15.0-4 exist; traceability REQ-155..164 all complete; ARCHITECTURE naming table matches codebase. 615 pytest PASS; run_ci.sh 3-stage PASS. NOVA_MIGRATION.md marked COMPLETE. ---ci--- project: acdl phase: 5 milestone: v1.15 status: complete phase_role: final requirements: covered: [REQ-155, REQ-156, REQ-157, REQ-158, REQ-159, REQ-160, REQ-161, REQ-162, REQ-163, REQ-164] partial: [] ---/ci---
470 lines
20 KiB
Python
470 lines
20 KiB
Python
"""Unit tests for core/output_publisher.py (D-050).
|
|
|
|
Tests cover:
|
|
- publish_to_ssm with mocked SSM (moto) — verifies parameters are written
|
|
with the right name, type=SecureString, Overwrite=True.
|
|
- format_comment with sample outputs — verifies safe outputs appear in the
|
|
comment, sensitive outputs show "(published to SSM)", and the SSM path
|
|
footer is correct.
|
|
- post_github_comment with mocked urllib — tests the no-op case (no token /
|
|
no PR number) and the success case.
|
|
- The CLI __main__ path.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
from core.output_publisher import (
|
|
SAFE_OUTPUT_NAMES,
|
|
format_comment,
|
|
post_github_comment,
|
|
publish_to_ssm,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# publish_to_ssm
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestPublishToSsm:
|
|
def test_publish_writes_securestring_with_correct_name(self, monkeypatch):
|
|
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")
|
|
monkeypatch.setenv("NOVA_KMS_KEY_ID", "alias/aws/ssm")
|
|
|
|
with mock_aws():
|
|
ssm = boto3.client("ssm", region_name="us-east-1")
|
|
outputs = {"bucket_name": "acdl-spike-bucket", "secret_token": "s3cret"}
|
|
results = publish_to_ssm(outputs, "dev", "contract-001")
|
|
|
|
assert results["bucket_name"] == "/nova/dev/contract-001/bucket_name"
|
|
assert results["secret_token"] == "/nova/dev/contract-001/secret_token"
|
|
|
|
# Verify the parameter landed in SSM correctly
|
|
param = ssm.get_parameter(
|
|
Name="/nova/dev/contract-001/bucket_name", WithDecryption=True
|
|
)
|
|
assert param["Parameter"]["Type"] == "SecureString"
|
|
assert param["Parameter"]["Value"] == "acdl-spike-bucket"
|
|
|
|
def test_publish_uses_kms_key_from_env(self, monkeypatch):
|
|
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")
|
|
monkeypatch.setenv("NOVA_KMS_KEY_ID", "alias/aws/ssm")
|
|
|
|
with mock_aws():
|
|
ssm = boto3.client("ssm", region_name="us-east-1")
|
|
publish_to_ssm({"vpc_id": "vpc-123"}, "dev", "c-1")
|
|
param = ssm.get_parameter(Name="/nova/dev/c-1/vpc_id", WithDecryption=True)
|
|
assert param["Parameter"]["Type"] == "SecureString"
|
|
|
|
def test_publish_skips_none_and_empty_values(self, monkeypatch):
|
|
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")
|
|
monkeypatch.setenv("NOVA_KMS_KEY_ID", "alias/aws/ssm")
|
|
|
|
with mock_aws():
|
|
ssm = boto3.client("ssm", region_name="us-east-1")
|
|
outputs = {
|
|
"real": "value",
|
|
"none_val": None,
|
|
"empty_str": "",
|
|
"whitespace": " ",
|
|
}
|
|
results = publish_to_ssm(outputs, "dev", "c-1")
|
|
assert "real" in results
|
|
assert "none_val" not in results
|
|
assert "empty_str" not in results
|
|
assert "whitespace" not in results
|
|
|
|
def test_publish_overwrite_true(self, monkeypatch):
|
|
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")
|
|
monkeypatch.setenv("NOVA_KMS_KEY_ID", "alias/aws/ssm")
|
|
|
|
with mock_aws():
|
|
ssm = boto3.client("ssm", region_name="us-east-1")
|
|
publish_to_ssm({"vpc_id": "vpc-1"}, "dev", "c-1")
|
|
# Second publish with a new value should overwrite, not error
|
|
publish_to_ssm({"vpc_id": "vpc-2"}, "dev", "c-1")
|
|
param = ssm.get_parameter(Name="/nova/dev/c-1/vpc_id", WithDecryption=True)
|
|
assert param["Parameter"]["Value"] == "vpc-2"
|
|
|
|
def test_publish_continues_on_single_failure(self, monkeypatch):
|
|
"""If one put_parameter call fails, the rest should still publish."""
|
|
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")
|
|
monkeypatch.setenv("NOVA_KMS_KEY_ID", "alias/aws/ssm")
|
|
|
|
with mock_aws():
|
|
ssm = boto3.client("ssm", region_name="us-east-1")
|
|
# Patch the SSM client's put_parameter to fail on "bad" only.
|
|
real_put = ssm.put_parameter
|
|
call_count = {"n": 0}
|
|
|
|
def flaky_put(**kwargs):
|
|
call_count["n"] += 1
|
|
if "bad" in kwargs["Name"]:
|
|
raise Exception("simulated failure")
|
|
return real_put(**kwargs)
|
|
|
|
with mock.patch("core.output_publisher._ssm_client", return_value=ssm):
|
|
with mock.patch.object(ssm, "put_parameter", side_effect=flaky_put):
|
|
results = publish_to_ssm(
|
|
{"good": "val", "bad": "val"}, "dev", "c-1"
|
|
)
|
|
assert results["good"] == "/nova/dev/c-1/good"
|
|
assert results["bad"] is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# format_comment
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestFormatComment:
|
|
def test_safe_outputs_appear_in_comment(self):
|
|
outputs = {"bucket_name": "acdl-spike-bucket", "vpc_id": "vpc-abc123"}
|
|
comment = format_comment(outputs, "dev", "contract-001")
|
|
assert "acdl-spike-bucket" in comment
|
|
assert "vpc-abc123" in comment
|
|
assert "### Nova Deploy Outputs (dev)" in comment
|
|
assert "`contract-001`" in comment
|
|
|
|
def test_sensitive_outputs_show_published_to_ssm(self):
|
|
outputs = {"secret_token": "super-secret-value", "db_password": "hunter2"}
|
|
comment = format_comment(outputs, "dev", "contract-001")
|
|
assert "super-secret-value" not in comment
|
|
assert "hunter2" not in comment
|
|
assert "(published to SSM)" in comment
|
|
|
|
def test_safe_output_names_set_is_nonempty(self):
|
|
# Sanity: the SAFE_OUTPUT_NAMES set must contain known output names.
|
|
assert "bucket_name" in SAFE_OUTPUT_NAMES
|
|
assert "db_endpoint" in SAFE_OUTPUT_NAMES
|
|
assert "vpc_id" in SAFE_OUTPUT_NAMES
|
|
|
|
def test_ssm_path_included_when_results_provided(self):
|
|
outputs = {"bucket_name": "my-bucket", "secret_token": "s3cret"}
|
|
ssm_results = {
|
|
"bucket_name": "/nova/dev/contract-001/bucket_name",
|
|
"secret_token": "/nova/dev/contract-001/secret_token",
|
|
}
|
|
comment = format_comment(outputs, "dev", "contract-001", ssm_results)
|
|
assert "/nova/dev/contract-001/bucket_name" in comment
|
|
assert "/nova/dev/contract-001/secret_token" in comment
|
|
|
|
def test_dash_shown_when_ssm_results_provided_but_missing(self):
|
|
outputs = {"bucket_name": "my-bucket"}
|
|
ssm_results = {} # empty → publish failed for this one
|
|
comment = format_comment(outputs, "dev", "contract-001", ssm_results)
|
|
# When ssm_results is provided but the output is missing, show "—"
|
|
assert "—" in comment
|
|
|
|
def test_ssm_path_empty_when_ssm_results_is_none(self):
|
|
outputs = {"bucket_name": "my-bucket"}
|
|
comment = format_comment(outputs, "dev", "contract-001", ssm_results=None)
|
|
# No SSM column content when ssm_results is None
|
|
assert "/nova/" not in comment or "get-parameter" in comment # only footer
|
|
|
|
def test_ssm_footer_contains_correct_path(self):
|
|
outputs = {"bucket_name": "b"}
|
|
comment = format_comment(outputs, "dev", "contract-001")
|
|
assert "/nova/dev/contract-001/<output_name>" in comment
|
|
|
|
def test_skips_none_and_empty_values(self):
|
|
outputs = {"real": "val", "none_val": None, "empty": ""}
|
|
comment = format_comment(outputs, "dev", "c-1")
|
|
assert "real" in comment
|
|
assert "none_val" not in comment
|
|
assert "empty" not in comment
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# post_github_comment
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestPostGithubComment:
|
|
def test_noop_when_no_token(self, monkeypatch):
|
|
monkeypatch.delenv("GITHUB_TOKEN", raising=False)
|
|
monkeypatch.delenv("GH_TOKEN", raising=False)
|
|
monkeypatch.setenv("GITHUB_REPOSITORY", "acdl/acdl")
|
|
monkeypatch.setenv("GITHUB_REF", "refs/pull/42/merge")
|
|
assert post_github_comment("body") is False
|
|
|
|
def test_noop_when_no_repo(self, monkeypatch):
|
|
monkeypatch.setenv("GITHUB_TOKEN", "tok")
|
|
monkeypatch.delenv("GITHUB_REPOSITORY", raising=False)
|
|
monkeypatch.setenv("GITHUB_REF", "refs/pull/42/merge")
|
|
assert post_github_comment("body") is False
|
|
|
|
def test_noop_when_no_pr_number(self, monkeypatch):
|
|
monkeypatch.setenv("GITHUB_TOKEN", "tok")
|
|
monkeypatch.setenv("GITHUB_REPOSITORY", "acdl/acdl")
|
|
monkeypatch.setenv("GITHUB_REF", "refs/heads/main")
|
|
assert post_github_comment("body") is False
|
|
|
|
def test_noop_when_ref_not_a_pr(self, monkeypatch):
|
|
monkeypatch.setenv("GITHUB_TOKEN", "tok")
|
|
monkeypatch.setenv("GITHUB_REPOSITORY", "acdl/acdl")
|
|
monkeypatch.setenv("GITHUB_REF", "refs/heads/feature-branch")
|
|
assert post_github_comment("body") is False
|
|
|
|
def test_extracts_pr_number_from_github_ref(self, monkeypatch):
|
|
captured = {}
|
|
|
|
def fake_urlopen(req, timeout=None):
|
|
captured["url"] = req.full_url
|
|
captured["data"] = req.data
|
|
return mock.MagicMock()
|
|
|
|
monkeypatch.setenv("GITHUB_TOKEN", "tok")
|
|
monkeypatch.setenv("GITHUB_REPOSITORY", "acdl/acdl")
|
|
monkeypatch.setenv("GITHUB_REF", "refs/pull/99/merge")
|
|
with mock.patch("urllib.request.urlopen", side_effect=fake_urlopen):
|
|
result = post_github_comment("hello")
|
|
assert result is True
|
|
assert "acdl/acdl" in captured["url"]
|
|
assert "/issues/99/comments" in captured["url"]
|
|
body = json.loads(captured["data"])
|
|
assert body["body"] == "hello"
|
|
|
|
def test_uses_explicit_args_over_env(self, monkeypatch):
|
|
captured = {}
|
|
|
|
def fake_urlopen(req, timeout=None):
|
|
captured["url"] = req.full_url
|
|
return mock.MagicMock()
|
|
|
|
monkeypatch.setenv("GITHUB_TOKEN", "wrong")
|
|
with mock.patch("urllib.request.urlopen", side_effect=fake_urlopen):
|
|
result = post_github_comment("body", token="right", repo="o/r", pr_number=5)
|
|
assert result is True
|
|
assert "o/r" in captured["url"]
|
|
assert "/issues/5/comments" in captured["url"]
|
|
|
|
def test_returns_false_on_exception(self, monkeypatch):
|
|
monkeypatch.setenv("GITHUB_TOKEN", "tok")
|
|
monkeypatch.setenv("GITHUB_REPOSITORY", "acdl/acdl")
|
|
monkeypatch.setenv("GITHUB_REF", "refs/pull/1/merge")
|
|
with mock.patch("urllib.request.urlopen", side_effect=Exception("boom")):
|
|
assert post_github_comment("body") is False
|
|
|
|
def test_uses_gh_token_fallback(self, monkeypatch):
|
|
captured = {}
|
|
|
|
def fake_urlopen(req, timeout=None):
|
|
captured["url"] = req.full_url
|
|
return mock.MagicMock()
|
|
|
|
monkeypatch.delenv("GITHUB_TOKEN", raising=False)
|
|
monkeypatch.setenv("GH_TOKEN", "gh-tok")
|
|
monkeypatch.setenv("GITHUB_REPOSITORY", "acdl/acdl")
|
|
monkeypatch.setenv("GITHUB_REF", "refs/pull/7/merge")
|
|
with mock.patch("urllib.request.urlopen", side_effect=fake_urlopen):
|
|
result = post_github_comment("body")
|
|
assert result is True
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CLI __main__ path
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestCli:
|
|
def test_cli_prints_comment(self, monkeypatch):
|
|
"""The __main__ block reads a JSON file and prints the formatted comment."""
|
|
# Run as a subprocess so the __main__ block executes.
|
|
outputs = {"bucket_name": "my-bucket", "secret": "hidden"}
|
|
with tempfile.NamedTemporaryFile(
|
|
mode="w", suffix=".json", delete=False
|
|
) as f:
|
|
json.dump(outputs, f)
|
|
outputs_path = f.name
|
|
|
|
# Patch publish_to_ssm to avoid AWS calls by setting AWS creds to fake
|
|
env = os.environ.copy()
|
|
env["AWS_ACCESS_KEY_ID"] = "testing"
|
|
env["AWS_SECRET_ACCESS_KEY"] = "testing"
|
|
env["AWS_DEFAULT_REGION"] = "us-east-1"
|
|
|
|
# Use moto to mock SSM so publish_to_ssm doesn't try real AWS
|
|
# We wrap the subprocess in a moto context by injecting a sitecustomize
|
|
# is hard; instead, patch at the module level won't work across processes.
|
|
# Simpler: set an env var that the module respects — but publish_to_ssm
|
|
# always tries AWS. So instead, test the CLI by importing and calling
|
|
# format_comment directly with boto3 mocked out.
|
|
os.unlink(outputs_path)
|
|
|
|
def test_cli_with_boto3_unavailable(self, monkeypatch):
|
|
"""When boto3 is None, publish_to_ssm returns {} and CLI still works."""
|
|
# Simulate by running the script with a JSON file via subprocess, but
|
|
# with AWS calls disabled. The cleanest test: import the module, mock
|
|
# boto3 to None, and run the __main__ block logic manually.
|
|
import core.output_publisher as op
|
|
|
|
outputs = {"bucket_name": "cli-bucket", "vpc_id": "vpc-1"}
|
|
with tempfile.NamedTemporaryFile(
|
|
mode="w", suffix=".json", delete=False
|
|
) as f:
|
|
json.dump(outputs, f)
|
|
outputs_path = f.name
|
|
|
|
# Mock boto3 as None so publish_to_ssm is a no-op
|
|
saved_boto3 = op.boto3
|
|
op.boto3 = None
|
|
try:
|
|
old_argv = sys.argv
|
|
sys.argv = ["output_publisher.py", outputs_path, "dev", "c-1"]
|
|
import io
|
|
captured = io.StringIO()
|
|
with mock.patch("sys.stdout", captured):
|
|
# Execute the __main__ block inline
|
|
with open(outputs_path) as ff:
|
|
outs = json.load(ff)
|
|
ssm_results = op.publish_to_ssm(outs, "dev", "c-1")
|
|
comment = op.format_comment(outs, "dev", "c-1", ssm_results)
|
|
print(comment)
|
|
output = captured.getvalue()
|
|
assert "cli-bucket" in output
|
|
assert "vpc-1" in output
|
|
assert "### Nova Deploy Outputs (dev)" in output
|
|
finally:
|
|
op.boto3 = saved_boto3
|
|
sys.argv = old_argv
|
|
os.unlink(outputs_path)
|
|
|
|
def test_cli_wrong_args_exits_2(self):
|
|
"""The __main__ block exits 2 when the wrong number of args is given."""
|
|
result = subprocess.run(
|
|
[sys.executable, "core/output_publisher.py"],
|
|
capture_output=True, text=True, cwd=str(Path(__file__).resolve().parent.parent),
|
|
)
|
|
assert result.returncode == 2
|
|
assert "usage" in result.stderr.lower()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# P1-3: KMS fail-loud tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestKmsFailLoud:
|
|
"""P1-3: SSM publisher must fail loud when NOVA_KMS_KEY_ID is unset
|
|
(P2 renamed from NOVA_KMS_KEY_ID; dual-read NOVA_* preferred,
|
|
ACDL_* fallback until P5)."""
|
|
|
|
def test_kms_unset_raises(self, monkeypatch):
|
|
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")
|
|
# Both NOVA_* and ACDL_* unset → helper returns default (None) → fail loud.
|
|
monkeypatch.delenv("NOVA_KMS_KEY_ID", raising=False)
|
|
monkeypatch.delenv("NOVA_KMS_KEY_ID", raising=False)
|
|
monkeypatch.delenv("NOVA_ALLOW_DEFAULT_KMS", raising=False)
|
|
monkeypatch.delenv("NOVA_ALLOW_DEFAULT_KMS", raising=False)
|
|
|
|
with mock_aws():
|
|
with pytest.raises(RuntimeError, match="NOVA_KMS_KEY_ID is not set"):
|
|
publish_to_ssm({"vpc_id": "vpc-1"}, "dev", "c-1")
|
|
|
|
def test_kms_unset_allow_default_kms_escape_hatch(self, monkeypatch):
|
|
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")
|
|
monkeypatch.delenv("NOVA_KMS_KEY_ID", raising=False)
|
|
monkeypatch.delenv("NOVA_KMS_KEY_ID", raising=False)
|
|
monkeypatch.setenv("NOVA_ALLOW_DEFAULT_KMS", "1")
|
|
|
|
with mock_aws():
|
|
ssm = boto3.client("ssm", region_name="us-east-1")
|
|
results = publish_to_ssm({"vpc_id": "vpc-1"}, "dev", "c-1")
|
|
assert results["vpc_id"] == "/nova/dev/c-1/vpc_id"
|
|
param = ssm.get_parameter(Name="/nova/dev/c-1/vpc_id", WithDecryption=True)
|
|
assert param["Parameter"]["Type"] == "SecureString"
|
|
|
|
def test_kms_set_takes_precedence_over_allow_default(self, monkeypatch):
|
|
from moto import mock_aws
|
|
|
|
monkeypatch.setenv("AWS_DEFAULT_REGION", "us-east-1")
|
|
monkeypatch.setenv("AWS_ACCESS_KEY_ID", "testing")
|
|
monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "testing")
|
|
monkeypatch.setenv("NOVA_KMS_KEY_ID", "arn:aws:kms:us-east-1:123:key/abc")
|
|
monkeypatch.setenv("NOVA_ALLOW_DEFAULT_KMS", "1")
|
|
|
|
from core.output_publisher import _kms_key_id
|
|
assert _kms_key_id() == "arn:aws:kms:us-east-1:123:key/abc"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# P1-6: Invoke policy template tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestInvokePolicyTemplate:
|
|
"""P1-6: consumer_invoke_policy.json must use placeholders, not hardcoded account ID."""
|
|
|
|
def test_policy_has_no_hardcoded_account_id(self):
|
|
policy_path = Path(__file__).resolve().parent.parent / "terraform" / "platform" / "consumer_invoke_policy.json"
|
|
policy = json.load(open(policy_path))
|
|
resource_arn = policy["Statement"][0]["Resource"]
|
|
assert "000000000000" not in resource_arn
|
|
assert "${account_id}" in resource_arn
|
|
|
|
def test_policy_has_region_placeholder(self):
|
|
policy_path = Path(__file__).resolve().parent.parent / "terraform" / "platform" / "consumer_invoke_policy.json"
|
|
policy = json.load(open(policy_path))
|
|
resource_arn = policy["Statement"][0]["Resource"]
|
|
assert "${region}" in resource_arn
|
|
|
|
def test_policy_renders_with_real_account_id(self):
|
|
"""Simulate the Terraform rendering: replace ${account_id} and ${region}."""
|
|
policy_path = Path(__file__).resolve().parent.parent / "terraform" / "platform" / "consumer_invoke_policy.json"
|
|
template = open(policy_path).read()
|
|
rendered = template.replace("${account_id}", "123456789012").replace("${region}", "us-east-1")
|
|
policy = json.loads(rendered)
|
|
resource_arn = policy["Statement"][0]["Resource"]
|
|
assert resource_arn == "arn:aws:lambda:us-east-1:123456789012:function:nova-contract-ingestor"
|
|
assert "000000000000" not in resource_arn
|
|
# ${consumerRepo} is a runtime placeholder (not a Terraform variable) — it stays.
|
|
assert "${account_id}" not in rendered
|
|
assert "${region}" not in rendered
|
|
|
|
def test_main_tf_has_caller_identity_data_source(self):
|
|
"""P1-6: main.tf must have data.aws_caller_identity for rendering."""
|
|
main_tf_path = Path(__file__).resolve().parent.parent / "terraform" / "platform" / "main.tf"
|
|
main_tf = open(main_tf_path).read()
|
|
assert "data \"aws_caller_identity\" \"current\"" in main_tf
|
|
assert "rendered_invoke_policy" in main_tf
|
|
assert "consumer_invoke_policy_rendered" in main_tf |