Best Python code snippet using localstack_python
storage_service.py
Source:storage_service.py
1from services.client_service import init_aws_client2from botocore.exceptions import ClientError3from helper import get_end_date_based_on_schedule, human_sorting_keys4from configs.constants import NO_SUCH_KEY_ERROR, CURRENT_DATE5from configs.configs import S3_BUCKET_NAME6import json7def upload_user_details_file_to_s3(file_content, is_update=False, file_name_to_update=None):8 print('S3 upload starting...')9 if file_name_to_update:10 file_name = file_name_to_update11 else:12 file_name = file_content['user_details']['id'] + '.json'13 # check if user file already exists14 if is_update:15 file_content_updated = file_content16 else:17 existing_user_schedule_details = is_file_available_in_s3(file_name)18 if not existing_user_schedule_details:19 file_content_updated = build_insert_details(file_content)20 else:21 file_content_updated = build_updated_details(file_content, existing_user_schedule_details)22 object_name = '/tmp/' + file_name23 file_open = open(object_name, 'w')24 file_open.write(json.dumps(file_content_updated))25 file_open.close()26 s3_client = init_aws_client('s3')27 s3_client.upload_file(object_name, S3_BUCKET_NAME, file_name)28 print('after s3 insert')29def is_file_available_in_s3(file_name):30 try:31 s3_client = init_aws_client('s3')32 user_scheduler_details = s3_client.get_object(Bucket=S3_BUCKET_NAME, Key=file_name)33 user_scheduler_details = user_scheduler_details['Body'].read()34 return json.loads(user_scheduler_details.decode("utf-8"))35 except ClientError as ex:36 if ex.response['Error']['Code'] == NO_SUCH_KEY_ERROR:37 return False38def build_insert_details(file_content):39 user_details_array = []40 # being the first user data start with 141 file_content['report_id'] = file_content['user_details']['id'] + '_1'42 file_content['is_alert_send'] = False43 file_content['active'] = True44 file_content['next_scheduled_alert_date'] = get_end_date_based_on_schedule(CURRENT_DATE, file_content['report_schedule'])45 user_details_array.append(file_content)46 return user_details_array47def build_updated_details(file_content, existing_content):48 report_id_list = [report_id['report_id'] for report_id in existing_content]49 report_id_list.sort(key=human_sorting_keys)50 current_report_id = report_id_list[-1]51 latest_report_id = str(int(current_report_id.split('_')[-1]) + 1)52 file_content['report_id'] = file_content['user_details']['id'] + '_' + latest_report_id53 file_content['is_alert_send'] = False54 file_content['active'] = True55 file_content['next_scheduled_alert_date'] = get_end_date_based_on_schedule(CURRENT_DATE, file_content['report_schedule'])56 existing_content.append(file_content)...
constants.py
Source:constants.py
1import datetime2CURRENT_DATE = datetime.datetime.today()3REPORT_TYPE_DAILY = '1'4REPORT_TYPE_WEEKLY = '2'5REPORT_TYPE_MONTHLY = '3'6REPORT_SCHEDULE = {7 REPORT_TYPE_DAILY: 'Daily',8 REPORT_TYPE_WEEKLY: 'Weekly',9 REPORT_TYPE_MONTHLY: 'Monthly'10}11REPORT_SCHEDULE_ALTERNATIVE = {12 REPORT_TYPE_DAILY: 'Day',13 REPORT_TYPE_WEEKLY: 'Week',14 REPORT_TYPE_MONTHLY: 'Month'15}16STATUS_ACTIVE = 117STATUS_INACTIVE = 018NO_SUCH_KEY_ERROR = 'NoSuchKey'19DEFAULT_ROUND_VALUE = 220INIT_AMOUNT = 0.0021DATE_FORMAT_DEFAULT = '%Y-%m-%d'22HTTP_STATUS_CODE_200 = 20023HTTP_STATUS_CODE_404 = 40424HTTP_STATUS_CODE_401 = 40125'''Modal constant values'''26MODAL_HEADER = 'Kmart Cost Notifier'27MODAL_TYPE_PLAIN_TEXT = "plain_text"28MODAL_TYPE_INPUT = "input"29MODAL_TYPE_SECTION = "section"30MODAL_TYPE_HEADER = "header"31MODAL_TYPE_ACTIONS = "actions"32MODAL_TYPE_DIVIDER = "divider"33MODAL_TYPE_CONTEXT = "context"34MODAL_TYPE_MRK_DWN = "mrkdwn" # Used to create visual highlights in messages35MODAL_INPUT_TYPE_EXTERNAL_SELECT = "external_select"36MODAL_INPUT_TYPE_CONVERSATION_SELECT = "conversations_select"37MODAL_INPUT_TYPE_PLAIN_TEXT = "plain_text_input"38MODAL_INPUT_TYPE_RADIO_BUTTON = "radio_buttons"39MODAL_INPUT_TYPE_CHECKBOX = "checkboxes"40MODAL_INPUT_TYPE_BUTTON = "button"41MODAL_VALUE_ALERT_NAME = "alert_name"42MODAL_VALUE_PROJECT_TAG = "project_tag"43MODAL_VALUE_BUDGET_LIMIT = "budget_limit"44MODAL_VALUE_BUDGET_THRESHOLD = "budget_threshold"45MODAL_VALUE_REPORT_SCHEDULE = "report_schedule"46MODAL_VALUE_CHANNEL_ID = "channel_id"47MODAL_VALUE_ALERT_LIST = "alert_list"48MODAL_VALUE_SEND_NOW = "send_now"49MODAL_VALUE_PROJECT_FAST = "project_fast"50MODAL_VALUE_PROJECT_SHARED = "project_shared"51MODAL_VALUE_UPDATE_PROJECT_SHARED = "update_project_shared"52MODAL_VALUE_INSTANT_ALERT = "instant_alert"53MODAL_VALUE_CREATE_ALERT = "create_alert"54MODAL_VALUE_UPDATE_ALERT = "update_alert"55MODAL_VALUE_DELETE_ALERT = "delete_alert"56MODAL_VALUE_FAST_ACCOUNT_NAME = "fast_account_name"57CLIENT_EMOJI_TREND_UP = ":arrow_up:"58CLIENT_EMOJI_TREND_DOWN = ":arrow_down:"59CLIENT_EMOJI_IF_BREACHED = ":exclamation:"60CLIENT_EMOJI_IF_NOT_BREACHED = ":white_check_mark:"61CLIENT_EMOJI_PRIVATE_CHANNEL = ':lock:'62CLIENT_EMOJI_PUBLIC_CHANNEL = ':hash:'...
test_artifacts.py
Source:test_artifacts.py
1# -*- coding: utf-8 -*-2import os3import unittest4import unittest.mock as mock5import platiagro6from platiagro.util import BUCKET_NAME, MINIO_CLIENT7import tests.util as util8class TestArtifacts(unittest.TestCase):9 @mock.patch.object(10 MINIO_CLIENT,11 "fget_object",12 side_effect=util.NO_SUCH_KEY_ERROR,13 )14 def test_download_artifact_not_found(self, mock_fget_object):15 """16 Should raise an exception when given an artifact name that does not exist.17 """18 bad_artifact_name = "unk.zip"19 local_path = "./unk.zip"20 with self.assertRaises(FileNotFoundError):21 platiagro.download_artifact(name=bad_artifact_name, path=local_path)22 mock_fget_object.assert_any_call(23 bucket_name=BUCKET_NAME,24 object_name=f"artifacts/{bad_artifact_name}",25 file_path=local_path,26 )27 @mock.patch.object(28 MINIO_CLIENT,29 "fget_object",30 side_effect=lambda **kwargs: open("unk.zip", "w").close(),31 )32 def test_download_artifact_success(self, mock_fget_object):33 """34 Should download an artifact to a given local path.35 """36 artifact_name = "unk.zip"37 local_path = "./unk.zip"38 platiagro.download_artifact(name=artifact_name, path=local_path)39 self.assertTrue(os.path.exists(local_path))40 mock_fget_object.assert_any_call(41 bucket_name=BUCKET_NAME,42 object_name=f"artifacts/{artifact_name}",43 file_path=local_path,...
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!