How to use list_associations method in localstack

Best Python code snippet using localstack_python

test_sagemaker_source.py

Source:test_sagemaker_source.py

1import json2from botocore.stub import Stubber3from freezegun import freeze_time4from datahub.ingestion.api.common import PipelineContext5from datahub.ingestion.source.aws.sagemaker import (6    SagemakerSource,7    SagemakerSourceConfig,8)9from datahub.ingestion.source.aws.sagemaker_processors.jobs import (10    job_type_to_info,11    job_types,12)13from tests.test_helpers import mce_helpers14from tests.unit.test_sagemaker_source_stubs import (15    describe_endpoint_response_1,16    describe_endpoint_response_2,17    describe_feature_group_response_1,18    describe_feature_group_response_2,19    describe_feature_group_response_3,20    describe_group_response,21    describe_model_response_1,22    describe_model_response_2,23    get_first_model_package_incoming_response,24    get_model_group_incoming_response,25    get_second_model_package_incoming_response,26    job_stubs,27    list_actions_response,28    list_artifacts_response,29    list_contexts_response,30    list_endpoints_response,31    list_feature_groups_response,32    list_first_endpoint_incoming_response,33    list_first_endpoint_outgoing_response,34    list_groups_response,35    list_models_response,36    list_second_endpoint_incoming_response,37    list_second_endpoint_outgoing_response,38)39FROZEN_TIME = "2020-04-14 07:00:00"40def sagemaker_source() -> SagemakerSource:41    return SagemakerSource(42        ctx=PipelineContext(run_id="sagemaker-source-test"),43        config=SagemakerSourceConfig(aws_region="us-west-2"),44    )45@freeze_time(FROZEN_TIME)46def test_sagemaker_ingest(tmp_path, pytestconfig):47    sagemaker_source_instance = sagemaker_source()48    with Stubber(sagemaker_source_instance.sagemaker_client) as sagemaker_stubber:49        sagemaker_stubber.add_response(50            "list_actions",51            list_actions_response,52            {},53        )54        sagemaker_stubber.add_response(55            "list_artifacts",56            list_artifacts_response,57            {},58        )59        sagemaker_stubber.add_response(60            "list_contexts",61            list_contexts_response,62            {},63        )64        sagemaker_stubber.add_response(65            "list_associations",66            list_first_endpoint_incoming_response,67            {68                "DestinationArn": "arn:aws:sagemaker:us-west-2:123412341234:action/deploy-the-first-endpoint"69            },70        )71        sagemaker_stubber.add_response(72            "list_associations",73            list_first_endpoint_outgoing_response,74            {75                "SourceArn": "arn:aws:sagemaker:us-west-2:123412341234:action/deploy-the-first-endpoint"76            },77        )78        sagemaker_stubber.add_response(79            "list_associations",80            list_second_endpoint_incoming_response,81            {82                "DestinationArn": "arn:aws:sagemaker:us-west-2:123412341234:action/deploy-the-second-endpoint"83            },84        )85        sagemaker_stubber.add_response(86            "list_associations",87            list_second_endpoint_outgoing_response,88            {89                "SourceArn": "arn:aws:sagemaker:us-west-2:123412341234:action/deploy-the-second-endpoint"90            },91        )92        sagemaker_stubber.add_response(93            "list_associations",94            get_model_group_incoming_response,95            {96                "DestinationArn": "arn:aws:sagemaker:us-west-2:123412341234:context/a-model-package-group-context"97            },98        )99        sagemaker_stubber.add_response(100            "list_associations",101            get_first_model_package_incoming_response,102            {103                "DestinationArn": "arn:aws:sagemaker:us-west-2:123412341234:artifact/the-first-model-package-artifact"104            },105        )106        sagemaker_stubber.add_response(107            "list_associations",108            get_second_model_package_incoming_response,109            {110                "DestinationArn": "arn:aws:sagemaker:us-west-2:123412341234:artifact/the-second-model-package-artifact"111            },112        )113        sagemaker_stubber.add_response(114            "list_feature_groups",115            list_feature_groups_response,116            {},117        )118        sagemaker_stubber.add_response(119            "describe_feature_group",120            describe_feature_group_response_1,121            {122                "FeatureGroupName": "test-2",123            },124        )125        sagemaker_stubber.add_response(126            "describe_feature_group",127            describe_feature_group_response_2,128            {129                "FeatureGroupName": "test-1",130            },131        )132        sagemaker_stubber.add_response(133            "describe_feature_group",134            describe_feature_group_response_3,135            {136                "FeatureGroupName": "test",137            },138        )139        for job_type in job_types:140            job = job_stubs[job_type.value]141            job_info = job_type_to_info[job_type]142            sagemaker_stubber.add_response(143                job_info.list_command,144                job["list"],145                {},146            )147        for job_type in job_types:148            job = job_stubs[job_type.value]149            job_info = job_type_to_info[job_type]150            sagemaker_stubber.add_response(151                job_info.describe_command,152                job["describe"],153                {job_info.describe_name_key: job["describe_name"]},154            )155        sagemaker_stubber.add_response(156            "list_endpoints",157            list_endpoints_response,158            {},159        )160        sagemaker_stubber.add_response(161            "describe_endpoint",162            describe_endpoint_response_1,163            {"EndpointName": "the-first-endpoint"},164        )165        sagemaker_stubber.add_response(166            "describe_endpoint",167            describe_endpoint_response_2,168            {"EndpointName": "the-second-endpoint"},169        )170        sagemaker_stubber.add_response(171            "list_model_package_groups",172            list_groups_response,173            {},174        )175        sagemaker_stubber.add_response(176            "describe_model_package_group",177            describe_group_response,178            {"ModelPackageGroupName": "a-model-package-group"},179        )180        sagemaker_stubber.add_response(181            "list_models",182            list_models_response,183            {},184        )185        sagemaker_stubber.add_response(186            "describe_model",187            describe_model_response_1,188            {"ModelName": "the-first-model"},189        )190        sagemaker_stubber.add_response(191            "describe_model",192            describe_model_response_2,193            {"ModelName": "the-second-model"},194        )195        mce_objects = [196            wu.metadata.to_obj() for wu in sagemaker_source_instance.get_workunits()197        ]198        with open(str(tmp_path / "sagemaker_mces.json"), "w") as f:199            json.dump(mce_objects, f, indent=2)200    # Verify the output.201    test_resources_dir = pytestconfig.rootpath / "tests/unit/sagemaker"202    mce_helpers.check_golden_file(203        pytestconfig,204        output_path=tmp_path / "sagemaker_mces.json",205        golden_path=test_resources_dir / "sagemaker_mces_golden.json",...

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.