KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use assert_response_status method in localstack
Best Python code snippet using localstack_python
app_fl_pipeline.py
Source:app_fl_pipeline.py 
...61 data_dict = {'workspaceId': workspace_id,62 'status': 'RUNNING', 'size': size}63 resp = sess.get(url, params=data_dict, timeout=default_timeout)64 assert(resp.status_code == 200)65 if not assert_response_status(resp):66 return []67 fl_list = resp.json()['data']['appList']68 return [{'instanceId': fl['instanceId'], 'status': fl['status'], 'appName': fl['appName']} for fl in fl_list]69@retry(stop_max_attempt_number=default_retry)70def list_fl_pipelines(sess, instance_id, template_id):71 '''72 List all pipelines for a running flowengine.73 '''74 url = base_url + \75 f'automl-engine/{instance_id}/automl/v1/pipeline/{template_id}/list'76 resp = sess.get(url, timeout=default_timeout)77 assert(resp.status_code == 200)78 if not assert_response_status(resp):79 return []80 pipelines = resp.json()['data']81 if profile == 'v2':82 pipelines = pipelines['engineJobPipelineTemplateList']83 return [{'template_id': p['engineTemplateId'], 'id': p['id'], 'status': p['data']['status']}84 for p in pipelines]85@retry(stop_max_attempt_number=default_retry)86def run_fl_pipeline(sess, instance_id, template_id, pipeline_id):87 url = base_url + \88 f'automl-engine/{instance_id}/automl/v1/pipeline/{template_id}/{pipeline_id}/start'89 resp = sess.post(url, data=r'{}', timeout=default_timeout)90 assert(resp.status_code == 200)91 if assert_response_status(resp):92 status = resp.json()['data']['status']93 if status == 'running':94 return True95 return False96@retry(stop_max_attempt_number=default_retry)97def list_fl_pipeline_tasks(sess, instance_id, pipeline_id):98 '''99 List all history tasks of a flowengine pipeline.100 '''101 url = base_url + \102 f'automl-engine/{instance_id}/automl/v1/pipeline/{pipeline_id}/historyList'103 resp = sess.get(url, timeout=default_timeout)104 assert(resp.status_code == 200)105 if not assert_response_status(resp):106 return []107 tasks = resp.json()['data']108 return sorted([{'id': t['id'], 'status':t['status']} for t in tasks], key=lambda x: x['id'])109@retry(stop_max_attempt_number=default_retry)110def stop_fl_pipeline_task(sess, instance_id, pipeline_id, task_id):111 url = base_url + \112 f'automl-engine/{instance_id}/automl/v1/pipeline/{task_id}/stopHistory?engineId={pipeline_id}'113 resp = sess.delete(url, data=r'{}', timeout=default_timeout)114 assert(resp.status_code == 200)115 return True if assert_response_status(resp) else False116@retry(stop_max_attempt_number=default_retry)117def resume_fl_pipeline_task(sess, instance_id, pipeline_id, task_id):118 url = base_url + \119 f'automl-engine/{instance_id}/automl/v1/pipeline/{task_id}/resumeHistory?engineId={pipeline_id}'120 resp = sess.post(url, data=r'{}', timeout=default_timeout)121 assert(resp.status_code == 200)122 return True if assert_response_status(resp) else False123def update_and_run_pipeline_demo(sess):124 '''125 Update param of a pipeline node, and run pipeline.126 '''127 fl_id = 60128 data_dict = {'page': 1, 'size': 20}129 url = base_url + f'automl-engine/{fl_id}/automl/v1/pipeline/1/list'130 resp = sess.get(url, params=data_dict, timeout=default_timeout)131 assert(resp.status_code == 200)132 # get pipeline133 pipeline_id = 1134 pipeline = {}135 pipelines = resp.json()['data']['engineJobPipelineTemplateList']136 for p in pipelines:137 if p['id'] == pipeline_id:138 pipeline = p139 assert(pipeline is not None)140 # get node (job)141 node_name = 'offline_job01'142 node = {}143 nodes = pipeline['data']['nodes']144 for n in nodes:145 if n['name'] == node_name:146 node = n147 assert(node is not None)148 # get node param149 key_name = 'param1'150 param = {}151 params = node['params']['values']152 for p in params:153 if p['key'] == key_name:154 param = p155 assert(param is not None)156 logger.info('to update: ' + str(param))157 # update node param of pipeline158 # url = base_url + f'automl-engine/{fl_id}/automl/v1/pipeline/template/1/update'159# -----------------------------------------------160# Get info and operator for flowengine template161# -----------------------------------------------162@retry(stop_max_attempt_number=default_retry)163def list_template_pipelines(sess, template_id, size=default_page_size):164 '''165 List all pipelines of template.166 '''167 url = base_url + \168 f'template-market/v1/pipeline/template/{template_id}/list/byPage?size={size}'169 resp = sess.get(url, timeout=default_timeout)170 assert(resp.status_code == 200)171 if not assert_response_status(resp):172 return []173 return [{'id': template['id'], 'describe': template['data']['describe']}174 for template in resp.json()['data']['engineJobPipelineTemplateList']]175@retry(stop_max_attempt_number=default_retry)176def list_template_jobs(sess, template_id, size=default_page_size):177 '''178 List all jobs of template.179 '''180 url = base_url + \181 f'template-market/v1/job/template/{template_id}/list/byPage?page=1&size={size}'182 resp = sess.get(url, timeout=default_timeout)183 assert(resp.status_code == 200)184 if not assert_response_status(resp):185 return []186 return [{'id': job['id'], 'name': job['name']}187 for job in resp.json()['data']['engineJobTemplateList']]188def list_template_pipeline_jobs(sess, template_id, pipeline_id, size=default_page_size):189 data = get_template_pipeline_data(sess, template_id, pipeline_id, size)190 return [{'name': node['name']} for node in data['data']['nodes']] if len(data) > 0 else []191@retry(stop_max_attempt_number=default_retry)192def get_template_pipeline_data(sess, template_id, pipeline_id, size=10):193 '''194 Get meta data of a pipeline.195 '''196 url = base_url + \197 f'template-market/v1/pipeline/template/{template_id}/list/byPage?size={size}'198 resp = sess.get(url, timeout=default_timeout)199 assert(resp.status_code == 200)200 if not assert_response_status(resp):201 return {}202 pipelines = [p for p in resp.json()['data']['engineJobPipelineTemplateList']203 if str(p['id']) == pipeline_id]204 return pipelines[0] if len(pipelines) > 0 else {}205@retry(stop_max_attempt_number=default_retry)206def get_template_job_data(sess, template_id, job_id, size=default_page_size):207 '''208 Get meta data of a job.209 '''210 url = base_url + \211 f'template-market/v1/job/template/{template_id}/list/byPage?page=1&size={size}'212 resp = sess.get(url, timeout=default_timeout)213 assert(resp.status_code == 200)214 if not assert_response_status(resp):215 return {}216 jobs = [job for job in resp.json()['data']['engineJobTemplateList']217 if str(job['id']) == job_id]218 return jobs[0] if len(jobs) > 0 else {}219@retry(stop_max_attempt_number=default_retry)220def create_template_pipeline_job(sess, template_id, pipeline_template_name):221 '''222 Create a pipeline job of template.223 '''224 url = base_url + \225 f'template-market/v1/job/template/{template_id}/create'226 data = build_pipeline_job_data(template_id, pipeline_template_name)227 resp = sess.post(url, data=json.dumps(data), timeout=default_timeout)228 assert(resp.status_code == 200)229 return True if assert_response_status(resp) else False230def build_pipeline_job_data(template_id, template_name):231 data = {'name': template_name, 'engineTemplateId': template_id,232 'indexType': 'DAG', 'type': 'TASK', 'system': False}233 outputCfg = data['outputConfig'] = {}234 outputCfg['outputType'] = 'Table'235 execCfg = data['executionConfig'] = {}236 execCfg['runMode'] = 'SINGLE'237 execCfg['interval'] = 0238 execCfg['intervalUnit'] = 'MINUTE'239 execCfg['trigger'] = 'ModelTrainFinishEvent'240 return data241@retry(stop_max_attempt_number=default_retry)242def delete_template_pipeline_job(sess, template_id, job_id):243 '''244 Delete a pipeline job of template.245 '''246 url = base_url + \247 f'template-market/v1/job/template/{template_id}/{job_id}/delete'248 resp = sess.delete(url, data=r'{}', timeout=default_timeout)249 assert(resp.status_code == 200)250 return True if assert_response_status(resp) else False251@retry(stop_max_attempt_number=default_retry)252def delete_template_pipeline(sess, template_id, pipeline_id):253 '''254 Delete a pipeline of template.255 '''256 url = base_url + \257 f'template-market/v1/pipeline/template/{template_id}/{pipeline_id}/delete'258 resp = sess.delete(url, data=r'{}', timeout=default_timeout)259 assert(resp.status_code == 200)260 return True if assert_response_status(resp) else False261# -----------------------------------------------262# Copy and edit pipeline and job of template263# -----------------------------------------------264@retry(stop_max_attempt_number=default_retry)265def copy_template_pipeline(sess, template_id, src_pipeline_id, pipeline_name):266 '''267 Create a new pipeline of template from a source pipeline.268 '''269 url = base_url + \270 f'template-market/v1/pipeline/template/{template_id}/create'271 data = get_template_pipeline_data(sess, template_id, src_pipeline_id)272 if len(data) == 0:273 logger.error(f'srouce pipeline(id={src_pipeline_id}) is not found!')274 return False275 else:276 data = data['data']277 data['name'] = pipeline_name278 data['describe'] = f'pipeline copied from pipelineID={src_pipeline_id}.'279 req_data = {'engineTemplateId': template_id,280 'pipelineKey': pipeline_name + '_' + str(uuid.uuid1())[:8]}281 req_data['data'] = data282 resp = sess.post(url, data=json.dumps(req_data), timeout=default_timeout)283 assert(resp.status_code == 200)284 return True if assert_response_status(resp) else False285@retry(stop_max_attempt_number=default_retry)286def copy_template_job(sess, template_id, src_job_id, job_name):287 '''288 Create a new job of template from a source job.289 '''290 url = base_url + f'template-market/v1/job/template/{template_id}/create'291 req_data = get_template_job_data(sess, template_id, src_job_id)292 if len(req_data) == 0:293 logger.error(f'srouce job(id={src_job_id}) is not found!')294 return False295 req_data['id'] = None296 req_data['name'] = job_name297 req_data['desc'] = f'job copied from jobID={src_job_id}.'298 resp = sess.post(url, data=json.dumps(req_data), timeout=default_timeout)299 assert(resp.status_code == 200)300 return True if assert_response_status(resp) else False301@retry(stop_max_attempt_number=default_retry)302def update_template_pipeline(sess, template_id, pipeline_id):303 '''304 Update existing pipeline of template.305 Example: update pipeline context key value.306 '''307 url = base_url + \308 f'template-market/v1/pipeline/template/{template_id}/update'309 req_data = get_template_pipeline_data(sess, template_id, pipeline_id)310 context_kv = req_data['data']['pipelineParams']['values']311 # for test312 if len(context_kv) > 0:313 context_kv[0]['describe'] = 'new test kv desc.'314 resp = sess.put(url, data=json.dumps(req_data), timeout=default_timeout)315 assert(resp.status_code == 200)316 return True if assert_response_status(resp) else False317@retry(stop_max_attempt_number=default_retry)318def update_template_job(sess, template_id, job_id, params):319 '''320 Update existing job of template.321 Example: update job name and description.322 '''323 url = base_url + f'template-market/v1/job/template/{template_id}/update'324 req_data = get_template_job_data(sess, template_id, job_id)325 for k, v in params.kv_map.items():326 req_data[k] = v327 resp = sess.put(url, data=json.dumps(req_data), timeout=default_timeout)328 assert(resp.status_code == 200)329 return True if assert_response_status(resp) else False330class JobParams(object):331 '''332 Job parameters for create and update a job of template.333 '''334 def __init__(self):335 self.kv_map = {}336 def set_name(self, name):337 self.kv_map['name'] = name338 return self339 def set_desc(self, text):340 self.kv_map['desc'] = text341 return self342# -----------------------------------------------343# Common344# -----------------------------------------------345def log_request_info(req):346 logger.debug('request url: %s' % req.url)347 logger.debug('request headers: %s' % req.headers)348 logger.debug('request method: %s' % req.method)349 logger.debug('request body: %s' % req.body)350def assert_response_status(resp):351 cond = (str(resp.json()['status']) == '0')352 if not cond:353 logger.error('***** assert response status FAILED.')354 logger.error('***** response:\n' + resp.text)355 return cond356class MyRequest(object):357 '''358 Http request with common logs and assert.359 '''360 def get(self, *args, **kwargs):361 resp = requests.get(*args, **kwargs)362 self._log_request_info(resp)363 self._log_response_info(resp)364 self._assert_success(resp)...
test_api.py
Source:test_api.py
...10from rest_framework.authtoken.models import Token11from nose.tools import eq_12from oneanddone.users.tests import UserFactory13class APITests(APITestCase):14 def assert_response_status(self, response, expected_status):15 eq_(response.status_code, expected_status,16 "Test Failed, got response status: %s, expected status: %s" %17 (response.status_code, expected_status))18 def setUp(self):19 self.client_user = UserFactory.create()20 # Add add/change/delete user permission for client21 delete_permission = Permission.objects.get(codename='delete_user')22 add_permission = Permission.objects.get(codename='add_user')23 change_permission = Permission.objects.get(codename='change_user')24 self.client_user.user_permissions.add(add_permission, change_permission,25 delete_permission)26 self.token = Token.objects.create(user=self.client_user)27 self.uri = '/api/v1/user/'28 def test_change_user_profile_data(self):29 """30 Test Change User Profile Data(name, username and privacy_policy_accepted)31 """32 self.client.credentials(HTTP_AUTHORIZATION='Token ' + self.token.key)33 # Create a new user34 user_data = {'username': 'testname', 'email': 'test@test.com',35 'profile': {'name': 'Test Name', 'username': 'testname', 'privacy_policy_accepted': True}}36 response = self.client.post(self.uri, user_data, format='json')37 # Change Profile Data(name, username, privacy_policy_accepted)38 changed_data = {'username': 'testname', 'email': 'test@test.com',39 'profile': {'name': 'Changed Test Name', 'username': 'testname123', 'privacy_policy_accepted': False}}40 user_uri = self.uri + user_data['email'] + '/'41 response = self.client.patch(user_uri, changed_data, format='json')42 self.assert_response_status(response, status.HTTP_200_OK)43 response_data = json.loads(response.content)44 eq_(response_data['profile'], changed_data['profile'])45 def test_create_new_user(self):46 """47 Test Create new user with Profile Data(name, username and privacy_policy_accepted)48 """49 self.client.credentials(HTTP_AUTHORIZATION='Token ' + self.token.key)50 user_data = {'username': 'testuser', 'email': 'test@test.com',51 'profile': {'name': 'Test Name', 'username': 'testuser', 'privacy_policy_accepted': True}}52 response = self.client.post(self.uri, user_data, format='json')53 self.assert_response_status(response, status.HTTP_201_CREATED)54 response_data = json.loads(response.content)55 eq_(response_data['profile'], user_data['profile'])56 def test_client_with_false_token(self):57 """58 Test user list, user details, user deletion for user with false token59 """60 invalid_key = 'd81e33c57b2d9471f4d6849bab3cb233b3b30468'61 false_token = ''.join(random.sample(invalid_key, len(invalid_key)))62 self.client.credentials(HTTP_AUTHORIZATION='Token ' + false_token)63 response = self.client.get(reverse('api-user'))64 self.assert_response_status(response, status.HTTP_401_UNAUTHORIZED)65 test_user = UserFactory.create()66 user_uri = self.uri + test_user.email + '/'67 get_response = self.client.get(user_uri)68 self.assert_response_status(get_response, status.HTTP_401_UNAUTHORIZED)69 delete_response = self.client.delete(user_uri)70 self.assert_response_status(delete_response, status.HTTP_401_UNAUTHORIZED)71 def test_delete_user(self):72 """73 Test DELETE user with particular email for authenticated user74 """75 self.client.credentials(HTTP_AUTHORIZATION='Token ' + self.token.key)76 test_user = UserFactory.create()77 user_uri = self.uri + test_user.email + '/'78 # Make a DELETE request79 delete_response = self.client.delete(user_uri)80 self.assert_response_status(delete_response, status.HTTP_204_NO_CONTENT)81 # Verify that the user has been deleted82 get_response = self.client.get(user_uri)83 self.assert_response_status(get_response, status.HTTP_404_NOT_FOUND)84 def test_forbidden_client(self):85 """86 Test user deletion by an authenticated but forbidden client87 """88 forbidden_user = UserFactory.create()89 forbidden_token = Token.objects.create(user=forbidden_user)90 self.client.credentials(HTTP_AUTHORIZATION='Token ' + forbidden_token.key)91 test_user = UserFactory.create()92 user_uri = self.uri + test_user.email + '/'93 # Make a DELETE request94 delete_response = self.client.delete(user_uri)95 self.assert_response_status(delete_response, status.HTTP_403_FORBIDDEN)96 def test_get_user_details(self):97 """98 Test GET details of a user with particular email for authenticated user99 """100 self.client.credentials(HTTP_AUTHORIZATION='Token ' + self.token.key)101 test_user = UserFactory.create()102 user_uri = self.uri + test_user.email + '/'103 response = self.client.get(user_uri)104 self.assert_response_status(response, status.HTTP_200_OK)105 def test_get_user_list(self):106 """107 Test GET user list for authenticated user108 """109 header = {'HTTP_AUTHORIZATION': 'Token {}'.format(self.token)}110 response = self.client.get(reverse('api-user'), {}, **header)111 self.assert_response_status(response, status.HTTP_200_OK)112 def test_unauthenticated_client(self):113 """114 Test user list, user details, user deletion for unauthenticated user115 """116 response = self.client.get(reverse('api-user'))117 self.assert_response_status(response, status.HTTP_401_UNAUTHORIZED)118 test_user = UserFactory.create()119 user_uri = self.uri + test_user.email + '/'120 get_response = self.client.get(user_uri)121 self.assert_response_status(get_response, status.HTTP_401_UNAUTHORIZED)122 delete_response = self.client.delete(user_uri)...
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 localstack 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?
