こんにちは。ノバセルでエンジニアをしている新開です。
この記事は ノバセル Advent Calendar 12日目です。
はじめに
QuickSightはAWSの強力なBIツールであり、データの可視化を効率的に行うことができます。 QuickSightを使ってダッシュボードを作成し、データ分析を効率化する事例が増えています。
しかしダッシュボードの変更が正しく反映されるかを手作業で確認するのは非効率です。 本記事では、この課題を解決するために継続的インテグレーション (CI) によって変更点の自動テストを導入するテンプレートを解説します。
自動テストの必要性
QuickSightのダッシュボードは視覚的なインターフェースで設計されることが多く、目視による手動のテストを行なっているパターンが多いです。 このパターンには以下の課題があると考えています。
- 非効率性: 変更のたびに手動で確認を行うとミスを誘発する可能性がある
- 信頼性: 意図しない影響や破壊的な変更を早期に検出する必要がある
- 迅速性: 目視による手動のテストプロセスにかかる時間を減らしたい
CIによる自動テストを導入することでこれらの課題を解決し、ダッシュボード変更の品質担保と効率向上が期待できます。
方針
QuickSightダッシュボードの変更を検知し、自動テストを実行するための全体的な実装方針を説明します。
1. QuickSightの変更検知
AWS Lambdaを使用してQuickSight APIからダッシュボードの変更を監視します。 変更が検知されるとCIパイプラインがトリガーされます。
2. CIパイプラインの設定
GitHub Actionsを使用してテストスクリプトを実行します。 GitHub Actionsは多くのプロジェクトにおいて無料枠内で運用可能であり、追加のインフラコストがかかりません。
3. テストの実行
QuickSightダッシュボードの変更を検証する際に使用できるテストケースの例を実装します。
実装
1. QuickSightの変更を検知してGitHub Actionsをトリガーする
AWS Lambdaを使用して変更を検知し、GitHub Actionsをトリガーしてテストを実行します。
import os import json import requests import boto3 from datetime import datetime def lambda_handler(event, context): quicksight = boto3.client('quicksight') response = quicksight.list_dashboards(AwsAccountId='<アカウントID>') dashboards = response['DashboardSummaryList'] # 前回の確認時刻を取得 env_value = os.environ.get('LAST_CHECKED_TIME') last_checked_time = datetime.fromisoformat(env_value) if env_value else datetime.min # 環境変数に最新のチェック時刻を保存 update_last_checked_time(datetime.now().isoformat()) for dashboard in dashboards: if dashboard['LastUpdatedTime'] > last_checked_time: # GitHub ActionsのWebhookをトリガー requests.post( url='https://api.github.com/repos/<owner>/<repo>/dispatches', headers={ 'Authorization': 'token <GitHubトークン>', 'Accept': 'application/vnd.github.everest-preview+json' }, json={ "event_type": "quicksight_change", "dashboard_id": dashboard['DashboardId']} ) return { 'statusCode': 200, 'body': json.dumps('Webhook triggered') }
今回、変更検知としてはQuickSight APIを活用しました。
ダッシュボードのLastUpdatedTimeが前回のテスト実行時間よりも後であれば、そのダッシュボードのテストを行います。
前回のテスト実行時間LAST_CHECKED_TIMEはLambdaの環境変数に格納します。Lambdaではファイルへの書き込みができないためです。（厳密には可能だが、/tmpディレクトリで一時領域になるため長期間のデータ保存や頻繁な書き込みには不適）
Lambdaの環境変数に書き込む場合は下記の関数が適切でしょう。
import boto3 import os import json from datetime import datetime def update_last_checked_time(last_checked_time): client = boto3.client('lambda') response = client.get_function_configuration(FunctionName='<Lambda関数名>') # Lambda関数の環境変数を更新 client.update_function_configuration( FunctionName=function_name, Environment={ 'LAST_CHECKED_TIME': last_checked_time } )
またAWS EventBridgeでQuickSightダッシュボードの変更を直接検知するネイティブな仕組みは現在導入されていません。 そのためEventBridgeのスケジュールルールを使用して、一定間隔でLambda関数をトリガーし、QuickSight APIを呼び出して変更をチェックする仕組みを構築すると良いでしょう。
2. GitHub Actionsの設定
変更検知後にテストを実行するパイプラインを構築します。
name: Test QuickSight Changes on: repository_dispatch: types: - quicksight_change jobs: test: runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - name: Set up Python uses: actions/setup-python@v5 with: python-version: '3.12' - name: Install dependencies run: | pip install boto3 pytest - name: Run tests env: DASHBOARD_ID: ${{ github.event.client_payload.dashboard_id }} run: pytest test_quicksight.py --dashboard-id $DASHBOARD_ID
Python3.12はGitHub Actionsとの互換性があり、boto3やpytestなどの主要ライブラリが安定して動作します。
テストスクリプトにダッシュボードIDを引数として与えることで、 ダッシュボードごとにテスト内容を入れ替えるように対処しました。
3. テストスクリプト
import boto3 import time client = boto3.client('quicksight') def describe_dashboard(dashboard_id): return client.describe_dashboard( AwsAccountId='<アカウントID>', DashboardId=dashboard_id ) # ダッシュボードの存在検証 def test_dashboard_exists(): response = describe_dashboard('<ダッシュボードID>') assert response['Dashboard']['Name'] == 'ExpectedDashboardName', "Dashboard name does not match." # フィルタ設定の検証 def test_filter_settings(): response = describe_dashboard('<ダッシュボードID>') filters = response['Dashboard']['Filters'] assert any(f['Name'] == 'expected_filter_name' for f in filters), "Expected filter is missing." # 計算フィールドの検証 def test_calculated_fields(): response = describe_dashboard('<ダッシュボードID>') calculated_fields = response['Dashboard']['CalculatedFields'] assert any(cf['Name'] == 'expected_calculated_field_name' for cf in calculated_fields), "Expected calculated field is missing." # データセットの整合性検証 def test_dataset_schema(): response = describe_dashboard('<ダッシュボードID>') columns = [col['Name'] for col in response['DataSet']['PhysicalTableMap']['PhysicalTable']['Schema']['Columns']] assert 'expected_column' in columns, "Expected column is missing in dataset schema."
テストスクリプトのexpected値はダッシュボードによって異なります。 expected値をJSONで管理し、テストスクリプト内で読み込むアプローチを取ります。
{
"dashboard_id_1": {
"expected_name": "Sales Dashboard",
"expected_filters": ["Region", "Product"],
"expected_calculated_fields": ["TotalSales", "ProfitMargin"]
},
"dashboard_id_2": {
"expected_name": "Marketing Dashboard",
"expected_filters": ["Campaign", "Channel"],
"expected_calculated_fields": ["ROI", "ClickThroughRate"]
}
}
import json with open('expected_values.json') as f: expected_values = json.load(f) def test_dashboard_exists(): dashboard_id = '<ダッシュボードID>' expected = expected_values[dashboard_id] response = describe_dashboard(dashboard_id) assert response['Dashboard']['Name'] == expected['expected_name'], "Dashboard name does not match."
最後に
この記事ではCIパイプラインによるQuickSightダッシュボードの変更を検知して自動テストする仕組みを解説しました。
AWS LambdaとQuickSight APIを組み合わせることで変更を検知できるようになります。 QuickSightダッシュボードの変更点をCIによる自動テストで管理することで、開発プロセスの効率化と信頼性向上が期待できます。 この記事の手順を参考に、QuickSight運用の品質向上を目指してみてください。
ここまでお読みいただきありがとうございました！
