"""EventBridge Pipes integration tests."""

import time

import pytest
from botocore.exceptions import ClientError

ACCOUNT_ID = "000000000000"
REGION = "us-east-1"
ROLE_ARN = f"arn:aws:iam::{ACCOUNT_ID}:role/pipe-role"


def sqs_arn(queue_name):
    return f"arn:aws:sqs:{REGION}:{ACCOUNT_ID}:{queue_name}"


class TestPipesCRUD:
    """Test Pipes create, describe, list, update, delete operations."""

    def test_create_pipe(self, pipes_client, sqs_client, unique_name):
        """Test CreatePipe creates a pipe in STOPPED state."""
        src_name = f"pipe-src-{unique_name}"
        tgt_name = f"pipe-tgt-{unique_name}"
        pipe_name = f"pipe-{unique_name}"

        sqs_client.create_queue(QueueName=src_name)
        sqs_client.create_queue(QueueName=tgt_name)
        try:
            response = pipes_client.create_pipe(
                Name=pipe_name,
                Source=sqs_arn(src_name),
                Target=sqs_arn(tgt_name),
                RoleArn=ROLE_ARN,
                DesiredState="STOPPED",
            )
            assert response["CurrentState"] == "STOPPED"
            assert pipe_name in response["Arn"]
        finally:
            pipes_client.delete_pipe(Name=pipe_name)
            sqs_client.delete_queue(QueueUrl=_queue_url(sqs_client, src_name))
            sqs_client.delete_queue(QueueUrl=_queue_url(sqs_client, tgt_name))

    def test_describe_pipe(self, pipes_client, sqs_client, unique_name):
        """Test DescribePipe returns pipe details."""
        src_name = f"pipe-src-{unique_name}"
        tgt_name = f"pipe-tgt-{unique_name}"
        pipe_name = f"pipe-{unique_name}"

        sqs_client.create_queue(QueueName=src_name)
        sqs_client.create_queue(QueueName=tgt_name)
        try:
            pipes_client.create_pipe(
                Name=pipe_name,
                Source=sqs_arn(src_name),
                Target=sqs_arn(tgt_name),
                RoleArn=ROLE_ARN,
                DesiredState="STOPPED",
            )
            response = pipes_client.describe_pipe(Name=pipe_name)
            assert response["Name"] == pipe_name
            assert response["Source"] == sqs_arn(src_name)
            assert response["Target"] == sqs_arn(tgt_name)
            assert response["CurrentState"] == "STOPPED"
        finally:
            pipes_client.delete_pipe(Name=pipe_name)
            sqs_client.delete_queue(QueueUrl=_queue_url(sqs_client, src_name))
            sqs_client.delete_queue(QueueUrl=_queue_url(sqs_client, tgt_name))

    def test_list_pipes(self, pipes_client, sqs_client, unique_name):
        """Test ListPipes returns created pipe."""
        src_name = f"pipe-src-{unique_name}"
        tgt_name = f"pipe-tgt-{unique_name}"
        pipe_name = f"pipe-{unique_name}"

        sqs_client.create_queue(QueueName=src_name)
        sqs_client.create_queue(QueueName=tgt_name)
        try:
            pipes_client.create_pipe(
                Name=pipe_name,
                Source=sqs_arn(src_name),
                Target=sqs_arn(tgt_name),
                RoleArn=ROLE_ARN,
                DesiredState="STOPPED",
            )
            response = pipes_client.list_pipes()
            names = [p["Name"] for p in response["Pipes"]]
            assert pipe_name in names
        finally:
            pipes_client.delete_pipe(Name=pipe_name)
            sqs_client.delete_queue(QueueUrl=_queue_url(sqs_client, src_name))
            sqs_client.delete_queue(QueueUrl=_queue_url(sqs_client, tgt_name))

    def test_update_pipe(self, pipes_client, sqs_client, unique_name):
        """Test UpdatePipe updates pipe description."""
        src_name = f"pipe-src-{unique_name}"
        tgt_name = f"pipe-tgt-{unique_name}"
        pipe_name = f"pipe-{unique_name}"

        sqs_client.create_queue(QueueName=src_name)
        sqs_client.create_queue(QueueName=tgt_name)
        try:
            pipes_client.create_pipe(
                Name=pipe_name,
                Source=sqs_arn(src_name),
                Target=sqs_arn(tgt_name),
                RoleArn=ROLE_ARN,
                DesiredState="STOPPED",
            )
            pipes_client.update_pipe(
                Name=pipe_name,
                RoleArn=ROLE_ARN,
                Description="updated via SDK",
                DesiredState="STOPPED",
            )
            response = pipes_client.describe_pipe(Name=pipe_name)
            assert response["Description"] == "updated via SDK"
        finally:
            pipes_client.delete_pipe(Name=pipe_name)
            sqs_client.delete_queue(QueueUrl=_queue_url(sqs_client, src_name))
            sqs_client.delete_queue(QueueUrl=_queue_url(sqs_client, tgt_name))

    def test_delete_pipe(self, pipes_client, sqs_client, unique_name):
        """Test DeletePipe removes the pipe."""
        src_name = f"pipe-src-{unique_name}"
        tgt_name = f"pipe-tgt-{unique_name}"
        pipe_name = f"pipe-{unique_name}"

        sqs_client.create_queue(QueueName=src_name)
        sqs_client.create_queue(QueueName=tgt_name)
        try:
            pipes_client.create_pipe(
                Name=pipe_name,
                Source=sqs_arn(src_name),
                Target=sqs_arn(tgt_name),
                RoleArn=ROLE_ARN,
                DesiredState="STOPPED",
            )
            pipes_client.delete_pipe(Name=pipe_name)

            with pytest.raises(ClientError) as exc_info:
                pipes_client.describe_pipe(Name=pipe_name)
            assert exc_info.value.response["Error"]["Code"] == "NotFoundException"
        finally:
            sqs_client.delete_queue(QueueUrl=_queue_url(sqs_client, src_name))
            sqs_client.delete_queue(QueueUrl=_queue_url(sqs_client, tgt_name))

    def test_describe_nonexistent_pipe(self, pipes_client):
        """Test DescribePipe for non-existent pipe returns NotFoundException."""
        with pytest.raises(ClientError) as exc_info:
            pipes_client.describe_pipe(Name="nonexistent-pipe")
        assert exc_info.value.response["Error"]["Code"] == "NotFoundException"


class TestPipesLifecycle:
    """Test Pipes start/stop operations."""

    def test_start_and_stop_pipe(self, pipes_client, sqs_client, unique_name):
        """Test StartPipe and StopPipe transitions."""
        src_name = f"pipe-src-{unique_name}"
        tgt_name = f"pipe-tgt-{unique_name}"
        pipe_name = f"pipe-{unique_name}"

        sqs_client.create_queue(QueueName=src_name)
        sqs_client.create_queue(QueueName=tgt_name)
        try:
            pipes_client.create_pipe(
                Name=pipe_name,
                Source=sqs_arn(src_name),
                Target=sqs_arn(tgt_name),
                RoleArn=ROLE_ARN,
                DesiredState="STOPPED",
            )

            response = pipes_client.start_pipe(Name=pipe_name)
            assert response["CurrentState"] == "RUNNING"

            response = pipes_client.stop_pipe(Name=pipe_name)
            assert response["CurrentState"] == "STOPPED"
        finally:
            pipes_client.delete_pipe(Name=pipe_name)
            sqs_client.delete_queue(QueueUrl=_queue_url(sqs_client, src_name))
            sqs_client.delete_queue(QueueUrl=_queue_url(sqs_client, tgt_name))


class TestPipesPolling:
    """Test Pipes source polling and target invocation."""

    def test_sqs_to_sqs_forwarding(self, pipes_client, sqs_client, unique_name):
        """Test that a RUNNING pipe forwards SQS messages to SQS target."""
        src_name = f"pipe-src-{unique_name}"
        tgt_name = f"pipe-tgt-{unique_name}"
        pipe_name = f"pipe-{unique_name}"

        src_resp = sqs_client.create_queue(QueueName=src_name)
        src_url = src_resp["QueueUrl"]
        tgt_resp = sqs_client.create_queue(QueueName=tgt_name)
        tgt_url = tgt_resp["QueueUrl"]
        try:
            pipes_client.create_pipe(
                Name=pipe_name,
                Source=sqs_arn(src_name),
                Target=sqs_arn(tgt_name),
                RoleArn=ROLE_ARN,
                DesiredState="RUNNING",
            )

            sqs_client.send_message(QueueUrl=src_url, MessageBody="hello from pipes")

            found = False
            for _ in range(15):
                time.sleep(1)
                resp = sqs_client.receive_message(
                    QueueUrl=tgt_url, MaxNumberOfMessages=1, WaitTimeSeconds=1
                )
                msgs = resp.get("Messages", [])
                if msgs and "hello from pipes" in msgs[0]["Body"]:
                    found = True
                    break

            assert found, "Target queue should receive forwarded message"
        finally:
            pipes_client.delete_pipe(Name=pipe_name)
            sqs_client.delete_queue(QueueUrl=src_url)
            sqs_client.delete_queue(QueueUrl=tgt_url)

    def test_filter_criteria_filters_messages(self, pipes_client, sqs_client, unique_name):
        """Test that FilterCriteria only forwards matching messages."""
        src_name = f"pipe-src-{unique_name}"
        tgt_name = f"pipe-tgt-{unique_name}"
        pipe_name = f"pipe-{unique_name}"

        src_resp = sqs_client.create_queue(QueueName=src_name)
        src_url = src_resp["QueueUrl"]
        tgt_resp = sqs_client.create_queue(QueueName=tgt_name)
        tgt_url = tgt_resp["QueueUrl"]
        try:
            pipes_client.create_pipe(
                Name=pipe_name,
                Source=sqs_arn(src_name),
                Target=sqs_arn(tgt_name),
                RoleArn=ROLE_ARN,
                DesiredState="RUNNING",
                SourceParameters={
                    "FilterCriteria": {
                        "Filters": [
                            {"Pattern": '{"body": {"status": ["active"]}}'},
                        ],
                    },
                },
            )

            sqs_client.send_message(
                QueueUrl=src_url,
                MessageBody='{"status": "active", "id": "match-1"}',
            )
            sqs_client.send_message(
                QueueUrl=src_url,
                MessageBody='{"status": "inactive", "id": "no-match"}',
            )

            found = False
            for _ in range(15):
                time.sleep(1)
                resp = sqs_client.receive_message(
                    QueueUrl=tgt_url, MaxNumberOfMessages=10, WaitTimeSeconds=1
                )
                msgs = resp.get("Messages", [])
                if any("match-1" in m["Body"] for m in msgs):
                    assert not any("no-match" in m["Body"] for m in msgs), \
                        "Non-matching message should not be forwarded"
                    found = True
                    break

            assert found, "Target queue should receive matching message"

            attrs = sqs_client.get_queue_attributes(
                QueueUrl=src_url, AttributeNames=["ApproximateNumberOfMessages"]
            )
            assert attrs["Attributes"]["ApproximateNumberOfMessages"] == "0"
        finally:
            pipes_client.delete_pipe(Name=pipe_name)
            sqs_client.delete_queue(QueueUrl=src_url)
            sqs_client.delete_queue(QueueUrl=tgt_url)

    def test_batch_size_in_source_parameters(self, pipes_client, sqs_client, unique_name):
        """Test that BatchSize in SourceParameters is respected."""
        src_name = f"pipe-src-{unique_name}"
        tgt_name = f"pipe-tgt-{unique_name}"
        pipe_name = f"pipe-{unique_name}"

        src_resp = sqs_client.create_queue(QueueName=src_name)
        src_url = src_resp["QueueUrl"]
        tgt_resp = sqs_client.create_queue(QueueName=tgt_name)
        tgt_url = tgt_resp["QueueUrl"]
        try:
            pipes_client.create_pipe(
                Name=pipe_name,
                Source=sqs_arn(src_name),
                Target=sqs_arn(tgt_name),
                RoleArn=ROLE_ARN,
                DesiredState="RUNNING",
                SourceParameters={
                    "SqsQueueParameters": {
                        "BatchSize": 1,
                    },
                },
            )

            for i in range(1, 4):
                sqs_client.send_message(QueueUrl=src_url, MessageBody=f"batch-msg-{i}")

            found_messages = set()
            for _ in range(20):
                if len(found_messages) >= 3:
                    break
                time.sleep(1)
                resp = sqs_client.receive_message(
                    QueueUrl=tgt_url, MaxNumberOfMessages=10, WaitTimeSeconds=1
                )
                for msg in resp.get("Messages", []):
                    for j in range(1, 4):
                        if f"batch-msg-{j}" in msg["Body"]:
                            found_messages.add(f"batch-msg-{j}")

            assert len(found_messages) == 3, "All 3 messages should arrive at target"
        finally:
            pipes_client.delete_pipe(Name=pipe_name)
            sqs_client.delete_queue(QueueUrl=src_url)
            sqs_client.delete_queue(QueueUrl=tgt_url)

    def test_stopped_pipe_does_not_forward(self, pipes_client, sqs_client, unique_name):
        """Test that a STOPPED pipe does not forward messages."""
        src_name = f"pipe-src-{unique_name}"
        tgt_name = f"pipe-tgt-{unique_name}"
        pipe_name = f"pipe-{unique_name}"

        src_resp = sqs_client.create_queue(QueueName=src_name)
        src_url = src_resp["QueueUrl"]
        tgt_resp = sqs_client.create_queue(QueueName=tgt_name)
        tgt_url = tgt_resp["QueueUrl"]
        try:
            pipes_client.create_pipe(
                Name=pipe_name,
                Source=sqs_arn(src_name),
                Target=sqs_arn(tgt_name),
                RoleArn=ROLE_ARN,
                DesiredState="STOPPED",
            )

            sqs_client.send_message(QueueUrl=src_url, MessageBody="should not forward")
            time.sleep(3)

            attrs = sqs_client.get_queue_attributes(
                QueueUrl=src_url, AttributeNames=["ApproximateNumberOfMessages"]
            )
            assert attrs["Attributes"]["ApproximateNumberOfMessages"] == "1"

            resp = sqs_client.receive_message(
                QueueUrl=tgt_url, MaxNumberOfMessages=1, WaitTimeSeconds=1
            )
            assert len(resp.get("Messages", [])) == 0
        finally:
            pipes_client.delete_pipe(Name=pipe_name)
            sqs_client.delete_queue(QueueUrl=src_url)
            sqs_client.delete_queue(QueueUrl=tgt_url)

    def test_source_queue_drained_after_forwarding(
        self, pipes_client, sqs_client, unique_name
    ):
        """Test that source queue messages are deleted after forwarding."""
        src_name = f"pipe-src-{unique_name}"
        tgt_name = f"pipe-tgt-{unique_name}"
        pipe_name = f"pipe-{unique_name}"

        src_resp = sqs_client.create_queue(QueueName=src_name)
        src_url = src_resp["QueueUrl"]
        tgt_resp = sqs_client.create_queue(QueueName=tgt_name)
        tgt_url = tgt_resp["QueueUrl"]
        try:
            pipes_client.create_pipe(
                Name=pipe_name,
                Source=sqs_arn(src_name),
                Target=sqs_arn(tgt_name),
                RoleArn=ROLE_ARN,
                DesiredState="RUNNING",
            )

            sqs_client.send_message(QueueUrl=src_url, MessageBody="drain test")

            drained = False
            for _ in range(15):
                time.sleep(1)
                attrs = sqs_client.get_queue_attributes(
                    QueueUrl=src_url, AttributeNames=["ApproximateNumberOfMessages"]
                )
                if attrs["Attributes"]["ApproximateNumberOfMessages"] == "0":
                    drained = True
                    break

            assert drained, "Source queue should be drained after pipe forwards messages"
        finally:
            pipes_client.delete_pipe(Name=pipe_name)
            sqs_client.delete_queue(QueueUrl=src_url)
            sqs_client.delete_queue(QueueUrl=tgt_url)


def _queue_url(sqs_client, queue_name):
    """Get queue URL by name, ignoring errors."""
    try:
        return sqs_client.get_queue_url(QueueName=queue_name)["QueueUrl"]
    except Exception:
        return ""