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

How to use check_invocation method in localstack

Best Python code snippet using localstack_python

test_procedure_runner.py

Source:test_procedure_runner.py

...36        yield "progress1"37        yield "final"38    invocation = mock.make_dummy_invocation()39    await AsyncGenRunner(invocation, gen())40    check_invocation(invocation,41                     ("progress0", "instant progress", "progress1"),42                     "final")43async def test_async_gen_runner_interrupt():44    async def gen():45        yield "a"46        await asyncio.sleep(0)47        yield "final"48    invocation = mock.make_dummy_invocation()49    runner = AsyncGenRunner(invocation, gen())50    await asyncio.sleep(0)51    await runner.interrupt(make_interrupt())52    await runner53    check_invocation(invocation, (), error=True)54async def test_async_gen_runner_cancel():55    async def gen():56        yield "a"57        await asyncio.sleep(0)58        yield "final"59    invocation = mock.make_dummy_invocation()60    runner = AsyncGenRunner(invocation, gen())61    await asyncio.sleep(0)62    runner.cancel()63    await runner64    assert not mock.get_messages(invocation)65async def test_async_gen_runner_interrupt_handle():66    async def gen():67        yield "a"68        yield "b"69        await asyncio.sleep(0)70        try:71            yield "c"72        except aiowamp.Interrupt:73            yield "plz no error"74    invocation = mock.make_dummy_invocation()75    runner = AsyncGenRunner(invocation, gen())76    await asyncio.sleep(0)77    await runner.interrupt(make_interrupt())78    await runner79    check_invocation(invocation, ("a",), "plz no error")80async def test_coro_runner():81    async def procedure():82        return "hello world"83    invocation = mock.make_dummy_invocation()84    await CoroRunner(invocation, procedure())85    check_invocation(invocation, (), "hello world")86async def test_coro_runner_interrupt():87    async def procedure():88        await asyncio.sleep(50)89        return "hello world"90    invocation = mock.make_dummy_invocation()91    runner = CoroRunner(invocation, procedure())92    await asyncio.sleep(0)93    await runner.interrupt(make_interrupt())94    await runner95    check_invocation(invocation, (), error=True)96async def test_coro_runner_cancel():97    async def procedure():98        await asyncio.sleep(50)99        return "hello world"100    invocation = mock.make_dummy_invocation()101    runner = CoroRunner(invocation, procedure())102    await asyncio.sleep(0)103    runner.cancel()104    await runner105    assert not mock.get_messages(invocation)106async def test_coro_runner_interrupt_handle():107    async def procedure():108        try:109            await asyncio.sleep(50)110        except aiowamp.Interrupt:111            pass112        return "hello world"113    invocation = mock.make_dummy_invocation()114    runner = CoroRunner(invocation, procedure())115    await asyncio.sleep(0)116    await runner.interrupt(make_interrupt())117    await runner118    check_invocation(invocation, (), "hello world")119async def test_awaitable_runner():120    def a():121        return asyncio.sleep(0, "hello world")122    invocation = mock.make_dummy_invocation()123    await AwaitableRunner(invocation, a())124    check_invocation(invocation, (), "hello world")125async def test_awaitable_runner_interrupt():126    def a():127        return asyncio.create_task(asyncio.sleep(3600, "hello world"))128    invocation = mock.make_dummy_invocation()129    runner = AwaitableRunner(invocation, a())130    await runner.interrupt(make_interrupt())131    await runner132    check_invocation(invocation, (), error=True)133async def test_awaitable_runner_cancel():134    def a():135        return asyncio.create_task(asyncio.sleep(3600, "hello world"))136    invocation = mock.make_dummy_invocation()137    runner = AwaitableRunner(invocation, a())138    runner.cancel()139    await runner...

test_logs.py

Source:test_logs.py

...96            ],97        )98        resp2 = log_client.describe_subscription_filters(logGroupName=log_group_name)99        self.assertEqual(len(resp2['subscriptionFilters']), 1)100        def check_invocation():101            events = testutil.get_lambda_log_events(TEST_LAMBDA_NAME_PY3)102            self.assertEqual(len(events), 2)...

check.py

Source:check.py

...19    assert_msg(b, aiowamp.msg.Error)20    assert a.error == b.error21    assert a.args == b.args22    assert a.kwargs == b.kwargs23def check_invocation(invocation: aiowamp.InvocationABC,24                     progress: Iterable[aiowamp.InvocationHandlerResult],25                     final: aiowamp.InvocationHandlerResult = None, *,26                     error: Union[aiowamp.msg.Error, bool] = None) -> None:27    sent_messages = mock.get_messages(invocation)28    print("\nSENT:", sent_messages)29    final_msg = sent_messages.pop()30    for expected, msg in itertools.zip_longest(progress, sent_messages):31        assert expected is not None, ("got unexpected progress", msg)32        assert msg is not None, ("missing progress", expected)33        assert assert_progress_yield(msg), "not a progress message"34        args, kwargs = get_return_values(expected)35        check_args_kwargs(msg, args, kwargs)36    if final is not None or error is not None:37        assert final_msg is not None, "no final message"...

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.