Differential D8639 Diff 40861 python_modules/libraries/dagster-aws/dagster_aws_tests/ecs_tests/test_launcher.py

Changeset View

Standalone View

python_modules/libraries/dagster-aws/dagster_aws_tests/ecs_tests/test_launcher.py

# pylint: disable=redefined-outer-name, protected-access		# pylint: disable=redefined-outer-name, protected-access, unused-argument
		import boto3
import pytest		import pytest
from dagster.core.definitions.reconstructable import ReconstructableRepository		from dagster.core.definitions.reconstructable import ReconstructableRepository
from dagster.core.host_representation.origin import InProcessRepositoryLocationOrigin		from dagster.core.host_representation.origin import InProcessRepositoryLocationOrigin
from dagster.core.test_utils import instance_for_test		from dagster.core.test_utils import instance_for_test

from . import repo		from . import repo


Show All 31 Lines	return ecs.run_task(
"subnets": [network_interface.subnet_id],		"subnets": [network_interface.subnet_id],
"securityGroups": [security_group.id],		"securityGroups": [security_group.id],
},		},
},		},
)["tasks"][0]		)["tasks"][0]


@pytest.fixture		@pytest.fixture
def instance(ecs, ec2, task, monkeypatch, requests_mock):		def stub_aws(ecs, ec2, monkeypatch):
		# Any call to boto3.client() will return ecs.
		# Any call to boto3.resource() will return ec2.
		# This only works because our launcher happens to use a client for ecs and
		# a resource for ec2 - if that were to change or if new aws objects were to
		# be introduced, this fixture would need to be refactored.
		monkeypatch.setattr(boto3, "client", lambda args, *kwargs: ecs)
		monkeypatch.setattr(boto3, "resource", lambda args, *kwargs: ec2)


		@pytest.fixture
		def instance(stub_aws, task, monkeypatch, requests_mock):
container_uri = "http://metadata_host"		container_uri = "http://metadata_host"
monkeypatch.setenv("ECS_CONTAINER_METADATA_URI_V4", container_uri)		monkeypatch.setenv("ECS_CONTAINER_METADATA_URI_V4", container_uri)
container = task["containers"][0]["name"]		container = task["containers"][0]["name"]
requests_mock.get(container_uri, json={"Name": container})		requests_mock.get(container_uri, json={"Name": container})

task_uri = container_uri + "/task"		task_uri = container_uri + "/task"
requests_mock.get(		requests_mock.get(
task_uri,		task_uri,
json={		json={
"Cluster": task["clusterArn"],		"Cluster": task["clusterArn"],
"TaskARN": task["taskArn"],		"TaskARN": task["taskArn"],
},		},
)		)
overrides = {"run_launcher": {"module": "dagster_aws.ecs", "class": "EcsRunLauncher"}}		overrides = {"run_launcher": {"module": "dagster_aws.ecs", "class": "EcsRunLauncher"}}
with instance_for_test(overrides) as instance:		with instance_for_test(overrides) as instance:
monkeypatch.setattr(instance.run_launcher, "ecs", ecs)
monkeypatch.setattr(instance.run_launcher, "ec2", ec2)
yield instance		yield instance


@pytest.fixture		@pytest.fixture
def pipeline():		def pipeline():
return repo.pipeline		return repo.pipeline


@pytest.fixture		@pytest.fixture
def external_pipeline(image):		def external_pipeline(image):
with InProcessRepositoryLocationOrigin(		with InProcessRepositoryLocationOrigin(
ReconstructableRepository.for_file(		ReconstructableRepository.for_file(
repo.__file__, repo.repository.__name__, container_image=image		repo.__file__, repo.repository.__name__, container_image=image
),		),
).create_location() as location:		).create_location() as location:
yield location.get_repository(repo.repository.__name__).get_full_external_pipeline(		yield location.get_repository(repo.repository.__name__).get_full_external_pipeline(
repo.pipeline.__name__		repo.pipeline.__name__
)		)


		dgibsonUnsubmitted Not Done Inline Actions as discussedin the other diff this pattern does not cleanup correctly, but we can sort that out on that diff dgibson: as discussedin the other diff this pattern does not cleanup correctly, but we can sort that out…
@pytest.fixture		@pytest.fixture
def run(instance, pipeline):		def run(instance, pipeline):
return instance.create_run_for_pipeline(pipeline)		return instance.create_run_for_pipeline(pipeline)


def test_launching(		def test_launching(
ecs, instance, run, external_pipeline, subnet, network_interface, image, environment		ecs, instance, run, external_pipeline, subnet, network_interface, image, environment
):		):
▲ Show 20 Lines • Show All 57 Lines • Show Last 20 Lines