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

How to use add_request method in localstack

Best Python code snippet using localstack_python

test_rate_limiters.py

Source:test_rate_limiters.py

...25        rate_limiter, result = create_rate_limiter(26            BatchMessageRateLimiter, name="Test", factory=create_message, delay=0.127        )28        await sleep(0.1)  # let the nursery start the rate limiter29        rate_limiter.add_request(1)30        await sleep(1)31        rate_limiter.add_request(2)32        rate_limiter.add_request(3)33        rate_limiter.add_request(4)34        await sleep(1)35        assert result == [(1,), (2, 3, 4)]36    async def test_yields_correctly_with_batching(37        self, create_rate_limiter, autojump_clock38    ):39        rate_limiter, result = create_rate_limiter(40            BatchMessageRateLimiter, name="Test", factory=create_message, delay=0.141        )42        await sleep(0.1)  # let the nursery start the rate limiter43        # batch 1 - this is always sent immediately ###44        rate_limiter.add_request(1)45        # batch 2 - gets delayed because batch 1 went out recently ###46        await sleep(0.05)47        rate_limiter.add_request(1)48        rate_limiter.add_request(3)49        rate_limiter.add_request(4)50        await sleep(0.01)51        rate_limiter.add_request(1)52        await sleep(0.02)53        rate_limiter.add_request(2)54        rate_limiter.add_request(4)55        await sleep(1)56        # batch 3 - this is sent immediately because there was 1s of silence ###57        rate_limiter.add_request(3)58        await sleep(0.05)59        # batch 4 - gets delayed because batch 3 went out recently ###60        rate_limiter.add_request(3)61        rate_limiter.add_request(4)62        await sleep(0.03)63        rate_limiter.add_request(6)64        await sleep(1)65        assert result == [(1,), (1, 2, 3, 4), (3,), (3, 4, 6)]66class TestUAVMessageRateLimiter:67    async def test_yields_nothing_by_default(self, create_rate_limiter, autojump_clock):68        _, result = create_rate_limiter(69            UAVMessageRateLimiter, name="Test", factory=create_message, delay=0.170        )71        assert result == []72    async def test_yields_correctly_without_batching(73        self, create_rate_limiter, autojump_clock74    ):75        rate_limiter, result = create_rate_limiter(76            UAVMessageRateLimiter, name="Test", factory=create_message, delay=0.177        )78        await sleep(0.1)  # let the nursery start the rate limiter79        rate_limiter.add_request((1, 2))80        await sleep(1)81        rate_limiter.add_request((3, 4, 5))82        await sleep(1)83        assert result == [(1, 2), (3, 4, 5)]84    async def test_yields_correctly_with_batching(85        self, create_rate_limiter, autojump_clock86    ):87        rate_limiter, result = create_rate_limiter(88            UAVMessageRateLimiter, name="Test", factory=create_message, delay=0.189        )90        await sleep(0.1)  # let the nursery start the rate limiter91        # batch 1 - this is always sent immediately ###92        rate_limiter.add_request((1, 2))93        # batch 2 - gets delayed because batch 1 went out recently ###94        await sleep(0.05)95        rate_limiter.add_request((1, 3, 4))96        await sleep(0.01)97        rate_limiter.add_request((1,))98        await sleep(0.02)99        rate_limiter.add_request((2, 4))100        await sleep(1)101        # batch 3 - this is sent immediately because there was 1s of silence ###102        rate_limiter.add_request((3, 4, 5))103        await sleep(0.05)104        # batch 4 - gets delayed because batch 3 went out recently ###105        rate_limiter.add_request((3, 4))106        await sleep(0.03)107        rate_limiter.add_request((6,))108        await sleep(1)...

pedf_spinlocks.py

Source:pedf_spinlocks.py

...30        self.trivial_ts[3].partition = 1;31        self.trivial_ts[4].partition = 1;32        # ---------[ Reuqests ]-----------------------33        # L_{0,0}=3, N_{0,0}=134        self.trivial_ts[0].resmodel[0].add_request(3)35        # L_{1,0}=3, N_{1,0}=136        self.trivial_ts[1].resmodel[0].add_request(1)37        #self.trivial_ts[1].resmodel[0].add_request(1)38        # L_{1,1}=5, N_{1,1}=139        self.trivial_ts[1].resmodel[1].add_request(5)40        # L_{2,0}=5, N_{2,0}=141        self.trivial_ts[2].resmodel[0].add_request(1)42        self.trivial_ts[2].resmodel[0].add_request(1)43        self.trivial_ts[2].resmodel[0].add_request(1)44        # L_{3,0}=1, N_{3,0}=245        self.trivial_ts[3].resmodel[0].add_request(1)46        self.trivial_ts[3].resmodel[0].add_request(1)47        # L_{4,0}=1, N_{4,0}=148        self.trivial_ts[4].resmodel[0].add_request(4)49        # L_{4,1}=1, N_{4,1}=250        self.trivial_ts[4].resmodel[1].add_request(1)51        self.trivial_ts[4].resmodel[1].add_request(1)52        # L_{4,2}=2, N_{4,2}=153        self.trivial_ts[4].resmodel[2].add_request(2)54    @unittest.skipIf(not schedcat.locking.bounds.lp_cpp_available, "no native LP solver available")55    def test_MSRP(self):56        if lb.pedf_msrp_is_schedulable(self.trivial_ts):57            print "[MSRP] SCHEDULABLE";58        else:59            print "[MSRP] NOT SCHEDULABLE";60    @unittest.skipIf(not schedcat.locking.bounds.lp_cpp_available, "no native LP solver available")61    def test_FIFO_preempt(self):62        if lb.pedf_fifo_preempt_is_schedulable(self.trivial_ts):63            print "[FIFO Preemptive] SCHEDULABLE";64        else:65            print "[FIFO Preemptive] NOT SCHEDULABLE";66    def test_MSRP_classic(self):67        if lb.pedf_msrp_classic_is_schedulable(self.trivial_ts, self.trivial_num_cpus):...

task_server.py

Source:task_server.py

1#!/usr/bin/env python2"""3Created on Apr 13, 20154Last updated on Apr 30, 20185@author: Luigi De Russis, Teodoro Montanaro, Alberto Monge Roffarello6"""7from flask import Flask, jsonify, abort, request, Response, render_template8from flask_bootstrap import Bootstrap9import db_interaction10app = Flask(__name__)11Bootstrap(app)12# ---------- FRONT-END Single-page application ------------13@app.route('/tasks.html')14def tasks():15    return render_template("tasks.html")16# ---------- REST SERVER ----------17@app.route('/api/v1.0/tasks', methods=['GET'])18def get_tasks():19    # init20    tasks = []21    # get the task list from the db22    task_list = db_interaction.get_tasks()23    # prepare the task list for jsonify24    for item in task_list:25        task = prepare_for_json(item)26        tasks.append(task)27    # return the task data28    return jsonify({'tasks': tasks})29@app.route('/api/v1.0/tasks/<int:task_id>', methods=['GET'])30def get_task(task_id):31    # get the task32    task = db_interaction.get_task(int(task_id))33    # return 404 not found if no task has the given id34    if task is None:35        abort(404)36    # convert the task in a JSON representation37    return jsonify({'task': prepare_for_json(task)})38@app.route('/api/v1.0/tasks', methods=['POST'])39def insert_task():40    # get the request body41    add_request = request.json42    # check whether a task is present in the request or not43    if (add_request is not None) and ('description' in add_request) and ('urgent' in add_request):44        text = add_request['description']45        urgent = add_request['urgent']46        # insert in the database47        db_interaction.insert_task(text, urgent)48        return Response(status=201)49    # return an error in case of problems50    abort(403)51@app.route('/api/v1.0/tasks/<int:task_id>', methods=['PUT'])52def update_task(task_id):53    # get the request body54    add_request = request.json55    # check whether a task is present in the request or not56    if add_request is not None and ('description' and 'urgent') in add_request:57        text = add_request['description']58        urgent = add_request['urgent']59        # update the task60        task = db_interaction.update_task(int(task_id),text,urgent)61        return Response(status=200)62    # return an error in case of problems63    abort(403)64@app.route('/api/v1.0/tasks/<int:task_id>', methods=['DELETE'])65def delete_task(task_id):66    # delete the task67    task = db_interaction.remove_task_by_id(int(task_id))68    return Response(status=200)69def prepare_for_json(item):70    """71    Convert the task in a dictionary for easing the JSON creation72    """73    task = dict()74    task['id'] = item[0]75    task['description'] = item[1]76    task['urgent'] = item[2]77    return task78if __name__ == '__main__':...

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.