from __future__ import annotations

import re
from pathlib import Path

import yaml

ROOT = Path(__file__).resolve().parents[3]


def _read(path: str) -> str:
    return (ROOT / path).read_text()


def _run_scripts(value: object) -> list[str]:
    scripts: list[str] = []
    if isinstance(value, dict):
        for key, child in value.items():
            if key == "run" and isinstance(child, str):
                scripts.append(child)
            else:
                scripts.extend(_run_scripts(child))
    elif isinstance(value, list):
        for child in value:
            scripts.extend(_run_scripts(child))
    return scripts


def test_oidc_trust_has_no_repo_wildcard_and_pins_subjects() -> None:
    terraform = _read("infra/aws/modules/oidc-github/main.tf")
    assert 'repo:${var.github_org}/${var.github_repo}:*' not in terraform
    assert 'repo:${local.github_repository}:ref:refs/heads/main' in terraform
    assert 'repo:${local.github_repository}:environment:production' in terraform
    assert 'values   = ["sts.amazonaws.com"]' in terraform
    assert 'token.actions.githubusercontent.com:repository' in terraform
    assert 'token.actions.githubusercontent.com:job_workflow_ref' in terraform
    assert 'token.actions.githubusercontent.com:workflow_ref' not in terraform
    assert 'token.actions.githubusercontent.com:workflow"' in terraform
    artifact_trust = terraform.split(
        'data "aws_iam_policy_document" "trust_artifact_main"', 1
    )[1].split('data "aws_iam_policy_document" "trust_dax_production"', 1)[0]
    assert "local.cli_workflow_name" in artifact_trust
    assert "local.plugin_workflow_name" in artifact_trust
    assert 'token.actions.githubusercontent.com:workflow"' in artifact_trust
    # The environment is configured main-only in GitHub, and IAM independently
    # requires main so repository-setting drift still fails closed.
    dax_trust = terraform.split(
        'data "aws_iam_policy_document" "trust_dax_production"', 1
    )[1].split(
        'data "aws_iam_policy_document" "trust_production_deploy"', 1
    )[0]
    deploy_trust = terraform.split(
        'data "aws_iam_policy_document" "trust_production_deploy"', 1
    )[1].split('resource "aws_iam_role" "frontend_config"', 1)[0]
    for trust in (dax_trust, deploy_trust):
        assert 'token.actions.githubusercontent.com:ref' in trust
        assert "local.github_main_ref" in trust
    assert "local.dax_feishu_workflow_name" in dax_trust
    assert "local.dax_bigquery_workflow_name" in dax_trust
    assert "local.dax_catalog_workflow_name" in dax_trust
    assert 'token.actions.githubusercontent.com:workflow"' in dax_trust
    assert "local.production_deploy_workflow_ref" in deploy_trust


def test_oidc_trust_uses_only_aws_supported_github_claims() -> None:
    terraform = _read("infra/aws/modules/oidc-github/main.tf")
    claims = set(re.findall(r"token\.actions\.githubusercontent\.com:([a-z_]+)", terraform))
    # Keep this allowlist aligned with AWS IAM's GitHub OIDC condition-key
    # mapping. In particular, GitHub emits workflow_ref but AWS does not map it.
    supported = {
        "actor",
        "actor_id",
        "aud",
        "enterprise_id",
        "environment",
        "job_workflow_ref",
        "ref",
        "repository",
        "repository_id",
        "repository_owner_id",
        "sub",
        "workflow",
    }
    assert claims <= supported


def test_direct_oidc_workflow_names_match_yaml_and_are_unique() -> None:
    terraform = _read("infra/aws/modules/oidc-github/main.tf")
    expected = {
        "cli_workflow_name": "cli-release.yml",
        "plugin_workflow_name": "plugin-release.yml",
        "dax_feishu_workflow_name": "deploy-feishu-dax-approval.yml",
        "dax_bigquery_workflow_name": "deploy-dax-bigquery-integration.yml",
        "dax_catalog_workflow_name": "seed-dax-dataset-catalog.yml",
    }
    configured_names: list[str] = []
    for local_name, filename in expected.items():
        match = re.search(rf'{local_name}\s*=\s*"([^"]+)"', terraform)
        assert match, local_name
        configured = match.group(1)
        workflow = yaml.safe_load(_read(f".github/workflows/{filename}"))
        assert workflow["name"] == configured
        configured_names.append(configured)
    assert len(configured_names) == len(set(configured_names))

    # IAM's direct-workflow condition is the display name, so an unrelated
    # workflow reusing a trusted name would silently broaden the trust policy.
    all_names = [
        yaml.safe_load(path.read_text())["name"]
        for path in (ROOT / ".github/workflows").glob("*.yml")
    ]
    for configured in configured_names:
        assert all_names.count(configured) == 1, configured


def test_roles_keep_production_mutations_out_of_publish_roles() -> None:
    terraform = _read("infra/aws/modules/oidc-github/main.tf")
    assert 'resource "aws_iam_role" "terraform_plan"' not in terraform
    assert "terraform-state-read-only" not in terraform

    image_block = terraform.split(
        'data "aws_iam_policy_document" "image_publish"', 1
    )[1]
    image_block = image_block.split(
        'resource "aws_iam_role_policy" "image_publish"', 1
    )[0]
    statements = re.findall(r"statement\s*\{([^{}]*)\}", image_block, re.DOTALL)
    assert len(statements) == 2

    token_statement, repository_statement = statements
    assert re.findall(r'"(ecr:[^"]+)"', token_statement) == [
        "ecr:GetAuthorizationToken"
    ]
    assert re.findall(r'"([^"]+)"', token_statement.split("resources", 1)[1]) == ["*"]
    assert set(re.findall(r'"(ecr:[^"]+)"', repository_statement)) == {
        "ecr:BatchCheckLayerAvailability",
        "ecr:BatchGetImage",
        "ecr:CompleteLayerUpload",
        "ecr:GetDownloadUrlForLayer",
        "ecr:InitiateLayerUpload",
        "ecr:PutImage",
        "ecr:UploadLayerPart",
    }
    assert re.search(r"resources\s*=\s*var\.ecr_arns", repository_statement)
    assert "ssm:" not in image_block
    assert "s3:" not in image_block

    artifact_block = terraform.split('resource "aws_iam_role" "artifact_publish"', 1)[1]
    artifact_block = artifact_block.split('resource "aws_iam_role" "dax_deploy"', 1)[0]
    assert "s3:PutObject" in artifact_block
    assert "plugin-dist/latest/manifest.json" in artifact_block
    assert "s3:DeleteObject" not in artifact_block
    assert "s3:ListBucket" not in artifact_block
    assert "ssm:" not in artifact_block
    assert "ecr:" not in artifact_block

    dax_block = terraform.split('resource "aws_iam_role" "dax_deploy"', 1)[1]
    dax_block = dax_block.split('resource "aws_iam_role" "production_deploy"', 1)[0]
    assert 'actions = ["ssm:GetParameter"]' in dax_block
    put_parameter_resources = re.search(
        r'actions\s*=\s*\["ssm:PutParameter"\]\s*resources\s*=\s*\[(.*?)\]',
        dax_block,
        flags=re.DOTALL,
    )
    assert put_parameter_resources is not None
    assert re.findall(r'"([^"\n]+)"', put_parameter_resources.group(1)) == [
        "${local.parameter_prefix_arn}/FEISHU_APPROVAL_DATA_ACCESS_GROUP_GRANT_CODE",
        "${local.parameter_prefix_arn}/FEISHU_APPROVAL_DATA_ACCESS_SERVICE_ACCOUNT_GRANT_CODE",
        "${local.parameter_prefix_arn}/FEISHU_APPROVAL_DATA_ACCESS_PERSONAL_GRANT_CODE",
    ]
    for forbidden in (
        "ssm:GetParameters",
        "ssm:DeleteParameter",
        "ssm:AddTagsToResource",
        "s3:GetObject",
        "s3:DeleteObject",
        "s3:ListBucket",
    ):
        assert forbidden not in dax_block
    assert dax_block.count('resources = ["*"]') == 1

    production_block = terraform.split(
        'resource "aws_iam_role" "production_deploy"', 1
    )[1]
    assert "ssm:SendCommand" in production_block
    assert "ecr:" not in production_block
    assert "ssm:GetParameter" not in production_block
    assert production_block.count('resources = ["*"]') == 1


def test_oidc_output_migration_preserves_unrelated_production_outputs() -> None:
    outputs = _read("infra/aws/environments/prod/outputs.tf")
    assert 'output "redis_endpoint"' in outputs
    assert "module.elasticache.primary_endpoint" in outputs


def test_oidc_runbook_allows_only_guarded_production_adoption() -> None:
    runbook = _read("docs/ops/github-oidc-roles.md")
    active = runbook.split("## Production adoption (guarded workflow only)", 1)[1]

    assert "stage01_guard.py plan" in active
    assert "stage01_guard.py apply" in active
    assert "--execute-reviewed-plan" in active
    assert "exactly the 14" in active
    assert "05-post-compute-iam" in active
    assert "terraform apply" not in active.lower()
    assert "normal full Terraform apply" not in runbook
    assert "-target=module." not in runbook


def test_workflows_use_purpose_specific_roles_and_production_gate() -> None:
    all_workflows = "\n".join(
        path.read_text() for path in (ROOT / ".github/workflows").glob("*.yml")
    )
    assert "vars.AWS_ROLE_ARN" not in all_workflows
    terraform_ci = _read(".github/workflows/terraform-plan.yml")
    assert "AWS_TERRAFORM_PLAN_ROLE_ARN" not in terraform_ci
    assert "configure-aws-credentials" not in terraform_ci
    assert "id-token: write" not in terraform_ci
    assert "terraform init -backend=false" in terraform_ci
    assert "terraform plan" not in terraform_ci
    assert "working-directory: infra/aws" in terraform_ci
    assert "run: terraform fmt -check -recursive" in terraform_ci
    image = _read(".github/workflows/image-release.yml")
    assert "AWS_IMAGE_PUBLISH_ROLE_ARN" in image
    assert "AWS_FRONTEND_CONFIG_ROLE_ARN" in image
    assert "unset-current-credentials: true" in image
    assert "aws ssm send-command" not in image
    assert "uses: ./.github/workflows/deploy-ec2.yml" not in image
    manual = _read(".github/workflows/prod-deploy.yml")
    assert "workflow_dispatch:" in manual
    assert "uses: ./.github/workflows/deploy-ec2.yml" in manual

    for path in (
        ".github/workflows/cli-release.yml",
        ".github/workflows/plugin-release.yml",
    ):
        assert "AWS_ARTIFACT_PUBLISH_ROLE_ARN" in _read(path)
    assert "AWS_PRODUCTION_DEPLOY_ROLE_ARN" in _read(
        ".github/workflows/deploy-ec2.yml"
    )
    for path in (
        ".github/workflows/deploy-feishu-dax-approval.yml",
        ".github/workflows/deploy-dax-bigquery-integration.yml",
        ".github/workflows/seed-dax-dataset-catalog.yml",
    ):
        workflow = _read(path)
        assert "AWS_DAX_DEPLOY_ROLE_ARN" in workflow
        assert "environment: production" in workflow
        assert "inputs.github_environment" not in workflow


def test_oidc_roles_have_non_mutating_identity_preflights() -> None:
    aggregate = _read(".github/workflows/iam-oidc-preflight.yml")
    assert "uses: ./.github/workflows/image-release.yml" in aggregate
    assert "uses: ./.github/workflows/deploy-ec2.yml" in aggregate
    assert aggregate.count("iam_preflight_only: true") == 2

    image = _read(".github/workflows/image-release.yml")
    prepare = image.split("  prepare:", 1)[1].split("\n  build-server:", 1)[0]
    assert prepare.count("aws sts get-caller-identity") == 2
    build_server = image.split("  build-server:", 1)[1].split(
        "\n  build-web:", 1
    )[0]
    build_web = image.split("  build-web:", 1)[1]
    for block in (build_server, build_web):
        assert "if: ${{ !inputs.iam_preflight_only }}" in block

    cli = _read(".github/workflows/cli-release.yml")
    cli_preflight = cli.split("  iam-preflight:", 1)[1].split(
        "\n  build-binaries:", 1
    )[0]
    assert "AWS_ARTIFACT_PUBLISH_ROLE_ARN" in cli_preflight
    assert "aws sts get-caller-identity" in cli_preflight
    assert "aws s3" not in cli_preflight

    dax = _read(".github/workflows/deploy-feishu-dax-approval.yml")
    dax_preflight = dax.split("  iam-preflight:", 1)[1].split("\n  deploy:", 1)[0]
    assert "AWS_DAX_DEPLOY_ROLE_ARN" in dax_preflight
    assert "aws sts get-caller-identity" in dax_preflight
    assert "aws s3" not in dax_preflight
    assert "aws ssm" not in dax_preflight

    deploy = _read(".github/workflows/deploy-ec2.yml")
    assert "aws sts get-caller-identity" in deploy
    for mutation in (
        "Verify browser upload CORS contract",
        "Upload deploy assets to S3",
        "Deploy via SSM Run Command",
    ):
        suffix = deploy.split(f"- name: {mutation}", 1)[1]
        assert "if: ${{ !inputs.iam_preflight_only }}" in suffix.split("\n\n", 1)[0]


def test_production_deploy_has_exactly_one_environment_gate() -> None:
    wrapper = _read(".github/workflows/prod-deploy.yml")
    reusable = _read(".github/workflows/deploy-ec2.yml")
    assert "environment: production" not in wrapper
    assert "Approval Gate" not in wrapper
    assert reusable.count("environment: production") == 1


def test_ec2_secret_and_parameter_reads_are_project_scoped() -> None:
    terraform = _read("infra/aws/modules/ec2/main.tf")
    assert "secretsmanager:" not in terraform
    assert "parameter/${var.project}/${var.environment}/*" in terraform
    boundary = terraform.split('sid    = "ParameterStoreBoundary"', 1)[1].split(
        "statement {", 1
    )[0]
    assert 'effect = "Deny"' in boundary
    assert "not_resources" in boundary
    assert set(re.findall(r'"(ssm:[^"]+)"', boundary)) == {
        "ssm:GetParameter",
        "ssm:GetParameters",
        "ssm:GetParametersByPath",
    }
    exact_not_resource = (
        "arn:${data.aws_partition.current.partition}:ssm:"
        '${data.aws_region.current.name}:'
        '${data.aws_caller_identity.current.account_id}:'
        "parameter/${var.project}/${var.environment}/*"
    )
    assert terraform.count('sid    = "ParameterStoreBoundary"') == 1
    not_resources_match = re.search(
        r"not_resources\s*=\s*\[(.*?)\]", boundary, flags=re.DOTALL
    )
    assert not_resources_match is not None
    assert re.findall(r'"([^"]+)"', not_resources_match.group(1)) == [
        exact_not_resource
    ]
    ecr_read = terraform.split('sid = "ECRRepositoryRead"', 1)[1].split(
        "statement {", 1
    )[0]
    assert "resources = var.ecr_arns" in ecr_read
    assert set(re.findall(r'"(ecr:[^"]+)"', ecr_read)) == {
        "ecr:BatchCheckLayerAvailability",
        "ecr:BatchGetImage",
        "ecr:GetDownloadUrlForLayer",
    }
    deploy_list = terraform.split('sid       = "S3DeployAssetsList"', 1)[1]
    assert 'values   = ["deploy-assets/*"]' in deploy_list


def test_id_tokens_are_granted_only_to_jobs_that_assume_aws_roles() -> None:
    image = _read(".github/workflows/image-release.yml")
    assert "permissions:\n  contents: read\n\nenv:" in image
    job_bounds = (
        ("prepare", "build-server"),
        ("build-server", "build-web"),
    )
    for job, next_job in job_bounds:
        block = image.split(f"  {job}:", 1)[1].split(f"\n  {next_job}:", 1)[0]
        assert "id-token: write" in block
    build_web = image.split("  build-web:", 1)[1]
    assert "id-token: write" in build_web
    assert "issues: write" not in image
    assert "prepull:" not in image

    cli = _read(".github/workflows/cli-release.yml")
    build_block = cli.split("  build-binaries:", 1)[1].split("  publish:", 1)[0]
    publish_block = cli.split("  publish:", 1)[1]
    assert "id-token: write" not in build_block
    assert "id-token: write" in publish_block


def test_oidc_workflow_inputs_never_interpolate_directly_into_shell() -> None:
    paths = (
        "deploy-ec2.yml",
        "deploy-feishu-dax-approval.yml",
        "deploy-dax-bigquery-integration.yml",
        "seed-dax-dataset-catalog.yml",
    )
    for filename in paths:
        workflow = yaml.safe_load(_read(f".github/workflows/{filename}"))
        scripts = _run_scripts(workflow)
        assert scripts, filename
        assert all("${{ inputs." not in script for script in scripts), filename

    validation_step = {
        "deploy-feishu-dax-approval.yml": (
            "Validate workflow inputs before obtaining AWS credentials"
        ),
        "deploy-dax-bigquery-integration.yml": (
            "Validate workflow inputs before obtaining AWS credentials"
        ),
        "seed-dax-dataset-catalog.yml": "Validate inventory input",
    }
    for filename, validation_name in validation_step.items():
        workflow_text = _read(f".github/workflows/{filename}")
        if filename == "deploy-feishu-dax-approval.yml":
            # The separate identity-only job intentionally assumes the role
            # without consuming workflow inputs. Keep this ordering assertion
            # scoped to the live deployment job.
            workflow_text = workflow_text.split("\n  deploy:", 1)[1]
        validation = workflow_text.index(validation_name)
        credentials = workflow_text.index("Configure AWS credentials (OIDC)")
        assert validation < credentials, filename


def test_dax_catalog_import_requires_explicit_manual_dispatch() -> None:
    workflow = yaml.safe_load(_read(".github/workflows/seed-dax-dataset-catalog.yml"))
    triggers = workflow.get("on", workflow.get(True))
    assert set(triggers) == {"workflow_dispatch"}
    inputs = triggers["workflow_dispatch"]["inputs"]
    assert inputs["dry_run"]["default"] == "true"
    assert workflow["env"]["INPUT_DRY_RUN"] == "${{ inputs.dry_run }}"
    assert workflow["env"]["INPUT_ALLOW_UNRESOLVED"] == "${{ inputs.allow_unresolved }}"