様々なインフラストラクチャの描画ツールである 『Cartography』 を使ってみた

タダです.

AWS の IAM ロールとポリシーがどのような関係になっているかを描画するツールを調べた時に,Cartography を使う機会があったのでこの記事で試した内容をまとめていきます.

Cartography とは

Cartography はインフラストラクチャーとその関係を Neo4j データベースによるグラフ表示で集約するPythonツールです.サポートされているプラットフォームは次のとおりです.

• Amazon Web Services
• Google Cloud Platform
• Google GSuite
• Duo CRXcavator
• Jamf
• Okta
• Github
• DigitalOcean
• Microsoft Azure
• Kubernetes
• PagerDuty
• Crowdstrike Falcon
• NIST CVE

Cartography のセットアップ

セットアップはドキュメントを参考に進めました.Neo4j を動かすために JDK/JRE が 11以上のものを入れる必要があるのですが,オプションとして Amazon Coretto 11 を紹介されていたため,ドキュメントに沿って Amazon Corretto 11 を入れてみました.

$/usr/libexec/java_home --verbose
Matching Java Virtual Machines (1):
    11.0.17 (x86_64) "Amazon.com Inc." - "Amazon Corretto 11" /Library/Java/JavaVirtualMachines/amazon-corretto-11.jdk/Contents/Home
/Library/Java/JavaVirtualMachines/amazon-corretto-11.jdk/Contents/Home

次に,Neo4j のコミュニティエディションの 4.4 系のものを入れるのですが,この記事では 4.4.15 のものを入れたことにします.サーバーを起動するとローカル 7474番ポートで使えるようになりました.

$./bin/neo4j console
~中略~
Starting Neo4j.
2023-01-11 23:15:29.713+0000 INFO  Starting...
2023-01-11 23:15:30.371+0000 INFO  This instance is ServerId{ae9c3997} (ae9c3997-16b4-49ef-bc31-d9f9577e9795)
2023-01-11 23:15:31.915+0000 INFO  ======== Neo4j 4.4.15 ========
2023-01-11 23:15:33.978+0000 INFO  Performing postInitialization step for component 'security-users' with version 3 and status CURRENT
2023-01-11 23:15:33.979+0000 INFO  Updating the initial password in component 'security-users'
2023-01-11 23:16:00.295+0000 INFO  Bolt enabled on localhost:7687.
2023-01-11 23:16:01.387+0000 INFO  Remote interface available at http://localhost:7474/
2023-01-11 23:16:01.401+0000 INFO  id: 1A231D4D04203A4C6DA1D525A2297D3B807DFDCC9B40D94B25CC3631D57EE0CA
2023-01-11 23:16:01.402+0000 INFO  name: system
2023-01-11 23:16:01.402+0000 INFO  creationDate: 2022-12-21T05:43:28.235Z
2023-01-11 23:16:01.402+0000 INFO  Started.

そして,リポジトリをローカルに持ってきて,pip install cartography を実行して Cartographyをインストールします.なお,この記事ではバージョン3.9.6` で確認しています.インストール完了後,コマンドが実行できるようになります.

$cartography --help
usage: cartography [-h] [-v] [-q] [--neo4j-uri NEO4J_URI] [--neo4j-user NEO4J_USER] [--neo4j-password-env-var NEO4J_PASSWORD_ENV_VAR] [--neo4j-password-prompt]
                   [--neo4j-max-connection-lifetime NEO4J_MAX_CONNECTION_LIFETIME] [--neo4j-database NEO4J_DATABASE] [--update-tag UPDATE_TAG] [--aws-sync-all-profiles] [--aws-best-effort-mode]
                   [--oci-sync-all-profiles] [--azure-sync-all-subscriptions] [--azure-sp-auth] [--azure-tenant-id AZURE_TENANT_ID] [--azure-client-id AZURE_CLIENT_ID]
                   [--azure-client-secret-env-var AZURE_CLIENT_SECRET_ENV_VAR] [--aws-requested-syncs AWS_REQUESTED_SYNCS] [--crxcavator-api-base-uri CRXCAVATOR_API_BASE_URI]
                   [--crxcavator-api-key-env-var CRXCAVATOR_API_KEY_ENV_VAR] [--analysis-job-directory ANALYSIS_JOB_DIRECTORY] [--okta-org-id OKTA_ORG_ID] [--okta-api-key-env-var OKTA_API_KEY_ENV_VAR]
                   [--okta-saml-role-regex OKTA_SAML_ROLE_REGEX] [--github-config-env-var GITHUB_CONFIG_ENV_VAR] [--digitalocean-token-env-var DIGITALOCEAN_TOKEN_ENV_VAR]
                   [--permission-relationships-file PERMISSION_RELATIONSHIPS_FILE] [--jamf-base-uri JAMF_BASE_URI] [--jamf-user JAMF_USER] [--jamf-password-env-var JAMF_PASSWORD_ENV_VAR]
                   [--k8s-kubeconfig K8S_KUBECONFIG] [--nist-cve-url NIST_CVE_URL] [--cve-enabled] [--statsd-enabled] [--statsd-prefix STATSD_PREFIX] [--statsd-host STATSD_HOST]
                   [--statsd-port STATSD_PORT] [--pagerduty-api-key-env-var PAGERDUTY_API_KEY_ENV_VAR] [--pagerduty-request-timeout PAGERDUTY_REQUEST_TIMEOUT]
                   [--crowdstrike-client-id-env-var CROWDSTRIKE_CLIENT_ID_ENV_VAR] [--crowdstrike-client-secret-env-var CROWDSTRIKE_CLIENT_SECRET_ENV_VAR] [--crowdstrike-api-url CROWDSTRIKE_API_URL]

cartography consolidates infrastructure assets and the relationships between them in an intuitive graph view. This application can be used to pull configuration data from multiple sources, load it in to
Neo4j, and run arbitrary enrichment and analysis on that data. Please make sure you have Neo4j running and have configured AWS credentials with the SecurityAudit IAM policy before getting started.
Running cartography with no parameters will execute a simple sync against a Neo4j instance running locally. It will use your default AWS credentials and will not execute and post-sync analysis jobs.
Please see the per-parameter documentation below for information on how to connect to different Neo4j instances, use auth when communicating with Neo4j, sync data from multiple AWS accounts, and execute
arbitrary analysis jobs after the conclusion of the sync.

optional arguments:
  -h, --help            show this help message and exit
  -v, --verbose         Enable verbose logging for cartography.
  -q, --quiet           Restrict cartography logging to warnings and errors only.
  --neo4j-uri NEO4J_URI
                        A valid Neo4j URI to sync against. See https://neo4j.com/docs/api/python-driver/current/driver.html#uri for complete documentation on the structure of a Neo4j URI.
  --neo4j-user NEO4J_USER
                        A username with which to authenticate to Neo4j.
  --neo4j-password-env-var NEO4J_PASSWORD_ENV_VAR
                        The name of an environment variable containing a password with which to authenticate to Neo4j.
  --neo4j-password-prompt
                        Present an interactive prompt for a password with which to authenticate to Neo4j. This parameter supersedes other methods of supplying a Neo4j password.
  --neo4j-max-connection-lifetime NEO4J_MAX_CONNECTION_LIFETIME
                        Time in seconds for the Neo4j driver to consider a TCP connection alive. cartography default = 3600, which is the same as the Neo4j driver default. See
                        https://neo4j.com/docs/driver-manual/1.7/client-applications/#driver-config-connection-pool-management.
  --neo4j-database NEO4J_DATABASE
                        The name of the database in Neo4j to connect to. If not specified, uses the config settings of your Neo4j database itself to infer which database is set to default. See
                        https://neo4j.com/docs/api/python-driver/4.4/api.html#database.
  --update-tag UPDATE_TAG
                        A unique tag to apply to all Neo4j nodes and relationships created or updated during the sync run. This tag is used by cleanup jobs to identify nodes and relationships that are
                        stale and need to be removed from the graph. By default, cartography will use a UNIX timestamp as the update tag.
  --aws-sync-all-profiles
                        Enable AWS sync for all discovered named profiles. When this parameter is supplied cartography will discover all configured AWS named profiles (see
                        https://docs.aws.amazon.com/cli/latest/userguide/cli-configure-profiles.html) and run the AWS sync job for each profile not named "default". If this parameter is not supplied,
                        cartography will use the default AWS credentials available in your environment to run the AWS sync once. When using this parameter it is suggested that you create an AWS config
                        file containing a named profile for each AWS account you want to sync and use the AWS_CONFIG_FILE environment variable to point to that config file (see
                        https://docs.aws.amazon.com/cli/latest/userguide/cli-configure-envvars.html). cartography respects the AWS CLI/SDK environment variables and does not override them.
  --aws-best-effort-mode
                        Enable AWS sync best effort mode when syncing AWS accounts. This will allow cartography to continue syncing other accounts and delay raising an exception until the very end.
  --oci-sync-all-profiles
                        Enable OCI sync for all discovered named profiles. When this parameter is supplied cartography will discover all configured OCI named profiles (see https://docs.oracle.com/en-
                        us/iaas/Content/API/Concepts/sdkconfig.htm) and run the OCI sync job for each profile not named "DEFAULT". If this parameter is not supplied, cartography will use the default OCI
                        credentials available in your environment to run the OCI sync once.
  --azure-sync-all-subscriptions
                        Enable Azure sync for all discovered subscriptions. When this parameter is supplied cartography will discover all configured Azure subscriptions.
  --azure-sp-auth       Use Service Principal authentication for Azure sync.
  --azure-tenant-id AZURE_TENANT_ID
                        Azure Tenant Id for Service Principal Authentication.
  --azure-client-id AZURE_CLIENT_ID
                        Azure Client Id for Service Principal Authentication.
  --azure-client-secret-env-var AZURE_CLIENT_SECRET_ENV_VAR
                        The name of environment variable containing Azure Client Secret for Service Principal Authentication.
  --aws-requested-syncs AWS_REQUESTED_SYNCS
                        Comma-separated list of AWS resources to sync. Example 1: "ecr,s3,ec2:instance" for ECR, S3, and all EC2 instance resources. See the full list available in source code at
                        cartography.intel.aws.resources. If not specified, cartography by default will run all AWS sync modules available.
  --crxcavator-api-base-uri CRXCAVATOR_API_BASE_URI
                        Base URI for the CRXcavator API. Defaults to public API endpoint.
  --crxcavator-api-key-env-var CRXCAVATOR_API_KEY_ENV_VAR
                        The name of an environment variable containing a key with which to auth to the CRXcavator API. Required if you are using the CRXcavator intel module. Ignored otherwise.
  --analysis-job-directory ANALYSIS_JOB_DIRECTORY
                        A path to a directory containing analysis jobs to run at the conclusion of the sync. cartography will discover all JSON files in the given directory (and its subdirectories) and
                        pass them to the GraphJob API to execute against the graph. This allows you to apply data transformation and augmentation at the end of a sync run without writing code.
                        cartography does not guarantee the order in which the jobs are executed.
  --okta-org-id OKTA_ORG_ID
                        Okta organizational id to sync. Required if you are using the Okta intel module. Ignored otherwise.
  --okta-api-key-env-var OKTA_API_KEY_ENV_VAR
                        The name of an environment variable containing a key with which to auth to the Okta API.Required if you are using the Okta intel module. Ignored otherwise.
  --okta-saml-role-regex OKTA_SAML_ROLE_REGEX
                        The regex used to map Okta groups to AWS roles when using okta as a SAML provider.The regex is the one entered in Step 5: Enabling Group Based Role Mapping in Oktahttps://saml-
                        doc.okta.com/SAML_Docs/How-to-Configure-SAML-2.0-for-Amazon-Web-Service#c-step5The regex must contain the {{role}} and {{accountid}} tags
  --github-config-env-var GITHUB_CONFIG_ENV_VAR
                        The name of an environment variable containing a Base64 encoded GitHub config object.Required if you are using the GitHub intel module. Ignored otherwise.
  --digitalocean-token-env-var DIGITALOCEAN_TOKEN_ENV_VAR
                        The name of an environment variable containing a DigitalOcean access token.Required if you are using the DigitalOcean intel module. Ignored otherwise.
  --permission-relationships-file PERMISSION_RELATIONSHIPS_FILE
                        The path to the permission relationships mapping file.If omitted the default permission relationships will be created
  --jamf-base-uri JAMF_BASE_URI
                        Your Jamf base URI, e.g. https://hostname.com/JSSResource.Required if you are using the Jamf intel module. Ignored otherwise.
  --jamf-user JAMF_USER
                        A username with which to authenticate to Jamf.
  --jamf-password-env-var JAMF_PASSWORD_ENV_VAR
                        The name of an environment variable containing a password with which to authenticate to Jamf.
  --k8s-kubeconfig K8S_KUBECONFIG
                        The path to kubeconfig file specifying context to access K8s cluster(s).
  --nist-cve-url NIST_CVE_URL
                        The base url for the NIST CVE data. Default = https://nvd.nist.gov/feeds/json/cve/1.1
  --cve-enabled         If set, CVE data will be synced from NIST.
  --statsd-enabled      If set, enables sending metrics using statsd to a server of your choice.
  --statsd-prefix STATSD_PREFIX
                        The string to prefix statsd metrics with. Only used if --statsd-enabled is on. Default = empty string.
  --statsd-host STATSD_HOST
                        The IP address of your statsd server. Only used if --statsd-enabled is on. Default = 127.0.0.1.
  --statsd-port STATSD_PORT
                        The port of your statsd server. Only used if --statsd-enabled is on. Default = UDP 8125.
  --pagerduty-api-key-env-var PAGERDUTY_API_KEY_ENV_VAR
                        The name of environment variable containing the pagerduty API key for authentication.
  --pagerduty-request-timeout PAGERDUTY_REQUEST_TIMEOUT
                        Seconds to timeout for pagerduty API sessions.
  --crowdstrike-client-id-env-var CROWDSTRIKE_CLIENT_ID_ENV_VAR
                        The name of environment variable containing the crowdstrike client id for authentication.
  --crowdstrike-client-secret-env-var CROWDSTRIKE_CLIENT_SECRET_ENV_VAR
                        The name of environment variable containing the crowdstrike secret key for authentication.
  --crowdstrike-api-url CROWDSTRIKE_API_URL
                        The crowdstrike URL, if using self-hosted. Defaults to the public crowdstrike API URL otherwise.

For more documentation please visit: https://github.com/lyft/cartography

Cartography を使って AWS の関連図を描画する

Neo4j のローカルサーバーを起動した状態で,Cartography のコマンドを実行していくのですが,Neo4j の初回起動ユーザーとパスワードをパラメータにセットして default クレデンシャルでセットされた AWS アカウントの構成情報を収集して Neo4j サーバーに投入してくれます.

$ cartography --neo4j-user neo4j --neo4j-password-prompt --neo4j-uri  bolt://localhost:7687

次に,収集したデータに対してクエリを実行してみます.IAM ポリシーでワイルドカードを使っている IAMロールの描画をしてみます.

MATCH (a:AWSPrincipal)-->(p:AWSPolicy)-[:STATEMENT]->(s)
WHERE s.resource = ["*"]
RETURN a, p, s

IAMロールに紐付くポリシーとポリシーの中でワイルドカードが指定されている権限が表示されました.

関連情報

lyft.github.io

まとめ

Cartography を使ってみたので,インストールと試しにクエリを叩いて描画してみました.IAM のような大量のリソースが作られてて関連図がわかり易くするためのツールとして良さそうでした.

CloudWatch Events の定数を使って Lambda の処理を分岐させる

タダです.

定期イベントを CloudWatch Events で設定し,定型処理を実行する Lambda をキックすることはよくあることだと思います.今回 CloudWatch Events の定数を使って Lambda の処理を分岐させるのをやってみたのでこの記事にまとめていきます.今回は ECS タスクの数を調整する処理を作ってみました.

ECS のタスク数を調整する Lambda

Lambda のコードは以下のようなコードを書いてます.CloudWatch Events から desired_count をパラメーターとして渡し,desired_countが0ならコンテナをなくして,desired_countを1ならコンテナを1台起動させます.

import json
import boto3
import logging
from botocore.exceptions import ClientError

logger = logging.getLogger()
logger.setLevel(logging.INFO)

def lambda_handler(event, context):
    logger.info("Update ECS Task Count")
    CLUSTER_NAME = 'cluster name'
    SERVICE_NAME =  'service name'
    desired_count = event['desired_count'] 
    try:
        client = boto3.client('ecs')
        
        if desired_count == 0:
            result = client.update_service(
                    cluster = 'cluster name',
                    service = 'service name',
                    desiredCount = desired_count
                )
            return "Stop Complete."
            
        elif desired_count == 1:
            result = client.update_service(
                    cluster = 'cluster name',
                    service = 'service name',
                    desiredCount = desired_count
                )
            return "Start Complete."
            
            
    except ClientError as e:
        logger.exception("Exception.")

CloudWatch Events の定数

Lambda のコードで設定している desired_count を定数で定義していきます.イベントのターゲットを指定する箇所で定数を {"desired_count": 0(1)}と設定します.

コンテナを0台にするイベント

コンテナを1台にするイベント

関連情報

docs.aws.amazon.com

動作テスト

Lambda のテストイベントで次のような設定をして実行してみます.下記のテストでコンテナを起動できるか確認します.1台のコンテナ起動できました.

{
  "desired_count": 1
}

次にコンテナをなくしてみます.意図通りコンテナを消せました.

{
  "desired_count": 0
}

まとめ

CloudWatch Events の定数を使って ECS タスク数を Lambda 側で処理を分岐させました.この機能を知るまでは Lambda を処理ごとに作ってしまっていたので,管理する関数の数が少なくなるので良かったです.

redash のデータソースで Python を使ってみる

タダです.

redash のデータソースとして Python を使ってみる機会があったので,この記事にまとめていきます.なお,redash のバージョンは 8.0.0+b32245 で確認しています.

Python をデータソースに追加

デフォルトでは,データソースに Python がないため追加の設定が必要です.自分の確認環境は EC2 の AMI を使っているため /opt/redash/env に下記の記述を追記後,docker-compse up -d で起動し直せばデータソースに Python が追加されます.

REDASH_ADDITIONAL_QUERY_RUNNERS=redash.query_runner.python

追加されたデータソース画面

データソース追加後のコードを書いてみる

別々のクエリの結果をテーブル表示したいみたいな場合は下記のようなコードで表示できました.この場合は ALB のログの中から0.5秒以上のログと全件のログを出して表示させるみたいなことができます.

queryResults1 = execute_query('データソース名',"SELECT COUNT(*) FROM alb_logs WHERE url LIKE '%/hoge/hoge?%' target_processing_time >= 0.5 AND target_status_code != '-'")
queryResults2 = execute_query('データソース名',"SELECT COUNT(*) FROM alb_logs WHERE url LIKE '%/hoge/hoge?%' AND target_status_code != '-'")

results = {}
for rows1 in queryResults1['rows']:
    for rows2 in queryResults2['rows']:
        add_result_row(result, {
            'over': rows1['_col0'],
            'all': rows2['_col0']
        })
    break

add_result_column(result, 'all', '', 'string')
add_result_column(result, 'over', '', 'string')

クエリ結果

また, 上記のクエリを複数保存していたのを合体して表にさせることも下記のようなクエリでできます.

queryResults1 = get_query_result(redash のクエリ ID)
queryResults2 = get_query_result(redash のクエリ ID)

results = {}
for rows1 in queryResults['rows']:
    add_result_row(result, {
        'over': rows1['over'],
        'all': rows1['all'] 
    })

for rows2 in queryResults2['rows']:
    add_result_row(result, {
        'over': rows2['over'],
        'all': rows2['all'] 
    })

add_result_column(result, 'all', '', 'string')
add_result_column(result, 'over', '', 'string')

クエリ結果

まとめ

redash のデータソースとして Python を使うことがあったのでまとめました.SQL だけでなく Python のコードを書いてデータを集計したりできるのは redash の特徴だと思いますので,今後も適宜使っていきます!

Lambda を使って Elasticsearch のエイリアスとインデックスを更新する

タダです.

Amazon OpenSearch Service のインデックス再設計に関わった際に,次の運用上の懸念がありました.

• インデックスは月ごとに生成し,生成したインデックスに対してアプリケーションから見た時のエイリアスとして参照用と書き込み用の2つを貼りたい
• 前月以前のインデックスにも参照用エイリアスを残す形で運用したい

上記の運用を実現するために Lambda で上記の処理を行えないか検証したためその内容をこの記事にまとめます.

Lambda の権限と実行コード

Lambda の権限

Lambda の権限は検証として下記の IAM を付与しています.

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": "logs:CreateLogGroup",
            "Resource": "arn:aws:logs:ap-northeast-1:xxx:*"
        },
        {
            "Effect": "Allow",
            "Action": [
                "logs:CreateLogStream",
                "logs:PutLogEvents"
            ],
            "Resource": [
                "arn:aws:logs:ap-northeast-1:xxx:log-group:/aws/lambda/xxx:*"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "es:ESHttpPost",
                "es:ESHttpGet",
                "es:ESHttpPut",
                "es:ESHttpDelete"
            ],
            "Resource": "*"
        }
    ]
}

実行コード

コードは Python 3.9 で確認しています.

import boto3
import os
from requests_aws4auth import AWS4Auth
from elasticsearch import Elasticsearch, RequestsHttpConnection
from urllib.request import Request, urlopen
from datetime import datetime, timedelta, timezone
from dateutil.relativedelta import relativedelta

ENDPOINT = os.environ['ES_ENDPOINT']
REGION = os.environ['ES_REGION']
SERVICE = 'es'
CREDINTIAL = boto3.Session().get_credentials()
AWSAUTH = AWS4Auth(CREDINTIAL.access_key, CREDINTIAL.secret_key, REGION, SERVICE, session_token=CREDINTIAL.token)
JST = timezone(timedelta(hours=+9), 'JST')

def lambda_handler(event, context):
    client = Elasticsearch(
        hosts = [{'host': ENDPOINT, 'port': 443}],
        http_auth = AWSAUTH,
        use_ssl = True,
        verify_certs = True,
        connection_class = RequestsHttpConnection
    )

    write_alias_name = "test-writer-alias"
    read_alias_name = "test-reader-alias"
    index_prefix = "test-index-"
    new_index = index_prefix + datetime.now(JST).strftime('%Y%m')
    old_index = index_prefix + datetime.strftime(datetime.today() - relativedelta( months = 1 ), '%Y%m')
   
    client.indices.update_aliases(
            body={
                "actions": [
                    {"remove": {"alias": write_alias_name, "index": old_index}},
                    {"add": {"alias": write_alias_name, "index": new_index}},
                    {"add": {"alias": read_alias_name, "index": new_index}},
                ]
            }
        )

Lambda の実行結果確認

Lambda 実行後の挙動を確認していきます.実行前に予め前月のインデックス test-index-202110 とエイリアスとして test-reader-ailias(参照用エイリアス)と test-writer-alias(書き込み用エイリアス)を作った状態で確認しています.本記事は2021年の11月に書いているので,インデックスとして test-index-202111 とエイリアスが書き込み,参照が貼られ,前月にも参照用エイリアスが残った状態にできていたら期待通りです.

Lambda実行前

% curl -XGET 'https://xxx.ap-northeast-1.es.amazonaws.com/_alias?pretty=true'
{
  ".kibana_1" : {
    "aliases" : {
      ".kibana" : { }
    }
  },
  "test-index-202110" : {
    "aliases" : {
      "test-reader-alias" : { }, 
      "test-writer-alias" : { }  
    }
  }
}

Lambda を実行してみたところ期待通りの結果が得られました.

Lambda実行後

% curl -XGET 'https://xxx.ap-northeast-1.es.amazonaws.com/_alias?pretty=true'
{
  ".kibana_1" : {
    "aliases" : {
      ".kibana" : { }
    }
  },
  "test-index-202110" : {
    "aliases" : {
      "test-reader-alias" : { }
    }
  },
  "test-index-202111" : {
    "aliases" : {
      "test-reader-alias" : { },
      "test-writer-alias" : { }
    }
  }

まとめ

Elasticsearch のインデックスとエイリアスの運用上の懸念に対して Lambda を使ったプローチで検証したのをまとめました.次は,インデックス再設計するためのインデックス移行について書いていければと思います.

関連記事

sadayoshi-tada.hatenablog.com

pytest と moto でモックを作ったテストを行う

タダです.

以前から気になっていた moto というツールを試す機会があったので,この記事で導入と pytest と組み合わせたテストをチュートリアルの内容を通して見ていきたいと思います.

moto って？

moto とはテストで AWS サービスをモックで作っていくための Python ライブラリです.

github.com

導入はpipでさくっとできます.なお,motoは全ての AWS サービスを呼び出す関数を網羅しているわけではなさそうなので,その点は注意です.

$ pip install moto

moto が対応しているサービス一覧 github.com

moto でモックを作り pytest でテストする

それでは, moto を使ってみようと思います.

$ pipenv install boto3
$ pipenv install --dev pytest moto

以下のようなディレクトリ切って,app.pyとそのテストをするためのtest.pyを作りました.

.
├── app
│   ├── app.py
│   └── __init__.py
└── tests
    ├── __init__.py
    └── test.py

チュートリアルとして S3 に put_object を実行する save 関数があるコードを用意しました.

import boto3

class MyModel(object):
    def __init__(self, name, value):
        self.name = name
        self.value = value

    def save(self):
        s3 = boto3.client('s3', region_name='ap-northeast-1')
        s3.put_object(Bucket='mybucket', Key=self.name, Body=self.value)

テスト用コードとしては次のものを用意しました.@mock_s3を記述することで S3 へのアクセスをモックしてくれてます.save関数を呼んで,オブジェクトを読みファイルの中身がis awesomeかを確認するようになっています.

import boto3
from moto import mock_s3
from app.app import MyModel

@mock_s3
def test_my_model_save():
    conn = boto3.resource('s3', region_name='ap-northeast-1')
    conn.create_bucket(Bucket='mybucket',CreateBucketConfiguration={
        'LocationConstraint':'ap-northeast-1'
    })

    model_instance = MyModel('steve', 'is awesome')
    model_instance.save()

    body = conn.Object('mybucket', 'steve').get()[
        'Body'].read().decode("utf-8")

    assert body == 'is awesome'

テストコードを走らせてみたところテストをパスしたのを確認できました.

$ pipenv run pytest tests/test.py 
=============================================================================== test session starts ===============================================================================
platform linux -- Python 3.7.9, pytest-6.2.3, py-1.10.0, pluggy-0.13.1
rootdir: /xxxx/xxxx/environment
collected 1 item                                                                                                                                                                  

tests/test.py .                                                                                                                                                             [100%]

================================================================================ warnings summary =================================================================================
../.local/share/virtualenvs/environment-QZ1wNgYc/lib64/python3.7/distutils/__init__.py:1
  /xxxx/xxxx/.local/share/virtualenvs/environment-QZ1wNgYc/lib64/python3.7/distutils/__init__.py:1: DeprecationWarning: the imp module is deprecated in favour of importlib; see the module's documentation for alternative uses
    import imp

-- Docs: https://docs.pytest.org/en/stable/warnings.html
========================================================================== 1 passed, 1 warning in 0.69s ===========================================================================

まとめ

簡単になりますが,motoを使ったテストコードを書くことをやってみました.AWS リソースを作ってまで確認するとコストがかかって確認が遅くなってしまうような状況ならこう言ったツールを作って素早く確認して行けると良さそうです.