procurement-ingest/tests/test_resource_policy_helper.py

"""Unit tests for common.make_slack_bot_read_policy (prod-migration PR).

Two layers, no full-app synth (PoIngestStack synth would trigger Docker
bundling in CI):

1. Synthesizes a minimal throwaway stack with one table using the helper and
   asserts the exact ResourcePolicy statement shape (sid, read-only actions,
   mgmt-account principal, PrincipalArn prefix condition, per-table ARN).
2. Static pins that the four bot-read tables in the real stack files pass
   resource_policy= and pending-site-review does NOT (no bot consumer).
"""

import importlib.util
import re
import sys
from pathlib import Path

import aws_cdk as cdk
import pytest
from aws_cdk import assertions
from aws_cdk import aws_dynamodb as dynamodb

REPO_ROOT = Path(__file__).resolve().parents[1]

_spec = importlib.util.spec_from_file_location(
    "cdk_common", REPO_ROOT / "cdk" / "common.py"
)
common = importlib.util.module_from_spec(_spec)
sys.modules["cdk_common"] = common
_spec.loader.exec_module(common)

MGMT_ACCOUNT = "328440206208"
PROD_ACCOUNT = "011934824531"


def _synth_policy(table_name):
    app = cdk.App()
    stack = cdk.Stack(
        app,
        "T",
        env=cdk.Environment(account=PROD_ACCOUNT, region="us-east-1"),
    )
    dynamodb.Table(
        stack,
        "Tbl",
        table_name=table_name,
        partition_key=dynamodb.Attribute(name="pk", type=dynamodb.AttributeType.STRING),
        billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST,
        resource_policy=common.make_slack_bot_read_policy(stack, table_name),
    )
    template = assertions.Template.from_stack(stack)
    tables = template.find_resources("AWS::DynamoDB::Table")
    (props,) = [r["Properties"] for r in tables.values()]
    statements = props["ResourcePolicy"]["PolicyDocument"]["Statement"]
    assert len(statements) == 1
    return statements[0]


def test_statement_shape_and_actions():
    s = _synth_policy("purchase-orders")
    assert s["Sid"] == "AllowMgmtSlackBotRead"
    assert s["Effect"] == "Allow"
    assert sorted(s["Action"]) == [
        "dynamodb:BatchGetItem",
        "dynamodb:DescribeTable",
        "dynamodb:GetItem",
        "dynamodb:Query",
        "dynamodb:Scan",
    ]


def test_no_write_actions():
    s = _synth_policy("purchase-orders")
    writes = {
        a
        for a in s["Action"]
        if re.search(r"Put|Update|Delete|Write|Create|Restore|Export|Import", a)
    }
    assert not writes


def test_principal_is_mgmt_account_root():
    s = _synth_policy("WorkOrders")
    principal = s["Principal"]["AWS"]
    # Token-joined root-principal ARN: {"Fn::Join": ["", ["arn:", {...}, ":iam::328440206208:root"]]}
    assert MGMT_ACCOUNT in str(principal)
    assert ":root" in str(principal)


def test_principal_arn_condition_pins_bot_stack_prefix():
    s = _synth_policy("WorkOrders")
    assert s["Condition"]["ArnLike"]["aws:PrincipalArn"] == (
        f"arn:aws:iam::{MGMT_ACCOUNT}:role/seahaven-slack-bot-*"
    )


@pytest.mark.parametrize("name", ["purchase-orders", "WorkOrderComments"])
def test_resource_is_exactly_the_named_table_arn(name):
    import json

    s = _synth_policy(name)
    # format_arn leaves the partition as {"Ref": "AWS::Partition"}, so the
    # resource is an Fn::Join; assert on the deterministic joined tail.
    dumped = json.dumps(s["Resource"])
    assert f"dynamodb:us-east-1:{PROD_ACCOUNT}:table/{name}" in dumped
    # Exactly one resource entry (no index wildcard, no extra ARNs).
    assert not isinstance(s["Resource"], list) or len(s["Resource"]) == 1


# --- static pins against the real stack files ---

_PO_STACK = (REPO_ROOT / "cdk" / "po_stack.py").read_text()
_WO_STACK = (REPO_ROOT / "cdk" / "wo_stack.py").read_text()


def _table_blocks(source):
    """Map table_name -> the dynamodb.Table(...) call block containing it."""
    blocks = {}
    for m in re.finditer(r"dynamodb\.Table\(", source):
        depth, i = 0, m.end() - 1
        while i < len(source):
            depth += {"(": 1, ")": -1}.get(source[i], 0)
            if depth == 0:
                break
            i += 1
        block = source[m.start() : i + 1]
        nm = re.search(r'table_name="([^"]+)"', block)
        if nm:
            blocks[nm.group(1)] = block
    return blocks


def test_bot_read_tables_carry_resource_policy():
    blocks = {**_table_blocks(_PO_STACK), **_table_blocks(_WO_STACK)}
    for name in [
        "purchase-orders",
        "verified-sites",
        "WorkOrders",
        "WorkOrderComments",
    ]:
        assert "make_slack_bot_read_policy" in blocks[name], name


def test_pending_site_review_has_no_resource_policy():
    blocks = _table_blocks(_PO_STACK)
    assert "resource_policy" not in blocks["pending-site-review"]