apm-wo-analysis/tests/test_pipeline_synth.py

"""Synth-level assertions for the analytics dataset (Phase 3) and Slack surfaces
(Phase 4).

Synthesizes ``apm-wo-analysis-pipeline`` and asserts: the Glue table carries
partition projection with the classifier's column schema; the Athena workgroup
enforces its result location; the classifier role has zero Glue access; and the
Phase 4 Slack post + interactions Lambdas exist behind an HTTP API with only the
scoped S3/Secrets/SSM permissions. No AWS, no Docker: ``aws:cdk:bundling-stacks=[]``
skips asset bundling so this is a fast offline gate.

Run with the repo venv:

    python -m pytest tests/test_pipeline_synth.py -q
"""

import json
import sys
from pathlib import Path

import aws_cdk as cdk
from aws_cdk.assertions import Match, Template

CDK_DIR = Path(__file__).resolve().parents[1] / "cdk"
sys.path.insert(0, str(CDK_DIR))

from stacks.pipeline_stack import PipelineStack  # noqa: E402


def _s3_path_ending(suffix: str):
    """Match an ``Fn::Join`` S3 path (bucket name is a Ref) ending in ``suffix``."""
    return {"Fn::Join": ["", Match.array_with([suffix])]}


def _cdk_context() -> dict:
    """The real cdk.json context (cert ARN, hosted zone, Slack domain) — the
    hand-built App below doesn't auto-load it the way `cdk synth` does."""
    ctx = json.loads((CDK_DIR / "cdk.json").read_text())["context"]
    ctx["aws:cdk:bundling-stacks"] = []
    return ctx


def _template() -> Template:
    app = cdk.App(context=_cdk_context())
    stack = PipelineStack(
        app,
        "apm-wo-analysis-pipeline",
        env=cdk.Environment(account="328440206208", region="us-east-1"),
    )
    return Template.from_stack(stack)


def test_snapshots_table_uses_partition_projection():
    _template().has_resource_properties(
        "AWS::Glue::Table",
        {
            "TableInput": {
                "Name": "apm_wo_snapshots",
                "PartitionKeys": [{"Name": "dt", "Type": "string"}],
                "Parameters": Match.object_like(
                    {
                        "projection.enabled": "true",
                        "projection.dt.type": "date",
                        "projection.dt.format": "yyyy-MM-dd",
                        "projection.dt.range": "2026-01-01,NOW",
                        "storage.location.template": _s3_path_ending(
                            "/analytics/dt=${dt}/"
                        ),
                    }
                ),
            }
        },
    )


def test_snapshots_table_column_schema_matches_classifier():
    # Booleans typed correctly and the columns the BUILD.md prose omitted
    # (contractor_description) are present — schema mirrors the writer.
    _template().has_resource_properties(
        "AWS::Glue::Table",
        {
            "TableInput": {
                "StorageDescriptor": Match.object_like(
                    {
                        # array_with is order-sensitive: list patterns in the
                        # same order the classifier writes them.
                        "Columns": Match.array_with(
                            [
                                {"Name": "contractor_description", "Type": "string"},
                                {"Name": "is_escalation", "Type": "boolean"},
                                {"Name": "is_action", "Type": "boolean"},
                                {"Name": "mismatch", "Type": "string"},
                            ]
                        ),
                    }
                ),
            }
        },
    )


def test_athena_workgroup_enforces_results_location():
    _template().has_resource_properties(
        "AWS::Athena::WorkGroup",
        {
            "Name": "apm-wo-analysis",
            "WorkGroupConfiguration": Match.object_like(
                {
                    "EnforceWorkGroupConfiguration": True,
                    "ResultConfiguration": {
                        "OutputLocation": _s3_path_ending("/athena-results/"),
                        "EncryptionConfiguration": {"EncryptionOption": "SSE_S3"},
                    },
                }
            ),
        },
    )


def test_classifier_role_has_no_glue_access():
    # Partition projection => the classifier only writes Parquet to S3. No IAM
    # policy in the stack should grant any glue:* action.
    policies = _template().find_resources("AWS::IAM::Policy")
    for policy in policies.values():
        for stmt in policy["Properties"]["PolicyDocument"]["Statement"]:
            actions = stmt.get("Action", [])
            actions = [actions] if isinstance(actions, str) else actions
            offending = [
                a for a in actions if isinstance(a, str) and a.startswith("glue:")
            ]
            assert not offending, f"unexpected Glue access: {offending}"


# ----- Phase 4 — Slack surfaces -----


def test_slack_lambdas_exist():
    t = _template()
    for fn_name, handler in (
        ("apm-wo-analysis-slack-post", "handler.handler"),
        ("apm-wo-analysis-slack-interactions", "interactions.handler"),
    ):
        t.has_resource_properties(
            "AWS::Lambda::Function",
            {
                "FunctionName": fn_name,
                "Handler": handler,
                "Runtime": "python3.12",
                "Architectures": ["arm64"],
            },
        )


def test_classifier_has_dlq():
    # Failed async invocations must surface, not silently drop a day's data.
    t = _template()
    t.resource_count_is("AWS::SQS::Queue", 1)
    t.has_resource_properties(
        "AWS::Lambda::Function",
        Match.object_like(
            {
                "FunctionName": "apm-wo-analysis-classifier",
                "DeadLetterConfig": Match.any_value(),
            }
        ),
    )


def test_classifier_daily_invocation_alarm_treats_missing_as_breaching():
    # A silent day publishes no Invocations datapoint. NOT_BREACHING would
    # hide the outage; BREACHING is the page. Dimension Value is a Ref to the
    # function (function_name is set, but CDK still Refs the resource).
    _template().has_resource_properties(
        "AWS::CloudWatch::Alarm",
        Match.object_like(
            {
                "AlarmName": "apm-wo-analysis-classifier-invocations",
                "Namespace": "AWS/Lambda",
                "MetricName": "Invocations",
                "Statistic": "Sum",
                "Period": 86400,
                "Threshold": 1,
                "ComparisonOperator": "LessThanThreshold",
                "EvaluationPeriods": 1,
                "TreatMissingData": "breaching",
            }
        ),
    )


def test_interactions_stage_is_throttled():
    # The public Slack interactions endpoint caps rate/burst.
    _template().has_resource_properties(
        "AWS::ApiGatewayV2::Stage",
        Match.object_like(
            {
                "DefaultRouteSettings": {
                    "ThrottlingRateLimit": 10,
                    "ThrottlingBurstLimit": 20,
                }
            }
        ),
    )


def test_interactions_api_routes_post_to_slack_endpoint():
    t = _template()
    t.resource_count_is("AWS::ApiGatewayV2::Api", 1)
    t.has_resource_properties(
        "AWS::ApiGatewayV2::Route", {"RouteKey": "POST /slack/interactions"}
    )
    # Custom domain on apm-wo.seahaven.com + a Route53 alias for it.
    t.has_resource_properties(
        "AWS::ApiGatewayV2::DomainName", {"DomainName": "apm-wo.seahaven.com"}
    )
    t.has_resource_properties("AWS::Route53::RecordSet", {"Type": "A"})


def test_slack_roles_have_no_broad_or_write_access():
    # The Slack Lambdas should only read analytics/, the Slack secret, and the
    # dashboard SSM param — never write S3, never s3:*/secretsmanager:* wildcards.
    # Scope to the Slack policies by logical ID (the classifier/drop-uploader
    # legitimately hold s3:PutObject).
    t = _template()
    slack_policies = {
        lid: p
        for lid, p in t.find_resources("AWS::IAM::Policy").items()
        if lid.startswith(("SlackPost", "SlackInteractions"))
    }
    assert slack_policies, "expected scoped policies for the Slack roles"
    for policy in slack_policies.values():
        for stmt in policy["Properties"]["PolicyDocument"]["Statement"]:
            actions = stmt.get("Action", [])
            actions = [actions] if isinstance(actions, str) else actions
            for a in actions:
                if not isinstance(a, str):
                    continue
                assert a not in ("s3:*", "secretsmanager:*", "*"), f"too broad: {a}"
                assert a != "s3:PutObject", "Slack roles must not write S3"