Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

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.