feat: AWS上のS3を利用する機能の追加

ecs/crm-datafetch/tests/test_walk_through.py

@@ -4,6 +4,7 @@ import os
 import os.path as path
 from datetime import datetime, timezone
 
+import boto3
 import pytest
 from src.controller import controller
 from src.parser.json_parser import JsonParser
@@ -11,9 +12,38 @@ from src.system_var.constants import YYYYMMDDTHHMMSSTZ
 from src.util.execute_datetime import ExecuteDateTime
 
 ROOT_DIR = path.abspath(path.dirname(__name__))
-DATA_BUCKET = 'mbj-newdwh2021-staging-data'
-CONFIG_BUCKET = 'mbj-newdwh2021-staging-config'
-BACKUP_BUCKET = 'mbj-newdwh2021-staging-backup-crm'
+
+
+# ↓↓↓モックテスト用(バケットをモック化する際、コメントアウトを外す。AWS利用テスト用はコメントアウトする)↓↓↓
+# DATA_BUCKET = 'mbj-newdwh2021-staging-data'
+# CONFIG_BUCKET = 'mbj-newdwh2021-staging-config'
+# BACKUP_BUCKET = 'mbj-newdwh2021-staging-backup-crm'
+#
+# @pytest.fixture
+# def s3_test(s3_client):
+#    s3_client.create_bucket(Bucket=DATA_BUCKET)
+#    s3_client.create_bucket(Bucket=CONFIG_BUCKET)
+#    s3_client.create_bucket(Bucket=BACKUP_BUCKET)
+#    yield
+# ↑↑↑モックテスト用↑↑↑
+
+# ↓↓↓AWS利用テスト用(AWS上のバケットを利用する際、コメントアウトを外す。モックテスト用はコメントアウトする)↓↓↓
+DATA_BUCKET = 'test-mbj-newdwh2021-test-data'
+CONFIG_BUCKET = 'test-mbj-newdwh2021-test-config'
+BACKUP_BUCKET = 'test-mbj-newdwh2021-test-backup-crm'
+
+
+@pytest.fixture
+def s3_test(s3_client):
+    yield
+
+
+@pytest.fixture
+def s3_client():
+    s3_client = boto3.client("s3")
+    yield s3_client
+# ↑↑↑AWS利用テスト用↑↑↑
+
 
 TARGET_FOLDER = 'crm/target'
 OBJECT_INFO_FOLDER = 'crm/object_info'
@@ -23,14 +53,6 @@ BACKUP_RESPONSE_JSON_FOLDER = 'response_json'
 PROCESS_RESULT_JSON_FILE_NAME = 'process_result.json'
 
 
-@pytest.fixture
-def s3_test(s3_client):
-    s3_client.create_bucket(Bucket=DATA_BUCKET)
-    s3_client.create_bucket(Bucket=CONFIG_BUCKET)
-    s3_client.create_bucket(Bucket=BACKUP_BUCKET)
-    yield
-
-
 @pytest.mark.walk_through
 def test_walk_through(s3_test, s3_client, monkeypatch, caplog):
     """
@@ -256,7 +278,7 @@ def assert_file_exist(s3_client, bucket_name, file_key):
 def get_object_config_list(folder_name: str):
     local_s3_path = path.join(ROOT_DIR, '..', '..', 's3', 'config', 'crm', folder_name)
     config_list = [os.path.join(local_s3_path, config) for config in os.listdir(local_s3_path) if config.endswith('.json')]
-    return config_list
+    return sorted(config_list, reverse=True)
 
 
 def read_json(json_path):