How to use validate_response method in tempest
Best Python code snippet using tempest_python
test_virtualnetwork.py
Source:test_virtualnetwork.py 
...35 vcn_dns_label = util.random_name('vcn', insert_underscore=False)36 result = self.invoke(37 ['vcn', 'create', '--compartment-id', util.COMPARTMENT_ID, '--display-name', vcn_name, '--cidr-block', cidr_block, '--dns-label', vcn_dns_label])38 self.vcn_ocid = util.find_id_in_response(result.output)39 util.validate_response(result, expect_etag=True)40 util.wait_until(['network', 'vcn', 'get', '--vcn-id', self.vcn_ocid], 'AVAILABLE',41 max_wait_seconds=300)42 result = self.invoke(['vcn', 'list', '--compartment-id', util.COMPARTMENT_ID])43 util.validate_response(result)44 vcn_name = vcn_name + "_updated"45 result = self.invoke(['vcn', 'update', '--vcn-id', self.vcn_ocid, '--display-name', vcn_name])46 util.validate_response(result, expect_etag=True)47 result = self.invoke(['vcn', 'get', '--vcn-id', self.vcn_ocid])48 util.validate_response(result, expect_etag=True)49 vcn_response = json.loads(result.output)50 assert vcn_response['data']['dns-label'] == vcn_dns_label51 @util.log_test52 def subtest_security_list_operations(self):53 sl_name = util.random_name('cli_test_security_list')54 egress_rules = util.remove_outer_quotes(oraclebmc_cli.core_cli_extended.network_create_security_list_egress_security_rules_example)55 ingress_rules = util.remove_outer_quotes(oraclebmc_cli.core_cli_extended.network_create_security_list_ingress_security_rules_example)56 result = self.invoke(57 ['security-list', 'create',58 '--compartment-id', util.COMPARTMENT_ID,59 '--display-name', sl_name,60 '--vcn-id', self.vcn_ocid,61 '--egress-security-rules', egress_rules,62 '--ingress-security-rules', ingress_rules63 ])64 self.sl_ocid = util.find_id_in_response(result.output)65 util.validate_response(result, expect_etag=True)66 util.wait_until(['network', 'security-list', 'get', '--security-list-id', self.sl_ocid], 'AVAILABLE',67 max_wait_seconds=300)68 result = self.invoke(69 ['security-list', 'list', '--compartment-id', util.COMPARTMENT_ID, '--vcn-id', self.vcn_ocid])70 util.validate_response(result)71 result = self.invoke(['security-list', 'get', '--security-list-id', self.sl_ocid])72 util.validate_response(result, expect_etag=True)73 sl_name = sl_name + "_updated"74 egress_rules_v2 = """[{"destination": "10.0.2.0/24", "protocol": "6", "tcpOptions": {"destinationPortRange": {"max": 1522, "min": 1522}}}]"""75 ingress_rules_v2 = """[{"protocol": "6", "source": "10.0.1.0/25", "tcpOptions": {"destinationPortRange": {"max": 1521, "min": 1521}}}]"""76 # TODO: A short sleep before every security list update to allow for replication.77 time.sleep(20)78 # Force update on all fields79 result = self.invoke(['security-list', 'update',80 '--security-list-id', self.sl_ocid,81 '--display-name', sl_name,82 '--egress-security-rules', egress_rules,83 '--ingress-security-rules', ingress_rules,84 '--force'])85 util.validate_response(result, expect_etag=True)86 time.sleep(20)87 # update display name only - does not show a prompt88 result = self.invoke(['security-list', 'update', '--security-list-id', self.sl_ocid, '--display-name', sl_name])89 util.validate_response(result, expect_etag=True)90 time.sleep(20)91 # update egress-rules, confirm y92 result = self.invoke(['security-list', 'update', '--security-list-id', self.sl_ocid, '--egress-security-rules', egress_rules_v2], input='y')93 util.validate_response(result, json_response_expected=False)94 time.sleep(20)95 # update ingress-rules, confirm y96 result = self.invoke(97 ['security-list', 'update', '--security-list-id', self.sl_ocid, '--ingress-security-rules', ingress_rules_v2], input='y')98 util.validate_response(result, json_response_expected=False)99 time.sleep(20)100 # update both, confirm y101 result = self.invoke(102 ['security-list', 'update', '--security-list-id', self.sl_ocid, '--ingress-security-rules',103 ingress_rules, '--egress-security-rules', egress_rules], input='y')104 util.validate_response(result, json_response_expected=False)105 time.sleep(20)106 # update egress-rules, confirm n107 result = self.invoke(108 ['security-list', 'update', '--security-list-id', self.sl_ocid, '--egress-security-rules', egress_rules_v2],109 input='n')110 assert result.exit_code != 0111 # update egress-rules, force112 result = self.invoke(113 ['security-list', 'update', '--security-list-id', self.sl_ocid, '--egress-security-rules', egress_rules_v2, '--force'])114 util.validate_response(result, expect_etag=True)115 @util.log_test116 def subtest_security_list_stateless_rules(self):117 time.sleep(10)118 stateless_egress_rule = """[{"destination": "10.0.2.0/24", "protocol": "6", "tcpOptions": {"destinationPortRange": {"max": 2, "min": 1}}, "isStateless":"true"}]"""119 result = self.invoke(120 ['security-list', 'update', '--security-list-id', self.sl_ocid, '--egress-security-rules', stateless_egress_rule,121 '--force'])122 util.validate_response(result, expect_etag=True)123 assert json.loads(result.output)["data"]["egress-security-rules"][0]["is-stateless"] is True124 time.sleep(20)125 explicit_stateful_egress_rule = """[{"destination": "10.0.2.0/24", "protocol": "6", "tcpOptions": {"destinationPortRange": {"max": 2, "min": 1}}, "isStateless":"false"}]"""126 result = self.invoke(127 ['security-list', 'update', '--security-list-id', self.sl_ocid, '--egress-security-rules',128 explicit_stateful_egress_rule,129 '--force'])130 util.validate_response(result, expect_etag=True)131 assert json.loads(result.output)["data"]["egress-security-rules"][0]["is-stateless"] is False132 time.sleep(20)133 implicit_stateful_egress_rule = """[{"destination": "10.0.2.0/24", "protocol": "17", "udpOptions": {"destinationPortRange": {"max": 2, "min": 1}, "sourcePortRange": {"max": 4, "min": 3}}}]"""134 result = self.invoke(135 ['security-list', 'update', '--security-list-id', self.sl_ocid, '--egress-security-rules',136 implicit_stateful_egress_rule,137 '--force'])138 util.validate_response(result, expect_etag=True)139 assert json.loads(result.output)["data"]["egress-security-rules"][0]["is-stateless"] is None140 @util.log_test141 def subtest_subnet_operations(self):142 subnet_name = util.random_name('cli_test_subnet')143 cidr_block = "10.0.0.0/16"144 security_list_ids = util.remove_outer_quotes(oraclebmc_cli.core_cli_extended.network_create_subnet_security_list_ids_example.format(sl_id=self.sl_ocid))145 subnet_dns_label = util.random_name('subnet', insert_underscore=False)146 result = self.invoke(147 ['subnet', 'create',148 '--compartment-id', util.COMPARTMENT_ID,149 '--availability-domain', util.AVAILABILITY_DOMAIN,150 '--display-name', subnet_name,151 '--vcn-id', self.vcn_ocid,152 '--cidr-block', cidr_block,153 '--security-list-ids', security_list_ids,154 '--dns-label', subnet_dns_label155 ])156 self.subnet_ocid = util.find_id_in_response(result.output)157 util.validate_response(result, expect_etag=True)158 util.wait_until(['network', 'subnet', 'get', '--subnet-id', self.subnet_ocid], 'AVAILABLE',159 max_wait_seconds=300)160 result = self.invoke(['subnet', 'list', '--compartment-id', util.COMPARTMENT_ID, '--vcn-id', self.vcn_ocid])161 util.validate_response(result)162 subnet_name = subnet_name + "_updated"163 result = self.invoke(['subnet', 'update', '--subnet-id', self.subnet_ocid, '--display-name', subnet_name])164 util.validate_response(result, expect_etag=True)165 result = self.invoke(['subnet', 'get', '--subnet-id', self.subnet_ocid])166 util.validate_response(result, expect_etag=True)167 subnet_response = json.loads(result.output)168 assert subnet_response['data']['dns-label'] == subnet_dns_label169 @util.log_test170 def subtest_internet_gateway_operations(self):171 ig_name = util.random_name('cli_test_ig')172 result = self.invoke(173 ['internet-gateway', 'create',174 '--compartment-id', util.COMPARTMENT_ID,175 '--is-enabled', "False",176 '--display-name', ig_name,177 '--vcn-id', self.vcn_ocid178 ])179 self.ig_ocid = util.find_id_in_response(result.output)180 util.validate_response(result, expect_etag=True)181 util.wait_until(['network', 'internet-gateway', 'get', '--ig-id', self.ig_ocid], 'AVAILABLE',182 max_wait_seconds=300)183 result = self.invoke(['internet-gateway', 'list', '--compartment-id', util.COMPARTMENT_ID, '--vcn-id', self.vcn_ocid])184 util.validate_response(result)185 ig_name = ig_name + "_updated"186 result = self.invoke(['internet-gateway', 'update', '--ig-id', self.ig_ocid, '--display-name', ig_name])187 util.validate_response(result, expect_etag=True)188 result = self.invoke(['internet-gateway', 'get', '--ig-id', self.ig_ocid])189 util.validate_response(result, expect_etag=True)190 @util.log_test191 def subtest_cpe_operations(self):192 cpe_name = util.random_name('cli_test_cpe')193 ip_address = "137.254.4.11"194 result = self.invoke(195 ['cpe', 'create',196 '--compartment-id', util.COMPARTMENT_ID,197 '--display-name', cpe_name,198 '--ip-address', ip_address,199 ])200 self.cpe_ocid = util.find_id_in_response(result.output)201 util.validate_response(result, expect_etag=True)202 result = self.invoke(['cpe', 'list', '--compartment-id', util.COMPARTMENT_ID])203 util.validate_response(result)204 cpe_name = cpe_name + "_updated"205 result = self.invoke(['cpe', 'update', '--cpe-id', self.cpe_ocid, '--display-name', cpe_name])206 util.validate_response(result, expect_etag=True)207 result = self.invoke(['cpe', 'get', '--cpe-id', self.cpe_ocid])208 util.validate_response(result, expect_etag=True)209 @util.log_test210 def subtest_dhcp_option_operations(self):211 dhcp_options_name = util.random_name('cli_test_dhcp_options')212 options = util.remove_outer_quotes(oraclebmc_cli.core_cli_extended.network_create_dhcp_options_options_example)213 result = self.invoke(214 ['dhcp-options', 'create',215 '--compartment-id', util.COMPARTMENT_ID,216 '--vcn-id', self.vcn_ocid,217 '--display-name', dhcp_options_name,218 '--options', options219 ])220 util.validate_response(result, expect_etag=True)221 self.dhcp_options_ocid = util.find_id_in_response(result.output)222 util.wait_until(['network', 'dhcp-options', 'get', '--dhcp-id', self.dhcp_options_ocid], 'AVAILABLE')223 result = self.invoke(['dhcp-options', 'list', '--compartment-id', util.COMPARTMENT_ID, '--vcn-id', self.vcn_ocid])224 util.validate_response(result)225 result = self.invoke(['dhcp-options', 'get', '--dhcp-id', self.dhcp_options_ocid])226 util.validate_response(result, expect_etag=True)227 dhcp_options_name = dhcp_options_name + "_updated"228 options_v2 = """[{"type": "DomainNameServer", "customDnsServers": ["202.44.61.10"], "serverType": "CustomDnsServer"},229 {"searchDomainNames": ["testvcn.oraclevcn.com"], "type":"SearchDomain"}]"""230 # update display name only - does not show a confirmation prompt231 result = self.invoke(['dhcp-options', 'update', '--dhcp-id', self.dhcp_options_ocid, '--display-name', dhcp_options_name])232 util.validate_response(result, expect_etag=True)233 # update options, confirm y234 result = self.invoke(235 ['dhcp-options', 'update', '--dhcp-id', self.dhcp_options_ocid, '--options', options_v2], input='y')236 util.validate_response(result, json_response_expected=False)237 # update options, confirm n238 result = self.invoke(239 ['dhcp-options', 'update', '--dhcp-id', self.dhcp_options_ocid, '--options', options], input='n')240 assert result.exit_code != 0241 time.sleep(20)242 # update options, force243 result = self.invoke(244 ['dhcp-options', 'update', '--dhcp-id', self.dhcp_options_ocid, '--options', options_v2, '--force'])245 util.validate_response(result, expect_etag=True)246 response = json.loads(result.output)247 # validate response contains SearchDomain option248 response_has_search_domain_option = False249 for option in response["data"]["options"]:250 if option["type"] == "SearchDomain":251 response_has_search_domain_option = True252 assert option["search-domain-names"][0] == "testvcn.oraclevcn.com"253 assert response_has_search_domain_option, "Options response should contain option of type 'SearchDomain'."254 @util.log_test255 def subtest_drg_operations(self):256 drg_name = util.random_name('cli_test_drg')257 result = self.invoke(258 ['drg', 'create',259 '--compartment-id', util.COMPARTMENT_ID,260 '--display-name', drg_name261 ])262 self.drg_ocid = util.find_id_in_response(result.output)263 util.validate_response(result, expect_etag=True)264 util.wait_until(['network', 'drg', 'get', '--drg-id', self.drg_ocid], 'AVAILABLE',265 max_wait_seconds=600)266 result = self.invoke(['drg', 'list', '--compartment-id', util.COMPARTMENT_ID])267 util.validate_response(result)268 drg_name = drg_name + "_updated"269 result = self.invoke(['drg', 'update', '--drg-id', self.drg_ocid, '--display-name', drg_name])270 util.validate_response(result, expect_etag=True)271 result = self.invoke(['drg', 'get', '--drg-id', self.drg_ocid])272 util.validate_response(result, expect_etag=True)273 @util.log_test274 def subtest_drg_attachment_operations(self):275 drg_attachment_name = util.random_name('cli_test_drg_attachment')276 result = self.invoke(277 ['drg-attachment', 'create',278 '--drg-id', self.drg_ocid,279 '--vcn-id', self.vcn_ocid,280 '--display-name', drg_attachment_name281 ])282 self.drg_attachment_ocid = util.find_id_in_response(result.output)283 util.validate_response(result, expect_etag=True)284 util.wait_until(['network', 'drg-attachment', 'get', '--drg-attachment-id', self.drg_attachment_ocid], 'ATTACHED')285 result = self.invoke(['drg-attachment', 'list', '--compartment-id', util.COMPARTMENT_ID])286 util.validate_response(result)287 drg_attachment_name = drg_attachment_name + "_updated"288 result = self.invoke(['drg-attachment', 'update', '--drg-attachment-id', self.drg_attachment_ocid, '--display-name', drg_attachment_name])289 util.validate_response(result, expect_etag=True)290 result = self.invoke(['drg-attachment', 'get', '--drg-attachment-id', self.drg_attachment_ocid])291 util.validate_response(result, expect_etag=True)292 @util.log_test293 def subtest_ip_sec_connection_operations(self):294 ipsc_name = util.random_name('cli_test_ipsc')295 routes = util.remove_outer_quotes(oraclebmc_cli.core_cli_extended.network_create_ip_sec_connection_static_routes_example)296 result = self.invoke(297 ['ip-sec-connection', 'create',298 '--compartment-id', util.COMPARTMENT_ID,299 '--display-name', ipsc_name,300 '--cpe-id', self.cpe_ocid,301 '--drg-id', self.drg_ocid,302 '--static-routes', routes303 ])304 self.ipsc_ocid = util.find_id_in_response(result.output)305 util.validate_response(result, expect_etag=True)306 util.wait_until(['network', 'ip-sec-connection', 'get', '--ipsc-id', self.ipsc_ocid], 'AVAILABLE',307 max_wait_seconds=300)308 result = self.invoke(['ip-sec-connection', 'list', '--compartment-id', util.COMPARTMENT_ID])309 util.validate_response(result)310 ipsc_name = ipsc_name + "_updated"311 result = self.invoke(['ip-sec-connection', 'update', '--ipsc-id', self.ipsc_ocid, '--display-name', ipsc_name])312 util.validate_response(result, expect_etag=True)313 result = self.invoke(['ip-sec-connection', 'get', '--ipsc-id', self.ipsc_ocid])314 util.validate_response(result, expect_etag=True)315 result = self.invoke(['ip-sec-connection', 'get-config', '--ipsc-id', self.ipsc_ocid])316 util.validate_response(result)317 result = self.invoke(['ip-sec-connection', 'get-status', '--ipsc-id', self.ipsc_ocid])318 util.validate_response(result)319 @util.log_test320 def subtest_route_table_operations(self):321 rt_name = util.random_name('cli_test_route_table')322 rules = util.remove_outer_quotes(oraclebmc_cli.core_cli_extended.network_create_route_table_route_rules_example.format(ig_id=self.ig_ocid))323 result = self.invoke(324 ['route-table', 'create',325 '--compartment-id', util.COMPARTMENT_ID,326 '--display-name', rt_name,327 '--vcn-id', self.vcn_ocid,328 '--route-rules', rules329 ])330 self.rt_ocid = util.find_id_in_response(result.output)331 util.validate_response(result, expect_etag=True)332 util.wait_until(['network', 'route-table', 'get', '--rt-id', self.rt_ocid], 'AVAILABLE',333 max_wait_seconds=300)334 result = self.invoke(['route-table', 'list', '--compartment-id', util.COMPARTMENT_ID, '--vcn-id', self.vcn_ocid])335 util.validate_response(result)336 result = self.invoke(['route-table', 'get', '--rt-id', self.rt_ocid])337 util.validate_response(result, expect_etag=True)338 rt_name = rt_name + "_updated"339 rules_v2 = """[{{"cidrBlock":"0.0.0.0/1","networkEntityId":"{ig_id}"}}]""".format(ig_id=self.ig_ocid)340 # update display name only - does not show a prompt341 result = self.invoke(['route-table', 'update', '--rt-id', self.rt_ocid, '--display-name', rt_name])342 util.validate_response(result, expect_etag=True)343 time.sleep(20)344 # update route-rules, confirm y345 result = self.invoke(346 ['route-table', 'update', '--rt-id', self.rt_ocid, '--route-rules', rules_v2], input='y')347 util.validate_response(result, json_response_expected=False)348 # update route-rules, confirm n349 result = self.invoke(350 ['route-table', 'update', '--rt-id', self.rt_ocid, '--route-rules', rules_v2], input='n')351 assert result.exit_code != 0352 time.sleep(20)353 # update route-rules, force354 result = self.invoke(355 ['route-table', 'update', '--rt-id', self.rt_ocid, '--route-rules', rules, '--force'])356 util.validate_response(result, expect_etag=True)357 @util.log_test358 def subtest_delete(self):359 error_count = 0360 if hasattr(self, 'rt_ocid'):361 max_retry = 3362 retry = 0363 try:364 while retry < max_retry:365 try:366 time.sleep(2)367 result = self.invoke(['route-table', 'delete', '--rt-id', self.rt_ocid, '--force'])368 util.validate_response(result)369 break370 except Exception as error:371 # if the route-table no longer exists then don't try to delete it again372 if '"status": 404' in result.output:373 break374 else:375 retry += 1376 print("Retrying route-table delete.")377 util.wait_until(['network', 'route-table', 'get', '--rt-id', self.rt_ocid], 'TERMINATED', succeed_if_not_found=True, max_wait_seconds=300)378 except Exception as error:379 util.print_latest_exception(error)380 error_count = error_count + 1381 if hasattr(self, 'ipsc_ocid'):382 try:383 result = self.invoke(['ip-sec-connection', 'delete', '--ipsc-id', self.ipsc_ocid, '--force'])384 util.validate_response(result)385 util.wait_until(['network', 'ip-sec-connection', 'get', '--ipsc-id', self.ipsc_ocid], 'TERMINATED', succeed_if_not_found=True, max_wait_seconds=300)386 except Exception as error:387 util.print_latest_exception(error)388 error_count = error_count + 1389 if hasattr(self, 'drg_attachment_ocid'):390 try:391 result = self.invoke(['drg-attachment', 'delete', '--drg-attachment-id', self.drg_attachment_ocid, '--force'])392 util.validate_response(result)393 util.wait_until(['network', 'drg-attachment', 'get', '--drg-attachment-id', self.drg_attachment_ocid], 'DETACHED', succeed_if_not_found=True, max_wait_seconds=600)394 except Exception as error:395 util.print_latest_exception(error)396 error_count = error_count + 1397 if hasattr(self, 'drg_ocid'):398 try:399 result = self.invoke(['drg', 'delete', '--drg-id', self.drg_ocid, '--force'])400 util.validate_response(result)401 util.wait_until(['network', 'drg', 'get', '--drg-id', self.drg_ocid], 'TERMINATED', succeed_if_not_found=True, max_wait_seconds=600)402 except Exception as error:403 util.print_latest_exception(error)404 error_count = error_count + 1405 if hasattr(self, 'dhcp_options_ocid'):406 try:407 result = self.invoke(['dhcp-options', 'delete', '--dhcp-id', self.dhcp_options_ocid, '--force'])408 util.validate_response(result)409 util.wait_until(['network', 'dhcp-options', 'get', '--dhcp-id', self.dhcp_options_ocid], 'TERMINATED', succeed_if_not_found=True, max_wait_seconds=600)410 except Exception as error:411 util.print_latest_exception(error)412 error_count = error_count + 1413 if hasattr(self, 'cpe_ocid'):414 try:415 result = self.invoke(['cpe', 'delete', '--cpe-id', self.cpe_ocid, '--force'])416 util.validate_response(result)417 except Exception as error:418 util.print_latest_exception(error)419 error_count = error_count + 1420 if hasattr(self, 'ig_ocid'):421 try:422 result = self.invoke(['internet-gateway', 'delete', '--ig-id', self.ig_ocid, '--force'])423 util.validate_response(result)424 util.wait_until(['network', 'internet-gateway', 'get', '--ig-id', self.ig_ocid], 'TERMINATED',425 max_wait_seconds=600, succeed_if_not_found=True)426 except Exception as error:427 util.print_latest_exception(error)428 error_count = error_count + 1429 if hasattr(self, 'subnet_ocid'):430 try:431 result = self.invoke(['subnet', 'delete', '--subnet-id', self.subnet_ocid, '--force'])432 util.validate_response(result)433 util.wait_until(['network', 'subnet', 'get', '--subnet-id', self.subnet_ocid], 'TERMINATED',434 max_wait_seconds=600, succeed_if_not_found=True)435 except Exception as error:436 util.print_latest_exception(error)437 error_count = error_count + 1438 if hasattr(self, 'sl_ocid'):439 try:440 result = self.invoke(['security-list', 'delete', '--security-list-id', self.sl_ocid, '--force'])441 util.validate_response(result)442 util.wait_until(['network', 'security-list', 'get', '--security-list-id', self.sl_ocid], 'TERMINATED', succeed_if_not_found=True, max_wait_seconds=300)443 except Exception as error:444 util.print_latest_exception(error)445 error_count = error_count + 1446 if hasattr(self, 'vcn_ocid'):447 try:448 result = self.invoke(['vcn', 'delete', '--vcn-id', self.vcn_ocid, '--force'])449 util.validate_response(result)450 except Exception as error:451 util.print_latest_exception(error)452 error_count = error_count + 1453 self.assertEquals(0, error_count)454 def invoke(self, params, debug=False, ** args):455 commands = ['network'] + params456 self.validator.register_call(commands)457 if debug is True:458 commands = ['--debug'] + commands459 return util.invoke_command(commands, ** args)460if __name__ == '__main__':...
test_edited_article.py
Source:test_edited_article.py
...54 def test_news_api_detail(self):55 a_id = self.edited_id_list[-1]56 ariticle = EditedArticle.get_by_id(a_id)57 response = test_app.get(f'/api/v1/news/{a_id}')58 json_resp = self.validate_response(response, 200)59 tools.assert_equals(json_resp.get('data'),60 ariticle.api_response())6162 ariticle.state = EditedAriticleConstant.STATE_PENDING63 ariticle.save()64 ariticle.reload()6566 response = test_app.get(f'/api/v1/news/{a_id}')67 json_resp = self.validate_response(response, 403)6869 @tools.nottest70 def test_news_published(self, amount, json_resp):71 for i in range(amount - 1):72 latest_publish = json_resp['data']['list'][i]['published']73 last_publish = json_resp['data']['list'][i + 1]['published']74 if last_publish > latest_publish:75 return False76 return True7778 def test_news_api(self):79 """80 æµè¯newsçgetæ¥å£8182 1ãæµè¯è½å¦æ¥è¯¢å°ææpublishedç¶æçæç« 83 2ãæµè¯æ¥è¯¢ç»æçå页åè½84 3ãæµè¯æ°é»æ¯å¦æpublishedæåº85 """86 query = {'state': EditedAriticleConstant.STATE_PUBLISHED}87 response = test_app.get('api/v1/news')88 json_resp = self.validate_response(response, 200)89 tools.assert_equals(json_resp['data']['count'], 25)9091 EditedArticle.objects().update(92 state=EditedAriticleConstant.STATE_PUBLISHED)9394 length = EditedArticle.objects(**query).count()95 response = test_app.get('api/v1/news')9697 json_resp = self.validate_response(response, 200)98 tools.assert_equals(json_resp['data']['count'], length)99 tools.assert_true(self.test_news_published(length, json_resp))100 tools.assert_true(length > 20)101 tools.assert_equals(len(json_resp['data']['list']), 20)102 json_item = json_resp['data']['list'][0]103 item = EditedArticle.get_by_id(id=json_item['id'])104 tools.assert_equals(json_item, item.api_base_response())105 article_count = EditedArticle.objects(state='published').count()106 self.test_search('news', article_count)107108 def test_search_day(self):109 """110 æµè¯æ¥è¯¢æä¸å¤©çæç« 111112 1ãæµè¯æ¥è¯¢æ²¡ææç« çä¸å¤©ï¼ç»æ为0113 2ãæµè¯æ¥è¯¢æä¸å¤©æç« 114 3ãæµè¯æ¥è¯¢ç»æçå页åè½115 4ãæµè¯æ¥è¯¢ç»ææ¯å¦æpublishedæåº116 """117 __time = datetime.now()118 day_str = __time.strftime('%Y%m%d')119 headers = {'Authorization': self.token}120 response = test_app.get('api/v1/news?day=20100101',121 headers=headers)122 json_resp = self.validate_response(response)123 tools.assert_equals(json_resp['data']['count'], 0)124125 response = test_app.get(f'api/v1/news?day={day_str}',126 headers=headers)127 length = EditedArticle.objects(state='published').count()128 json_resp = self.validate_response(response)129 tools.assert_true(self.test_news_published(130 len(json_resp['data']['list']), json_resp))131 tools.assert_equals(json_resp['data']['page_sum'], length//20+1)132 tools.assert_equals(json_resp['data']['current_page'], 1)133134 response = test_app.get(f'api/v1/news?day={day_str}&page=2',135 headers=headers)136 json_resp = self.validate_response(response)137 tools.assert_true(self.test_news_published(138 len(json_resp['data']['list']), json_resp))139 tools.assert_equals(json_resp['data']['page_sum'], length//20+1)140 tools.assert_equals(json_resp['data']['current_page'], 2)141142 @tools.nottest143 def test_search(self, article_group, count):144 """145 æµè¯æç´¢åè½146147 @param article_group: æç« æå¨çgroup(editedæè
news)148 @tpye search_word: str149 @param count: ç¶æ为publishedæç« çæ°é150 @tpye count: int151 @return: None152 @rtype: None153 """154 self.change_user_level(9)155 headers = {'Authorization': self.token}156 url = f'/api/v1/{article_group}?search=a'157 response = test_app.get(url, headers=headers)158 json_resp = self.validate_response(response)159 tools.assert_equals(json_resp['data']['count'], 0)160161 url = f'/api/v1/{article_group}?search=s'162 response = test_app.get(url, headers=headers)163 json_resp = self.validate_response(response)164 tools.assert_equals(json_resp['data']['count'], count)165166 url = f'/api/v1/{article_group}?state=published&search=s'167 response = test_app.get(url, headers=headers)168 json_resp = self.validate_response(response)169 tools.assert_equals(json_resp['data']['count'], count)170171 edited_articles = EditedArticle.objects().limit(3)172 for edited_article in edited_articles:173 edited_article.state = 'edited'174 edited_article.save()175 url = f'/api/v1/{article_group}?state=published&search=s'176 response = test_app.get(url, headers=headers)177 json_resp = self.validate_response(response)178 tools.assert_equals(json_resp['data']['count'], count - 3)179180 edited_articles = EditedArticle.objects().limit(3)181 for edited_article in edited_articles:182 edited_article.state = 'published'183 edited_article.save()184185 def test_edited_article_api(self):186 """187 æµè¯edited_articleçgetæ¥å£ï¼åé请æ±æ
åµ188189 1ãæµè¯ç»å½è®¤è¯190 2ãæµè¯æ¹åorder_byï¼éªè¯æ¥è¯¢ç»æ191 3ãæµè¯æ¥è¯¢ç»æçå页åè½åæç¶ææ¥è¯¢192 4ãæµè¯æ¥è¯¢è¯¦æ
193 """194 headers = {'Authorization': self.token}195 response = test_app.get('/api/v1/edited',196 headers=headers)197 json_resp = self.validate_response(response, 200)198 length = EditedArticle.objects(state='published').count()199 tools.assert_equals(len(json_resp['data']['list']), 20)200 tools.assert_equals(json_resp['data']['count'], length)201 tools.assert_equals(json_resp['data']['page_sum'], length//20+1)202 tools.assert_equals(json_resp['data']['current_page'], 1)203 tools.assert_equals(json_resp['data']['list'][0].get(204 'id'), str(self.edited_id_list[-1]))205 item = EditedArticle.get_by_id(self.edited_id_list[-1])206 tools.assert_equals(207 json_resp['data']['list'][0], item.api_base_response())208209 headers = {'Authorization': self.token}210 response = test_app.get('/api/v1/edited?order_by=added',211 headers=headers)212 json_resp = self.validate_response(response)213 tools.assert_not_equals(json_resp['data']['list'][0],214 EditedArticle.get_by_id215 (self.edited_id_list[-1]).api_base_response())216 tools.assert_equals(json_resp['data']['list'][0],217 EditedArticle.get_by_id(self.edited_id_list[0]).218 api_base_response())219220 url = '/api/v1/edited?page=2&page_size=20&state=published'221 response = test_app.get(url, headers=headers)222 json_resp = self.validate_response(response, 200)223 length = EditedArticle.objects(state='published').count()224 tools.assert_equals(len(json_resp['data']['list']), length-20)225 tools.assert_equals(json_resp['data']['count'], length)226 tools.assert_equals(json_resp['data']['page_sum'], length//20+1)227 tools.assert_equals(json_resp['data']['current_page'], 2)228229 url = '/api/v1/edited?state=all'230 response = test_app.get(url, headers=headers)231 json_resp = self.validate_response(response, 200)232 length = EditedArticle.objects().count()233 tools.assert_equals(len(json_resp['data']['list']), 20)234 tools.assert_equals(json_resp['data']['count'], length)235 tools.assert_equals(json_resp['data']['page_sum'], length//20+1)236 tools.assert_equals(json_resp['data']['current_page'], 1)237238 test_article = EditedArticle.get_by_id(self.edited_id_list[1])239 url = f'/api/v1/edited/{str(self.edited_id_list[1])}'240 response = test_app.get(url, headers=headers)241 json_resp = self.validate_response(response, 200)242 tools.assert_is_not_none(json_resp['data'])243 tools.assert_equals(json_resp['data'], test_article.api_response())244 edited_count = EditedArticle.objects(state='published').count()245 self.test_search('edited', edited_count)246247 def test_edited_article_post(self):248 """249 æµè¯edited_articleçpostæ¥å£250251 1ãæµè¯ä¼ é误æè
ä¸ä¼ tokenè½å¦ä¿®æ¹æç« 252 2ãæµè¯ç»å½è®¤è¯253 3ãæµè¯æéï¼æ®éç¨æ·ä¸è½æ°å¢254 4ãä¸ä¼ raw_articleæ°å¢ï¼article_from为1255 5ãä¼ raw_articleæ°å¢ï¼article_from为0256 4ãä¸ä¼ æ¶é´æ°å¢257 """258 api_url = 'api/v1/edited'259 api_method = test_app.post260 self.test_api_jwt(api_url, api_method)261262 headers = {'Authorization': self.token}263 self.change_user_level(1)264 response = test_app.post('/api/v1/edited',265 data=json.dumps(self.edited_data),266 headers=headers,267 content_type='application/json')268 json_resp = self.validate_response(response, 500)269 tools.assert_equals(json_resp.get("data"),270 {'msg': "user don't has authority"})271272 self.change_user_level(9)273 data = deepcopy(self.edited_data)274 data['state'] = 'edited'275 data['published'] = datetime.now().timestamp()276 response = test_app.post('/api/v1/edited',277 data=json.dumps(data),278 headers=headers,279 content_type='application/json')280 json_resp = self.validate_response(response, 200)281 tools.assert_is_not_none(json_resp.get('data'))282 tools.assert_is_not_none(json_resp.get('data').get('id'))283 tools.assert_equals(json_resp.get('data').get('article_from'), 1)284285 data = deepcopy(self.edited_data)286 data['state'] = 'edited'287 data['published'] = datetime.now().timestamp() + 10000288 data['added'] = datetime.now().timestamp() - 1000289 data['date'] = datetime.now().timestamp() - 2000290 data['updated'] = datetime.now().timestamp() - 2000291 data['url'] = 'http://www.sohu.com'292 data['raw_article'] = str(self.raw_id)293 response = test_app.post('/api/v1/edited',294 data=json.dumps(data),295 headers=headers,296 content_type='application/json')297 json_resp = self.validate_response(response, 200).get('data')298 tools.assert_is_not_none(json_resp.get('id'))299 tools.assert_not_equals(json_resp.get('published'), data['published'])300 tools.assert_not_equals(json_resp.get('added'), data['added'])301 tools.assert_not_equals(json_resp.get('date'), data['date'])302 tools.assert_not_equals(json_resp.get('updated'), data['updated'])303 tools.assert_equals(json_resp.get('article_from'), 0)304 tools.assert_equals(json_resp.get('raw_article'), str(self.raw_id))305 article = RawArticle.get_by_id(str(self.raw_id))306 tools.assert_equals(article.is_edited, True)307308 data.pop('added')309 data.pop('date')310 data['url'] = 'http://www.sohu'311 response = test_app.post('/api/v1/edited',312 data=json.dumps(data),313 headers=headers,314 content_type='application/json')315 json_resp = self.validate_response(response).get('data')316 tools.assert_is_not_none(json_resp.get('added'))317 tools.assert_is_not_none(json_resp.get('date'))318 tools.assert_equals(json_resp.get('cn_title'),319 data['cn_title'])320 tools.assert_equals(json_resp.get('cn_content'),321 data['cn_content'])322 tools.assert_equals(json_resp.get('state'),323 data['state'])324 EditedArticle.objects(state='edited').delete()325326 def test_put_edited_article(self):327 """328 æµè¯edited_articleçputæ¥å£329330 1ãæµè¯ä¼ é误æè
ä¸ä¼ tokenè½å¦ä¿®æ¹æç« 331 2ãæµè¯ç»å½è®¤è¯332 3ãæµè¯ä¿®æ¹æé333 4ãä¸ä¼ æ¶é´ä¿®æ¹334 """335 api_url = 'api/v1/edited'336 api_method = test_app.put337 self.test_api_jwt(api_url, api_method)338339 headers = {'Authorization': self.token}340 data = deepcopy(self.edited_data)341 data['cn_title'] = 'æµè¯111'342 data['state'] = 'pending'343 data['cn_summary'] = 'æµè¯aaa'344 data['published'] = datetime.now().timestamp() + 1000345 data['added'] = datetime.now().timestamp() - 1000346 data['date'] = datetime.now().timestamp() - 1000347 data['updated'] = datetime.now().timestamp() - 1000348 response = test_app.put('/api/v1/edited',349 headers=headers,350 data=json.dumps(data),351 content_type='application/json')352 self.validate_response(response, 400)353354 self.change_user_level(1)355 response = test_app.put(356 f'api/v1/edited/{str(self.edited_id_list[1])}',357 data=json.dumps(data),358 headers=headers,359 content_type='application/json')360 self.validate_response(response, 500)361362 self.change_user_level(9)363 response = test_app.put(364 f'api/v1/edited/{str(self.edited_id_list[1])}',365 data=json.dumps(data),366 headers=headers,367 content_type='application/json')368 json_resp = self.validate_response(response, 200)369 tools.assert_equals(json_resp['data']['cn_title'],370 data['cn_title'])371 tools.assert_equals(json_resp['data']['state'],372 data['state'])373 tools.assert_equals(json_resp['data']['cn_summary'],374 data['cn_summary'])375376 data.pop('added')377 data.pop('date')378 data.pop('published')379 response = test_app.put(380 f'/api/v1/edited/{str(self.edited_id_list[1])}',381 data=json.dumps(data),382 headers=headers,383 content_type='application/json')384 json_resp = json.loads(response.data).get('data')385 tools.assert_is_not_none(json_resp.get('added'))386 tools.assert_is_not_none(json_resp.get('date'))387 tools.assert_is_not_none(json_resp.get('published'))388389 response = test_app.get(390 f'/api/v1/edited/{str(self.edited_id_list[1])}',391 headers=headers)392 json_resp = self.validate_response(response, 200)393 tools.assert_not_equals(json_resp.get('cn_title'),394 data['cn_title'])395 tools.assert_not_equals(json_resp.get('cn_content'),396 data['cn_content'])397 tools.assert_not_equals(json_resp.get('state'),398 data['state'])399400 def test_zdelete_edited_article(self):401 """402 æµè¯edited_articleçdeleteæ¥å£403 """404 api_url = 'api/v1/edited'405 api_method = test_app.delete406 self.test_api_jwt(api_url, api_method)407 408 amount = EditedArticle.objects().count()409 cache_id = self.edited_id_list[1]410411 # print(type(EditedArticle.objects(id=cache_id)))412 # print(type(EditedArticle.objects(id=cache_id).first()))413 headers = {'Authorization': self.token}414 response = test_app.delete(415 f'/api/v1/edited/{str(self.edited_id_list[1])}',416 headers=headers417 )418 json_resp = self.validate_response(response, 200)419 # print(EditedArticle.objects().count())420 tools.assert_equals(json_resp.get('data'), {'msg': 'edited_article deleted.'})421 tools.assert_is_none(EditedArticle.objects(id=cache_id).first())422 tools.assert_equals(amount-1, EditedArticle.objects().count())423424425 def test_holoread_article_get(self):426 """427 æµè¯holoread_articleçgetæ¥å£428429 1ãæµè¯publishedæåº430 2ãæµè¯æ¥è¯¢åæ°laståcount431 """432 api_url = 'api/v1/articles'433 count = 10434 response = test_app.get(f'{api_url}?count={count}')435 json_resp = self.validate_response(response, 200)436 data = json_resp.get('data')437 tools.assert_equals(count, len(data))438 tools.assert_equals(str(self.edited_id_list[0]),439 data[0]['_id'])440441 last = int(time.time() + 100)442 response = test_app.get(f'{api_url}?last={last}')443 json_resp = self.validate_response(response, 200)444 data = json_resp.get('data')445 tools.assert_equals(0, len(data))446447 articles = EditedArticle.objects().limit(3)448 for article in articles:449 article.state = 'pending'450 article.save()451 response = test_app.get(f'{api_url}')452 json_resp = self.validate_response(response, 200)453 data = json_resp.get('data')454 tools.assert_equals(len(data),455 len(self.edited_id_list) - 3)456 for article in articles:457 article.state = 'published'
...
test_http_api.py
Source:test_http_api.py
...5from roboto.http_api import AnyAPIResponse, validate_response6def test_validate_good_response() -> None:7 """Ensure validate_response returns result if ok is True."""8 response: Any = AnyAPIResponse(ok=True, result=[])9 assert validate_response(response) == []10def test_validate_bad_response() -> None:11 """Ensure validate_response raises if ok is False."""12 response: Any = AnyAPIResponse(13 ok=False, error_code=400, description='There was error.'14 )15 try:16 validate_response(response)17 except BotAPIError as e:18 assert e.error_code == 40019 assert e.description == 'There was error.'20 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 tempest 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?
