KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use precheck method in autotest
Best Python code snippet using autotest_python
test_precheck_conditions.py
Source:test_precheck_conditions.py 
1import pytest2from src.Constant.constant import constant3from src.modules.irisv2.message.createMessage import CreateMessage4from src.modules.irisv2.message.preCheck import PreCheck5from src.modules.irisv2.helper.irisHelper import IrisHelper6from src.utilities.assertion import Assertion7from src.dbCalls.campaignShard import list_Calls8@pytest.mark.skipif(constant.config['cluster'] not in ['nightly'], reason='Precheck Credit Tests Authorized Only For Nightly')9class Test_Precheck_Conditions():10 @pytest.mark.parametrize('campaignType,testControlType,listType,channel,messageInfo,precheckErrors', [11 ('LIVE', 'ORG', 'UPLOAD', 'MOBILE',12 {'scheduleType': {'type': 'IMMEDIATE'}, 'offerType': 'PLAIN', 'messageStrategy': {'type': 'DEFAULT'},13 'channels': ['SMS', 'EMAIL'], 'useTinyUrl': False, 'encryptUrl': False, 'skipRateLimit': True},14 [{'errorType': 'CAMPAIGN_NOT_AUTHORIZED', 'parameters': {}}]),15 ('UPCOMING', 'ORG', 'UPLOAD', 'MOBILE',16 {'scheduleType': {'type': 'IMMEDIATE'}, 'offerType': 'PLAIN', 'messageStrategy': {'type': 'DEFAULT'},17 'channels': ['SMS', 'EMAIL'], 'useTinyUrl': False, 'encryptUrl': False, 'skipRateLimit': True},18 [{'errorType': 'CAMPAIGN_NOT_AUTHORIZED', 'parameters': {}}])19 ])20 def test_irisv2_message_precheck_create_upload_mobile_immediate_NotAuthorized(self, campaignType, testControlType,21 listType, channel, messageInfo,22 precheckErrors):23 messageDetails = CreateMessage.create(campaignType, testControlType, listType, channel, messageInfo,24 updateNode=True, lockNode=True)25 CreateMessage.assertResponse(messageDetails['RESPONSE'], 200)26 preCheckResponse = PreCheck.executePrecheck(27 constant.config['node'][campaignType][testControlType]['CAMPAIGN']['ID'],28 messageDetails['RESPONSE']['json']['entity']['id'])29 PreCheck.assertPreCheckResponse(preCheckResponse, 200)30 PreCheck.assertPrecheckStatus(preCheckResponse, precheckErrors)31 @pytest.mark.parametrize('campaignType,testControlType,listType,channel,messageInfo,precheckErrors', [32 ('LIVE', 'ORG', 'UPLOAD', 'MOBILE',33 {'scheduleType': {'type': 'IMMEDIATE'}, 'offerType': 'PLAIN', 'messageStrategy': {'type': 'DEFAULT'},34 'channels': ['SMS', 'EMAIL'], 'useTinyUrl': False, 'encryptUrl': False, 'skipRateLimit': True},35 [{'errorType': 'CAMPAIGN_NOT_AUTHORIZED', 'parameters': {}}]),36 ('UPCOMING', 'ORG', 'UPLOAD', 'MOBILE',37 {'scheduleType': {'type': 'IMMEDIATE'}, 'offerType': 'PLAIN', 'messageStrategy': {'type': 'DEFAULT'},38 'channels': ['SMS', 'EMAIL'], 'useTinyUrl': False, 'encryptUrl': False, 'skipRateLimit': True},39 [{'errorType': 'CAMPAIGN_NOT_AUTHORIZED', 'parameters': {}}])40 ])41 def test_irisv2_message_precheck_create_upload_mobile_particularDate_NotAuthorized(self, campaignType,42 testControlType, listType,43 channel, messageInfo,44 precheckErrors):45 messageDetails = CreateMessage.create(campaignType, testControlType, listType, channel, messageInfo,46 updateNode=True, lockNode=True)47 CreateMessage.assertResponse(messageDetails['RESPONSE'], 200)48 preCheckResponse = PreCheck.executePrecheck(49 constant.config['node'][campaignType][testControlType]['CAMPAIGN']['ID'],50 messageDetails['RESPONSE']['json']['entity']['id'])51 PreCheck.assertPreCheckResponse(preCheckResponse, 200)52 PreCheck.assertPrecheckStatus(preCheckResponse, precheckErrors)53 @pytest.mark.parametrize('campaignType,testControlType,listType,channel,messageInfo,precheckErrors', [54 ('LIVE', 'ORG', 'LOYALTY', 'MOBILE',55 {'scheduleType': {'type': 'RECURRING'}, 'offerType': 'PLAIN', 'messageStrategy': {'type': 'DEFAULT'},56 'channels': ['SMS', 'EMAIL'], 'useTinyUrl': False, 'encryptUrl': False, 'skipRateLimit': True},57 [{'errorType': 'CAMPAIGN_NOT_AUTHORIZED', 'parameters': {}}]),58 ('UPCOMING', 'ORG', 'LOYALTY', 'MOBILE',59 {'scheduleType': {'type': 'RECURRING'}, 'offerType': 'PLAIN', 'messageStrategy': {'type': 'DEFAULT'},60 'channels': ['SMS', 'EMAIL'], 'useTinyUrl': False, 'encryptUrl': False, 'skipRateLimit': True},61 [{'errorType': 'CAMPAIGN_NOT_AUTHORIZED', 'parameters': {}}])62 ])63 def est_irisv2_message_precheck_create_filter_mobile_Recurring_NotAuthorized(self, campaignType,64 testControlType,65 listType,66 channel, messageInfo,67 precheckErrors):68 messageDetails = CreateMessage.create(campaignType, testControlType, listType, channel, messageInfo,69 updateNode=True, lockNode=True)70 CreateMessage.assertResponse(messageDetails['RESPONSE'], 200)71 preCheckResponse = PreCheck.executePrecheck(72 constant.config['node'][campaignType][testControlType]['CAMPAIGN']['ID'],73 messageDetails['RESPONSE']['json']['entity']['id'])74 PreCheck.assertPreCheckResponse(preCheckResponse, 200)75 PreCheck.assertPrecheckStatus(preCheckResponse, precheckErrors)76 @pytest.mark.parametrize('campaignType,testControlType,listType,channel,messageInfo,precheckErrors', [77 ('LIVE', 'ORG', 'UPLOAD', 'MOBILE',78 {'scheduleType': {'type': 'IMMEDIATE'}, 'offerType': 'PLAIN', 'messageStrategy': {'type': 'DEFAULT'},79 'channels': ['SMS', 'EMAIL'], 'useTinyUrl': False, 'encryptUrl': False, 'skipRateLimit': True},80 [{'errorType': 'CAMPAIGN_NOT_AUTHORIZED', 'parameters': {}},81 {'errorType': 'BULK_CREDITS_NOT_AVAILABLE',82 'parameters': {'availableCredits': 0, 'balanceRequiredCredits': 10}}]),83 ('UPCOMING', 'ORG', 'UPLOAD', 'MOBILE',84 {'scheduleType': {'type': 'IMMEDIATE'}, 'offerType': 'PLAIN', 'messageStrategy': {'type': 'DEFAULT'},85 'channels': ['SMS', 'EMAIL'], 'useTinyUrl': False, 'encryptUrl': False, 'skipRateLimit': True},86 [{'errorType': 'CAMPAIGN_NOT_AUTHORIZED', 'parameters': {}},87 {'errorType': 'BULK_CREDITS_NOT_AVAILABLE',88 'parameters': {'availableCredits': 0, 'balanceRequiredCredits': 10}}])89 ])90 def test_irisv2_message_precheck_create_upload_mobile_immediate_CreditAvialable(self, campaignType,91 testControlType,92 listType,93 channel, messageInfo,94 precheckErrors):95 try:96 messageDetails = CreateMessage.create(campaignType, testControlType, listType, channel, messageInfo,97 updateNode=True, lockNode=True)98 precheckErrors[1]['parameters']['balanceRequiredCredits'] = int(list_Calls().getCustomerCountInGVD(99 messageDetails['PAYLOAD']['targetAudience']['include'][0]))100 CreateMessage.assertResponse(messageDetails['RESPONSE'], 200)101 IrisHelper.updateCredit(0, channel)102 preCheckResponse = PreCheck.executePrecheck(103 constant.config['node'][campaignType][testControlType]['CAMPAIGN']['ID'],104 messageDetails['RESPONSE']['json']['entity']['id'])105 PreCheck.assertPreCheckResponse(preCheckResponse, 200)106 PreCheck.assertPrecheckStatus(preCheckResponse,precheckErrors)107 except Exception, exp:108 Assertion.constructAssertion(False, 'Failure Due to Exception :{}'.format(exp))109 finally:110 IrisHelper.updateCredit(99999, channel)111 @pytest.mark.parametrize('campaignType,testControlType,listType,channel,messageInfo,precheckErrors', [112 ('LIVE', 'ORG', 'UPLOAD', 'MOBILE',113 {'scheduleType': {'type': 'PARTICULARDATE'}, 'offerType': 'PLAIN', 'messageStrategy': {'type': 'DEFAULT'},114 'channels': ['SMS', 'EMAIL'], 'useTinyUrl': False, 'encryptUrl': False, 'skipRateLimit': True},115 [{'errorType': 'CAMPAIGN_NOT_AUTHORIZED', 'parameters': {}},116 {'errorType': 'BULK_CREDITS_NOT_AVAILABLE',117 'parameters': {'availableCredits': 0, 'balanceRequiredCredits': 10}}]),118 ('UPCOMING', 'ORG', 'UPLOAD', 'MOBILE',119 {'scheduleType': {'type': 'PARTICULARDATE'}, 'offerType': 'PLAIN', 'messageStrategy': {'type': 'DEFAULT'},120 'channels': ['SMS', 'EMAIL'], 'useTinyUrl': False, 'encryptUrl': False, 'skipRateLimit': True},121 [{'errorType': 'CAMPAIGN_NOT_AUTHORIZED', 'parameters': {}},122 {'errorType': 'BULK_CREDITS_NOT_AVAILABLE',123 'parameters': {'availableCredits': 0, 'balanceRequiredCredits': 10}}])124 ])125 def test_irisv2_message_precheck_create_upload_mobile_particularDate_CreditAvialable(self, campaignType,126 testControlType,127 listType,128 channel, messageInfo,129 precheckErrors):130 try:131 messageDetails = CreateMessage.create(campaignType, testControlType, listType, channel, messageInfo,132 updateNode=True, lockNode=True)133 precheckErrors[1]['parameters']['balanceRequiredCredits'] = int(list_Calls().getCustomerCountInGVD(134 messageDetails['PAYLOAD']['targetAudience']['include'][0]))135 CreateMessage.assertResponse(messageDetails['RESPONSE'], 200)136 IrisHelper.updateCredit(0, channel)137 preCheckResponse = PreCheck.executePrecheck(138 constant.config['node'][campaignType][testControlType]['CAMPAIGN']['ID'],139 messageDetails['RESPONSE']['json']['entity']['id'])140 PreCheck.assertPreCheckResponse(preCheckResponse, 200)141 PreCheck.assertPrecheckStatus(preCheckResponse,precheckErrors)142 except Exception, exp:143 Assertion.constructAssertion(False, 'Failure Due to Exception :{}'.format(exp))144 finally:145 IrisHelper.updateCredit(99999, channel)146 @pytest.mark.parametrize('campaignType,testControlType,listType,channel,messageInfo,precheckErrors', [147 ('LIVE', 'ORG', 'UPLOAD', 'MOBILE',148 {'scheduleType': {'type': 'PARTICULARDATE'}, 'offerType': 'PLAIN', 'messageStrategy': {'type': 'DEFAULT'},149 'channels': ['SMS', 'EMAIL'], 'useTinyUrl': False, 'encryptUrl': False, 'skipRateLimit': True},150 [{'errorType': 'CAMPAIGN_NOT_AUTHORIZED', 'parameters': {}},151 {'errorType': 'BULK_CREDITS_NOT_AVAILABLE',152 'parameters': {'availableCredits': 1, 'balanceRequiredCredits': 9}}]),153 ('LIVE', 'ORG', 'UPLOAD', 'MOBILE',154 {'scheduleType': {'type': 'PARTICULARDATE'}, 'offerType': 'PLAIN', 'messageStrategy': {'type': 'DEFAULT'},155 'channels': ['SMS', 'EMAIL'], 'useTinyUrl': False, 'encryptUrl': False, 'skipRateLimit': True},156 [{'errorType': 'CAMPAIGN_NOT_AUTHORIZED', 'parameters': {}},157 {'errorType': 'BULK_CREDITS_NOT_AVAILABLE',158 'parameters': {'availableCredits': 7, 'balanceRequiredCredits': 3}}]),159 ('LIVE', 'ORG', 'UPLOAD', 'MOBILE',160 {'scheduleType': {'type': 'PARTICULARDATE'}, 'offerType': 'PLAIN', 'messageStrategy': {'type': 'DEFAULT'},161 'channels': ['SMS', 'EMAIL'], 'useTinyUrl': False, 'encryptUrl': False, 'skipRateLimit': True},162 [{'errorType': 'CAMPAIGN_NOT_AUTHORIZED', 'parameters': {}},163 {'errorType': 'BULK_CREDITS_NOT_AVAILABLE',164 'parameters': {'availableCredits': 9, 'balanceRequiredCredits': 1}}])165 ])166 def test_irisv2_message_precheck_create_upload_mobile_particularDate_CreditAvialable_CheckAvialable_BalanceRequiredCredit_Count(167 self, campaignType,168 testControlType,169 listType,170 channel, messageInfo,171 precheckErrors):172 try:173 messageDetails = CreateMessage.create(campaignType, testControlType, listType, channel, messageInfo,174 updateNode=True, lockNode=True)175 precheckErrors[1]['parameters']['balanceRequiredCredits'] = int(list_Calls().getCustomerCountInGVD(176 messageDetails['PAYLOAD']['targetAudience']['include'][0])) - precheckErrors[1]['parameters'][177 'availableCredits']178 CreateMessage.assertResponse(messageDetails['RESPONSE'], 200)179 IrisHelper.updateCredit(0, channel)180 IrisHelper.updateCredit(precheckErrors[1]['parameters']['availableCredits'], channel)181 preCheckResponse = PreCheck.executePrecheck(182 constant.config['node'][campaignType][testControlType]['CAMPAIGN']['ID'],183 messageDetails['RESPONSE']['json']['entity']['id'])184 PreCheck.assertPreCheckResponse(preCheckResponse, 200)185 PreCheck.assertPrecheckStatus(preCheckResponse,precheckErrors)186 except Exception, exp:187 Assertion.constructAssertion(False, 'Failure Due to Exception :{}'.format(exp))188 finally:189 IrisHelper.updateCredit(99999, channel)190 @pytest.mark.parametrize('campaignType,testControlType,listType,channel,messageInfo,precheckErrors', [191 ('LIVE', 'ORG', 'LOYALTY', 'MOBILE',192 {'scheduleType': {'type': 'RECURRING'}, 'offerType': 'PLAIN', 'messageStrategy': {'type': 'DEFAULT'},193 'channels': ['SMS', 'EMAIL'], 'useTinyUrl': False, 'encryptUrl': False, 'skipRateLimit': True},194 [{'errorType': 'CAMPAIGN_NOT_AUTHORIZED', 'parameters': {}},195 {'errorType': 'BULK_CREDITS_NOT_AVAILABLE',196 'parameters': {'availableCredits': 0, 'balanceRequiredCredits': 10}}])197 ])198 def test_irisv2_message_precheck_create_filter_mobile_recurring_CreditAvialable(self, campaignType,199 testControlType,200 listType,201 channel, messageInfo,202 precheckErrors):203 try:204 messageDetails = CreateMessage.create(campaignType, testControlType, listType, channel, messageInfo,205 updateNode=True, lockNode=True)206 precheckErrors[1]['parameters']['balanceRequiredCredits'] = list_Calls().getCustomerCountInGVD(207 messageDetails['PAYLOAD']['targetAudience']['include'][0])208 CreateMessage.assertResponse(messageDetails['RESPONSE'], 200)209 IrisHelper.updateCredit(0, channel)210 preCheckResponse = PreCheck.executePrecheck(211 constant.config['node'][campaignType][testControlType]['CAMPAIGN']['ID'],212 messageDetails['RESPONSE']['json']['entity']['id'])213 PreCheck.assertPreCheckResponse(preCheckResponse, 200)214 PreCheck.assertPrecheckStatus(preCheckResponse,precheckErrors)215 except Exception, exp:216 Assertion.constructAssertion(False, 'Failure Due to Exception :{}'.format(exp))217 finally:218 IrisHelper.updateCredit(99999, channel)219 @pytest.mark.parametrize('campaignType,testControlType,listType,channel,messageInfo,precheckErrors', [220 ('LIVE', 'ORG', 'LOYALTY', 'MOBILE',221 {'scheduleType': {'type': 'RECURRING'}, 'offerType': 'PLAIN', 'messageStrategy': {'type': 'DEFAULT'},222 'channels': ['SMS', 'EMAIL'], 'useTinyUrl': False, 'encryptUrl': False, 'skipRateLimit': True},223 [{'errorType': 'CAMPAIGN_NOT_AUTHORIZED', 'parameters': {}},224 {'errorType': 'MAX_USER_LIMIT_EXCEEDED', 'parameters': {'totalCustomerCount': 22, 'maxUsersAllowed': 1}}]),225 ('LIVE', 'CUSTOM', 'LOYALTY', 'MOBILE',226 {'scheduleType': {'type': 'RECURRING'}, 'offerType': 'PLAIN', 'messageStrategy': {'type': 'DEFAULT'},227 'channels': ['SMS', 'EMAIL'], 'useTinyUrl': False, 'encryptUrl': False, 'skipRateLimit': True},228 [{'errorType': 'CAMPAIGN_NOT_AUTHORIZED', 'parameters': {}},229 {'errorType': 'MAX_USER_LIMIT_EXCEEDED', 'parameters': {'totalCustomerCount': 22, 'maxUsersAllowed': 1}}])230 ])231 def test_irisv2_message_precheck_create_filter_mobile_Recurring_MaxUser(self, campaignType,232 testControlType,233 listType,234 channel, messageInfo,235 precheckErrors):236 messageDetails = CreateMessage.create(campaignType, testControlType, listType, channel, messageInfo,237 updateNode=True, lockNode=True, maxUser=[1])238 precheckErrors[1]['parameters']['totalCustomerCount'] = list_Calls().getCustomerCountInGVD(239 messageDetails['PAYLOAD']['targetAudience']['include'][0])240 CreateMessage.assertResponse(messageDetails['RESPONSE'], 200)241 preCheckResponse = PreCheck.executePrecheck(242 constant.config['node'][campaignType][testControlType]['CAMPAIGN']['ID'],243 messageDetails['RESPONSE']['json']['entity']['id'])244 PreCheck.assertPreCheckResponse(preCheckResponse, 200)245 PreCheck.assertPrecheckStatus(preCheckResponse,precheckErrors)246 @pytest.mark.parametrize('campaignType,testControlType,listType,channel,messageInfo,precheckErrors', [247 ('LIVE', 'ORG', 'UPLOAD', 'EMAIL',248 {'scheduleType': {'type': 'IMMEDIATE'}, 'offerType': 'PLAIN', 'messageStrategy': {'type': 'DEFAULT'},249 'channels': ['SMS', 'EMAIL'], 'useTinyUrl': False, 'encryptUrl': False, 'skipRateLimit': True},250 [{'errorType': 'CAMPAIGN_NOT_AUTHORIZED', 'parameters': {}}]),251 ('UPCOMING', 'ORG', 'UPLOAD', 'EMAIL',252 {'scheduleType': {'type': 'IMMEDIATE'}, 'offerType': 'PLAIN', 'messageStrategy': {'type': 'DEFAULT'},253 'channels': ['SMS', 'EMAIL'], 'useTinyUrl': False, 'encryptUrl': False, 'skipRateLimit': True},254 [{'errorType': 'CAMPAIGN_NOT_AUTHORIZED', 'parameters': {}}]),255 ('LIVE', 'CUSTOM', 'UPLOAD', 'EMAIL',256 {'scheduleType': {'type': 'IMMEDIATE'}, 'offerType': 'PLAIN', 'messageStrategy': {'type': 'DEFAULT'},257 'channels': ['SMS', 'EMAIL'], 'useTinyUrl': False, 'encryptUrl': False, 'skipRateLimit': True},258 [{'errorType': 'CAMPAIGN_NOT_AUTHORIZED', 'parameters': {}}])259 ])260 def test_irisv2_message_precheck_create_upload_email_immediate_NotAuthorized(self, campaignType,261 testControlType,262 listType,263 channel, messageInfo,264 precheckErrors):265 messageDetails = CreateMessage.create(campaignType, testControlType, listType, channel, messageInfo,266 updateNode=True, lockNode=True)267 CreateMessage.assertResponse(messageDetails['RESPONSE'], 200)268 preCheckResponse = PreCheck.executePrecheck(269 constant.config['node'][campaignType][testControlType]['CAMPAIGN']['ID'],270 messageDetails['RESPONSE']['json']['entity']['id'])271 PreCheck.assertPreCheckResponse(preCheckResponse, 200)272 PreCheck.assertPrecheckStatus(preCheckResponse,precheckErrors)273 @pytest.mark.parametrize('campaignType,testControlType,listType,channel,messageInfo,precheckErrors', [274 ('LIVE', 'ORG', 'LOYALTY', 'EMAIL',275 {'scheduleType': {'type': 'RECURRING'}, 'offerType': 'PLAIN', 'messageStrategy': {'type': 'DEFAULT'},276 'channels': ['SMS', 'EMAIL'], 'useTinyUrl': False, 'encryptUrl': False, 'skipRateLimit': True},277 [{'errorType': 'CAMPAIGN_NOT_AUTHORIZED', 'parameters': {}}]),278 ('LIVE', 'CUSTOM', 'LOYALTY', 'EMAIL',279 {'scheduleType': {'type': 'RECURRING'}, 'offerType': 'PLAIN', 'messageStrategy': {'type': 'DEFAULT'},280 'channels': ['SMS', 'EMAIL'], 'useTinyUrl': False, 'encryptUrl': False, 'skipRateLimit': True},281 [{'errorType': 'CAMPAIGN_NOT_AUTHORIZED', 'parameters': {}}]),282 ('UPCOMING', 'ORG', 'LOYALTY', 'EMAIL',283 {'scheduleType': {'type': 'RECURRING'}, 'offerType': 'PLAIN', 'messageStrategy': {'type': 'DEFAULT'},284 'channels': ['SMS', 'EMAIL'], 'useTinyUrl': False, 'encryptUrl': False, 'skipRateLimit': True},285 [{'errorType': 'CAMPAIGN_NOT_AUTHORIZED', 'parameters': {}}])286 ])287 def test_irisv2_message_precheck_create_filter_email_Recurring_NotAuthorized(self, campaignType,288 testControlType,289 listType,290 channel, messageInfo,291 precheckErrors):292 messageDetails = CreateMessage.create(campaignType, testControlType, listType, channel, messageInfo,293 updateNode=True, lockNode=True)294 CreateMessage.assertResponse(messageDetails['RESPONSE'], 200)295 preCheckResponse = PreCheck.executePrecheck(296 constant.config['node'][campaignType][testControlType]['CAMPAIGN']['ID'],297 messageDetails['RESPONSE']['json']['entity']['id'])298 PreCheck.assertPreCheckResponse(preCheckResponse, 200)299 PreCheck.assertPrecheckStatus(preCheckResponse,precheckErrors)300 @pytest.mark.parametrize('campaignType,testControlType,listType,channel,messageInfo,precheckErrors', [301 ('LIVE', 'ORG', 'LOYALTY', 'EMAIL',302 {'scheduleType': {'type': 'RECURRING'}, 'offerType': 'PLAIN', 'messageStrategy': {'type': 'DEFAULT'},303 'channels': ['SMS', 'EMAIL'], 'useTinyUrl': False, 'encryptUrl': False, 'skipRateLimit': True},304 [{'errorType': 'CAMPAIGN_NOT_AUTHORIZED', 'parameters': {}},305 {'errorType': 'MAX_USER_LIMIT_EXCEEDED', 'parameters': {'totalCustomerCount': 22, 'maxUsersAllowed': 1}}])306 ])307 def test_irisv2_message_precheck_create_filter_email_Recurring_MaxUser(self, campaignType,308 testControlType,309 listType,310 channel, messageInfo,311 precheckErrors):312 messageDetails = CreateMessage.create(campaignType, testControlType, listType, channel, messageInfo,313 updateNode=True, lockNode=True, maxUser=[1])314 precheckErrors[1]['parameters']['totalCustomerCount'] = list_Calls().getCustomerCountInGVD(315 messageDetails['PAYLOAD']['targetAudience']['include'][0])316 CreateMessage.assertResponse(messageDetails['RESPONSE'], 200)317 preCheckResponse = PreCheck.executePrecheck(318 constant.config['node'][campaignType][testControlType]['CAMPAIGN']['ID'],319 messageDetails['RESPONSE']['json']['entity']['id'])320 PreCheck.assertPreCheckResponse(preCheckResponse, 200)...
test_precheck_conditions_gateway.py
Source:test_precheck_conditions_gateway.py
1import pytest2from src.Constant.constant import constant3from src.modules.irisv2.helper.irisHelper import IrisHelper4from src.modules.irisv2.message.createMessage import CreateMessage5from src.modules.irisv2.message.preCheck import PreCheck6from src.utilities.assertion import Assertion7from src.utilities.logger import Logger8@pytest.mark.skipif(constant.config['cluster'] not in ['nightly'], reason='Precheck Gateway Tests Authorized Only For Nightly')9class Test_Gateway_PreCheck():10 def setup_class(self):11 self.actualOrgId = IrisHelper.updateOrgId(constant.config['mobilepush']['orgId'])12 self.actualOrgName = IrisHelper.updateOrgName(constant.config['mobilepush']['orgName'])13 def setup_method(self, method):14 Logger.logMethodName(method.__name__)15 @pytest.mark.parametrize('campaignType,testControlType,listType,channel,messageInfo,preCheckErrorList', [16 ('LIVE', 'ORG', 'UPLOAD', 'MOBILE',17 {'scheduleType': {'type': 'IMMEDIATE'}, 'offerType': 'PLAIN', 'messageStrategy': {'type': 'DEFAULT'},18 'channels': ['SMS', 'EMAIL'], 'useTinyUrl': False, 'encryptUrl': False, 'skipRateLimit': True},19 [{'errorType': 'CAMPAIGN_NOT_AUTHORIZED', 'parameters': {}},20 {'errorType': 'GATEWAY_NOT_AVAILABLE', 'parameters': {}}])21 ])22 def test_irisv2_message_precheck_create_upload_mobile_immediate_GatewayNotAvialable(self, campaignType,23 testControlType,24 listType,25 channel,26 messageInfo,27 preCheckErrorList):28 try:29 messageDetails = CreateMessage.create(campaignType, testControlType, listType, channel, messageInfo,30 updateNode=True, lockNode=True)31 CreateMessage.assertResponse(messageDetails['RESPONSE'], 200)32 IrisHelper.disableDomainGatewayMapId(channel)33 preCheckResponse = PreCheck.executePrecheck(34 constant.config['node'][campaignType][testControlType]['CAMPAIGN']['ID'],35 messageDetails['RESPONSE']['json']['entity']['id'])36 PreCheck.assertPreCheckResponse(preCheckResponse, 200)37 PreCheck.assertPrecheckStatus(preCheckResponse,preCheckErrorList)38 finally:39 IrisHelper.createNewDummyGateway(channel)40 @pytest.mark.parametrize('campaignType,testControlType,listType,channel,messageInfo,preCheckErrorList', [41 ('LIVE', 'ORG', 'UPLOAD', 'MOBILE',42 {'scheduleType': {'type': 'PARTICULARDATE'}, 'offerType': 'PLAIN', 'messageStrategy': {'type': 'DEFAULT'},43 'channels': ['SMS', 'EMAIL'], 'useTinyUrl': False, 'encryptUrl': False, 'skipRateLimit': True},44 [{'errorType': 'CAMPAIGN_NOT_AUTHORIZED', 'parameters': {}},45 {'errorType': 'GATEWAY_NOT_AVAILABLE', 'parameters': {}}])46 ])47 def test_irisv2_message_precheck_create_upload_mobile_particularDate_GatewayNotAvialable(self, campaignType,48 testControlType,49 listType,50 channel,51 messageInfo,52 preCheckErrorList):53 try:54 messageDetails = CreateMessage.create(campaignType, testControlType, listType, channel, messageInfo,55 updateNode=True, lockNode=True)56 CreateMessage.assertResponse(messageDetails['RESPONSE'], 200)57 IrisHelper.disableDomainGatewayMapId(channel)58 preCheckResponse = PreCheck.executePrecheck(59 constant.config['node'][campaignType][testControlType]['CAMPAIGN']['ID'],60 messageDetails['RESPONSE']['json']['entity']['id'])61 PreCheck.assertPreCheckResponse(preCheckResponse, 200)62 PreCheck.assertPrecheckStatus(preCheckResponse,preCheckErrorList)63 finally:64 IrisHelper.createNewDummyGateway(channel)65 @pytest.mark.parametrize('campaignType,testControlType,listType,channel,messageInfo,preCheckErrorList', [66 ('LIVE', 'ORG', 'UPLOAD', 'EMAIL',67 {'scheduleType': {'type': 'IMMEDIATE'}, 'offerType': 'PLAIN', 'messageStrategy': {'type': 'DEFAULT'},68 'channels': ['SMS', 'EMAIL'], 'useTinyUrl': False, 'encryptUrl': False, 'skipRateLimit': True},69 [{'errorType': 'CAMPAIGN_NOT_AUTHORIZED', 'parameters': {}},70 {'errorType': 'GATEWAY_NOT_AVAILABLE', 'parameters': {}}])71 ])72 def test_irisv2_message_precheck_create_upload_email_immediate_GatewayNotAvialable(self, campaignType,73 testControlType,74 listType,75 channel,76 messageInfo,77 preCheckErrorList):78 try:79 messageDetails = CreateMessage.create(campaignType, testControlType, listType, channel, messageInfo,80 updateNode=True, lockNode=True)81 CreateMessage.assertResponse(messageDetails['RESPONSE'], 200)82 IrisHelper.disableDomainGatewayMapId(channel)83 preCheckResponse = PreCheck.executePrecheck(84 constant.config['node'][campaignType][testControlType]['CAMPAIGN']['ID'],85 messageDetails['RESPONSE']['json']['entity']['id'])86 PreCheck.assertPreCheckResponse(preCheckResponse, 200)87 PreCheck.assertPrecheckStatus(preCheckResponse,preCheckErrorList)88 finally:89 IrisHelper.createNewDummyGateway(channel)90 @pytest.mark.parametrize('campaignType,testControlType,listType,channel,messageInfo,preCheckErrorList', [91 ('LIVE', 'ORG', 'UPLOAD', 'EMAIL',92 {'scheduleType': {'type': 'PARTICULARDATE'}, 'offerType': 'PLAIN', 'messageStrategy': {'type': 'DEFAULT'},93 'channels': ['SMS', 'EMAIL'], 'useTinyUrl': False, 'encryptUrl': False, 'skipRateLimit': True},94 [{'errorType': 'CAMPAIGN_NOT_AUTHORIZED', 'parameters': {}},95 {'errorType': 'GATEWAY_NOT_AVAILABLE', 'parameters': {}}])96 ])97 def test_irisv2_message_precheck_create_upload_email_particularDate_GatewayNotAvialable(self, campaignType,98 testControlType,99 listType,100 channel,101 messageInfo,102 preCheckErrorList):103 try:104 messageDetails = CreateMessage.create(campaignType, testControlType, listType, channel, messageInfo,105 updateNode=True, lockNode=True)106 CreateMessage.assertResponse(messageDetails['RESPONSE'], 200)107 IrisHelper.disableDomainGatewayMapId(channel)108 preCheckResponse = PreCheck.executePrecheck(109 constant.config['node'][campaignType][testControlType]['CAMPAIGN']['ID'],110 messageDetails['RESPONSE']['json']['entity']['id'])111 PreCheck.assertPreCheckResponse(preCheckResponse, 200)112 PreCheck.assertPrecheckStatus(preCheckResponse,preCheckErrorList)113 finally:114 IrisHelper.createNewDummyGateway(channel)115 @pytest.mark.parametrize('campaignType,testControlType,listType,channel,messageInfo,preCheckErrorList', [116 ('LIVE', 'ORG', 'UPLOAD', 'MOBILE_PUSH',117 {'scheduleType': {'type': 'IMMEDIATE'}, 'offerType': 'PLAIN', 'messageStrategy': {'type': 'DEFAULT',118 "android": {119 "contentType": "TEXT",120 "secondary_cta": {121 "enable": False,122 "value": None},123 "primary_cta": {124 "enable": False,125 "value": None},126 "custom": False},127 "ios": {128 "contentType": "TEXT",129 "secondary_cta": {130 "enable": False,131 "value": None},132 "primary_cta": {133 "enable": False,134 "value": None},135 "custom": False}},136 'channels': ['MOBILE_PUSH'], 'useTinyUrl': False, 'encryptUrl': False, 'skipRateLimit': True},137 [{'errorType': 'CAMPAIGN_NOT_AUTHORIZED', 'parameters': {}}])138 ])139 def test_irisv2_message_precheck_create_upload_mobilePush_immediate_NotAuthorized(self, campaignType,140 testControlType,141 listType,142 channel, messageInfo,143 preCheckErrorList):144 messageDetails = CreateMessage.create(campaignType, testControlType, listType, channel, messageInfo,145 updateNode=True, lockNode=True)146 CreateMessage.assertResponse(messageDetails['RESPONSE'], 200)147 preCheckResponse = PreCheck.executePrecheck(148 constant.config['node'][campaignType][testControlType]['CAMPAIGN']['ID'],149 messageDetails['RESPONSE']['json']['entity']['id'])150 PreCheck.assertPreCheckResponse(preCheckResponse, 200)151 PreCheck.assertPrecheckStatus(preCheckResponse,preCheckErrorList)152 @pytest.mark.parametrize('campaignType,testControlType,listType,channel,messageInfo,preCheckErrorList', [153 ('LIVE', 'ORG', 'UPLOAD', 'MOBILE_PUSH',154 {'scheduleType': {'type': 'PARTICULARDATE'}, 'offerType': 'PLAIN', 'messageStrategy': {'type': 'DEFAULT',155 "android": {156 "contentType": "TEXT",157 "secondary_cta": {158 "enable": False,159 "value": None},160 "primary_cta": {161 "enable": False,162 "value": None},163 "custom": False},164 "ios": {165 "contentType": "TEXT",166 "secondary_cta": {167 "enable": False,168 "value": None},169 "primary_cta": {170 "enable": False,171 "value": None},172 "custom": False}},173 'channels': ['MOBILE_PUSH'], 'useTinyUrl': False, 'encryptUrl': False, 'skipRateLimit': True},174 [{'errorType': 'CAMPAIGN_NOT_AUTHORIZED', 'parameters': {}}])175 ])176 def test_irisv2_message_precheck_create_upload_mobilePush_particularDate_NotAuthorized(self, campaignType,177 testControlType,178 listType,179 channel, messageInfo,180 preCheckErrorList):181 messageDetails = CreateMessage.create(campaignType, testControlType, listType, channel, messageInfo,182 updateNode=True, lockNode=True)183 CreateMessage.assertResponse(messageDetails['RESPONSE'], 200)184 preCheckResponse = PreCheck.executePrecheck(185 constant.config['node'][campaignType][testControlType]['CAMPAIGN']['ID'],186 messageDetails['RESPONSE']['json']['entity']['id'])187 PreCheck.assertPreCheckResponse(preCheckResponse, 200)...
preCheckDbValidation.py
Source:preCheckDbValidation.py
1import time2from src.dbCalls.messageInfo import message_calls3from src.dbCalls.messageInfo import message_info4from src.dbCalls.preCheck import precheck_calls5from src.utilities.assertion import Assertion6from src.utilities.logger import Logger7class PreCheckDBValidation():8 def __init__(self, campaignId, messageId, messageType, cronStatus, messageVersion=0, remindCheck=True,9 executeCheck=True, variantCheck=True, basicCheckStatus=None,10 basicRecurringCheck=None, reloadCheck=None, precheck=None, basicExecuteCheck=None, cdDetail=None,11 messageStatus_remind=None, messageStatus_execute=None, variantStatus='SUCCESS', messageDBDetails=None,12 byPassPrecheckValidation=False):13 Logger.log(14 'Initializing Precheck Validation for campaign Id :{} and MessageId :{}'.format(campaignId, messageId))15 self.campaignId = campaignId16 self.messageID = messageId17 self.messageType = messageType18 self.cronStatus = cronStatus19 self.version = messageVersion20 self.remindCheck = remindCheck21 self.executeCheck = executeCheck22 self.variantCheck = variantCheck23 self.basicCheckStatus = basicCheckStatus24 self.basicRecurringCheck = basicRecurringCheck25 self.reloadCheck = reloadCheck26 self.precheck = precheck27 self.basicExecuteCheck = basicExecuteCheck28 self.cdDetail = cdDetail29 self.messageStatus_remind = messageStatus_remind30 self.messageStatus_execute = messageStatus_execute31 self.varaintStatus = variantStatus32 self.messageDbDetails = messageDBDetails33 self.byPassPrecheckValidation = byPassPrecheckValidation34 def validateMessageFlow(self):35 self.getMessageDBDetails()36 CronValidator(self.messageDbDetails, self.cronStatus).validateCronInfo()37 if self.remindCheck: ReminderValidator(self.campaignId, self.messageID, self.messageType,38 basicCheckStatus=self.basicCheckStatus,39 reloadCheck=self.reloadCheck,40 precheck=self.precheck,41 messageStatus=self.messageStatus_remind,42 byPassPrecheckValidation=self.byPassPrecheckValidation).validate()43 if self.executeCheck: ExecutionValidator(self.campaignId, self.messageID, self.messageType,44 basicExecuteCheck=self.basicExecuteCheck,45 basicRecurringCheck=self.basicRecurringCheck,46 precheck=self.precheck, cdDetail=self.cdDetail,47 messageStatus_execute=self.messageStatus_execute).validate()48 if self.variantCheck: self.validateVariantStatus()49 def getMessageDBDetails(self):50 self.messageDbDetails = message_info(self.messageID,51 version=self.version).messageDbDetail if self.messageDbDetails is None else self.messageDbDetails52 def validateVariantStatus(self):53 dictOfVariantsInDB = precheck_calls().getVariantExecutionStatus(self.campaignId, self.messageID)54 for eachJobDetails in self.messageDbDetails['messageJobDetails_collection']['VARIANT_CREATION']:55 for varient in eachJobDetails['variant_detail']:56 Assertion.constructAssertion(varient['_id'] in dictOfVariantsInDB,57 'Varient ID :{} found in PrecheckVariantStatus DB :{}'.format(58 varient['_id'], dictOfVariantsInDB))59 Assertion.constructAssertion(dictOfVariantsInDB[varient['_id']] == self.varaintStatus,60 'For VariantId :{} , status :{} and expected :{}'.format(varient['_id'],61 dictOfVariantsInDB[62 varient[63 '_id']],64 self.varaintStatus))65class CronValidator():66 def __init__(self, messageDBDetails, cronStatus):67 Logger.log('Initializing Cron Validation with CronStatus :{}'.format(cronStatus))68 self.cronStatus = cronStatus69 self.messageDBDetails = messageDBDetails70 def getCronInfoWithVariants(self):71 mapOfVariantWithCron = dict()72 for eachJobDetails in self.messageDBDetails['messageJobDetails_collection']['VARIANT_CREATION']:73 for varient in eachJobDetails['variant_detail']:74 cronInfo = message_calls().getCronIdWithVariantId(varient['_id'], self.cronStatus)75 if cronInfo is None: raise Exception('NoCronEntryFoundForVariantId :{}'.format(varient))76 mapOfVariantWithCron.update({varient['_id']: cronInfo})77 return mapOfVariantWithCron78 def validateCronInfo(self):79 mapOfVariantWithCron = self.getCronInfoWithVariants()80 Logger.log('MapOfVariantWithStatus :{}'.format(mapOfVariantWithCron))81 for eachVariant in mapOfVariantWithCron:82 if mapOfVariantWithCron[eachVariant] is None:83 Assertion.addValidationMessage('For VariantId :{} cronInfo not Found'.format(eachVariant))84 Assertion.constructAssertion(mapOfVariantWithCron[eachVariant]['status'] in self.cronStatus,85 'Actual CronStatus is {} for varient :{} and expected :{}'.format(86 mapOfVariantWithCron[eachVariant]['status'], eachVariant, self.cronStatus))87class ReminderValidator():88 def __init__(self, campaignId, messageId, messageType, basicCheckStatus=None,89 reloadCheck=None, precheck=None, messageStatus=None, byPassPrecheckValidation=False):90 Logger.log('Remnder Validation Initialized with campaignId :{} and MessageId :{}'.format(campaignId, messageId))91 self.campaignId = campaignId92 self.messageId = messageId93 self.messageType = messageType94 self.byPassPrecheckValidation = byPassPrecheckValidation95 self.basicCheckStatus = 'SUCCESS' if basicCheckStatus is None else basicCheckStatus96 self.reloadCheck = {'GROUP_RELOAD_NFS': 'SUCCESS',97 'GROUP_RELOAD_CREATE_AUDIENCE': 'SUCCESS'} if reloadCheck is None else reloadCheck98 self.precheck = {'status': 'SUCCESS', 'errorDescription': None} if precheck is None else precheck99 self.messageStatus = 'SUCCESS' if messageStatus is None else messageStatus100 def validate(self):101 self.validateBasicCheck()102 if self.messageType.upper() == 'RECURRING':103 self.validateAudienceGroupReload()104 if self.messageType.upper() != 'IMMEDIATE': self.validatePrecheck()105 def validateBasicCheck(self):106 basicRemindCheck = precheck_calls('REMIND').getJobDetailFromPreExecutionJobStatus(self.campaignId,107 self.messageId,108 'BASIC_REMIND_CHECKS')109 Assertion.constructAssertion(basicRemindCheck['job_status'] == self.basicCheckStatus,110 'For BasicCheck actual Status :{} and expected Status :{}'.format(111 basicRemindCheck['job_status'], self.basicCheckStatus))112 def validateAudienceGroupReload(self):113 for eachJobType in ['GROUP_RELOAD_NFS', 'GROUP_RELOAD_CREATE_AUDIENCE']:114 reloadCheck = precheck_calls('REMIND').getJobDetailFromPreExecutionJobStatus(self.campaignId,115 self.messageId,116 eachJobType)117 Assertion.constructAssertion(reloadCheck['job_status'] == self.reloadCheck[eachJobType],118 'For ReloadCheck actual Status :{} and expected Status :{}'.format(119 reloadCheck['job_status'], self.reloadCheck[eachJobType]))120 def validatePrecheck(self):121 if self.byPassPrecheckValidation: return122 preCheck = precheck_calls('REMIND').getJobDetailFromPreExecutionJobStatus(self.campaignId, self.messageId,123 'PRECHECK')124 Assertion.constructAssertion(preCheck['job_status'] == self.precheck['status'],125 'For precheck actual Status :{} and expected Status :{}'.format(126 preCheck['job_status'], self.precheck['status']))127 if self.precheck['errorDescription'] is not None:128 Assertion.constructAssertion(sorted(preCheck['error_description'].split(',')) == sorted(129 self.precheck['errorDescription'].split(',')),130 'For precheck actual errorDescription :{} and expected Status :{}'.format(131 preCheck['error_description'], self.precheck['errorDescription']))132 def validateMessageStatus(self):133 eventStatus = precheck_calls('REMIND').getMsgDetailFromPreExecutionMessageStatus(self.campaignId,134 self.messageId,135 'REMIND')136 Assertion.constructAssertion(eventStatus['status'] == self.messageStatus,137 'For Event REMIND in preCheckMessageStatus status is :{} and expected is :{}'.format(138 eventStatus['status'], self.messageStatus))139class ExecutionValidator():140 def __init__(self, campaignId, messageId, messageType, basicExecuteCheck=None, precheck=None, cdDetail=None,141 basicRecurringCheck=None, messageStatus_execute=None):142 Logger.log(143 'Execution Validator Initialized with campaignId :{} and MessageId :{}'.format(campaignId, messageId))144 self.campaignId = campaignId145 self.messageId = messageId146 self.messageType = messageType147 self.basicRecurringCheck = 'SUCCESS' if basicRecurringCheck is None else basicRecurringCheck148 self.basicExecuteCheck = 'SUCCESS' if basicExecuteCheck is None else basicExecuteCheck149 self.precheck = {'status': 'SUCCESS', 'errorDescription': None} if precheck is None else precheck150 self.cdDetail = 'SUCCESS' if cdDetail is None else cdDetail151 self.messageStatus = 'SUCCESS' if messageStatus_execute is None else messageStatus_execute152 def validate(self):153 self.validateBasicExecuteCheck()154 if self.messageType.upper() == 'RECURRING':155 self.validateBasicRecurringCheck()156 if self.messageType.upper() == 'IMMEDIATE':157 self.validatePrecheck()158 if self.precheck['status'] == 'SUCCESS': self.validateCommunicationDetailsCreation()159 def validateBasicExecuteCheck(self):160 basicExecuteCheck = precheck_calls('EXECUTE').getJobDetailFromPreExecutionJobStatus(self.campaignId,161 self.messageId,162 'BASIC_EXECUTE_CHECKS')163 Assertion.constructAssertion(basicExecuteCheck['job_status'] == self.basicExecuteCheck,164 'For BasicCheck actual Status :{} and expected Status :{}'.format(165 basicExecuteCheck['job_status'], self.basicExecuteCheck))166 def validatePrecheck(self):167 preCheck = precheck_calls('EXECUTE').getJobDetailFromPreExecutionJobStatus(self.campaignId, self.messageId,168 'PRECHECK')169 Assertion.constructAssertion(preCheck['job_status'] == self.precheck['status'],170 'For precheck actual Status :{} and expected Status :{}'.format(171 preCheck['job_status'], self.precheck['status']))172 if self.precheck['errorDescription'] is not None:173 Assertion.constructAssertion(sorted(preCheck['error_description'].split(',')) == sorted(174 self.precheck['errorDescription'].split(',')),175 'For precheck actual errorDescription :{} and expected Status :{}'.format(176 preCheck['error_description'], self.precheck['errorDescription']))177 def validateCommunicationDetailsCreation(self):178 cdDetail = precheck_calls('EXECUTE').getJobDetailFromPreExecutionJobStatus(self.campaignId, self.messageId,179 'COMMUNICATION_DETAIL_CREATION')180 Assertion.constructAssertion(cdDetail['job_status'] == self.cdDetail,181 'For BasicCheck actual Status :{} and expected Status :{}'.format(182 cdDetail['job_status'], self.cdDetail))183 def validateMessageStatus(self):184 eventStatus = precheck_calls('EXECUTE').getMsgDetailFromPreExecutionMessageStatus(self.campaignId,185 self.messageId,186 'EXECUTE')187 Assertion.constructAssertion(eventStatus['status'] == self.messageStatus,188 'For Event EXECUTE in preCheckMessageStatus status is :{} and expected is :{}'.format(189 eventStatus['status'], self.messageStatus))190 def validateBasicRecurringCheck(self):191 basicRecurringCheck = precheck_calls('EXECUTE').getJobDetailFromPreExecutionJobStatus(self.campaignId,192 self.messageId,193 'RECURRING_CHECKS')194 Assertion.constructAssertion(basicRecurringCheck['job_status'] == self.basicRecurringCheck,195 'For BasicRecurringCheck actual Status :{} and expected Status :{}'.format(196 basicRecurringCheck['job_status'], self.basicRecurringCheck))197class Precheck_calls():198 def waitForJobTypeUpdate(self, campaignId, messageId, event, job, expectedStatus, expectedError=None):199 for _ in range(30):200 jobDetail = precheck_calls(event).getJobDetailFromPreExecutionJobStatus(campaignId, messageId, job)201 status = jobDetail['job_status']202 error = jobDetail['error_description']203 if status == expectedStatus:204 if expectedError is None:205 break206 elif expectedError == error:207 break208 else:209 time.sleep(10)210 else:...
Automation Testing Tutorials
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run autotest automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
