import boto3
import pytest
from botocore.exceptions import ClientError

from moto import mock_aws


@mock_aws
def test_filter_with_primary_region():
    conn = boto3.client("secretsmanager", region_name="us-east-1")

    conn.create_secret(Name="simple", SecretString="secret")
    conn.create_secret(
        Name="dup", SecretString="s", AddReplicaRegions=[{"Region": "us-east-2"}]
    )

    secrets = conn.list_secrets(
        Filters=[{"Key": "primary-region", "Values": ["us-east-1"]}]
    )["SecretList"]
    assert len(secrets) == 1

    secrets = conn.list_secrets(
        Filters=[{"Key": "primary-region", "Values": ["us-east-2"]}]
    )["SecretList"]
    assert len(secrets) == 0

    secrets = conn.list_secrets()["SecretList"]
    assert len(secrets) == 2

    # If our entry point is us-east-2, but filter in us-east-1
    # We can see the replicas in us-east-2 that have Primary Secrets in us-east-1
    use2 = boto3.client("secretsmanager", region_name="us-east-2")
    secrets = use2.list_secrets(
        Filters=[{"Key": "primary-region", "Values": ["us-east-1"]}]
    )["SecretList"]
    assert len(secrets) == 1
    assert secrets[0]["PrimaryRegion"] == "us-east-1"


@mock_aws
def test_create_secret_that_duplicates():
    use1 = boto3.client("secretsmanager", region_name="us-east-1")
    use2 = boto3.client("secretsmanager", region_name="us-east-2")

    resp = use1.create_secret(
        Name="dup", SecretString="s", AddReplicaRegions=[{"Region": "us-east-2"}]
    )
    primary_arn = resp["ARN"]
    assert resp["ReplicationStatus"][0]["Region"] == "us-east-2"
    assert resp["ReplicationStatus"][0]["Status"] == "InSync"

    primary_secret = use1.describe_secret(SecretId=primary_arn)
    assert primary_secret["ARN"] == primary_arn

    replica_arn = primary_arn.replace("us-east-1", "us-east-2")
    replica_secret = use2.describe_secret(SecretId=replica_arn)
    assert replica_secret["ARN"] == replica_arn
    assert replica_secret["Name"] == "dup"
    assert replica_secret["VersionIdsToStages"] == primary_secret["VersionIdsToStages"]

    replica_value = use2.get_secret_value(SecretId=replica_arn)
    assert replica_value["Name"] == "dup"
    assert replica_value["SecretString"] == "s"


@mock_aws
def test_create_secret_that_duplicates__in_same_region():
    conn = boto3.client("secretsmanager", region_name="us-east-1")

    with pytest.raises(ClientError) as exc:
        conn.create_secret(
            Name="dup_same_region",
            SecretString="s",
            AddReplicaRegions=[{"Region": "us-east-1"}],
        )
    err = exc.value.response["Error"]
    assert err["Code"] == "InvalidParameterException"
    assert err["Message"] == "Invalid replica region."


@mock_aws
def test_replicate_secret():
    us1 = boto3.client("secretsmanager", region_name="us-east-1")
    us2 = boto3.client("secretsmanager", region_name="us-east-2")

    resp = us1.create_secret(Name="dup", SecretString="s")
    primary_arn = resp["ARN"]

    resp = us1.replicate_secret_to_regions(
        SecretId=primary_arn, AddReplicaRegions=[{"Region": "us-east-2"}]
    )
    assert resp["ReplicationStatus"][0]["Region"] == "us-east-2"
    assert resp["ReplicationStatus"][0]["Status"] == "InSync"

    replica_arn = primary_arn.replace("us-east-1", "us-east-2")
    replica_secret = us2.describe_secret(SecretId=replica_arn)
    assert replica_secret["ARN"] == replica_arn


@mock_aws
def test_replicate_secret__in_same_region():
    conn = boto3.client("secretsmanager", region_name="us-east-1")

    resp = conn.create_secret(Name="dup", SecretString="s")
    with pytest.raises(ClientError) as exc:
        conn.replicate_secret_to_regions(
            SecretId=resp["ARN"], AddReplicaRegions=[{"Region": "us-east-1"}]
        )
    err = exc.value.response["Error"]
    assert err["Code"] == "InvalidParameterException"
    assert err["Message"] == "Invalid replica region."


@mock_aws
def test_replicate_secret__existing_secret():
    use1 = boto3.client("secretsmanager", region_name="us-east-1")
    use2 = boto3.client("secretsmanager", region_name="us-east-2")

    secret1 = use1.create_secret(Name="dup", SecretString="s1")["ARN"]
    secret2 = use2.create_secret(Name="dup", SecretString="s2")["ARN"]

    resp = use1.replicate_secret_to_regions(
        SecretId=secret1, AddReplicaRegions=[{"Region": "us-east-2"}]
    )
    assert resp["ReplicationStatus"][0]["Region"] == "us-east-2"
    assert resp["ReplicationStatus"][0]["Status"] == "Failed"
    assert (
        resp["ReplicationStatus"][0]["StatusMessage"]
        == "Replication failed: Secret name simple already exists in region us-east-2."
    )

    # us-east-2 still only has one secret
    assert len(use2.list_secrets()["SecretList"]) == 1

    # Both secrets still have original values
    assert use1.get_secret_value(SecretId=secret1)["SecretString"] == "s1"
    assert use2.get_secret_value(SecretId=secret2)["SecretString"] == "s2"


@mock_aws
def test_replicate_secret__overwrite_existing_secret():
    use1 = boto3.client("secretsmanager", region_name="us-east-1")
    use2 = boto3.client("secretsmanager", region_name="us-east-2")

    secret1 = use1.create_secret(Name="dup", SecretString="s1")["ARN"]
    secret2 = use2.create_secret(Name="dup", SecretString="s2")["ARN"]

    resp = use1.replicate_secret_to_regions(
        SecretId=secret1,
        AddReplicaRegions=[{"Region": "us-east-2"}],
        ForceOverwriteReplicaSecret=True,
    )
    assert resp["ReplicationStatus"][0]["Status"] == "InSync"

    # us-east-2 still only has one secret
    assert len(use2.list_secrets()["SecretList"]) == 1

    # Original Secret was yeeted
    with pytest.raises(ClientError):
        assert use2.get_secret_value(SecretId=secret2)["SecretString"] == "s1"

    # Use ListSecrets to get the new ARN
    # And verify it returns a copy of the us-east-1 secret
    secret2 = use2.list_secrets()["SecretList"][0]["ARN"]
    assert use2.get_secret_value(SecretId=secret2)["SecretString"] == "s1"


@mock_aws
def test_remove_regions_from_replication():
    use1 = boto3.client("secretsmanager", region_name="us-east-1")
    use2 = boto3.client("secretsmanager", region_name="us-east-2")
    usw2 = boto3.client("secretsmanager", region_name="us-west-2")

    arn1 = use1.create_secret(Name="dup", SecretString="s1")["ARN"]
    arn2 = arn1.replace("us-east-1", "us-east-2")
    arn3 = arn1.replace("us-east-1", "us-west-2")

    # Replicate to multiple regions
    use1.replicate_secret_to_regions(
        SecretId=arn1,
        AddReplicaRegions=[{"Region": "us-east-2"}, {"Region": "us-west-2"}],
    )

    # Replications exist
    use2.describe_secret(SecretId=arn2)
    usw2.describe_secret(SecretId=arn3)

    # Primary Secret knows about replicas
    replications = use1.describe_secret(SecretId=arn1)["ReplicationStatus"]
    assert set([rep["Region"] for rep in replications]) == {"us-east-2", "us-west-2"}

    # Remove us-east-2 replication
    use1.remove_regions_from_replication(
        SecretId=arn1, RemoveReplicaRegions=["us-east-2"]
    )
    with pytest.raises(ClientError):
        use2.describe_secret(SecretId=arn2)

    # us-west still exists
    usw2.describe_secret(SecretId=arn3)

    # Primary Secret no longer knows about us-east-2
    replications = use1.describe_secret(SecretId=arn1)["ReplicationStatus"]
    assert set([rep["Region"] for rep in replications]) == {"us-west-2"}

    # Removing region is idempotent - invoking it again does not result in any errors
    use1.remove_regions_from_replication(
        SecretId=arn1, RemoveReplicaRegions=["us-east-2"]
    )


@mock_aws
def test_update_replica():
    use1 = boto3.client("secretsmanager", region_name="us-east-1")
    use2 = boto3.client("secretsmanager", region_name="us-east-2")

    arn1 = use1.create_secret(
        Name="dup", SecretString="s1", AddReplicaRegions=[{"Region": "us-east-2"}]
    )["ARN"]
    arn2 = arn1.replace("us-east-1", "us-east-2")

    OP_NOT_PERMITTED = "Operation not permitted on a replica secret. Call must be made in primary secret's region."

    with pytest.raises(ClientError) as exc:
        use2.update_secret(SecretId=arn2, SecretString="asdf")
    err = exc.value.response["Error"]
    assert err["Code"] == "InvalidParameterException"
    assert err["Message"] == OP_NOT_PERMITTED

    with pytest.raises(ClientError) as exc:
        use2.put_secret_value(SecretId=arn2, SecretString="asdf")
    err = exc.value.response["Error"]
    assert err["Code"] == "InvalidParameterException"
    assert err["Message"] == OP_NOT_PERMITTED

    with pytest.raises(ClientError) as exc:
        use2.tag_resource(SecretId=arn2, Tags=[{"Key": "k", "Value": "v"}])
    err = exc.value.response["Error"]
    assert err["Code"] == "InvalidParameterException"
    assert err["Message"] == OP_NOT_PERMITTED

    with pytest.raises(ClientError) as exc:
        use2.untag_resource(SecretId=arn2, TagKeys=["k1"])
    err = exc.value.response["Error"]
    assert err["Code"] == "InvalidParameterException"
    assert err["Message"] == OP_NOT_PERMITTED

    with pytest.raises(ClientError) as exc:
        use2.put_resource_policy(SecretId=arn2, ResourcePolicy="k1")
    err = exc.value.response["Error"]
    assert err["Code"] == "InvalidParameterException"
    assert err["Message"] == OP_NOT_PERMITTED

    with pytest.raises(ClientError) as exc:
        use2.get_resource_policy(SecretId=arn2)
    err = exc.value.response["Error"]
    assert err["Code"] == "InvalidParameterException"
    assert err["Message"] == OP_NOT_PERMITTED

    with pytest.raises(ClientError) as exc:
        use2.delete_resource_policy(SecretId=arn2)
    err = exc.value.response["Error"]
    assert err["Code"] == "InvalidParameterException"
    assert err["Message"] == OP_NOT_PERMITTED

    with pytest.raises(ClientError) as exc:
        use2.delete_secret(SecretId=arn2)
    err = exc.value.response["Error"]
    assert err["Code"] == "InvalidParameterException"
    assert err["Message"] == OP_NOT_PERMITTED

    with pytest.raises(ClientError) as exc:
        use2.cancel_rotate_secret(SecretId=arn2)
    err = exc.value.response["Error"]
    assert err["Code"] == "InvalidParameterException"
    assert err["Message"] == OP_NOT_PERMITTED

    with pytest.raises(ClientError) as exc:
        use2.rotate_secret(SecretId=arn2)
    err = exc.value.response["Error"]
    assert err["Code"] == "InvalidParameterException"
    assert err["Message"] == OP_NOT_PERMITTED

    with pytest.raises(ClientError) as exc:
        use2.restore_secret(SecretId=arn2)
    err = exc.value.response["Error"]
    assert err["Code"] == "InvalidParameterException"
    assert err["Message"] == OP_NOT_PERMITTED

    with pytest.raises(ClientError) as exc:
        use2.replicate_secret_to_regions(SecretId=arn2, AddReplicaRegions=[{}])
    err = exc.value.response["Error"]
    assert err["Code"] == "InvalidParameterException"
    assert err["Message"] == OP_NOT_PERMITTED

    with pytest.raises(ClientError) as exc:
        use2.remove_regions_from_replication(SecretId=arn2, RemoveReplicaRegions=["a"])
    err = exc.value.response["Error"]
    assert err["Code"] == "InvalidParameterException"
    assert err["Message"] == OP_NOT_PERMITTED


@mock_aws
def test_delete_while_duplication_exist():
    use1 = boto3.client("secretsmanager", region_name="us-east-1")

    region = "us-east-2"
    arn = use1.create_secret(
        Name="dup", SecretString="s1", AddReplicaRegions=[{"Region": region}]
    )["ARN"]

    # unable to delete
    with pytest.raises(ClientError) as exc:
        use1.delete_secret(SecretId=arn)
    err = exc.value.response["Error"]
    assert err["Code"] == "InvalidParameterException"
    assert (
        err["Message"]
        == f"You can't delete secret {arn} that still has replica regions [{region}]"
    )

    # Remove us-east-2 replication
    use1.remove_regions_from_replication(SecretId=arn, RemoveReplicaRegions=[region])

    # Now we can delete
    use1.delete_secret(SecretId=arn)