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",...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!