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

How to use get_resources method in localstack

Best Python code snippet using localstack_python

test_util.py

Source:test_util.py

...26        channel.remote_eval.assert_called_once_with(code)27        rct.assert_called_once_with(channel.remote_eval(code), event_loop)28@pytest.mark.asyncio29async def test_get_resources_success_on_no_query(event_loop):30    assert get_resources(event_loop, AsyncTinyDB(), {}) == {}31@pytest.mark.asyncio32async def test_get_resources_returns_none_when_spec_not_satisfied(event_loop):33    store = AsyncTinyDB()34    spec = {'foo': body.foo == 'bar'}35    assert get_resources(event_loop, store, spec) is None36@given(keys=st.sets(st.text()))37@pytest.mark.asyncio38@pytest.mark.slowtest39async def test_get_resources_calls_store_search_for_each_spec(keys, event_loop):40    store = Mock()41    spec = {k: body[k] == k for k in keys}42    get_resources(event_loop, store, spec)43    assert store.search.call_count == len(keys)44@given(keys=st.sets(st.text()))45@pytest.mark.asyncio46@pytest.mark.slowtest47async def test_get_resources_calls_from_agent_info_for_each_doc_found(keys, event_loop):48    store = AsyncTinyDB()49    for key in keys:50        await store.insert({key: key})51    spec = {k: body[k] == k for k in keys}52    with patch('brainslug.remote.Remote.from_agent_info') as from_agent_info:53        get_resources(event_loop, store, spec)54        assert from_agent_info.call_count == len(keys)55@given(keys=st.sets(st.text()))56@pytest.mark.asyncio57@pytest.mark.slowtest58async def test_get_resources_calls_from_agent_info_and_returns_in_dict_value(keys, event_loop):59    store = AsyncTinyDB()60    for key in keys:61        await store.insert({key: key})62    spec = {k: body[k] == k for k in keys}63    def _from_agent_info(loop, doc):64        for k in doc:65            return k66    with patch('brainslug.remote.Remote.from_agent_info', _from_agent_info) as from_agent_info:67        resources = get_resources(event_loop, store, spec)68        assert resources == {k: k for k in keys}69async def test_wait_for_resources_return_resources(event_loop):70    resources = object()71    with patch('brainslug.runtime.get_resources') as get_resources:72        get_resources.return_value = resources73        assert await wait_for_resources(event_loop, None, None) is resources74@pytest.mark.asyncio75@pytest.mark.slowtest76async def test_wait_for_resources_hangs_if_no_resources(event_loop):77    class store:78        wait_for_insert = staticmethod(lambda: asyncio.sleep(0))79    spec = {'foo': body['foo'] == 'bar'}80    with patch('brainslug.runtime.get_resources') as get_resources:81        get_resources.side_effect = itertools.repeat(None)82        with pytest.raises(asyncio.TimeoutError):83            await asyncio.wait_for(wait_for_resources(event_loop, store, spec), 1)84        get_resources.assert_called()85@given(num_fails=st.integers(min_value=0, max_value=100))86@pytest.mark.asyncio87@pytest.mark.slowtest88async def test_wait_for_resources_calls_wait_for_every_failed_get_resources(event_loop, num_fails):89    store = Mock()90    store.wait_for_insert.side_effect = lambda: asyncio.sleep(0)91    spec = {'foo': body['foo'] == 'bar'}92    result = {'foo': object()}93    with patch('brainslug.runtime.get_resources') as get_resources:94        get_resources.side_effect = itertools.chain(95            itertools.repeat(None, num_fails),96            [result])97        await asyncio.wait_for(wait_for_resources(event_loop, store, spec), 1)98        assert get_resources.call_count == num_fails + 1...

example_tag_check.py

Source:example_tag_check.py

...17def main():18    tagcheck = AwsResources.TagCheck(config_path)19    tagcheck.load_tag_defs()20    print("EC2 Instances")21    ec2instance_tags = AwsResources.Ec2InstanceTags().get_resources().get_tags()22    display_tag_failure(tagcheck.get_noncomp_resources(ec2instance_tags))23    # print(json.dumps(elb_tags, indent=4))24    print("EC2 Volumes")25    ec2volume_tags = AwsResources.Ec2VolumeTags().get_resources().get_tags()26    display_tag_failure(tagcheck.get_noncomp_resources(ec2volume_tags))27    # print(json.dumps(elb_tags, indent=4))28    print("Elastic Load Balancers")29    elb_tags = AwsResources.ElbTags().get_resources().get_tags()30    display_tag_failure(tagcheck.get_noncomp_resources(elb_tags))31    # print(json.dumps(elb_tags, indent=4))32    print("Elastic Map Reduce")33    emr_tags = AwsResources.EmrTags().get_resources().get_tags()34    display_tag_failure(tagcheck.get_noncomp_resources(emr_tags))35    print('Data Pipelines')36    dp_tags = AwsResources.DataPipelineTags().get_resources().get_tags()37    display_tag_failure(tagcheck.get_noncomp_resources(dp_tags))38    # print(json.dumps(dp_tags, indent=4))39    print("Glacier")40    glacier_tags = AwsResources.GlacierTags().get_resources().get_tags()41    display_tag_failure(tagcheck.get_noncomp_resources(glacier_tags))42    # print(json.dumps(glacier_tags, indent=4))43    print("Kinesis")44    kinesis_tags = AwsResources.KinesisTags().get_resources().get_tags()45    display_tag_failure(tagcheck.get_noncomp_resources(kinesis_tags))46    # print(json.dumps(kinesis_tags, indent=4))47    print('Redshift')48    rs_tags = AwsResources.RedshiftTags().get_resources().get_tags()49    display_tag_failure(tagcheck.get_noncomp_resources(rs_tags))50    # print(json.dumps(rs_tags, indent=4))51    print('RDS')52    rds_tags = AwsResources.RdsTags().get_resources().get_tags()53    display_tag_failure(tagcheck.get_noncomp_resources(rds_tags))54    # print(json.dumps(rds_tags, indent=4))55    print('S3')56    s3_tags = AwsResources.S3Tags().get_resources().get_tags()57    display_tag_failure(tagcheck.get_noncomp_resources(s3_tags))58    # print(json.dumps(s3_tags, indent=4))59if __name__ == '__main__':...

test.py

Source:test.py

2import json3BASE_URL='http://127.0.0.1:8000/testapp/'4ENDPOINT='GET/'5fileType="song/"6def get_resources(id=None):7    data={}8    if id is not None:9        resp = requests.get(BASE_URL+ENDPOINT+fileType+str(id))10    else:11        resp = requests.get(BASE_URL+ENDPOINT+fileType)12    print(resp.status_code)13    if resp.status_code==200:14        print(resp.json())15get_resources(2)16get_resources(-2)17get_resources("sahil")18get_resources("../song")19get_resources("?id=3")20get_resources("?id!=3")21get_resources("?www.google.com")22get_resources("/www.google.com")2324#I had tested CURD operations with postman
...

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.