اجرای آزمایش خودکار پایان به پایان: استفاده از برنامه CI/CD خود با برنامه Cloud Development Clit (CDK)

from aws_cdk import aws_iam as iam, CfnOutput
from constructs import Construct

class IAMRoleStack(Stack):
    """Create the IAM deployment."""

    def __init__(
        self,
        scope: Construct,
        construct_id: str,
        **kwargs,
    ) -> None:
        """Create the IAM deployment.

        Args:
            scope (Construct): CDK App scope
        """
        role = iam.Role(
            self,
            id="TestIAMRole",
            assumed_by=iam.ServicePrincipal("ec2.amazonaws.com"),
            description="IAM Role for E2E testing",
        )

        role.add_to_policy(
            iam.PolicyStatement(
                actions=[
                    "states:ListExecutions",
                    "states:DescribeExecution",
                ],
                resources=["*"],
            )
        )

        CfnOutput(
            self.scope,
            id="RoleArn",
            value=role.role_arn,
            export_name="E2eTestIAMRole",
            description="ARN of the created IAM Role",
        )

export AWS_DEFAULT_REGION=${{ parameters.AwsRegion }}
KST=$( aws sts assume-role --endpoint-url https://sts.${{ parameters.AwsRegion }}.amazonaws.com --role-arn arn:aws:iam::${{ parameters.AccountID }}:role/${{ parameters.CDKDeploymentRole }}-${{ parameters.AccountID }}-${{ parameters.AwsRegion }} --role-session-name $(Build.SourceVersion) --duration-seconds 3600)
export AWS_ACCESS_KEY_ID=$(echo $KST | jq -r .Credentials.AccessKeyId)
export AWS_SECRET_ACCESS_KEY=$(echo $KST | jq -r .Credentials.SecretAccessKey)
export AWS_SESSION_TOKEN=$(echo $KST | jq -r .Credentials.SessionToken)
export EMAIL_PROCESSOR=$(aws cloudformation describe-stacks --stack-name mrht-developer-posteingang-stack --query "Stacks[0].Outputs[?ExportName=='ExportEmailProcessor'].OutputValue" --output text)
export STATE_MACHINE_ARN=$(aws cloudformation describe-stacks --stack-name mrht-developer-posteingang-stack --query "Stacks[0].Outputs[?ExportName=='ExportStateMachine'].OutputValue" --output text)
export IAM_ROLE_ARN=$(aws cloudformation describe-stacks --stack-name mrht-developer-posteingang-stack --query "Stacks[0].Outputs[?ExportName=='E2eTestIAMRole'].OutputValue" --output text)

KST=$( aws sts assume-role --endpoint-url https://sts.${{ parameters.AwsRegion }}.amazonaws.com --role-arn $IAM_ROLE_ARN --role-session-name $(Build.SourceVersion) --duration-seconds 3600)
unset AWS_ACCESS_KEY_ID AWS_SECRET_ACCESS_KEY AWS_SESSION_TOKEN
export AWS_ACCESS_KEY_ID=$(echo $KST | jq -r .Credentials.AccessKeyId)
export AWS_SECRET_ACCESS_KEY=$(echo $KST | jq -r .Credentials.SecretAccessKey)
export AWS_SESSION_TOKEN=$(echo $KST | jq -r .Credentials.SessionToken)
python3 src/e2e_test/e2e_test.py

import os
import time
import logging
import boto3
from botocore.exceptions import ClientError, ParamValidationError
import smbclient
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.mime.application import MIMEApplication

logging.basicConfig(
    level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

class SMBClient:
    """Manage an SMB connection to access shared files."""

    def __init__(self, username: str, password: str, servername: str):
        """Initialize the SMBClient with connection details.

        Args:
            username (str): The username for the SMB connection.
            password (str): The password for the SMB connection.
            servername (str): The hostname or IP address of the SMB server.
        """
        self.username = username
        self.password = password
        self.port = 445
        self.host_name = servername

    def connect_smb(self, smb_client: smbclient) -> None:
        """Connect to the SMB server.

        Args:
            smb_client (smbclient): The SMB client instance.
        """
        logger.info(f"Trying to connect to {self.host_name} as {self.username}.")
        try:
            smb_client.register_session(
                server=self.host_name, username=self.username, password=self.password
            )
            logger.info("Session established successfully.")
        except Exception as e:
            logger.exception(f"Failed to connect to SMB share: {e}")
            raise
        return True

    def list_directory(self, path: str, smb_client: smbclient) -> None:
        """List the contents of an SMB directory.

        Args:
            path (str): The path to the directory.
            smb_client (smbclient): The SMB client instance.

        Returns:
            List[str]: A list of the directory contents, or an empty list if an error occurs.
        """
        entries = []
        try:
            for entry in smb_client.listdir(path):
                logger.info(f"Found entry: {entry}")
                entries.append(entry)
            return entries
        except Exception as e:
            logger.exception(f"Error listing directory: {e}")
            return []

def send_email(
    sender: str,
    recipient: str,
    subject: str,
    body_path: str,
    attachment_path: str,
    smtp_server: str,
    smtp_port: int,
    smtp_username: str,
    smtp_password: str,
):
    """Send an email with a PDF attachment using SMTP.

    Args:
        sender (str): The email address of the sender.
        recipient (str): The email address of the recipient.
        subject (str): The subject of the email.
        body_path (str): The path to the text file containing the body of the email.
        attachment_path (str): The path to the PDF file to be attached.
        smtp_server (str): The SMTP server address.
        smtp_port (int): The SMTP server port.
        smtp_username (str): The SMTP username.
        smtp_password (str): The SMTP password.
    """
    msg = MIMEMultipart()
    msg["Subject"] = subject
    msg["From"] = sender
    msg["To"] = recipient

    with open(body_path, "r", encoding="utf-8") as text_file:
        text = MIMEText(text_file.read())
        msg.attach(text)

    with open(attachment_path, "rb") as pdf_file:
        attachment = MIMEApplication(pdf_file.read(), _subtype="pdf")
        attachment.add_header(
            "Content-Disposition", "attachment", filename="attachment.pdf"
        )
        msg.attach(attachment)

    try:
        with smtplib.SMTP(smtp_server, smtp_port) as s:
            s.starttls()
            s.login(smtp_username, smtp_password)
            s.send_message(msg)
            logger.info("Email sent successfully.")
    except Exception as e:
        logger.error(f"Error sending email: {str(e)}")
        raise


def main():
    """Run E2E test."""
    try:
        state_machine_arn = os.getenv("STATE_MACHINE_ARN")
        stage = os.getenv("STAGE")
        smb_user = os.getenv("SMB_USERNAME")
        smb_password = os.getenv("SMB_PASSWORD")
        smb_host_ip = os.getenv("SMB_SERVER")
        smb_path = f"{smb_host_ip}\\folder$\\folder\\"
        smtp_server = os.getenv("SMTP_SERVER")
        smtp_port = int(os.getenv("SMTP_PORT"))
        smtp_username = os.getenv("SMTP_USERNAME")
        smtp_password = os.getenv("SMTP_PASSWORD")
    except KeyError as e:
        logger.exception(e)
        raise

    try:
        sender = "scan2mail@example.com"
        recipient = "scan2mail@example.com"
        subject = "Test Email with Attachment"
        body_path = "./src/e2e_test/email_body.txt"
        attachment_path = "./src/e2e_test/test.pdf"

        send_email(
            sender=sender,
            recipient=recipient,
            subject=subject,
            body_path=body_path,
            attachment_path=attachment_path,
            smtp_server=smtp_server,
            smtp_port=smtp_port,
            smtp_username=smtp_username,
            smtp_password=smtp_password,
        )

        time.sleep(15)

        stepfunctions_client = boto3.client("stepfunctions")

        list_executions_response = stepfunctions_client.list_executions(
            stateMachineArn=state_machine_arn, maxResults=1, statusFilter="RUNNING"
        )

        if list_executions_response["executions"]:
            execution_arn = list_executions_response["executions"][0]["executionArn"]
            logger.info(f"Found execution: {execution_arn}")

            while True:
                response = stepfunctions_client.describe_execution(
                    executionArn=execution_arn
                )
                status = response["status"]

                if status != "RUNNING":
                    logger.info(f"Step Function execution status: {status}")
                    break

                logger.info(
                    "Step Function is still running. Checking again in 20 seconds..."
                )
                time.sleep(20)
        else:
            logger.warning("No running execution found.")
            raise

    except (ClientError, ParamValidationError) as e:
        logger.error(f"An error occurred: {str(e)}")
        raise

    counter = 0
    try:
        smb = SMBClient(
            username=smb_user,
            password=smb_password,
            servername=smb_host_ip,
        )
        smb.connect_smb(smbclient)
        response = smb.list_directory(smb_path, smbclient)
        desired_contract_id = "__102865946013"
        for file_name in response:
            if desired_contract_id in file_name:
                logger.info(f"Found the ID '{desired_contract_id}' in the directory contents.")
                counter += 1
    except Exception as e:
        logger.exception(e)
        raise

    if counter == 0:
        raise ValueError("Target ID not found")

    logger.info("Finished E2E test")

if __name__ == "__main__":
    try:
        main()
    except Exception as e:
        logger.error(f"An error occurred: {str(e)}")
        raise