Changeset View
Changeset View
Standalone View
Standalone View
integration_tests/test_suites/k8s-integration-test-suite/conftest.py
| # pylint: disable=unused-import | # pylint: disable=unused-import | ||||
| import os | import os | ||||
| import docker | import docker | ||||
| import kubernetes | import kubernetes | ||||
| import pytest | import pytest | ||||
| from dagster_k8s.launcher import K8sRunLauncher | from dagster_k8s.launcher import K8sRunLauncher | ||||
| from dagster_k8s.scheduler import K8sScheduler | from dagster_k8s.scheduler import K8sScheduler | ||||
| from dagster_k8s_test_infra.cluster import ( | from dagster_k8s_test_infra.cluster import ( | ||||
| dagster_instance, | dagster_instance_for_k8s_run_launcher, | ||||
| dagster_instance_with_k8s_scheduler, | dagster_instance_with_k8s_scheduler, | ||||
| define_cluster_provider_fixture, | define_cluster_provider_fixture, | ||||
| ) | ) | ||||
| from dagster_k8s_test_infra.helm import helm_namespace | from dagster_k8s_test_infra.helm import helm_namespace_for_k8s_run_launcher | ||||
| from dagster_k8s_test_infra.integration_utils import image_pull_policy | from dagster_k8s_test_infra.integration_utils import image_pull_policy | ||||
| from dagster_test.test_project import build_and_tag_test_image, test_project_docker_image | from dagster_test.test_project import build_and_tag_test_image, test_project_docker_image | ||||
| from dagster import seven | from dagster import seven | ||||
| from dagster.core.instance import DagsterInstance | from dagster.core.instance import DagsterInstance | ||||
| IS_BUILDKITE = os.getenv("BUILDKITE") is not None | IS_BUILDKITE = os.getenv("BUILDKITE") is not None | ||||
| Show All 15 Lines | |||||
| @pytest.yield_fixture | @pytest.yield_fixture | ||||
| def schedule_tempdir(): | def schedule_tempdir(): | ||||
| with seven.TemporaryDirectory() as tempdir: | with seven.TemporaryDirectory() as tempdir: | ||||
| yield tempdir | yield tempdir | ||||
| @pytest.fixture | @pytest.fixture | ||||
| def k8s_scheduler( | def k8s_scheduler( | ||||
| cluster_provider, helm_namespace | cluster_provider, helm_namespace_for_k8s_run_launcher | ||||
| ): # pylint: disable=redefined-outer-name,unused-argument | ): # pylint: disable=redefined-outer-name,unused-argument | ||||
| return K8sScheduler( | return K8sScheduler( | ||||
| scheduler_namespace=helm_namespace, | scheduler_namespace=helm_namespace_for_k8s_run_launcher, | ||||
| image_pull_secrets=[{"name": "element-dev-key"}], | image_pull_secrets=[{'name': 'element-dev-key'}], | ||||
| service_account_name="dagit-admin", | service_account_name='dagit-admin', | ||||
| instance_config_map="dagster-instance", | instance_config_map='dagster-instance', | ||||
| postgres_password_secret="dagster-postgresql-secret", | postgres_password_secret='dagster-postgresql-secret', | ||||
| dagster_home="/opt/dagster/dagster_home", | dagster_home='/opt/dagster/dagster_home', | ||||
| job_image=test_project_docker_image(), | job_image=test_project_docker_image(), | ||||
| load_incluster_config=False, | load_incluster_config=False, | ||||
| kubeconfig_file=cluster_provider.kubeconfig_file, | kubeconfig_file=cluster_provider.kubeconfig_file, | ||||
| image_pull_policy=image_pull_policy(), | image_pull_policy=image_pull_policy(), | ||||
| env_config_maps=["dagster-pipeline-env", "test-env-configmap"], | env_config_maps=["dagster-pipeline-env", "test-env-configmap"], | ||||
| env_secrets=["test-env-secret"], | env_secrets=["test-env-secret"], | ||||
| ) | ) | ||||
| @pytest.fixture(scope="function") | @pytest.fixture(scope='function') | ||||
| def restore_k8s_cron_tab(helm_namespace): # pylint: disable=redefined-outer-name | def restore_k8s_cron_tab( | ||||
| helm_namespace_for_k8s_run_launcher, | |||||
| ): # pylint: disable=redefined-outer-name | |||||
| kube_api = kubernetes.client.BatchV1beta1Api() | kube_api = kubernetes.client.BatchV1beta1Api() | ||||
| # Doubly make sure CronJobs are deleted pre-test and post-test | # Doubly make sure CronJobs are deleted pre-test and post-test | ||||
| kube_api.delete_collection_namespaced_cron_job(namespace=helm_namespace) | kube_api.delete_collection_namespaced_cron_job(namespace=helm_namespace_for_k8s_run_launcher) | ||||
| yield | yield | ||||
| kube_api.delete_collection_namespaced_cron_job(namespace=helm_namespace) | kube_api.delete_collection_namespaced_cron_job(namespace=helm_namespace_for_k8s_run_launcher) | ||||
| @pytest.fixture(scope="session") | @pytest.fixture(scope="session") | ||||
| def run_launcher( | def run_launcher( | ||||
| cluster_provider, helm_namespace | cluster_provider, helm_namespace_for_k8s_run_launcher | ||||
| ): # pylint: disable=redefined-outer-name,unused-argument | ): # pylint: disable=redefined-outer-name,unused-argument | ||||
| return K8sRunLauncher( | return K8sRunLauncher( | ||||
| image_pull_secrets=[{"name": "element-dev-key"}], | image_pull_secrets=[{"name": "element-dev-key"}], | ||||
| service_account_name="dagit-admin", | service_account_name="dagit-admin", | ||||
| instance_config_map="dagster-instance", | instance_config_map="dagster-instance", | ||||
| postgres_password_secret="dagster-postgresql-secret", | postgres_password_secret="dagster-postgresql-secret", | ||||
| dagster_home="/opt/dagster/dagster_home", | dagster_home="/opt/dagster/dagster_home", | ||||
| job_image=test_project_docker_image(), | job_image=test_project_docker_image(), | ||||
| load_incluster_config=False, | load_incluster_config=False, | ||||
| kubeconfig_file=cluster_provider.kubeconfig_file, | kubeconfig_file=cluster_provider.kubeconfig_file, | ||||
| image_pull_policy=image_pull_policy(), | image_pull_policy=image_pull_policy(), | ||||
| job_namespace=helm_namespace, | job_namespace=helm_namespace_for_k8s_run_launcher, | ||||
| env_config_maps=["dagster-pipeline-env", "test-env-configmap"], | env_config_maps=['dagster-pipeline-env', 'test-env-configmap'], | ||||
| env_secrets=["test-env-secret"], | env_secrets=['test-env-secret'], | ||||
| ) | ) | ||||
| @pytest.fixture(scope="session") | @pytest.fixture(scope="session") | ||||
| def dagster_docker_image(): | def dagster_docker_image(): | ||||
| docker_image = test_project_docker_image() | docker_image = test_project_docker_image() | ||||
| if not IS_BUILDKITE: | if not IS_BUILDKITE: | ||||
| ▲ Show 20 Lines • Show All 43 Lines • Show Last 20 Lines | |||||