KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use test_json_payload method in locust
Best Python code snippet using locust
test_notification.py
Source:test_notification.py 
1from datetime import datetime, timezone2from unittest.mock import MagicMock, patch3import callee4import requests5import responses6from freezegun import freeze_time7from chaoslib.exceptions import ChaosException8from chaoslib.notification import (9 DiscoverFlowEvent,10 InitFlowEvent,11 RunFlowEvent,12 ValidateFlowEvent,13 notify,14 notify_with_http,15)16def test_no_settings_is_okay() -> None:17 assert notify(None, DiscoverFlowEvent.DiscoverStarted) is None18def test_no_notifications_in_settings_is_okay() -> None:19 assert notify({"things": "stuff"}, DiscoverFlowEvent.DiscoverStarted) is None20@freeze_time("2020-01-01 00:00")21@patch("chaoslib.notification.notify_with_http")22def test_notify_calls_notify_with_http_when_type_is_http(23 mock_notify_with_http: MagicMock,24) -> None:25 now = datetime.utcnow().replace(tzinfo=timezone.utc).timestamp()26 payload = {"test-key": "test-value", "test-dict": {"test-dict-key": "test"}}27 channel = {"type": "http", "url": "http://example.com"}28 notify(29 settings={"notifications": [channel]},30 event=RunFlowEvent.RunStarted,31 payload=payload,32 )33 mock_notify_with_http.assert_called_once_with(34 channel,35 {36 "name": RunFlowEvent.RunStarted.value,37 "payload": payload,38 "phase": "run",39 "ts": now,40 },41 )42@patch("chaoslib.notification.logger", autospec=True)43def test_notify_with_http_requires_a_url(mock_logger: MagicMock) -> None:44 notify_with_http(channel={"type": "http", "url": ""}, payload={})45 mock_logger.debug.assert_called_with("missing url in notification channel")46@patch("chaoslib.notification.logger")47@patch("chaoslib.notification.requests")48def test_notify_with_http_gracefully_handles_exceptions(49 mock_requests: MagicMock, mock_logger: MagicMock50) -> None:51 exception = requests.exceptions.RequestException("An Exception")52 mock_requests.get.side_effect = exception53 notify_with_http(54 channel={55 "type": "http",56 "url": "http://test-url.com",57 "forward_event_payload": False,58 },59 payload={},60 )61 mock_logger.debug.assert_called_once_with(62 "failed calling notification endpoint", exc_info=exception63 )64@responses.activate65@patch("chaoslib.notification.logger")66def test_notify_with_http_handles_400_500_responses(mock_logger: MagicMock) -> None:67 test_400_url = "http://test-400-url.com"68 test_500_url = "http://test-500-url.com"69 responses.add(method=responses.GET, url=test_400_url, status=400)70 responses.add(method=responses.GET, url=test_500_url, status=500)71 notify_with_http(72 channel={"type": "http", "url": test_400_url, "forward_event_payload": False},73 payload={},74 )75 mock_logger.debug.assert_called_with(76 (77 f"notification sent to {test_400_url} failed with: "78 "400 Client Error: Bad Request for url: http://test-400-url.com/"79 )80 )81 notify_with_http(82 channel={"type": "http", "url": test_500_url, "forward_event_payload": False},83 payload={},84 )85 mock_logger.debug.assert_called_with(86 (87 f"notification sent to {test_500_url} failed with: "88 "500 Server Error: Internal Server Error for url: http://test-500-url.com/"89 )90 )91@responses.activate92@patch("chaoslib.notification.logger")93def test_notify_with_http_will_forward_event_payload(mock_logger: MagicMock) -> None:94 test_url = "http://test-post-url.com"95 test_payload = {"test-key": "test-val", "test-dict": {"test-key-2": "test-val-2"}}96 responses.add(97 method=responses.POST,98 url=test_url,99 match=[responses.matchers.json_params_matcher(test_payload)],100 )101 notify_with_http(channel={"type": "http", "url": test_url}, payload=test_payload)102 mock_logger.debug.assert_not_called()103@responses.activate104@patch("chaoslib.notification.logger")105def test_notify_with_http_wont_forward_event_payload(mock_logger: MagicMock) -> None:106 test_url = "http://test-post-url.com"107 test_payload = {"test-key": "test-val", "test-dict": {"test-key-2": "test-val-2"}}108 responses.add(109 method=responses.GET,110 url=test_url,111 )112 notify_with_http(113 channel={"type": "http", "url": test_url, "forward_event_payload": False},114 payload=test_payload,115 )116 mock_logger.debug.assert_not_called()117@responses.activate118@freeze_time("2021-01-01 00:00")119@patch("chaoslib.notification.logger")120def test_notify_with_http_handles_datetimes_present_in_payload(121 mock_logger: MagicMock,122) -> None:123 test_url = "http://test-datetime-url.com"124 now = datetime.now()125 now_timestamp = now.isoformat()126 test_payload = {"test-key": "test-val", "test-datetime": now}127 test_json_payload = {"test-key": "test-val", "test-datetime": now_timestamp}128 responses.add(129 method=responses.POST,130 url=test_url,131 match=[responses.matchers.json_params_matcher(test_json_payload)],132 )133 notify_with_http(channel={"type": "http", "url": test_url}, payload=test_payload)134 mock_logger.debug.assert_not_called()135@responses.activate136@patch("chaoslib.notification.logger")137def test_notify_with_http_handles_error_present_in_payload(138 mock_logger: MagicMock,139) -> None:140 test_url = "http://test-error-url.com"141 exception = ChaosException("Something went wrong here")142 test_payload = {"test-key": "test-val", "error": exception}143 test_json_payload = {144 "test-key": "test-val",145 "error": "An exception was raised: ChaosException('Something went wrong here')",146 }147 responses.add(148 method=responses.POST,149 url=test_url,150 match=[responses.matchers.json_params_matcher(test_json_payload)],151 )152 notify_with_http(channel={"type": "http", "url": test_url}, payload=test_payload)153 mock_logger.debug.assert_not_called()154@freeze_time("2021-01-01 00:00")155@patch("chaoslib.notification.notify_with_http")156def test_notify_correctly_assigns_phase_from_event_class(157 mock_notify_with_http: MagicMock,158) -> None:159 channel = {"type": "http", "url": "http://example.com"}160 for phase, event_class in [161 ("discovery", DiscoverFlowEvent.DiscoverStarted),162 ("init", InitFlowEvent.InitStarted),163 ("run", RunFlowEvent.RunStarted),164 ("validate", ValidateFlowEvent.ValidateStarted),165 ]:166 mock_notify_with_http.reset_mock()167 notify(settings={"notifications": [channel]}, event=event_class, payload=None)168 mock_notify_with_http.assert_called_once_with(169 channel,170 {171 "name": event_class.value,172 "payload": None,173 "phase": phase,174 "ts": datetime.utcnow().replace(tzinfo=timezone.utc).timestamp(),175 },176 )177@freeze_time("2021-01-01 00:00")178@patch("chaoslib.notification.notify_with_http")179def test_notify_appends_error_to_event_payload_if_provided(180 mock_notify_with_http: MagicMock,181) -> None:182 channel = {"type": "http", "url": "http://example.com"}183 exception = ChaosException("Something went wrong")184 notify(185 settings={"notifications": [channel]},186 event=DiscoverFlowEvent.DiscoverStarted,187 payload=None,188 error=exception,189 )190 mock_notify_with_http.assert_called_once_with(191 channel,192 {193 "name": DiscoverFlowEvent.DiscoverStarted.value,194 "payload": None,195 "phase": "discovery",196 "ts": datetime.utcnow().replace(tzinfo=timezone.utc).timestamp(),197 "error": exception,198 },199 )200@freeze_time("2021-01-01 00:00")201@patch("chaoslib.notification.notify_with_http")202def test_notify_only_notifies_on_events_specified(203 mock_notify_with_http: MagicMock,204) -> None:205 channel = {206 "type": "http",207 "url": "http://example.com",208 "events": ["discover-started"],209 }210 notify(211 settings={"notifications": [channel]},212 event=RunFlowEvent.RunFailed,213 payload=None,214 )215 mock_notify_with_http.assert_not_called()216@freeze_time("2020-01-01 00:00")217@patch("chaoslib.notification.notify_via_plugin")218def test_notify_calls_notify_via_plugin_when_type_is_plugin(219 mock_notify_via_plugin: MagicMock,220) -> None:221 now = datetime.utcnow().replace(tzinfo=timezone.utc).timestamp()222 payload = {"test-key": "test-value", "test-dict": {"test-dict-key": "test"}}223 channel = {"type": "plugin", "module": "fixtures.notifier"}224 notify(225 settings={"notifications": [channel]},226 event=RunFlowEvent.RunStarted,227 payload=payload,228 )229 mock_notify_via_plugin.assert_called_once_with(230 channel,231 {232 "name": RunFlowEvent.RunStarted.value,233 "payload": payload,234 "phase": "run",235 "ts": now,236 },237 )238@patch("fixtures.notifier.logger", autospec=True)239def test_notify_via_plugin_correctly_invokes_notify_func(logger: MagicMock) -> None:240 notify(241 {"notifications": [{"type": "plugin", "module": "fixtures.notifier"}]},242 RunFlowEvent.RunStarted,243 )244 logger.debug.assert_called_with("boom")245@patch("fixtures.notifier.logger", autospec=True)246def test_notify_via_plugin_correctly_invokes_notify_func_with_non_default_func_name(247 logger: MagicMock,248) -> None:249 notify(250 {251 "notifications": [252 {253 "type": "plugin",254 "module": "fixtures.notifier",255 "func": "notify_other",256 }257 ]258 },259 RunFlowEvent.RunStarted,260 )261 logger.debug.assert_called_with("doh")262@patch("chaoslib.notification.logger", autospec=True)263def test_notify_via_plugin_gracefully_handles_failure_to_import_plugin(264 logger: MagicMock,265) -> None:266 notify(267 {"notifications": [{"type": "plugin", "module": "fixtures.notifier___"}]},268 RunFlowEvent.RunStarted,269 )270 logger.debug.assert_called_with(271 "could not find Python plugin '{mod}' for notification".format(272 mod="fixtures.notifier___"273 )274 )275@patch("chaoslib.notification.logger", autospec=True)276def test_notify_via_plugin_gracefully_handles_failure_to_import_func(277 logger: MagicMock,278) -> None:279 notify(280 {281 "notifications": [282 {"type": "plugin", "module": "fixtures.notifier", "func": "blah"}283 ]284 },285 RunFlowEvent.RunStarted,286 )287 logger.debug.assert_called_with(288 "could not find function '{f}' in plugin '{mod}' "289 "for notification".format(mod="fixtures.notifier", f="blah")290 )291@patch("chaoslib.notification.logger", autospec=True)292def test_notify_via_plugin_gracefully_handles_failure_in_invoked_func(293 logger: MagicMock,294) -> None:295 notify(296 {297 "notifications": [298 {299 "type": "plugin",300 "module": "fixtures.notifier",301 "func": "notify_broken",302 }303 ]304 },305 RunFlowEvent.RunStarted,306 )307 logger.debug.assert_called_with(308 "failed calling notification plugin", exc_info=callee.InstanceOf(Exception)...
restqa.py
Source:restqa.py
1from __future__ import print_function2from __future__ import unicode_literals3from __future__ import division4from __future__ import absolute_import5from builtins import open6from future import standard_library7standard_library.install_aliases()8from builtins import object9import os10import re11import glob12import yaml13import jmespath14import requests15JSON_SELECTOR_REGEX = r'(resp(\.[a-zA-Z0-9]+)+)'16def none_function(ctxt):17 return ctxt18class RestTestsuiteDriver(object):19 def __init__(self, test_suite_dir, function_dict):20 self.test_suite_dir = test_suite_dir21 self.function_dict = function_dict22 def run_tests(self):23 test_suite_files_expr = "{}/*_test_suite.yml".format(self.test_suite_dir)24 suites = glob.glob(test_suite_files_expr)25 for suite in suites:26 with open(suite, 'r') as yaml_in:27 suite_settings = yaml.safe_load(yaml_in)28 suite_settings = suite_settings["suite"]29 suite_base_url = suite_settings.get("base_url", None)30 print("suite_base_url: {}".format(suite_base_url))31 suite_name = suite_settings.get("name", None)32 suite_setup = suite_settings.get("setup", None)33 suite_teardown = suite_settings.get("teardown", None)34 print("suite: {}::setup_function: {}::teardown_function: {}".format(suite_name, suite_setup, suite_teardown))35 try:36 ctxt = {}37 setup_function = self.function_dict.get(suite_setup, none_function)38 ctxt = setup_function(ctxt)39 tests = suite_settings.get("tests", [])40 for test_settings in tests:41 try:42 test_name = test_settings.get("name", "None")43 test_json_payload = test_settings.get("json", False)44 test_http_headers = test_settings.get("headers", [])45 test_http_method = test_settings.get("method", "GET")46 test_http_path = test_settings.get("path", None)47 test_http_payload = test_settings.get("payload", None)48 test_pre = test_settings.get("pre_test", None)49 test_post = test_settings.get("post_test", None)50 test_assertions = test_settings.get("assertions", None)51 headers = {}52 for test_http_header in test_http_headers:53 for header_name, header_value in test_http_header.items():54 headers[header_name] = header_value55 optional_params = {56 "headers": headers57 }58 if test_http_method == "GET" and test_http_payload != None:59 optional_params["params"] = test_http_payload60 if test_http_method == "POST":61 if test_http_payload != None:62 if test_json_payload:63 optional_params["json"] = json.dumps(test_http_payload)64 else:65 optional_params["data"] = test_http_payload66 request_url = "{}{}".format(suite_base_url, test_http_path)67 print("request_url: {}".format(request_url))68 test_pre_function = self.function_dict.get(test_pre, none_function)69 ctxt = test_pre_function(ctxt)70 try:71 resp = requests.request(test_http_method, request_url, **optional_params)72 resp = resp.json()73 locals = {74 "ctxt": ctxt,75 "resp": resp76 }77 for expression in test_assertions:78 matches = re.findall(JSON_SELECTOR_REGEX, expression)79 print("expression: {}/properties: {}".format(expression, matches))80 for match in matches:81 replace_expr = match[0].replace('resp\.', '')82 replace_expr = "{}'{}'{}".format("jmespath.search(", replace_expr ,", resp)")83 expression = expression.replace(match[0], replace_expr)84 print("New expression: {}".format(expression))85 eval(expression, None, locals)86 except Exception as e2:87 print("Exception occured while executing suite: {} / test: {} / {}".format(suite_name, test_name, e2))88 test_post_function = self.function_dict.get(test_post, none_function)89 ctxt = test_post_function(ctxt)90 except Exception as e1:91 print("Exception occured while executing suite: {} / test: {} / {}".format(suite_name, test_name, e1))92 teardown_function = self.function_dict.get(suite_teardown, none_function)93 ctxt = teardown_function(ctxt)94 except Exception as e:...
test_request.py
Source:test_request.py
2import jsonpickle3from telegrambotapiwrapper.request import json_payload4from telegrambotapiwrapper.typelib import *5class TestRequest(unittest.TestCase):6 def test_json_payload(self):7 btn1 = InlineKeyboardButton(text='add', url='http://lenta.ru')8 btn2 = InlineKeyboardButton(text='sub', url='http://topwar.ru')9 btn3 = InlineKeyboardButton(text='mul', url='http://waralbum.ru')10 btn4 = InlineKeyboardButton(text='div', url='http://anio.ru')11 inline_kb = InlineKeyboardMarkup([[btn1, btn2], [btn3, btn4]])12 payload = json_payload(inline_kb)13 res = {14 "inline_keyboard": [[{15 "text": "add",16 "url": "http://lenta.ru"17 }, {18 "text": "sub",19 "url": "http://topwar.ru"20 }],...
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 locust 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?
