Unlock 30% off on Manual Testing Annual Plans this Holiday Season. 
Use Code: LCMYEARENDEXCL30
Use Code: LCMYEARENDEXCL30Check icon
Copied to Clipboard!
How to use _get_clients method in tempest
Best Python code snippet using tempest_python
test_batch.py
Source:test_batch.py 
...7from moto import mock_batch, mock_iam, mock_ec2, mock_ecs, mock_logs8import functools9import nose10DEFAULT_REGION = "eu-central-1"11def _get_clients():12 return (13 boto3.client("ec2", region_name=DEFAULT_REGION),14 boto3.client("iam", region_name=DEFAULT_REGION),15 boto3.client("ecs", region_name=DEFAULT_REGION),16 boto3.client("logs", region_name=DEFAULT_REGION),17 boto3.client("batch", region_name=DEFAULT_REGION),18 )19def _setup(ec2_client, iam_client):20 """21 Do prerequisite setup22 :return: VPC ID, Subnet ID, Security group ID, IAM Role ARN23 :rtype: tuple24 """25 resp = ec2_client.create_vpc(CidrBlock="172.30.0.0/24")26 vpc_id = resp["Vpc"]["VpcId"]27 resp = ec2_client.create_subnet(28 AvailabilityZone="eu-central-1a", CidrBlock="172.30.0.0/25", VpcId=vpc_id29 )30 subnet_id = resp["Subnet"]["SubnetId"]31 resp = ec2_client.create_security_group(32 Description="test_sg_desc", GroupName="test_sg", VpcId=vpc_id33 )34 sg_id = resp["GroupId"]35 resp = iam_client.create_role(36 RoleName="TestRole", AssumeRolePolicyDocument="some_policy"37 )38 iam_arn = resp["Role"]["Arn"]39 iam_client.create_instance_profile(InstanceProfileName="TestRole")40 iam_client.add_role_to_instance_profile(41 InstanceProfileName="TestRole", RoleName="TestRole"42 )43 return vpc_id, subnet_id, sg_id, iam_arn44# Yes, yes it talks to all the things45@mock_ec246@mock_ecs47@mock_iam48@mock_batch49def test_create_managed_compute_environment():50 ec2_client, iam_client, ecs_client, logs_client, batch_client = _get_clients()51 vpc_id, subnet_id, sg_id, iam_arn = _setup(ec2_client, iam_client)52 compute_name = "test_compute_env"53 resp = batch_client.create_compute_environment(54 computeEnvironmentName=compute_name,55 type="MANAGED",56 state="ENABLED",57 computeResources={58 "type": "EC2",59 "minvCpus": 5,60 "maxvCpus": 10,61 "desiredvCpus": 5,62 "instanceTypes": ["t2.small", "t2.medium"],63 "imageId": "some_image_id",64 "subnets": [subnet_id],65 "securityGroupIds": [sg_id],66 "ec2KeyPair": "string",67 "instanceRole": iam_arn.replace("role", "instance-profile"),68 "tags": {"string": "string"},69 "bidPercentage": 123,70 "spotIamFleetRole": "string",71 },72 serviceRole=iam_arn,73 )74 resp.should.contain("computeEnvironmentArn")75 resp["computeEnvironmentName"].should.equal(compute_name)76 # Given a t2.medium is 2 vcpu and t2.small is 1, therefore 2 mediums and 1 small should be created77 resp = ec2_client.describe_instances()78 resp.should.contain("Reservations")79 len(resp["Reservations"]).should.equal(3)80 # Should have created 1 ECS cluster81 resp = ecs_client.list_clusters()82 resp.should.contain("clusterArns")83 len(resp["clusterArns"]).should.equal(1)84@mock_ec285@mock_ecs86@mock_iam87@mock_batch88def test_create_unmanaged_compute_environment():89 ec2_client, iam_client, ecs_client, logs_client, batch_client = _get_clients()90 vpc_id, subnet_id, sg_id, iam_arn = _setup(ec2_client, iam_client)91 compute_name = "test_compute_env"92 resp = batch_client.create_compute_environment(93 computeEnvironmentName=compute_name,94 type="UNMANAGED",95 state="ENABLED",96 serviceRole=iam_arn,97 )98 resp.should.contain("computeEnvironmentArn")99 resp["computeEnvironmentName"].should.equal(compute_name)100 # Its unmanaged so no instances should be created101 resp = ec2_client.describe_instances()102 resp.should.contain("Reservations")103 len(resp["Reservations"]).should.equal(0)104 # Should have created 1 ECS cluster105 resp = ecs_client.list_clusters()106 resp.should.contain("clusterArns")107 len(resp["clusterArns"]).should.equal(1)108# TODO create 1000s of tests to test complex option combinations of create environment109@mock_ec2110@mock_ecs111@mock_iam112@mock_batch113def test_describe_compute_environment():114 ec2_client, iam_client, ecs_client, logs_client, batch_client = _get_clients()115 vpc_id, subnet_id, sg_id, iam_arn = _setup(ec2_client, iam_client)116 compute_name = "test_compute_env"117 batch_client.create_compute_environment(118 computeEnvironmentName=compute_name,119 type="UNMANAGED",120 state="ENABLED",121 serviceRole=iam_arn,122 )123 resp = batch_client.describe_compute_environments()124 len(resp["computeEnvironments"]).should.equal(1)125 resp["computeEnvironments"][0]["computeEnvironmentName"].should.equal(compute_name)126 # Test filtering127 resp = batch_client.describe_compute_environments(computeEnvironments=["test1"])128 len(resp["computeEnvironments"]).should.equal(0)129@mock_ec2130@mock_ecs131@mock_iam132@mock_batch133def test_delete_unmanaged_compute_environment():134 ec2_client, iam_client, ecs_client, logs_client, batch_client = _get_clients()135 vpc_id, subnet_id, sg_id, iam_arn = _setup(ec2_client, iam_client)136 compute_name = "test_compute_env"137 batch_client.create_compute_environment(138 computeEnvironmentName=compute_name,139 type="UNMANAGED",140 state="ENABLED",141 serviceRole=iam_arn,142 )143 batch_client.delete_compute_environment(computeEnvironment=compute_name)144 resp = batch_client.describe_compute_environments()145 len(resp["computeEnvironments"]).should.equal(0)146 resp = ecs_client.list_clusters()147 len(resp.get("clusterArns", [])).should.equal(0)148@mock_ec2149@mock_ecs150@mock_iam151@mock_batch152def test_delete_managed_compute_environment():153 ec2_client, iam_client, ecs_client, logs_client, batch_client = _get_clients()154 vpc_id, subnet_id, sg_id, iam_arn = _setup(ec2_client, iam_client)155 compute_name = "test_compute_env"156 batch_client.create_compute_environment(157 computeEnvironmentName=compute_name,158 type="MANAGED",159 state="ENABLED",160 computeResources={161 "type": "EC2",162 "minvCpus": 5,163 "maxvCpus": 10,164 "desiredvCpus": 5,165 "instanceTypes": ["t2.small", "t2.medium"],166 "imageId": "some_image_id",167 "subnets": [subnet_id],168 "securityGroupIds": [sg_id],169 "ec2KeyPair": "string",170 "instanceRole": iam_arn.replace("role", "instance-profile"),171 "tags": {"string": "string"},172 "bidPercentage": 123,173 "spotIamFleetRole": "string",174 },175 serviceRole=iam_arn,176 )177 batch_client.delete_compute_environment(computeEnvironment=compute_name)178 resp = batch_client.describe_compute_environments()179 len(resp["computeEnvironments"]).should.equal(0)180 resp = ec2_client.describe_instances()181 resp.should.contain("Reservations")182 len(resp["Reservations"]).should.equal(3)183 for reservation in resp["Reservations"]:184 reservation["Instances"][0]["State"]["Name"].should.equal("terminated")185 resp = ecs_client.list_clusters()186 len(resp.get("clusterArns", [])).should.equal(0)187@mock_ec2188@mock_ecs189@mock_iam190@mock_batch191def test_update_unmanaged_compute_environment_state():192 ec2_client, iam_client, ecs_client, logs_client, batch_client = _get_clients()193 vpc_id, subnet_id, sg_id, iam_arn = _setup(ec2_client, iam_client)194 compute_name = "test_compute_env"195 batch_client.create_compute_environment(196 computeEnvironmentName=compute_name,197 type="UNMANAGED",198 state="ENABLED",199 serviceRole=iam_arn,200 )201 batch_client.update_compute_environment(202 computeEnvironment=compute_name, state="DISABLED"203 )204 resp = batch_client.describe_compute_environments()205 len(resp["computeEnvironments"]).should.equal(1)206 resp["computeEnvironments"][0]["state"].should.equal("DISABLED")207@mock_ec2208@mock_ecs209@mock_iam210@mock_batch211def test_create_job_queue():212 ec2_client, iam_client, ecs_client, logs_client, batch_client = _get_clients()213 vpc_id, subnet_id, sg_id, iam_arn = _setup(ec2_client, iam_client)214 compute_name = "test_compute_env"215 resp = batch_client.create_compute_environment(216 computeEnvironmentName=compute_name,217 type="UNMANAGED",218 state="ENABLED",219 serviceRole=iam_arn,220 )221 arn = resp["computeEnvironmentArn"]222 resp = batch_client.create_job_queue(223 jobQueueName="test_job_queue",224 state="ENABLED",225 priority=123,226 computeEnvironmentOrder=[{"order": 123, "computeEnvironment": arn}],227 )228 resp.should.contain("jobQueueArn")229 resp.should.contain("jobQueueName")230 queue_arn = resp["jobQueueArn"]231 resp = batch_client.describe_job_queues()232 resp.should.contain("jobQueues")233 len(resp["jobQueues"]).should.equal(1)234 resp["jobQueues"][0]["jobQueueArn"].should.equal(queue_arn)235 resp = batch_client.describe_job_queues(jobQueues=["test_invalid_queue"])236 resp.should.contain("jobQueues")237 len(resp["jobQueues"]).should.equal(0)238 # Create job queue which already exists239 try:240 resp = batch_client.create_job_queue(241 jobQueueName="test_job_queue",242 state="ENABLED",243 priority=123,244 computeEnvironmentOrder=[{"order": 123, "computeEnvironment": arn}],245 )246 except ClientError as err:247 err.response["Error"]["Code"].should.equal("ClientException")248 # Create job queue with incorrect state249 try:250 resp = batch_client.create_job_queue(251 jobQueueName="test_job_queue2",252 state="JUNK",253 priority=123,254 computeEnvironmentOrder=[{"order": 123, "computeEnvironment": arn}],255 )256 except ClientError as err:257 err.response["Error"]["Code"].should.equal("ClientException")258 # Create job queue with no compute env259 try:260 resp = batch_client.create_job_queue(261 jobQueueName="test_job_queue3",262 state="JUNK",263 priority=123,264 computeEnvironmentOrder=[],265 )266 except ClientError as err:267 err.response["Error"]["Code"].should.equal("ClientException")268@mock_ec2269@mock_ecs270@mock_iam271@mock_batch272def test_job_queue_bad_arn():273 ec2_client, iam_client, ecs_client, logs_client, batch_client = _get_clients()274 vpc_id, subnet_id, sg_id, iam_arn = _setup(ec2_client, iam_client)275 compute_name = "test_compute_env"276 resp = batch_client.create_compute_environment(277 computeEnvironmentName=compute_name,278 type="UNMANAGED",279 state="ENABLED",280 serviceRole=iam_arn,281 )282 arn = resp["computeEnvironmentArn"]283 try:284 batch_client.create_job_queue(285 jobQueueName="test_job_queue",286 state="ENABLED",287 priority=123,288 computeEnvironmentOrder=[289 {"order": 123, "computeEnvironment": arn + "LALALA"}290 ],291 )292 except ClientError as err:293 err.response["Error"]["Code"].should.equal("ClientException")294@mock_ec2295@mock_ecs296@mock_iam297@mock_batch298def test_update_job_queue():299 ec2_client, iam_client, ecs_client, logs_client, batch_client = _get_clients()300 vpc_id, subnet_id, sg_id, iam_arn = _setup(ec2_client, iam_client)301 compute_name = "test_compute_env"302 resp = batch_client.create_compute_environment(303 computeEnvironmentName=compute_name,304 type="UNMANAGED",305 state="ENABLED",306 serviceRole=iam_arn,307 )308 arn = resp["computeEnvironmentArn"]309 resp = batch_client.create_job_queue(310 jobQueueName="test_job_queue",311 state="ENABLED",312 priority=123,313 computeEnvironmentOrder=[{"order": 123, "computeEnvironment": arn}],314 )315 queue_arn = resp["jobQueueArn"]316 batch_client.update_job_queue(jobQueue=queue_arn, priority=5)317 resp = batch_client.describe_job_queues()318 resp.should.contain("jobQueues")319 len(resp["jobQueues"]).should.equal(1)320 resp["jobQueues"][0]["priority"].should.equal(5)321 batch_client.update_job_queue(jobQueue="test_job_queue", priority=5)322 resp = batch_client.describe_job_queues()323 resp.should.contain("jobQueues")324 len(resp["jobQueues"]).should.equal(1)325 resp["jobQueues"][0]["priority"].should.equal(5)326@mock_ec2327@mock_ecs328@mock_iam329@mock_batch330def test_update_job_queue():331 ec2_client, iam_client, ecs_client, logs_client, batch_client = _get_clients()332 vpc_id, subnet_id, sg_id, iam_arn = _setup(ec2_client, iam_client)333 compute_name = "test_compute_env"334 resp = batch_client.create_compute_environment(335 computeEnvironmentName=compute_name,336 type="UNMANAGED",337 state="ENABLED",338 serviceRole=iam_arn,339 )340 arn = resp["computeEnvironmentArn"]341 resp = batch_client.create_job_queue(342 jobQueueName="test_job_queue",343 state="ENABLED",344 priority=123,345 computeEnvironmentOrder=[{"order": 123, "computeEnvironment": arn}],346 )347 queue_arn = resp["jobQueueArn"]348 batch_client.delete_job_queue(jobQueue=queue_arn)349 resp = batch_client.describe_job_queues()350 resp.should.contain("jobQueues")351 len(resp["jobQueues"]).should.equal(0)352@mock_ec2353@mock_ecs354@mock_iam355@mock_batch356def test_register_task_definition():357 ec2_client, iam_client, ecs_client, logs_client, batch_client = _get_clients()358 vpc_id, subnet_id, sg_id, iam_arn = _setup(ec2_client, iam_client)359 resp = batch_client.register_job_definition(360 jobDefinitionName="sleep10",361 type="container",362 containerProperties={363 "image": "busybox",364 "vcpus": 1,365 "memory": 128,366 "command": ["sleep", "10"],367 },368 )369 resp.should.contain("jobDefinitionArn")370 resp.should.contain("jobDefinitionName")371 resp.should.contain("revision")372 assert resp["jobDefinitionArn"].endswith(373 "{0}:{1}".format(resp["jobDefinitionName"], resp["revision"])374 )375@mock_ec2376@mock_ecs377@mock_iam378@mock_batch379def test_reregister_task_definition():380 # Reregistering task with the same name bumps the revision number381 ec2_client, iam_client, ecs_client, logs_client, batch_client = _get_clients()382 vpc_id, subnet_id, sg_id, iam_arn = _setup(ec2_client, iam_client)383 resp1 = batch_client.register_job_definition(384 jobDefinitionName="sleep10",385 type="container",386 containerProperties={387 "image": "busybox",388 "vcpus": 1,389 "memory": 128,390 "command": ["sleep", "10"],391 },392 )393 resp1.should.contain("jobDefinitionArn")394 resp1.should.contain("jobDefinitionName")395 resp1.should.contain("revision")396 assert resp1["jobDefinitionArn"].endswith(397 "{0}:{1}".format(resp1["jobDefinitionName"], resp1["revision"])398 )399 resp1["revision"].should.equal(1)400 resp2 = batch_client.register_job_definition(401 jobDefinitionName="sleep10",402 type="container",403 containerProperties={404 "image": "busybox",405 "vcpus": 1,406 "memory": 68,407 "command": ["sleep", "10"],408 },409 )410 resp2["revision"].should.equal(2)411 resp2["jobDefinitionArn"].should_not.equal(resp1["jobDefinitionArn"])412 resp3 = batch_client.register_job_definition(413 jobDefinitionName="sleep10",414 type="container",415 containerProperties={416 "image": "busybox",417 "vcpus": 1,418 "memory": 42,419 "command": ["sleep", "10"],420 },421 )422 resp3["revision"].should.equal(3)423 resp3["jobDefinitionArn"].should_not.equal(resp1["jobDefinitionArn"])424 resp3["jobDefinitionArn"].should_not.equal(resp2["jobDefinitionArn"])425 resp4 = batch_client.register_job_definition(426 jobDefinitionName="sleep10",427 type="container",428 containerProperties={429 "image": "busybox",430 "vcpus": 1,431 "memory": 41,432 "command": ["sleep", "10"],433 },434 )435 resp4["revision"].should.equal(4)436 resp4["jobDefinitionArn"].should_not.equal(resp1["jobDefinitionArn"])437 resp4["jobDefinitionArn"].should_not.equal(resp2["jobDefinitionArn"])438 resp4["jobDefinitionArn"].should_not.equal(resp3["jobDefinitionArn"])439@mock_ec2440@mock_ecs441@mock_iam442@mock_batch443def test_delete_task_definition():444 ec2_client, iam_client, ecs_client, logs_client, batch_client = _get_clients()445 vpc_id, subnet_id, sg_id, iam_arn = _setup(ec2_client, iam_client)446 resp = batch_client.register_job_definition(447 jobDefinitionName="sleep10",448 type="container",449 containerProperties={450 "image": "busybox",451 "vcpus": 1,452 "memory": 128,453 "command": ["sleep", "10"],454 },455 )456 batch_client.deregister_job_definition(jobDefinition=resp["jobDefinitionArn"])457 resp = batch_client.describe_job_definitions()458 len(resp["jobDefinitions"]).should.equal(0)459@mock_ec2460@mock_ecs461@mock_iam462@mock_batch463def test_describe_task_definition():464 ec2_client, iam_client, ecs_client, logs_client, batch_client = _get_clients()465 vpc_id, subnet_id, sg_id, iam_arn = _setup(ec2_client, iam_client)466 batch_client.register_job_definition(467 jobDefinitionName="sleep10",468 type="container",469 containerProperties={470 "image": "busybox",471 "vcpus": 1,472 "memory": 128,473 "command": ["sleep", "10"],474 },475 )476 batch_client.register_job_definition(477 jobDefinitionName="sleep10",478 type="container",479 containerProperties={480 "image": "busybox",481 "vcpus": 1,482 "memory": 64,483 "command": ["sleep", "10"],484 },485 )486 batch_client.register_job_definition(487 jobDefinitionName="test1",488 type="container",489 containerProperties={490 "image": "busybox",491 "vcpus": 1,492 "memory": 64,493 "command": ["sleep", "10"],494 },495 )496 resp = batch_client.describe_job_definitions(jobDefinitionName="sleep10")497 len(resp["jobDefinitions"]).should.equal(2)498 resp = batch_client.describe_job_definitions()499 len(resp["jobDefinitions"]).should.equal(3)500 resp = batch_client.describe_job_definitions(jobDefinitions=["sleep10", "test1"])501 len(resp["jobDefinitions"]).should.equal(3)502 for job_definition in resp["jobDefinitions"]:503 job_definition["status"].should.equal("ACTIVE")504@mock_logs505@mock_ec2506@mock_ecs507@mock_iam508@mock_batch509def test_submit_job_by_name():510 ec2_client, iam_client, ecs_client, logs_client, batch_client = _get_clients()511 vpc_id, subnet_id, sg_id, iam_arn = _setup(ec2_client, iam_client)512 compute_name = "test_compute_env"513 resp = batch_client.create_compute_environment(514 computeEnvironmentName=compute_name,515 type="UNMANAGED",516 state="ENABLED",517 serviceRole=iam_arn,518 )519 arn = resp["computeEnvironmentArn"]520 resp = batch_client.create_job_queue(521 jobQueueName="test_job_queue",522 state="ENABLED",523 priority=123,524 computeEnvironmentOrder=[{"order": 123, "computeEnvironment": arn}],525 )526 queue_arn = resp["jobQueueArn"]527 job_definition_name = "sleep10"528 batch_client.register_job_definition(529 jobDefinitionName=job_definition_name,530 type="container",531 containerProperties={532 "image": "busybox",533 "vcpus": 1,534 "memory": 128,535 "command": ["sleep", "10"],536 },537 )538 batch_client.register_job_definition(539 jobDefinitionName=job_definition_name,540 type="container",541 containerProperties={542 "image": "busybox",543 "vcpus": 1,544 "memory": 256,545 "command": ["sleep", "10"],546 },547 )548 resp = batch_client.register_job_definition(549 jobDefinitionName=job_definition_name,550 type="container",551 containerProperties={552 "image": "busybox",553 "vcpus": 1,554 "memory": 512,555 "command": ["sleep", "10"],556 },557 )558 job_definition_arn = resp["jobDefinitionArn"]559 resp = batch_client.submit_job(560 jobName="test1", jobQueue=queue_arn, jobDefinition=job_definition_name561 )562 job_id = resp["jobId"]563 resp_jobs = batch_client.describe_jobs(jobs=[job_id])564 # batch_client.terminate_job(jobId=job_id)565 len(resp_jobs["jobs"]).should.equal(1)566 resp_jobs["jobs"][0]["jobId"].should.equal(job_id)567 resp_jobs["jobs"][0]["jobQueue"].should.equal(queue_arn)568 resp_jobs["jobs"][0]["jobDefinition"].should.equal(job_definition_arn)569# SLOW TESTS570@mock_logs571@mock_ec2572@mock_ecs573@mock_iam574@mock_batch575def test_submit_job():576 ec2_client, iam_client, ecs_client, logs_client, batch_client = _get_clients()577 vpc_id, subnet_id, sg_id, iam_arn = _setup(ec2_client, iam_client)578 compute_name = "test_compute_env"579 resp = batch_client.create_compute_environment(580 computeEnvironmentName=compute_name,581 type="UNMANAGED",582 state="ENABLED",583 serviceRole=iam_arn,584 )585 arn = resp["computeEnvironmentArn"]586 resp = batch_client.create_job_queue(587 jobQueueName="test_job_queue",588 state="ENABLED",589 priority=123,590 computeEnvironmentOrder=[{"order": 123, "computeEnvironment": arn}],591 )592 queue_arn = resp["jobQueueArn"]593 resp = batch_client.register_job_definition(594 jobDefinitionName="sayhellotomylittlefriend",595 type="container",596 containerProperties={597 "image": "busybox:latest",598 "vcpus": 1,599 "memory": 128,600 "command": ["echo", "hello"],601 },602 )603 job_def_arn = resp["jobDefinitionArn"]604 resp = batch_client.submit_job(605 jobName="test1", jobQueue=queue_arn, jobDefinition=job_def_arn606 )607 job_id = resp["jobId"]608 future = datetime.datetime.now() + datetime.timedelta(seconds=30)609 while datetime.datetime.now() < future:610 time.sleep(1)611 resp = batch_client.describe_jobs(jobs=[job_id])612 if resp["jobs"][0]["status"] == "FAILED":613 raise RuntimeError("Batch job failed")614 if resp["jobs"][0]["status"] == "SUCCEEDED":615 break616 else:617 raise RuntimeError("Batch job timed out")618 resp = logs_client.describe_log_streams(logGroupName="/aws/batch/job")619 len(resp["logStreams"]).should.equal(1)620 ls_name = resp["logStreams"][0]["logStreamName"]621 resp = logs_client.get_log_events(622 logGroupName="/aws/batch/job", logStreamName=ls_name623 )624 [event["message"] for event in resp["events"]].should.equal(["hello"])625@mock_logs626@mock_ec2627@mock_ecs628@mock_iam629@mock_batch630def test_list_jobs():631 ec2_client, iam_client, ecs_client, logs_client, batch_client = _get_clients()632 vpc_id, subnet_id, sg_id, iam_arn = _setup(ec2_client, iam_client)633 compute_name = "test_compute_env"634 resp = batch_client.create_compute_environment(635 computeEnvironmentName=compute_name,636 type="UNMANAGED",637 state="ENABLED",638 serviceRole=iam_arn,639 )640 arn = resp["computeEnvironmentArn"]641 resp = batch_client.create_job_queue(642 jobQueueName="test_job_queue",643 state="ENABLED",644 priority=123,645 computeEnvironmentOrder=[{"order": 123, "computeEnvironment": arn}],646 )647 queue_arn = resp["jobQueueArn"]648 resp = batch_client.register_job_definition(649 jobDefinitionName="sleep5",650 type="container",651 containerProperties={652 "image": "busybox:latest",653 "vcpus": 1,654 "memory": 128,655 "command": ["sleep", "5"],656 },657 )658 job_def_arn = resp["jobDefinitionArn"]659 resp = batch_client.submit_job(660 jobName="test1", jobQueue=queue_arn, jobDefinition=job_def_arn661 )662 job_id1 = resp["jobId"]663 resp = batch_client.submit_job(664 jobName="test2", jobQueue=queue_arn, jobDefinition=job_def_arn665 )666 job_id2 = resp["jobId"]667 future = datetime.datetime.now() + datetime.timedelta(seconds=30)668 resp_finished_jobs = batch_client.list_jobs(669 jobQueue=queue_arn, jobStatus="SUCCEEDED"670 )671 # Wait only as long as it takes to run the jobs672 while datetime.datetime.now() < future:673 resp = batch_client.describe_jobs(jobs=[job_id1, job_id2])674 any_failed_jobs = any([job["status"] == "FAILED" for job in resp["jobs"]])675 succeeded_jobs = all([job["status"] == "SUCCEEDED" for job in resp["jobs"]])676 if any_failed_jobs:677 raise RuntimeError("A Batch job failed")678 if succeeded_jobs:679 break680 time.sleep(0.5)681 else:682 raise RuntimeError("Batch jobs timed out")683 resp_finished_jobs2 = batch_client.list_jobs(684 jobQueue=queue_arn, jobStatus="SUCCEEDED"685 )686 len(resp_finished_jobs["jobSummaryList"]).should.equal(0)687 len(resp_finished_jobs2["jobSummaryList"]).should.equal(2)688@mock_logs689@mock_ec2690@mock_ecs691@mock_iam692@mock_batch693def test_terminate_job():694 ec2_client, iam_client, ecs_client, logs_client, batch_client = _get_clients()695 vpc_id, subnet_id, sg_id, iam_arn = _setup(ec2_client, iam_client)696 compute_name = "test_compute_env"697 resp = batch_client.create_compute_environment(698 computeEnvironmentName=compute_name,699 type="UNMANAGED",700 state="ENABLED",701 serviceRole=iam_arn,702 )703 arn = resp["computeEnvironmentArn"]704 resp = batch_client.create_job_queue(705 jobQueueName="test_job_queue",706 state="ENABLED",707 priority=123,708 computeEnvironmentOrder=[{"order": 123, "computeEnvironment": arn}],...
tables.py
Source:tables.py
...48 def _ensure_table_client(self) -> None:49 if not self._table_client:50 raise TableError("Table client not initialized. Use as a context manager.")51 def __enter__(self) -> TableClient:52 self._service_client, self._table_client = self._get_clients()53 return self._table_client54 def __exit__(self, *args: Any) -> None:55 if self._table_client:56 self._table_client.close()57 self._table_client = None58 if self._service_client:59 self._service_client.close()60 self._service_client = None61 @classmethod62 def from_sas_token(63 cls: Type[T], account_url: str, sas_token: str, table_name: str64 ) -> T:65 def _get_clients(66 _url: str = account_url, _token: str = sas_token, _table: str = table_name67 ) -> Tuple[Optional[TableServiceClient], TableClient]:68 table_service_client = TableServiceClient(69 endpoint=_url,70 credential=AzureSasCredential(_token),71 )72 return (73 table_service_client,74 table_service_client.get_table_client(table_name=_table),75 )76 return cls(_get_clients)77 @classmethod78 def from_connection_string(79 cls: Type[T], connection_string: str, table_name: str80 ) -> T:81 def _get_clients(82 _conn_str: str = connection_string, _table: str = table_name83 ) -> Tuple[Optional[TableServiceClient], TableClient]:84 table_service_client = TableServiceClient.from_connection_string(85 conn_str=_conn_str86 )87 return (88 table_service_client,89 table_service_client.get_table_client(table_name=_table),90 )91 return cls(_get_clients)92 @classmethod93 def from_account_key(94 cls: Type[T],95 account_name: str,96 account_key: str,97 table_name: str,98 account_url: Optional[str] = None,99 ttl: Optional[int] = None,100 ) -> T:101 def _get_clients(102 _name: str = account_name,103 _key: str = account_key,104 _url: Optional[str] = account_url,105 _table: str = table_name,106 ) -> Tuple[Optional[TableServiceClient], TableClient]:107 _url = _url or f"https://{_name}.table.core.windows.net"108 credential = AzureNamedKeyCredential(name=_name, key=_key)109 table_service_client = TableServiceClient(110 endpoint=_url, credential=credential111 )112 return (113 table_service_client,114 table_service_client.get_table_client(table_name=_table),115 )...
queues.py
Source:queues.py
...19 def _ensure_queue_client(self) -> None:20 if not self._queue_client:21 raise QueueError("Table client not initialized. Use as a context manager.")22 def __enter__(self) -> QueueClient:23 self._service_client, self._queue_client = self._get_clients()24 return self._queue_client25 def __exit__(self, *args: Any) -> None:26 if self._queue_client:27 self._queue_client.close()28 self._queue_client = None29 if self._service_client:30 self._service_client.close()31 self._service_client = None32 @classmethod33 def from_sas_token(34 cls,35 account_url: str,36 sas_token: str,37 queue_name: str,38 ) -> "QueueService":39 def _get_clients(40 _url: str = account_url, _token: str = sas_token, _queue: str = queue_name41 ) -> Tuple[Optional[QueueServiceClient], QueueClient]:42 service_client = QueueServiceClient(43 account_url=_url,44 credential=_token,45 )46 return (47 service_client,48 service_client.get_queue_client(49 queue=_queue,50 message_encode_policy=BinaryBase64EncodePolicy(),51 message_decode_policy=BinaryBase64DecodePolicy(),52 ),53 )54 return cls(_get_clients)55 @classmethod56 def from_connection_string(57 cls, connection_string: str, queue_name: str58 ) -> "QueueService":59 def _get_clients(60 _conn_str: str = connection_string, _queue: str = queue_name61 ) -> Tuple[Optional[QueueServiceClient], QueueClient]:62 service_client = QueueServiceClient.from_connection_string(63 conn_str=_conn_str64 )65 return (66 service_client,67 service_client.get_queue_client(68 queue=_queue,69 message_encode_policy=BinaryBase64EncodePolicy(),70 message_decode_policy=BinaryBase64DecodePolicy(),71 ),72 )73 return cls(_get_clients)74 @classmethod75 def from_account_key(76 cls,77 account_url: str,78 account_key: str,79 queue_name: str,80 ) -> "QueueService":81 def _get_clients(82 _key: str = account_key,83 _url: str = account_url,84 _queue: str = queue_name,85 ) -> Tuple[Optional[QueueServiceClient], QueueClient]:86 service_client = QueueServiceClient(account_url=_url, credential=_key)87 return (88 service_client,89 service_client.get_queue_client(90 queue=_queue,91 message_encode_policy=BinaryBase64EncodePolicy(),92 message_decode_policy=BinaryBase64DecodePolicy(),93 ),94 )95 return cls(_get_clients)...
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run tempest automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
