mock_provider "aws" {
  mock_data "aws_caller_identity" {
    defaults = {
      account_id = "118688039167"
      arn        = "arn:aws:iam::118688039167:root"
      user_id    = "118688039167"
    }
  }

  mock_data "aws_availability_zones" {
    defaults = {
      names    = ["us-west-2a", "us-west-2b", "us-west-2c", "us-west-2d"]
      zone_ids = ["usw2-az2", "usw2-az1", "usw2-az3", "usw2-az4"]
    }
  }

  mock_data "aws_ami" {
    defaults = {
      id = "ami-0123456789abcdef0"
    }
  }

  mock_data "aws_ec2_instance_type" {
    defaults = {
      instance_type           = "c8g.large"
      supported_architectures = ["arm64"]
    }
  }
}

override_resource {
  target          = random_integer.port_base
  override_during = plan
  values = {
    result = 30000
  }
}

override_resource {
  target = random_password.mysql_root
  values = {
    result = "testmysqlrootpassword00000000000"
  }
}

override_resource {
  target = random_password.mysql
  values = {
    result = "testmysqluserpassword00000000000"
  }
}

override_resource {
  target = random_string.kafka_username
  values = {
    result = "username"
  }
}

override_resource {
  target = random_password.kafka
  values = {
    result = "testkafkapassword000000000000000"
  }
}

override_resource {
  target = random_string.schema_registry_username
  values = {
    result = "registry"
  }
}

override_resource {
  target = random_password.schema_registry
  values = {
    result = "testschemaregistrypassword00000000"
  }
}

override_resource {
  target = random_id.pulsar_secret
  values = {
    b64_std = "MDEyMzQ1Njc4OWFiY2RlZjAxMjM0NTY3ODlhYmNkZWY="
  }
}

override_resource {
  target = random_id.kafka_cluster_id
  values = {
    b64_url = "MDEyMzQ1Njc4OWFiY2RlZg"
  }
}

override_resource {
  target          = random_id.deployment
  override_during = plan
  values = {
    hex = "010203040506"
  }
}

override_resource {
  target          = aws_s3_bucket.test_data
  override_during = plan
  values = {
    arn = "arn:aws:s3:::date-integration-test-010203040506-118688039167-us-west-2"
    id  = "date-integration-test-010203040506-118688039167-us-west-2"
  }
}

override_resource {
  target          = aws_iam_role.s3_access
  override_during = plan
  values = {
    arn  = "arn:aws:iam::118688039167:role/date-integration-test-010203040506-s3-access"
    name = "date-integration-test-010203040506-s3-access"
  }
}

run "running_defaults" {
  command = plan

  variables {
    s3_backup_export_access_grants = {
      backup-primary = {
        account_id       = "111111111111"
        role_arn_pattern = "arn:aws:iam::111111111111:role/test-backup-primary-*"
      }
      backup-secondary = {
        account_id       = "222222222222"
        role_arn_pattern = "arn:aws:iam::222222222222:role/test-backup-secondary-*"
      }
    }
    s3_assume_role_grants = {
      auto-deployment-test = {
        account_id  = "024848481520"
        external_id = "1369847559694040894"
      }
      cdc-primary = {
        account_id  = "333333333333"
        external_id = "test-external-id-primary"
      }
      cdc-secondary = {
        account_id  = "444444444444"
        external_id = "test-external-id-secondary"
      }
    }
  }

  assert {
    condition     = aws_instance.this.instance_type == "c8g.large"
    error_message = "The default instance must be the ARM64 c8g.large."
  }

  assert {
    condition     = length(aws_lb.kafka) == 1 && length(aws_vpc_endpoint_service.kafka) == 1
    error_message = "Running mode must create one shared NLB and endpoint service."
  }

  assert {
    condition     = length(local.subnets) == 4 && length(aws_subnet.public) == 4
    error_message = "The shared NLB VPC must create one subnet in every available AZ."
  }

  assert {
    condition     = aws_lb.kafka[0].enable_cross_zone_load_balancing
    error_message = "The single shared NLB must enable cross-zone load balancing."
  }

  assert {
    condition     = length(terraform_data.endpoint_service_availability_zones) == 1
    error_message = "Running mode must reconcile all NLB zones into the Endpoint Service."
  }

  assert {
    condition     = terraform_data.wait_after_state_change.triggers_replace[0] == "running"
    error_message = "Running mode must key the post-state-change service waiter on desired_state."
  }

  assert {
    condition     = length(aws_lb_listener.mysql) == 1 && length(aws_lb_listener.kafka_public) == 1 && length(aws_lb_listener.kafka_sasl) == 1
    error_message = "The shared NLB must expose MySQL and both Kafka listeners."
  }

  assert {
    condition     = local.ports.mysql >= 20000 && local.ports.pulsar_admin_http <= 65535
    error_message = "All external ports must remain in the valid high-port range."
  }

  assert {
    condition     = length(toset(values(local.ports))) == 8
    error_message = "Every external endpoint must use a unique port."
  }

  assert {
    condition     = local.deployment_name == "date-integration-test-010203040506" && local.common_tags.usedby == local.deployment_name
    error_message = "Every deployment must have a stable unique usedby tag."
  }

  assert {
    condition     = alltrue([for dependency in local.e2e_kafka_dependencies : dependency.networkType != "NetworkType_Private_Link" || dependency.projectID == "1369847559694040763,1369847559694040575"])
    error_message = "Generated Kafka PrivateLink dependencies must target the configured TiDB Cloud E2E projects."
  }

  assert {
    condition = local.allowed_principal_arns == toset([
      "arn:aws:iam::024848481520:root",
      "arn:aws:iam::118688039167:root",
      "arn:aws:iam::131464424160:root",
      "arn:aws:iam::143458967504:root",
      "arn:aws:iam::200683210682:root",
      "arn:aws:iam::380838443567:root",
      "arn:aws:iam::385595570414:root",
      "arn:aws:iam::676546228356:root",
      "arn:aws:iam::730335266318:root",
    ])
    error_message = "The AWS Endpoint Service must retain the approved consumer account allowlist."
  }

  assert {
    condition     = aws_s3_bucket.test_data.bucket == "date-integration-test-010203040506-118688039167-us-west-2"
    error_message = "The default S3 bucket must be unique to the deployment and account."
  }

  assert {
    condition     = aws_iam_user_policy_attachment.s3_access.user == aws_iam_user.s3_access.name
    error_message = "The AK/SK user must retain its fixture S3 access policy."
  }

  assert {
    condition     = aws_iam_role_policy_attachment.s3_access.role == aws_iam_role.s3_access.name
    error_message = "The assumable S3 role must retain the fixture S3 access policy even without external assume-role grants."
  }

  assert {
    condition = (
      toset(one([for statement in jsondecode(aws_iam_policy.s3_access.policy).Statement : statement if statement.Sid == "BucketMetadata"]).Action) == toset(["s3:GetBucketLocation", "s3:ListBucket", "s3:ListBucketMultipartUploads"]) &&
      contains(one([for statement in jsondecode(aws_iam_policy.s3_access.policy).Statement : statement if statement.Sid == "CDCAndBackupExportObjects"]).Action, "s3:AbortMultipartUpload") &&
      contains(one([for statement in jsondecode(aws_iam_policy.s3_access.policy).Statement : statement if statement.Sid == "CDCAndBackupExportObjects"]).Action, "s3:ListMultipartUploadParts")
    )
    error_message = "The shared S3 role policy must retain bucket discovery and multipart permissions."
  }

  assert {
    condition = (
      toset(one([for statement in jsondecode(aws_iam_policy.s3_access.policy).Statement : statement if statement.Sid == "ImportReadObjects"]).Action) == toset(["s3:GetObject", "s3:GetObjectAttributes", "s3:GetObjectTagging", "s3:ListMultipartUploadParts"]) &&
      one([for statement in jsondecode(aws_iam_policy.s3_access.policy).Statement : statement if statement.Sid == "ImportReadObjects"]).Resource == "arn:aws:s3:::date-integration-test-010203040506-118688039167-us-west-2/import/*"
    )
    error_message = "The shared S3 role policy must grant read access to the fixture import/ prefix."
  }

  assert {
    condition = (
      length(var.s3_backup_export_access_grants) == 2 &&
      toset(keys(var.s3_backup_export_access_grants)) == toset(["backup-primary", "backup-secondary"])
    )
    error_message = "The test must exercise multiple backup-export role grants."
  }

  assert {
    condition = (
      length(var.s3_assume_role_grants) == 3 &&
      toset(keys(var.s3_assume_role_grants)) == toset(["auto-deployment-test", "cdc-primary", "cdc-secondary"])
    )
    error_message = "The test must exercise multiple account and external-ID assume-role grants."
  }

  assert {
    condition = alltrue([
      for grant in values(var.s3_assume_role_grants) : (
        one([for statement in jsondecode(aws_iam_role.s3_access.assume_role_policy).Statement : statement if statement.Sid == "AllowCloudAssumeRole${grant.account_id}"]).Principal.AWS == "arn:aws:iam::${grant.account_id}:root" &&
        contains(tolist(one([for statement in jsondecode(aws_iam_role.s3_access.assume_role_policy).Statement : statement if statement.Sid == "AllowCloudAssumeRole${grant.account_id}"]).Condition.StringEquals["sts:ExternalId"]), grant.external_id)
      )
    ])
    error_message = "The shared S3 access role must let each configured account assume it only with its configured external ID."
  }

  assert {
    condition     = length(jsondecode(aws_s3_bucket_policy.test_data.policy).Statement) == 4
    error_message = "The S3 bucket policy must retain TLS enforcement and contain all three external-access statements."
  }

  assert {
    condition = (
      toset(one([for statement in jsondecode(aws_s3_bucket_policy.test_data.policy).Statement : statement if statement.Sid == "AllowBackupExportWriters"]).Action) == toset(["s3:PutObject", "s3:PutObjectAcl"]) &&
      one([for statement in jsondecode(aws_s3_bucket_policy.test_data.policy).Statement : statement if statement.Sid == "AllowBackupExportWriters"]).Resource == "arn:aws:s3:::date-integration-test-010203040506-118688039167-us-west-2/backup-export/*" &&
      toset(one([for statement in jsondecode(aws_s3_bucket_policy.test_data.policy).Statement : statement if statement.Sid == "AllowBackupExportWriters"]).Principal.AWS) == toset([for grant in values(var.s3_backup_export_access_grants) : "arn:aws:iam::${grant.account_id}:root"]) &&
      toset(one([for statement in jsondecode(aws_s3_bucket_policy.test_data.policy).Statement : statement if statement.Sid == "AllowBackupExportWriters"]).Condition.ArnLike["aws:PrincipalArn"]) == toset([for grant in values(var.s3_backup_export_access_grants) : grant.role_arn_pattern])
    )
    error_message = "Backup-export access must match only the requested role pattern, prefix, and write actions."
  }

  assert {
    condition = (
      one([for statement in jsondecode(aws_s3_bucket_policy.test_data.policy).Statement : statement if statement.Sid == "AllowCDCList"]).Action == "s3:ListBucket" &&
      one([for statement in jsondecode(aws_s3_bucket_policy.test_data.policy).Statement : statement if statement.Sid == "AllowCDCList"]).Resource == "arn:aws:s3:::date-integration-test-010203040506-118688039167-us-west-2" &&
      one([for statement in jsondecode(aws_s3_bucket_policy.test_data.policy).Statement : statement if statement.Sid == "AllowCDCList"]).Principal.AWS == "arn:aws:iam::118688039167:role/date-integration-test-010203040506-s3-access" &&
      toset(one([for statement in jsondecode(aws_s3_bucket_policy.test_data.policy).Statement : statement if statement.Sid == "AllowCDCList"]).Condition.StringLike["s3:prefix"]) == toset(["cdc/", "cdc/*"])
    )
    error_message = "CDC ListBucket access must be limited to the CDC prefix."
  }

  assert {
    condition = (
      toset(one([for statement in jsondecode(aws_s3_bucket_policy.test_data.policy).Statement : statement if statement.Sid == "AllowCDCObjects"]).Action) == toset(["s3:PutObject", "s3:GetObject", "s3:DeleteObject"]) &&
      one([for statement in jsondecode(aws_s3_bucket_policy.test_data.policy).Statement : statement if statement.Sid == "AllowCDCObjects"]).Resource == "arn:aws:s3:::date-integration-test-010203040506-118688039167-us-west-2/cdc/*" &&
      one([for statement in jsondecode(aws_s3_bucket_policy.test_data.policy).Statement : statement if statement.Sid == "AllowCDCObjects"]).Principal.AWS == "arn:aws:iam::118688039167:role/date-integration-test-010203040506-s3-access"
    )
    error_message = "CDC object access must use the shared S3 role and only the requested CDC object actions."
  }

  assert {
    condition = (
      local.e2e_s3_dependencies[0].roleARN == "arn:aws:iam::118688039167:role/date-integration-test-010203040506-s3-access" &&
      !can(local.e2e_s3_dependencies[0].accessKey) &&
      !can(local.e2e_s3_dependencies[0].secretKey)
    )
    error_message = "Generated CDC S3 configuration must use the shared IAM role instead of static AK/SK credentials."
  }
}

run "normal_sleep_preserves_privatelink" {
  command = plan

  variables {
    desired_state                   = "stopped"
    remove_privatelink_when_stopped = false
  }

  assert {
    condition     = length(aws_lb.kafka) == 1 && length(aws_vpc_endpoint_service.kafka) == 1
    error_message = "Normal sleep must preserve the shared NLB and endpoint service."
  }

  assert {
    condition     = terraform_data.wait_after_state_change.triggers_replace[0] == "stopped"
    error_message = "Sleep must replace the post-state-change waiter so the next wake runs it again."
  }
}

run "empty_assume_role_grants_keep_role_access" {
  command = plan

  variables {
    s3_assume_role_grants = {}
  }

  assert {
    condition     = aws_iam_role_policy_attachment.s3_access.role == aws_iam_role.s3_access.name
    error_message = "The fixture S3 role must keep its S3 policy when no external assume-role consumer is configured."
  }
}

run "rejects_x86_instance_type" {
  command = plan

  variables {
    instance_type = "t3.xlarge"
  }

  override_data {
    target = data.aws_ec2_instance_type.services
    values = {
      instance_type           = "t3.xlarge"
      supported_architectures = ["x86_64"]
    }
  }

  expect_failures = [aws_instance.this]
}

run "deep_sleep_removes_privatelink" {
  command = plan

  variables {
    desired_state                   = "stopped"
    remove_privatelink_when_stopped = true
  }

  assert {
    condition     = length(aws_lb.kafka) == 0 && length(aws_vpc_endpoint_service.kafka) == 0 && length(terraform_data.endpoint_service_availability_zones) == 0
    error_message = "Deep sleep must remove the shared NLB and endpoint service."
  }
}