nik/newdwh2021
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: バッチ実行管理テーブルのユニットテスト実装完了

Browse Source
This commit is contained in:
shimoda.m@nds-tyo.co.jp 2025-05-27 11:19:15 +09:00
parent 61ff9f1159
commit a6b17b0c78
2 changed files with 54 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
ecs/jskult-batch/src/manager/jskult_batch_run_manager.py
View File

@ -1,4 +1,4 @@
from datetime import datetime, timedelta, timezone
from datetime import UTC, datetime, timedelta
import boto3
@ -28,7 +28,7 @@ class JskultBatchRunManager:
batch_run_status (str): バッチ処理ステータス
"""
# レコードの有効期限を現在時刻+24hのタイムスタンプで生成
now = datetime.now(timezone.utc)
now = datetime.now(UTC)
later = now + timedelta(hours=24)
timestamp_24h = int(later.timestamp())

55
ecs/jskult-batch/tests/manager/test_jskult_batch_run_manager.py
View File

@ -14,12 +14,19 @@ class TestJskultBatchRunManager:
yield dynamodb
dynamodb.delete_item(Key={
'execution_id': UNITTEST_EXECUTION_ID
})
dynamodb.delete_item(
TableName=BATCH_MANAGE_DYNAMODB_TABLE_NAME,
Key={
'execution_id': {'S': UNITTEST_EXECUTION_ID}
}
)
def test_batch_success(self, dynamodb_client):
"""バッチ実行管理テーブルに成功のステータスが書き込まれること
Args:
dynamodb_client (boto3.Client): DynamoDB クライアント
"""
# Arrange
# Act
sut = JskultBatchRunManager(BATCH_MANAGE_DYNAMODB_TABLE_NAME, UNITTEST_EXECUTION_ID)
@ -34,3 +41,45 @@ class TestJskultBatchRunManager:
}
actual = dynamodb_client.get_item(**options)
assert actual['Item']['batch_run_status']['S'] == 'success'
def test_batch_failed(self, dynamodb_client):
"""バッチ実行管理テーブルに失敗のステータスが書き込まれること
Args:
dynamodb_client (boto3.Client): DynamoDB クライアント
"""
# Arrange
# Act
sut = JskultBatchRunManager(BATCH_MANAGE_DYNAMODB_TABLE_NAME, UNITTEST_EXECUTION_ID)
sut.batch_failed()
# Assert
options = {
'TableName': BATCH_MANAGE_DYNAMODB_TABLE_NAME,
'Key': {
'execution_id': {'S': UNITTEST_EXECUTION_ID},
},
}
actual = dynamodb_client.get_item(**options)
assert actual['Item']['batch_run_status']['S'] == 'failed'
def test_batch_retry(self, dynamodb_client):
"""バッチ実行管理テーブルにリトライのステータスが書き込まれること
Args:
dynamodb_client (boto3.Client): DynamoDB クライアント
"""
# Arrange
# Act
sut = JskultBatchRunManager(BATCH_MANAGE_DYNAMODB_TABLE_NAME, UNITTEST_EXECUTION_ID)
sut.batch_retry()
# Assert
options = {
'TableName': BATCH_MANAGE_DYNAMODB_TABLE_NAME,
'Key': {
'execution_id': {'S': UNITTEST_EXECUTION_ID},
},
}
actual = dynamodb_client.get_item(**options)
assert actual['Item']['batch_run_status']['S'] == 'retry'