Zenn
Open4

AWS Lambda

nabetsunabetsu

PythonでSQSのアプリ作成

プロジェクトの作成

sam initでテンプレートを作成する。

sam init

SQSの定義を追加。

Resources:
  HelloWorldFunction:
    Type: AWS::Serverless::Function # More info about Function Resource: https://github.com/awslabs/serverless-application-model/blob/master/versions/2016-10-31.md#awsserverlessfunction
    Properties:
      PackageType: Image
      Events:
        MySQSEvent:
          Type: SQS
          Properties:
            Queue: !GetAtt MySqsQueue.Arn
            BatchSize: 1
    Metadata:
      Dockerfile: Dockerfile
      DockerContext: ./make_queue
      DockerTag: python3.8-v1

  MySqsQueue:
    Type: AWS::SQS::Queue

ローカル稼働

SAMのgenerate-event機能を利用してテスト用のEvent情報を作成する。

% sam local generate-event sqs receive-message
{
  "Records": [
    {
      "messageId": "19dd0b57-b21e-4ac1-bd88-01bbb068cb78",
      "receiptHandle": "MessageReceiptHandle",
      "body": "Hello from SQS!",
      "attributes": {
        "ApproximateReceiveCount": "1",
        "SentTimestamp": "1523232000000",
        "SenderId": "123456789012",
        "ApproximateFirstReceiveTimestamp": "1523232000001"
      },
      "messageAttributes": {},
      "md5OfBody": "7b270e59b47ff90a553787216d55d91d",
      "eventSource": "aws:sqs",
      "eventSourceARN": "arn:aws:sqs:us-east-1:123456789012:MyQueue",
      "awsRegion": "us-east-1"
    }
  ]
}

あとはsam local invoke機能を使ってLambdaのローカル稼働を実施する。

-eオプションで指定したEvent情報のファイルをインプットとして稼働ができる。

% sam local invoke HelloWorldFunction -e ./events/sqs.json
Invoking Container created from helloworldfunction:python3.8-v1
Building image..........
Skip pulling image and use local one: helloworldfunction:rapid-1.19.1.

START RequestId: d830ac67-0913-4a78-bb40-4de79086c077 Version: $LATEST
END RequestId: d830ac67-0913-4a78-bb40-4de79086c077
REPORT RequestId: d830ac67-0913-4a78-bb40-4de79086c077  Init Duration: 0.13 ms  Duration: 125.69 ms     Billed Duration: 200 ms Memory Size: 128 MB     Max Memory Used: 128 MB
{"statusCode": 200, "body": "{\"messages\": [\"Hello from SQS!\"], \"input\": {\"Records\": [{\"messageId\": \"19dd0b57-b21e-4ac1-bd88-01bbb068cb78\", \"receiptHandle\": \"MessageReceiptHandle\", \"body\": \"Hello from SQS!\", \"attributes\": {\"ApproximateReceiveCount\": \"1\", \"SentTimestamp\": \"1523232000000\", \"SenderId\": \"123456789012\", \"ApproximateFirstReceiveTimestamp\": \"1523232000001\"}, \"messageAttributes\": {}, \"md5OfBody\": \"7b270e59b47ff90a553787216d55d91d\", \"eventSource\": \"aws:sqs\", \"eventSourceARN\": \"arn:aws:sqs:us-east-1:123456789012:MyQueue\", \"awsRegion\": \"us-east-1\"}]}}"}

参考資料

nabetsunabetsu

Lambdaにテストの追加

importの問題

以下のディレクトリ構成の時

.
├── .aws-sam
│   └── build.toml
├── .gitignore
├── README.md
├── __init__.py
├── events
│   ├── event.json
│   └── sqs.json
├── poetry.lock
├── pyproject.toml
├── src
│   ├── beer
│   │   ├── Dockerfile
│   │   ├── __init__.py
│   │   └── app.py
│   └── twitter_to_db
│       ├── Dockerfile
│       ├── __init__.py
│       ├── app.py
│       └── requirements.txt
├── template.yaml
└── tests
    ├── __init__.py
    └── unit
        ├── __init__.py
        ├── _test_handler.py
        ├── get_beer_data
        │   └── __init__.py
        └── test_get_beer_data.py

以下の通りpython -mをつけるとテスト対象のモジュールのimportが行える。
python -m をつけるとルートディレクトリの設定が変わる？

# pytestを直接実行した時
$ pytest tests
...
E   ModuleNotFoundError: No module named 'src'
================================================================== 1 error in 0.05s ==================================================================

# python -mをつけた時
$ python -m pytest t
...
E   ModuleNotFoundError: No module named 'boto3'
============================================================== short test summary info ===============================================================
ERROR tests/unit/test_get_beer_data.py
!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! Interrupted: 1 error during collection !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
================================================================== 1 error in 0.06s ==================================================================

この方法もある

Testing AWS CDK and Lambda Functions with pytest

参考資料

nabetsunabetsu

テスト全般

  • Unit、Integration
    • コード単位
  • Function単位
    • sam local start-lambdaを使ってFunctionを起動する

Lambda自体を呼び出す

レスポンスの結果から以下のようにしてLambdaからReturnされた値にアクセスできる。

response = boto3.client('lambda').invoke(
        FunctionName='test0820',
        InvocationType='RequestResponse',
        Payload=Payload
    )
body = json.loads(response['Payload'].read())

参考資料