KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use is_empty_response method in localstack

Best Python code snippet using localstack_python

test_server.py

Source:test_server.py

...30                source=config.node_mapping[node_id]["addr"],31            ),32        )33    return inner34def is_empty_response(resp: server.ResponsesEvents) -> bool:35    return not resp.responses and not resp.events36# Testing Conversions from one Server type to another37def test_candidate_validate_conversions(candidate):38    # any other server type is valid39    assert candidate.validate_conversion(server.Follower) is True40    assert candidate.validate_conversion(server.Candidate) is True41    assert candidate.validate_conversion(server.Leader) is True42def test_follower_validate_conversions(follower):43    with pytest.raises(ValueError):44        follower.validate_conversion(server.Follower)45    with pytest.raises(ValueError):46        follower.validate_conversion(server.Leader)47    with pytest.raises(ValueError):48        # Some random value...49        follower.validate_conversion(server.Leader.mro()[1])50    # only this is valid51    assert follower.validate_conversion(server.Candidate) is True52def test_leader_validate_conversions(leader):53    with pytest.raises(ValueError):54        leader.validate_conversion(server.Candidate)55    with pytest.raises(ValueError):56        leader.validate_conversion(server.Leader)57    with pytest.raises(ValueError):58        # Some random value...59        leader.validate_conversion(server.Leader.mro()[1])60    # only this is valid61    assert leader.validate_conversion(server.Follower) is True62# Testing Follower Log Append63@pytest.mark.parametrize(64    "log_to_pull,success_expected",65    (66        ("a_log", False),67        ("b_log", False),68        ("c_log", True),69        ("d_log", True),70        ("e_log", False),71        ("f_log", False),72    ),73)74def test_follower_msg_append(75    log_to_pull, success_expected, follower, figure7_logs, fig7_sample_message76):77    follower.log = figure7_logs[log_to_pull]78    follower.commit_index = 479    follower.current_term = follower.log[-1].term80    event = Event(EventType.LeaderAppendLogEntryRpc, fig7_sample_message)81    result, more_events = follower.handle_append_entries_message(event)82    # We expect an event to trigger election timeout restart in runtime83    assert len(more_events) == 184    assert more_events[0].type == EventType.ResetElectionTimeout85    # we also expect a response message86    assert len(result) == 187    resp = result[0]88    assert resp.success is success_expected89    assert resp.source == fig7_sample_message.dest90    assert resp.dest == fig7_sample_message.source91    expected_match = 10 if success_expected else -192    assert resp.match_index == expected_match93    # test the higher-level function94    inst, (resps, more_events) = follower.handle_event(event)95    assert inst is follower96    # We expect an event to trigger election timeout restart in runtime97    assert len(more_events) == 198    assert more_events[0].type == EventType.ResetElectionTimeout99    resp = resps[0]100    assert resp.success is success_expected101    assert resp.source == fig7_sample_message.dest102    assert resp.dest == fig7_sample_message.source103# Testing Candidate conversions104@pytest.mark.parametrize(105    "etype,new_class,events",106    (107        (108            EventType.SelfWinElection,109            server.Leader,110            [models.EVENT_CONVERSION_TO_LEADER, models.EVENT_START_HEARTBEAT],111        ),112        (113            EventType.LeaderAppendLogEntryRpc,114            server.Follower,115            [models.EVENT_CONVERSION_TO_FOLLOWER],116        ),117        (EventType.Tick, None, []),118    ),119)120def test_follower_candidate_convert(etype, new_class, events, candidate):121    event = Event(etype, None)122    inst, resps = candidate.handle_event(event)123    if etype == EventType.Tick:124        assert is_empty_response(resps)125    else:126        assert not is_empty_response(resps)127        for left, right in zip(events, resps.events):128            assert left == right129    if new_class is None:130        assert inst == candidate131        return None  # Further tests below are about conversion132    assert isinstance(inst, new_class)133    assert id(inst) != id(candidate)134    if isinstance(inst, server.Candidate):135        assert inst.current_term == candidate.current_term + 1136    for key in filter(lambda el: el != "current_term", inst.transfer_attrs):137        assert getattr(inst, key) == getattr(candidate, key)138# Election starts (or restarts for Candidate) -> Returns new candidate and Sends Vote Requests139def test_candidate_construct_request_vote_rpcs(follower, candidate):140    """There's a lot going on in this test"""141    for raft_server in (follower, candidate):142        event = Event(EventType.ElectionTimeoutStartElection, None)143        new_candidate, resp_events = raft_server.handle_event(event)144        assert new_candidate is not raft_server145        resps, empty = new_candidate.handle_start_election(event)146        assert resp_events and resp_events == (resps, empty)147        assert resps148        assert not empty  # no successive events149        assert new_candidate.current_term == raft_server.current_term + 1150        for key in filter(lambda el: el != "current_term", raft_server.transfer_attrs):151            assert getattr(raft_server, key) == getattr(new_candidate, key)152        for node_id, msg in zip(new_candidate.all_node_ids, resps):153            # make sure leader is not including itself in recipients154            assert new_candidate.address != msg.dest155            # Check recipient is correct156            assert new_candidate.config.node_mapping[node_id]["addr"] == msg.dest157            # make sure leader is telling nodes where to send replies158            assert new_candidate.address == msg.source159            # Get the message and confirm its values make sense160            msg.last_log_index == len(new_candidate.log) - 1161            msg.last_log_term == new_candidate.log[-1].term if new_candidate.log else 0162def test_follower_handle_request_vote_rpc(163    follower, fig7_a_log, candidate_request_vote_event164):165    follower.log = fig7_a_log166    follower.current_term = follower.log[-1].term167    # Grant vote if log long enough and nobody voted for yet168    event = candidate_request_vote_event(1, 8, ("127.0.0.1", 3112))169    event.msg.last_log_index = len(follower.log) + 1170    event.msg.last_log_term = follower.log[-1].term171    follower.node_id = 2172    results = follower.handle_request_vote_rpc(event)173    assert not is_empty_response(results)174    resp = results.responses[0]175    assert resp.vote_granted176    assert follower.voted_for == 1177    assert resp.dest == event.msg.source178    assert resp.source == event.msg.dest179    assert len(results.events) == 1180    assert results.events[0].type == EventType.ResetElectionTimeout181    # mark it something else and see vote rejected182    follower.voted_for = 3183    results = follower.handle_request_vote_rpc(event)184    assert not is_empty_response(results)185    resp = results.responses[0]186    assert not resp.vote_granted187    assert follower.voted_for == 3188    assert resp.dest == event.msg.source189    assert resp.source == event.msg.dest190    assert not results.events191    # reset to None and check the log_index is long enough for vote192    follower.voted_for = None193    event.msg.last_log_index = 2194    results = follower.handle_request_vote_rpc(event)195    assert not is_empty_response(results)196    resp = results.responses[0]197    assert not resp.vote_granted198    assert follower.voted_for is None199    assert resp.dest == event.msg.source200    assert resp.source == event.msg.dest201    assert not results.events202    inst, resps_events = follower.handle_event(event)203    assert inst is follower204    assert inst.current_term == event.msg.term205    result, empty = resps_events206    assert not empty207    assert not result[0].vote_granted208    assert not results.events209def test_handle_vote_response_greater_term(210    candidate, follower, fig7_a_log, candidate_request_vote_event211):212    # First, we try it with a term that's < Candidate's term213    candidate.current_term = 20214    follower.log = fig7_a_log215    follower.current_term = follower.log[-1].term216    # Reusing logic from follower-request vote tests217    event = candidate_request_vote_event(1, 8, ("127.0.0.1", 3112))218    event.msg.last_log_index = len(follower.log) + 1219    event.msg.last_log_term = follower.log[-1].term220    follower.node_id = 2221    responses, _ = follower.handle_request_vote_rpc(event)222    # sanity check223    assert responses224    resp_event = Event(EventType.ReceiveServerCandidateVote, responses[0])225    inst, responses = candidate.handle_event(resp_event)226    assert is_empty_response(responses)227    assert inst == candidate228    # Now we try it again and turn the candidate's current_term down229    # Calling handle-event with a reponse with greater term creates conversion230    candidate.current_term = 1231    inst, responses = candidate.handle_event(resp_event)232    assert inst != candidate233    assert not is_empty_response(responses)234    assert isinstance(inst, server.Follower)235    assert responses.events[0] == models.EVENT_CONVERSION_TO_FOLLOWER236def test_handle_vote_response(237    candidate, follower, fig7_a_log, candidate_request_vote_event238):239    follower.log = fig7_a_log240    follower.current_term = follower.log[-1].term241    # Reusing logic from follower-request vote tests242    event = candidate_request_vote_event(1, 8, ("127.0.0.1", 3112))243    event.msg.last_log_index = len(follower.log) + 1244    event.msg.last_log_term = follower.log[-1].term245    follower.node_id = 2246    responses, _ = follower.handle_request_vote_rpc(event)247    # sanity check248    assert responses249    resp_event = Event(EventType.ReceiveServerCandidateVote, responses[0])250    result = candidate.handle_vote_response(resp_event)251    assert is_empty_response(result)252    # Idempotent: same response doesn't magically make them win253    result = candidate.handle_vote_response(resp_event)254    assert is_empty_response(result)255    # After they have more votes, they can win256    candidate.votes_received = set((3, 4))257    result = candidate.handle_vote_response(resp_event)258    assert not is_empty_response(result)259    _, events = result260    assert events261    assert events[0].type == EventType.SelfWinElection262# Testing Leader events263def test_leader_handle_event(leader, sample_append_confirm_rpc):264    etype = EventType.AppendEntryConfirm265    event = Event(etype, sample_append_confirm_rpc)266    leader.current_term = 20267    # should not convert268    inst, maybe_resp_evs = leader.handle_event(event)269    assert is_empty_response(maybe_resp_evs)270    assert inst is leader271    # Try to coerce it to convert272    event.msg.term = 100273    inst2, maybe_resp_evs = leader.handle_event(event)274    assert inst2 is not leader275    assert not is_empty_response(maybe_resp_evs)276    assert len(maybe_resp_evs.events) == 1277    assert maybe_resp_evs.events[0] == models.EVENT_CONVERSION_TO_FOLLOWER278    assert isinstance(inst2, server.Follower)279# Testing Leader Steps down280def check_that_leader_steps_down_if_out_of_date(leader, sample_append_confirm_rpc):281    event = Event(EventType.AppendEntryConfirm, sample_append_confirm_rpc)282    event.msg.term = 20283    # In this case, the leader realizes it's out of date and immediately converts284    inst, maybe_resp_evs = leader.handle_event(event)285    assert maybe_resp_evs is not None286    assert inst is not leader287    assert isinstance(inst, server.Follower)288# Testing Send Append Entries RPC289def test_leader_handle_heartbeat(leader):290    """There's a lot going on in this test"""291    event = Event(EventType.HeartbeatTime, None)292    inst, (resps, empty) = leader.handle_event(event)293    # no conversion happened294    assert not empty295    assert inst is leader296    for node_id, msg in zip(leader.all_node_ids, resps):297        # make sure leader is not including itself in recipients298        assert leader.address != msg.dest299        # Check recipient is correct300        assert leader.config.node_mapping[node_id]["addr"] == msg.dest301        # make sure leader is telling nodes where to send replies302        assert leader.address == msg.source303        # Get the message and confirm its values make sense304        msg.prev_log_index == leader.next_index[node_id] - 1305        len(msg.entries) + leader.next_index[node_id] == len(leader.log)306# This should be a *known* client!307APPEND_SUCCESS = Event(308    EventType.AppendEntryConfirm,309    rpc.AppendEntriesResponse(310        term=6, match_index=10, source_node_id=2, success=True, dest=("127.0.0.1", 3112)311    ),312)313# This should be a *known* client!314APPEND_FAIL = Event(315    EventType.AppendEntryConfirm,316    rpc.AppendEntriesResponse(317        term=1, match_index=2, source_node_id=4, success=False, dest=("127.0.0.1", 3114)318    ),319)320# Testing Leader handling append321def test_leader_handle_append_response(leader):322    """There's a lot going on in this test"""323    inst, resps = leader.handle_event(APPEND_SUCCESS)324    assert is_empty_response(resps)325    assert inst is leader326    assert leader.commit_index == 8327    assert leader.match_index[APPEND_SUCCESS.msg.source_node_id] == len(leader.log)328    assert leader.next_index[APPEND_SUCCESS.msg.source_node_id] == len(leader.log) + 1329    # before failure: grab current expected next_index330    current_next_for_node = leader.next_index[APPEND_FAIL.msg.source_node_id]331    current_match_for_node = leader.next_index[APPEND_FAIL.msg.source_node_id]332    inst, resps = leader.handle_event(APPEND_FAIL)333    assert is_empty_response(resps)334    assert inst is leader335    assert leader.commit_index == 8336    assert leader.match_index[APPEND_FAIL.msg.source_node_id] == current_match_for_node337    assert (338        leader.next_index[APPEND_FAIL.msg.source_node_id] == current_next_for_node - 1...

test_template_versions.py

Source:test_template_versions.py

1#!/usr/bin/python2# -*- coding: utf-8 -*-3import sys4import unittest5import requests6if sys.version_info >= (3, 0):7    from unittest.mock import MagicMock, patch8else:9    from mock import patch, MagicMock10from shellfoundry.exceptions import NoVersionsHaveBeenFoundException11from shellfoundry.utilities.template_versions import TemplateVersions12def mock_get_branches_from_github(is_empty_response=False):13    if is_empty_response:14        response_content = "[]"15    else:16        response_content = """[17      {18        "name": "5.0.0",19        "commit": {20          "sha": "efe253280a3346c2be23b1a9af4113f2b989f26c",21          "url": "https://api.github.com/repos/user/repo/commits/efe253280a3346c2be23b1a9af4113f2b989f26c"22        }23      },24      {25        "name": "5.0.1",26        "commit": {27          "sha": "e67ec4de2f00fabfe7be35fe412ec400847ccc7d",28          "url": "https://api.github.com/repos/user/repo/commits/e67ec4de2f00fabfe7be35fe412ec400847ccc7d"29        }30      },31      {32        "name": "5.0.2",33        "commit": {34          "sha": "421c20b231b11672411f39f813c93d0f82723f3a",35          "url": "https://api.github.com/repos/user/repo/commits/421c20b231b11672411f39f813c93d0f82723f3a"36        }37      },38      {39        "name": "master",40        "commit": {41          "sha": "b8687aef6a15a4fd9c6daa6b7549470e9e3c4c11",42          "url": "https://api.github.com/repos/user/repo/commits/b8687aef6a15a4fd9c6daa6b7549470e9e3c4c11"43        }44      }45    ]46    """  # noqa E50147    response = requests.models.Response()48    response.status_code = 20049    response._content = str.encode(response_content)50    return response51class TestTemplateVersions(unittest.TestCase):52    def setUp(self):53        self.template_version = TemplateVersions("user", "repo")54    @patch(55        "shellfoundry.utilities.template_versions.requests.get",56        MagicMock(side_effect=requests.HTTPError("Failed to receive versions")),57    )58    def test_get_versions_of_template_error_due_to_request_failed(self):59        with self.assertRaises(requests.HTTPError) as context:60            self.template_version.get_versions_of_template()61        # Assert62        self.assertEqual(str(context.exception), "Failed to receive versions")63    @patch(64        "shellfoundry.utilities.template_versions.requests.get",65        MagicMock(return_value=mock_get_branches_from_github(is_empty_response=True)),66    )67    def test_get_versions_of_template_and_has_no_versions_failure(self):68        with self.assertRaises(NoVersionsHaveBeenFoundException) as context:69            self.template_version.get_versions_of_template()70            # Assert71            self.assertEqual(72                str(context.exception), "No versions have been found for this template"73            )74    @patch(75        "shellfoundry.utilities.template_versions.requests.get",76        MagicMock(return_value=mock_get_branches_from_github()),77    )78    def test_get_versions_of_template_reversed_success(self):79        # Act80        versions = self.template_version.get_versions_of_template()81        # Assert...

val_user.py

Source:val_user.py

1from rest_framework.exceptions import ErrorDetail2from rest_framework.response import Response3from ..models import User4# return (user, response)5def validate_username(username):6    if not username:7        return None, Response({8            'username': ErrorDetail(string='Username cannot be empty.',9                                    code='blank')10        }, 400)11    try:12        user = User.objects.get(username=username)13    except User.DoesNotExist:14        return None, Response({15            'username': ErrorDetail(string='User not found.',16                                    code='not_found')17        }, 404)18    return user, None19# return (is_active, response)20def validate_is_active(is_active):21    is_empty_response = Response({22        'is_active': ErrorDetail(string='Is Active cannot be empty.',23                                 code='blank')24    }, 400)25    try:26        if not str(is_active):27            return None, is_empty_response28    except ValueError:29        return None, is_empty_response30    if not isinstance(is_active, bool):31        return None, Response({32            'is_active': ErrorDetail(string='Is Active must be a boolean.',33                                     code='invalid')34        }, 400)...

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.