| # Licensed to the Apache Software Foundation (ASF) under one |
| # or more contributor license agreements. See the NOTICE file |
| # distributed with this work for additional information |
| # regarding copyright ownership. The ASF licenses this file |
| # to you under the Apache License, Version 2.0 (the |
| # "License"); you may not use this file except in compliance |
| # with the License. You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, |
| # software distributed under the License is distributed on an |
| # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| # KIND, either express or implied. See the License for the |
| # specific language governing permissions and limitations |
| # under the License. |
| from __future__ import annotations |
| |
| import os |
| import unittest.mock |
| from datetime import datetime |
| |
| import pytest |
| |
| from airflow import DAG |
| from airflow.api_connexion.exceptions import EXCEPTIONS_LINK_MAP |
| from airflow.models import DagBag, DagModel |
| from airflow.models.serialized_dag import SerializedDagModel |
| from airflow.operators.empty import EmptyOperator |
| from airflow.security import permissions |
| from airflow.utils.session import provide_session |
| from tests.test_utils.api_connexion_utils import assert_401, create_user, delete_user |
| from tests.test_utils.config import conf_vars |
| from tests.test_utils.db import clear_db_dags, clear_db_runs, clear_db_serialized_dags |
| |
| |
| @pytest.fixture() |
| def current_file_token(url_safe_serializer) -> str: |
| return url_safe_serializer.dumps(__file__) |
| |
| |
| DAG_ID = "test_dag" |
| TASK_ID = "op1" |
| DAG2_ID = "test_dag2" |
| DAG3_ID = "test_dag3" |
| |
| |
| @pytest.fixture(scope="module") |
| def configured_app(minimal_app_for_api): |
| app = minimal_app_for_api |
| |
| create_user( |
| app, # type: ignore |
| username="test", |
| role_name="Test", |
| permissions=[ |
| (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG), |
| (permissions.ACTION_CAN_EDIT, permissions.RESOURCE_DAG), |
| (permissions.ACTION_CAN_DELETE, permissions.RESOURCE_DAG), |
| ], |
| ) |
| create_user(app, username="test_no_permissions", role_name="TestNoPermissions") # type: ignore |
| create_user(app, username="test_granular_permissions", role_name="TestGranularDag") # type: ignore |
| app.appbuilder.sm.sync_perm_for_dag( # type: ignore |
| "TEST_DAG_1", |
| access_control={"TestGranularDag": [permissions.ACTION_CAN_EDIT, permissions.ACTION_CAN_READ]}, |
| ) |
| app.appbuilder.sm.sync_perm_for_dag( # type: ignore |
| "TEST_DAG_1", |
| access_control={"TestGranularDag": [permissions.ACTION_CAN_EDIT, permissions.ACTION_CAN_READ]}, |
| ) |
| |
| with DAG( |
| DAG_ID, |
| start_date=datetime(2020, 6, 15), |
| doc_md="details", |
| params={"foo": 1}, |
| tags=["example"], |
| ) as dag: |
| EmptyOperator(task_id=TASK_ID) |
| |
| with DAG(DAG2_ID, start_date=datetime(2020, 6, 15)) as dag2: # no doc_md |
| EmptyOperator(task_id=TASK_ID) |
| |
| with DAG(DAG3_ID) as dag3: # DAG start_date set to None |
| EmptyOperator(task_id=TASK_ID, start_date=datetime(2019, 6, 12)) |
| |
| dag_bag = DagBag(os.devnull, include_examples=False) |
| dag_bag.dags = {dag.dag_id: dag, dag2.dag_id: dag2, dag3.dag_id: dag3} |
| |
| app.dag_bag = dag_bag |
| |
| yield app |
| |
| delete_user(app, username="test") # type: ignore |
| delete_user(app, username="test_no_permissions") # type: ignore |
| delete_user(app, username="test_granular_permissions") # type: ignore |
| |
| |
| class TestDagEndpoint: |
| @staticmethod |
| def clean_db(): |
| clear_db_runs() |
| clear_db_dags() |
| clear_db_serialized_dags() |
| |
| @pytest.fixture(autouse=True) |
| def setup_attrs(self, configured_app) -> None: |
| self.clean_db() |
| self.app = configured_app |
| self.client = self.app.test_client() # type:ignore |
| self.dag_id = DAG_ID |
| self.dag2_id = DAG2_ID |
| self.dag3_id = DAG3_ID |
| |
| def teardown_method(self) -> None: |
| self.clean_db() |
| |
| @provide_session |
| def _create_dag_models(self, count, dag_id_prefix="TEST_DAG", is_paused=False, session=None): |
| for num in range(1, count + 1): |
| dag_model = DagModel( |
| dag_id=f"{dag_id_prefix}_{num}", |
| fileloc=f"/tmp/dag_{num}.py", |
| schedule_interval="2 2 * * *", |
| is_active=True, |
| is_paused=is_paused, |
| ) |
| session.add(dag_model) |
| |
| @provide_session |
| def _create_deactivated_dag(self, session=None): |
| dag_model = DagModel( |
| dag_id="TEST_DAG_DELETED_1", |
| fileloc="/tmp/dag_del_1.py", |
| schedule_interval="2 2 * * *", |
| is_active=False, |
| ) |
| session.add(dag_model) |
| |
| |
| class TestGetDag(TestDagEndpoint): |
| @conf_vars({("webserver", "secret_key"): "mysecret"}) |
| def test_should_respond_200(self): |
| self._create_dag_models(1) |
| response = self.client.get("/api/v1/dags/TEST_DAG_1", environ_overrides={"REMOTE_USER": "test"}) |
| assert response.status_code == 200 |
| assert { |
| "dag_id": "TEST_DAG_1", |
| "description": None, |
| "fileloc": "/tmp/dag_1.py", |
| "file_token": "Ii90bXAvZGFnXzEucHki.EnmIdPaUPo26lHQClbWMbDFD1Pk", |
| "is_paused": False, |
| "is_active": True, |
| "is_subdag": False, |
| "owners": [], |
| "root_dag_id": None, |
| "schedule_interval": {"__type": "CronExpression", "value": "2 2 * * *"}, |
| "tags": [], |
| "next_dagrun": None, |
| "has_task_concurrency_limits": True, |
| "next_dagrun_data_interval_start": None, |
| "next_dagrun_data_interval_end": None, |
| "max_active_runs": 16, |
| "next_dagrun_create_after": None, |
| "last_expired": None, |
| "max_active_tasks": 16, |
| "last_pickled": None, |
| "default_view": None, |
| "last_parsed_time": None, |
| "scheduler_lock": None, |
| "timetable_description": None, |
| "has_import_errors": False, |
| "pickle_id": None, |
| } == response.json |
| |
| @conf_vars({("webserver", "secret_key"): "mysecret"}) |
| def test_should_respond_200_with_schedule_interval_none(self, session): |
| dag_model = DagModel( |
| dag_id="TEST_DAG_1", |
| fileloc="/tmp/dag_1.py", |
| schedule_interval=None, |
| is_paused=False, |
| ) |
| session.add(dag_model) |
| session.commit() |
| response = self.client.get("/api/v1/dags/TEST_DAG_1", environ_overrides={"REMOTE_USER": "test"}) |
| assert response.status_code == 200 |
| assert { |
| "dag_id": "TEST_DAG_1", |
| "description": None, |
| "fileloc": "/tmp/dag_1.py", |
| "file_token": "Ii90bXAvZGFnXzEucHki.EnmIdPaUPo26lHQClbWMbDFD1Pk", |
| "is_paused": False, |
| "is_active": False, |
| "is_subdag": False, |
| "owners": [], |
| "root_dag_id": None, |
| "schedule_interval": None, |
| "tags": [], |
| "next_dagrun": None, |
| "has_task_concurrency_limits": True, |
| "next_dagrun_data_interval_start": None, |
| "next_dagrun_data_interval_end": None, |
| "max_active_runs": 16, |
| "next_dagrun_create_after": None, |
| "last_expired": None, |
| "max_active_tasks": 16, |
| "last_pickled": None, |
| "default_view": None, |
| "last_parsed_time": None, |
| "scheduler_lock": None, |
| "timetable_description": None, |
| "has_import_errors": False, |
| "pickle_id": None, |
| } == response.json |
| |
| def test_should_respond_200_with_granular_dag_access(self): |
| self._create_dag_models(1) |
| response = self.client.get( |
| "/api/v1/dags/TEST_DAG_1", environ_overrides={"REMOTE_USER": "test_granular_permissions"} |
| ) |
| assert response.status_code == 200 |
| |
| def test_should_respond_404(self): |
| response = self.client.get("/api/v1/dags/INVALID_DAG", environ_overrides={"REMOTE_USER": "test"}) |
| assert response.status_code == 404 |
| |
| def test_should_raises_401_unauthenticated(self): |
| self._create_dag_models(1) |
| |
| response = self.client.get("/api/v1/dags/TEST_DAG_1") |
| |
| assert_401(response) |
| |
| def test_should_raise_403_forbidden(self): |
| response = self.client.get( |
| f"/api/v1/dags/{self.dag_id}/details", environ_overrides={"REMOTE_USER": "test_no_permissions"} |
| ) |
| assert response.status_code == 403 |
| |
| def test_should_respond_403_with_granular_access_for_different_dag(self): |
| self._create_dag_models(3) |
| response = self.client.get( |
| "/api/v1/dags/TEST_DAG_2", environ_overrides={"REMOTE_USER": "test_granular_permissions"} |
| ) |
| assert response.status_code == 403 |
| |
| |
| class TestGetDagDetails(TestDagEndpoint): |
| def test_should_respond_200(self, current_file_token): |
| response = self.client.get( |
| f"/api/v1/dags/{self.dag_id}/details", environ_overrides={"REMOTE_USER": "test"} |
| ) |
| assert response.status_code == 200 |
| last_parsed = response.json["last_parsed"] |
| expected = { |
| "catchup": True, |
| "concurrency": 16, |
| "max_active_tasks": 16, |
| "dag_id": "test_dag", |
| "dag_run_timeout": None, |
| "default_view": "grid", |
| "description": None, |
| "doc_md": "details", |
| "fileloc": __file__, |
| "file_token": current_file_token, |
| "is_paused": None, |
| "is_active": None, |
| "is_subdag": False, |
| "orientation": "LR", |
| "owners": ["airflow"], |
| "params": { |
| "foo": { |
| "__class": "airflow.models.param.Param", |
| "value": 1, |
| "description": None, |
| "schema": {}, |
| } |
| }, |
| "schedule_interval": { |
| "__type": "TimeDelta", |
| "days": 1, |
| "microseconds": 0, |
| "seconds": 0, |
| }, |
| "start_date": "2020-06-15T00:00:00+00:00", |
| "tags": [{"name": "example"}], |
| "timezone": "Timezone('UTC')", |
| "max_active_runs": 16, |
| "pickle_id": None, |
| "end_date": None, |
| "is_paused_upon_creation": None, |
| "last_parsed": last_parsed, |
| "render_template_as_native_obj": False, |
| } |
| assert response.json == expected |
| |
| def test_should_response_200_with_doc_md_none(self, current_file_token): |
| response = self.client.get( |
| f"/api/v1/dags/{self.dag2_id}/details", environ_overrides={"REMOTE_USER": "test"} |
| ) |
| assert response.status_code == 200 |
| last_parsed = response.json["last_parsed"] |
| expected = { |
| "catchup": True, |
| "concurrency": 16, |
| "max_active_tasks": 16, |
| "dag_id": "test_dag2", |
| "dag_run_timeout": None, |
| "default_view": "grid", |
| "description": None, |
| "doc_md": None, |
| "fileloc": __file__, |
| "file_token": current_file_token, |
| "is_paused": None, |
| "is_active": None, |
| "is_subdag": False, |
| "orientation": "LR", |
| "owners": ["airflow"], |
| "params": {}, |
| "schedule_interval": { |
| "__type": "TimeDelta", |
| "days": 1, |
| "microseconds": 0, |
| "seconds": 0, |
| }, |
| "start_date": "2020-06-15T00:00:00+00:00", |
| "tags": [], |
| "timezone": "Timezone('UTC')", |
| "max_active_runs": 16, |
| "pickle_id": None, |
| "end_date": None, |
| "is_paused_upon_creation": None, |
| "last_parsed": last_parsed, |
| "render_template_as_native_obj": False, |
| } |
| assert response.json == expected |
| |
| def test_should_response_200_for_null_start_date(self, current_file_token): |
| response = self.client.get( |
| f"/api/v1/dags/{self.dag3_id}/details", environ_overrides={"REMOTE_USER": "test"} |
| ) |
| assert response.status_code == 200 |
| last_parsed = response.json["last_parsed"] |
| expected = { |
| "catchup": True, |
| "concurrency": 16, |
| "max_active_tasks": 16, |
| "dag_id": "test_dag3", |
| "dag_run_timeout": None, |
| "default_view": "grid", |
| "description": None, |
| "doc_md": None, |
| "fileloc": __file__, |
| "file_token": current_file_token, |
| "is_paused": None, |
| "is_active": None, |
| "is_subdag": False, |
| "orientation": "LR", |
| "owners": ["airflow"], |
| "params": {}, |
| "schedule_interval": { |
| "__type": "TimeDelta", |
| "days": 1, |
| "microseconds": 0, |
| "seconds": 0, |
| }, |
| "start_date": None, |
| "tags": [], |
| "timezone": "Timezone('UTC')", |
| "max_active_runs": 16, |
| "pickle_id": None, |
| "end_date": None, |
| "is_paused_upon_creation": None, |
| "last_parsed": last_parsed, |
| "render_template_as_native_obj": False, |
| } |
| assert response.json == expected |
| |
| def test_should_respond_200_serialized(self, current_file_token): |
| # Get the dag out of the dagbag before we patch it to an empty one |
| SerializedDagModel.write_dag(self.app.dag_bag.get_dag(self.dag_id)) |
| |
| # Create empty app with empty dagbag to check if DAG is read from db |
| dag_bag = DagBag(os.devnull, include_examples=False, read_dags_from_db=True) |
| patcher = unittest.mock.patch.object(self.app, "dag_bag", dag_bag) |
| patcher.start() |
| |
| expected = { |
| "catchup": True, |
| "concurrency": 16, |
| "max_active_tasks": 16, |
| "dag_id": "test_dag", |
| "dag_run_timeout": None, |
| "default_view": "grid", |
| "description": None, |
| "doc_md": "details", |
| "fileloc": __file__, |
| "file_token": current_file_token, |
| "is_paused": None, |
| "is_active": None, |
| "is_subdag": False, |
| "orientation": "LR", |
| "owners": ["airflow"], |
| "params": { |
| "foo": { |
| "__class": "airflow.models.param.Param", |
| "value": 1, |
| "description": None, |
| "schema": {}, |
| } |
| }, |
| "schedule_interval": { |
| "__type": "TimeDelta", |
| "days": 1, |
| "microseconds": 0, |
| "seconds": 0, |
| }, |
| "start_date": "2020-06-15T00:00:00+00:00", |
| "tags": [{"name": "example"}], |
| "timezone": "Timezone('UTC')", |
| "max_active_runs": 16, |
| "pickle_id": None, |
| "end_date": None, |
| "is_paused_upon_creation": None, |
| "render_template_as_native_obj": False, |
| } |
| response = self.client.get( |
| f"/api/v1/dags/{self.dag_id}/details", environ_overrides={"REMOTE_USER": "test"} |
| ) |
| |
| assert response.status_code == 200 |
| expected.update({"last_parsed": response.json["last_parsed"]}) |
| |
| assert response.json == expected |
| |
| patcher.stop() |
| |
| response = self.client.get( |
| f"/api/v1/dags/{self.dag_id}/details", environ_overrides={"REMOTE_USER": "test"} |
| ) |
| assert response.status_code == 200 |
| expected = { |
| "catchup": True, |
| "concurrency": 16, |
| "max_active_tasks": 16, |
| "dag_id": "test_dag", |
| "dag_run_timeout": None, |
| "default_view": "grid", |
| "description": None, |
| "doc_md": "details", |
| "fileloc": __file__, |
| "file_token": current_file_token, |
| "is_paused": None, |
| "is_active": None, |
| "is_subdag": False, |
| "orientation": "LR", |
| "owners": ["airflow"], |
| "params": { |
| "foo": { |
| "__class": "airflow.models.param.Param", |
| "value": 1, |
| "description": None, |
| "schema": {}, |
| } |
| }, |
| "schedule_interval": {"__type": "TimeDelta", "days": 1, "microseconds": 0, "seconds": 0}, |
| "start_date": "2020-06-15T00:00:00+00:00", |
| "tags": [{"name": "example"}], |
| "timezone": "Timezone('UTC')", |
| "max_active_runs": 16, |
| "pickle_id": None, |
| "end_date": None, |
| "is_paused_upon_creation": None, |
| "render_template_as_native_obj": False, |
| } |
| expected.update({"last_parsed": response.json["last_parsed"]}) |
| assert response.json == expected |
| |
| def test_should_raises_401_unauthenticated(self): |
| response = self.client.get(f"/api/v1/dags/{self.dag_id}/details") |
| |
| assert_401(response) |
| |
| def test_should_raise_404_when_dag_is_not_found(self): |
| response = self.client.get( |
| "/api/v1/dags/non_existing_dag_id/details", environ_overrides={"REMOTE_USER": "test"} |
| ) |
| assert response.status_code == 404 |
| assert response.json == { |
| "detail": "The DAG with dag_id: non_existing_dag_id was not found", |
| "status": 404, |
| "title": "DAG not found", |
| "type": EXCEPTIONS_LINK_MAP[404], |
| } |
| |
| |
| class TestGetDags(TestDagEndpoint): |
| @provide_session |
| def test_should_respond_200(self, session, url_safe_serializer): |
| self._create_dag_models(2) |
| self._create_deactivated_dag() |
| |
| dags_query = session.query(DagModel).filter(~DagModel.is_subdag) |
| assert len(dags_query.all()) == 3 |
| |
| response = self.client.get("api/v1/dags", environ_overrides={"REMOTE_USER": "test"}) |
| file_token = url_safe_serializer.dumps("/tmp/dag_1.py") |
| file_token2 = url_safe_serializer.dumps("/tmp/dag_2.py") |
| |
| assert response.status_code == 200 |
| assert { |
| "dags": [ |
| { |
| "dag_id": "TEST_DAG_1", |
| "description": None, |
| "fileloc": "/tmp/dag_1.py", |
| "file_token": file_token, |
| "is_paused": False, |
| "is_active": True, |
| "is_subdag": False, |
| "owners": [], |
| "root_dag_id": None, |
| "schedule_interval": { |
| "__type": "CronExpression", |
| "value": "2 2 * * *", |
| }, |
| "tags": [], |
| "next_dagrun": None, |
| "has_task_concurrency_limits": True, |
| "next_dagrun_data_interval_start": None, |
| "next_dagrun_data_interval_end": None, |
| "max_active_runs": 16, |
| "next_dagrun_create_after": None, |
| "last_expired": None, |
| "max_active_tasks": 16, |
| "last_pickled": None, |
| "default_view": None, |
| "last_parsed_time": None, |
| "scheduler_lock": None, |
| "timetable_description": None, |
| "has_import_errors": False, |
| "pickle_id": None, |
| }, |
| { |
| "dag_id": "TEST_DAG_2", |
| "description": None, |
| "fileloc": "/tmp/dag_2.py", |
| "file_token": file_token2, |
| "is_paused": False, |
| "is_active": True, |
| "is_subdag": False, |
| "owners": [], |
| "root_dag_id": None, |
| "schedule_interval": { |
| "__type": "CronExpression", |
| "value": "2 2 * * *", |
| }, |
| "tags": [], |
| "next_dagrun": None, |
| "has_task_concurrency_limits": True, |
| "next_dagrun_data_interval_start": None, |
| "next_dagrun_data_interval_end": None, |
| "max_active_runs": 16, |
| "next_dagrun_create_after": None, |
| "last_expired": None, |
| "max_active_tasks": 16, |
| "last_pickled": None, |
| "default_view": None, |
| "last_parsed_time": None, |
| "scheduler_lock": None, |
| "timetable_description": None, |
| "has_import_errors": False, |
| "pickle_id": None, |
| }, |
| ], |
| "total_entries": 2, |
| } == response.json |
| |
| def test_only_active_true_returns_active_dags(self, url_safe_serializer): |
| self._create_dag_models(1) |
| self._create_deactivated_dag() |
| response = self.client.get("api/v1/dags?only_active=True", environ_overrides={"REMOTE_USER": "test"}) |
| file_token = url_safe_serializer.dumps("/tmp/dag_1.py") |
| assert response.status_code == 200 |
| assert { |
| "dags": [ |
| { |
| "dag_id": "TEST_DAG_1", |
| "description": None, |
| "fileloc": "/tmp/dag_1.py", |
| "file_token": file_token, |
| "is_paused": False, |
| "is_active": True, |
| "is_subdag": False, |
| "owners": [], |
| "root_dag_id": None, |
| "schedule_interval": { |
| "__type": "CronExpression", |
| "value": "2 2 * * *", |
| }, |
| "tags": [], |
| "next_dagrun": None, |
| "has_task_concurrency_limits": True, |
| "next_dagrun_data_interval_start": None, |
| "next_dagrun_data_interval_end": None, |
| "max_active_runs": 16, |
| "next_dagrun_create_after": None, |
| "last_expired": None, |
| "max_active_tasks": 16, |
| "last_pickled": None, |
| "default_view": None, |
| "last_parsed_time": None, |
| "scheduler_lock": None, |
| "timetable_description": None, |
| "has_import_errors": False, |
| "pickle_id": None, |
| } |
| ], |
| "total_entries": 1, |
| } == response.json |
| |
| def test_only_active_false_returns_all_dags(self, url_safe_serializer): |
| self._create_dag_models(1) |
| self._create_deactivated_dag() |
| response = self.client.get("api/v1/dags?only_active=False", environ_overrides={"REMOTE_USER": "test"}) |
| file_token = url_safe_serializer.dumps("/tmp/dag_1.py") |
| file_token_2 = url_safe_serializer.dumps("/tmp/dag_del_1.py") |
| assert response.status_code == 200 |
| assert { |
| "dags": [ |
| { |
| "dag_id": "TEST_DAG_1", |
| "description": None, |
| "fileloc": "/tmp/dag_1.py", |
| "file_token": file_token, |
| "is_paused": False, |
| "is_active": True, |
| "is_subdag": False, |
| "owners": [], |
| "root_dag_id": None, |
| "schedule_interval": { |
| "__type": "CronExpression", |
| "value": "2 2 * * *", |
| }, |
| "tags": [], |
| "next_dagrun": None, |
| "has_task_concurrency_limits": True, |
| "next_dagrun_data_interval_start": None, |
| "next_dagrun_data_interval_end": None, |
| "max_active_runs": 16, |
| "next_dagrun_create_after": None, |
| "last_expired": None, |
| "max_active_tasks": 16, |
| "last_pickled": None, |
| "default_view": None, |
| "last_parsed_time": None, |
| "scheduler_lock": None, |
| "timetable_description": None, |
| "has_import_errors": False, |
| "pickle_id": None, |
| }, |
| { |
| "dag_id": "TEST_DAG_DELETED_1", |
| "description": None, |
| "fileloc": "/tmp/dag_del_1.py", |
| "file_token": file_token_2, |
| "is_paused": False, |
| "is_active": False, |
| "is_subdag": False, |
| "owners": [], |
| "root_dag_id": None, |
| "schedule_interval": { |
| "__type": "CronExpression", |
| "value": "2 2 * * *", |
| }, |
| "tags": [], |
| "next_dagrun": None, |
| "has_task_concurrency_limits": True, |
| "next_dagrun_data_interval_start": None, |
| "next_dagrun_data_interval_end": None, |
| "max_active_runs": 16, |
| "next_dagrun_create_after": None, |
| "last_expired": None, |
| "max_active_tasks": 16, |
| "last_pickled": None, |
| "default_view": None, |
| "last_parsed_time": None, |
| "scheduler_lock": None, |
| "timetable_description": None, |
| "has_import_errors": False, |
| "pickle_id": None, |
| }, |
| ], |
| "total_entries": 2, |
| } == response.json |
| |
| @pytest.mark.parametrize( |
| "url, expected_dag_ids", |
| [ |
| ("api/v1/dags?tags=t1", ["TEST_DAG_1", "TEST_DAG_3"]), |
| ("api/v1/dags?tags=t2", ["TEST_DAG_2", "TEST_DAG_3"]), |
| ("api/v1/dags?tags=t1,t2", ["TEST_DAG_1", "TEST_DAG_2", "TEST_DAG_3"]), |
| ("api/v1/dags", ["TEST_DAG_1", "TEST_DAG_2", "TEST_DAG_3", "TEST_DAG_4"]), |
| ], |
| ) |
| def test_filter_dags_by_tags_works(self, url, expected_dag_ids): |
| # test filter by tags |
| dag1 = DAG(dag_id="TEST_DAG_1", tags=["t1"]) |
| dag2 = DAG(dag_id="TEST_DAG_2", tags=["t2"]) |
| dag3 = DAG(dag_id="TEST_DAG_3", tags=["t1", "t2"]) |
| dag4 = DAG(dag_id="TEST_DAG_4") |
| dag1.sync_to_db() |
| dag2.sync_to_db() |
| dag3.sync_to_db() |
| dag4.sync_to_db() |
| |
| response = self.client.get(url, environ_overrides={"REMOTE_USER": "test"}) |
| assert response.status_code == 200 |
| dag_ids = [dag["dag_id"] for dag in response.json["dags"]] |
| |
| assert expected_dag_ids == dag_ids |
| |
| @pytest.mark.parametrize( |
| "url, expected_dag_ids", |
| [ |
| ("api/v1/dags?dag_id_pattern=DAG_1", {"TEST_DAG_1", "SAMPLE_DAG_1"}), |
| ("api/v1/dags?dag_id_pattern=SAMPLE_DAG", {"SAMPLE_DAG_1", "SAMPLE_DAG_2"}), |
| ( |
| "api/v1/dags?dag_id_pattern=_DAG_", |
| {"TEST_DAG_1", "TEST_DAG_2", "SAMPLE_DAG_1", "SAMPLE_DAG_2"}, |
| ), |
| ], |
| ) |
| def test_filter_dags_by_dag_id_works(self, url, expected_dag_ids): |
| # test filter by tags |
| dag1 = DAG(dag_id="TEST_DAG_1") |
| dag2 = DAG(dag_id="TEST_DAG_2") |
| dag3 = DAG(dag_id="SAMPLE_DAG_1") |
| dag4 = DAG(dag_id="SAMPLE_DAG_2") |
| dag1.sync_to_db() |
| dag2.sync_to_db() |
| dag3.sync_to_db() |
| dag4.sync_to_db() |
| |
| response = self.client.get(url, environ_overrides={"REMOTE_USER": "test"}) |
| assert response.status_code == 200 |
| dag_ids = {dag["dag_id"] for dag in response.json["dags"]} |
| |
| assert expected_dag_ids == dag_ids |
| |
| def test_should_respond_200_with_granular_dag_access(self): |
| self._create_dag_models(3) |
| response = self.client.get( |
| "/api/v1/dags", environ_overrides={"REMOTE_USER": "test_granular_permissions"} |
| ) |
| assert response.status_code == 200 |
| assert len(response.json["dags"]) == 1 |
| assert response.json["dags"][0]["dag_id"] == "TEST_DAG_1" |
| |
| @pytest.mark.parametrize( |
| "url, expected_dag_ids", |
| [ |
| ("api/v1/dags?limit=1", ["TEST_DAG_1"]), |
| ("api/v1/dags?limit=2", ["TEST_DAG_1", "TEST_DAG_10"]), |
| ( |
| "api/v1/dags?offset=5", |
| ["TEST_DAG_5", "TEST_DAG_6", "TEST_DAG_7", "TEST_DAG_8", "TEST_DAG_9"], |
| ), |
| ( |
| "api/v1/dags?offset=0", |
| [ |
| "TEST_DAG_1", |
| "TEST_DAG_10", |
| "TEST_DAG_2", |
| "TEST_DAG_3", |
| "TEST_DAG_4", |
| "TEST_DAG_5", |
| "TEST_DAG_6", |
| "TEST_DAG_7", |
| "TEST_DAG_8", |
| "TEST_DAG_9", |
| ], |
| ), |
| ("api/v1/dags?limit=1&offset=5", ["TEST_DAG_5"]), |
| ("api/v1/dags?limit=1&offset=1", ["TEST_DAG_10"]), |
| ("api/v1/dags?limit=2&offset=2", ["TEST_DAG_2", "TEST_DAG_3"]), |
| ], |
| ) |
| def test_should_respond_200_and_handle_pagination(self, url, expected_dag_ids): |
| self._create_dag_models(10) |
| |
| response = self.client.get(url, environ_overrides={"REMOTE_USER": "test"}) |
| |
| assert response.status_code == 200 |
| |
| dag_ids = [dag["dag_id"] for dag in response.json["dags"]] |
| |
| assert expected_dag_ids == dag_ids |
| assert 10 == response.json["total_entries"] |
| |
| def test_should_respond_200_default_limit(self): |
| self._create_dag_models(101) |
| |
| response = self.client.get("api/v1/dags", environ_overrides={"REMOTE_USER": "test"}) |
| |
| assert response.status_code == 200 |
| |
| assert 100 == len(response.json["dags"]) |
| assert 101 == response.json["total_entries"] |
| |
| def test_should_raises_401_unauthenticated(self): |
| response = self.client.get("api/v1/dags") |
| |
| assert_401(response) |
| |
| def test_should_respond_403_unauthorized(self): |
| self._create_dag_models(1) |
| |
| response = self.client.get("api/v1/dags", environ_overrides={"REMOTE_USER": "test_no_permissions"}) |
| |
| assert response.status_code == 403 |
| |
| def test_paused_true_returns_paused_dags(self, url_safe_serializer): |
| self._create_dag_models(1, dag_id_prefix="TEST_DAG_PAUSED", is_paused=True) |
| self._create_dag_models(1, dag_id_prefix="TEST_DAG_UNPAUSED", is_paused=False) |
| response = self.client.get("api/v1/dags?paused=True", environ_overrides={"REMOTE_USER": "test"}) |
| file_token = url_safe_serializer.dumps("/tmp/dag_1.py") |
| assert response.status_code == 200 |
| assert { |
| "dags": [ |
| { |
| "dag_id": "TEST_DAG_PAUSED_1", |
| "description": None, |
| "fileloc": "/tmp/dag_1.py", |
| "file_token": file_token, |
| "is_paused": True, |
| "is_active": True, |
| "is_subdag": False, |
| "owners": [], |
| "root_dag_id": None, |
| "schedule_interval": { |
| "__type": "CronExpression", |
| "value": "2 2 * * *", |
| }, |
| "tags": [], |
| "next_dagrun": None, |
| "has_task_concurrency_limits": True, |
| "next_dagrun_data_interval_start": None, |
| "next_dagrun_data_interval_end": None, |
| "max_active_runs": 16, |
| "next_dagrun_create_after": None, |
| "last_expired": None, |
| "max_active_tasks": 16, |
| "last_pickled": None, |
| "default_view": None, |
| "last_parsed_time": None, |
| "scheduler_lock": None, |
| "timetable_description": None, |
| "has_import_errors": False, |
| "pickle_id": None, |
| } |
| ], |
| "total_entries": 1, |
| } == response.json |
| |
| def test_paused_false_returns_unpaused_dags(self, url_safe_serializer): |
| self._create_dag_models(1, dag_id_prefix="TEST_DAG_PAUSED", is_paused=True) |
| self._create_dag_models(1, dag_id_prefix="TEST_DAG_UNPAUSED", is_paused=False) |
| response = self.client.get("api/v1/dags?paused=False", environ_overrides={"REMOTE_USER": "test"}) |
| file_token = url_safe_serializer.dumps("/tmp/dag_1.py") |
| assert response.status_code == 200 |
| assert { |
| "dags": [ |
| { |
| "dag_id": "TEST_DAG_UNPAUSED_1", |
| "description": None, |
| "fileloc": "/tmp/dag_1.py", |
| "file_token": file_token, |
| "is_paused": False, |
| "is_active": True, |
| "is_subdag": False, |
| "owners": [], |
| "root_dag_id": None, |
| "schedule_interval": { |
| "__type": "CronExpression", |
| "value": "2 2 * * *", |
| }, |
| "tags": [], |
| "next_dagrun": None, |
| "has_task_concurrency_limits": True, |
| "next_dagrun_data_interval_start": None, |
| "next_dagrun_data_interval_end": None, |
| "max_active_runs": 16, |
| "next_dagrun_create_after": None, |
| "last_expired": None, |
| "max_active_tasks": 16, |
| "last_pickled": None, |
| "default_view": None, |
| "last_parsed_time": None, |
| "scheduler_lock": None, |
| "timetable_description": None, |
| "has_import_errors": False, |
| "pickle_id": None, |
| } |
| ], |
| "total_entries": 1, |
| } == response.json |
| |
| def test_paused_none_returns_all_dags(self, url_safe_serializer): |
| self._create_dag_models(1, dag_id_prefix="TEST_DAG_PAUSED", is_paused=True) |
| self._create_dag_models(1, dag_id_prefix="TEST_DAG_UNPAUSED", is_paused=False) |
| response = self.client.get("api/v1/dags", environ_overrides={"REMOTE_USER": "test"}) |
| file_token = url_safe_serializer.dumps("/tmp/dag_1.py") |
| assert response.status_code == 200 |
| assert { |
| "dags": [ |
| { |
| "dag_id": "TEST_DAG_PAUSED_1", |
| "description": None, |
| "fileloc": "/tmp/dag_1.py", |
| "file_token": file_token, |
| "is_paused": True, |
| "is_active": True, |
| "is_subdag": False, |
| "owners": [], |
| "root_dag_id": None, |
| "schedule_interval": { |
| "__type": "CronExpression", |
| "value": "2 2 * * *", |
| }, |
| "tags": [], |
| "next_dagrun": None, |
| "has_task_concurrency_limits": True, |
| "next_dagrun_data_interval_start": None, |
| "next_dagrun_data_interval_end": None, |
| "max_active_runs": 16, |
| "next_dagrun_create_after": None, |
| "last_expired": None, |
| "max_active_tasks": 16, |
| "last_pickled": None, |
| "default_view": None, |
| "last_parsed_time": None, |
| "scheduler_lock": None, |
| "timetable_description": None, |
| "has_import_errors": False, |
| "pickle_id": None, |
| }, |
| { |
| "dag_id": "TEST_DAG_UNPAUSED_1", |
| "description": None, |
| "fileloc": "/tmp/dag_1.py", |
| "file_token": file_token, |
| "is_paused": False, |
| "is_active": True, |
| "is_subdag": False, |
| "owners": [], |
| "root_dag_id": None, |
| "schedule_interval": { |
| "__type": "CronExpression", |
| "value": "2 2 * * *", |
| }, |
| "tags": [], |
| "next_dagrun": None, |
| "has_task_concurrency_limits": True, |
| "next_dagrun_data_interval_start": None, |
| "next_dagrun_data_interval_end": None, |
| "max_active_runs": 16, |
| "next_dagrun_create_after": None, |
| "last_expired": None, |
| "max_active_tasks": 16, |
| "last_pickled": None, |
| "default_view": None, |
| "last_parsed_time": None, |
| "scheduler_lock": None, |
| "timetable_description": None, |
| "has_import_errors": False, |
| "pickle_id": None, |
| }, |
| ], |
| "total_entries": 2, |
| } == response.json |
| |
| |
| class TestPatchDag(TestDagEndpoint): |
| def test_should_respond_200_on_patch_is_paused(self, url_safe_serializer): |
| file_token = url_safe_serializer.dumps("/tmp/dag_1.py") |
| dag_model = self._create_dag_model() |
| response = self.client.patch( |
| f"/api/v1/dags/{dag_model.dag_id}", |
| json={ |
| "is_paused": False, |
| }, |
| environ_overrides={"REMOTE_USER": "test"}, |
| ) |
| assert response.status_code == 200 |
| expected_response = { |
| "dag_id": "TEST_DAG_1", |
| "description": None, |
| "fileloc": "/tmp/dag_1.py", |
| "file_token": file_token, |
| "is_paused": False, |
| "is_active": False, |
| "is_subdag": False, |
| "owners": [], |
| "root_dag_id": None, |
| "schedule_interval": { |
| "__type": "CronExpression", |
| "value": "2 2 * * *", |
| }, |
| "tags": [], |
| "next_dagrun": None, |
| "has_task_concurrency_limits": True, |
| "next_dagrun_data_interval_start": None, |
| "next_dagrun_data_interval_end": None, |
| "max_active_runs": 16, |
| "next_dagrun_create_after": None, |
| "last_expired": None, |
| "max_active_tasks": 16, |
| "last_pickled": None, |
| "default_view": None, |
| "last_parsed_time": None, |
| "scheduler_lock": None, |
| "timetable_description": None, |
| "has_import_errors": False, |
| "pickle_id": None, |
| } |
| assert response.json == expected_response |
| |
| def test_should_respond_200_on_patch_with_granular_dag_access(self): |
| self._create_dag_models(1) |
| response = self.client.patch( |
| "/api/v1/dags/TEST_DAG_1", |
| json={ |
| "is_paused": False, |
| }, |
| environ_overrides={"REMOTE_USER": "test_granular_permissions"}, |
| ) |
| assert response.status_code == 200 |
| |
| def test_should_respond_400_on_invalid_request(self): |
| patch_body = { |
| "is_paused": True, |
| "schedule_interval": { |
| "__type": "CronExpression", |
| "value": "1 1 * * *", |
| }, |
| } |
| dag_model = self._create_dag_model() |
| response = self.client.patch(f"/api/v1/dags/{dag_model.dag_id}", json=patch_body) |
| assert response.status_code == 400 |
| assert response.json == { |
| "detail": "Property is read-only - 'schedule_interval'", |
| "status": 400, |
| "title": "Bad Request", |
| "type": EXCEPTIONS_LINK_MAP[400], |
| } |
| |
| def test_should_respond_404(self): |
| response = self.client.get("/api/v1/dags/INVALID_DAG", environ_overrides={"REMOTE_USER": "test"}) |
| assert response.status_code == 404 |
| |
| @provide_session |
| def _create_dag_model(self, session=None): |
| dag_model = DagModel( |
| dag_id="TEST_DAG_1", fileloc="/tmp/dag_1.py", schedule_interval="2 2 * * *", is_paused=True |
| ) |
| session.add(dag_model) |
| return dag_model |
| |
| def test_should_raises_401_unauthenticated(self): |
| dag_model = self._create_dag_model() |
| response = self.client.patch( |
| f"/api/v1/dags/{dag_model.dag_id}", |
| json={ |
| "is_paused": False, |
| }, |
| ) |
| |
| assert_401(response) |
| |
| def test_should_respond_200_with_update_mask(self, url_safe_serializer): |
| file_token = url_safe_serializer.dumps("/tmp/dag_1.py") |
| dag_model = self._create_dag_model() |
| payload = { |
| "is_paused": False, |
| } |
| response = self.client.patch( |
| f"/api/v1/dags/{dag_model.dag_id}?update_mask=is_paused", |
| json=payload, |
| environ_overrides={"REMOTE_USER": "test"}, |
| ) |
| |
| assert response.status_code == 200 |
| expected_response = { |
| "dag_id": "TEST_DAG_1", |
| "description": None, |
| "fileloc": "/tmp/dag_1.py", |
| "file_token": file_token, |
| "is_paused": False, |
| "is_active": False, |
| "is_subdag": False, |
| "owners": [], |
| "root_dag_id": None, |
| "schedule_interval": { |
| "__type": "CronExpression", |
| "value": "2 2 * * *", |
| }, |
| "tags": [], |
| "next_dagrun": None, |
| "has_task_concurrency_limits": True, |
| "next_dagrun_data_interval_start": None, |
| "next_dagrun_data_interval_end": None, |
| "max_active_runs": 16, |
| "next_dagrun_create_after": None, |
| "last_expired": None, |
| "max_active_tasks": 16, |
| "last_pickled": None, |
| "default_view": None, |
| "last_parsed_time": None, |
| "scheduler_lock": None, |
| "timetable_description": None, |
| "has_import_errors": False, |
| "pickle_id": None, |
| } |
| assert response.json == expected_response |
| |
| @pytest.mark.parametrize( |
| "payload, update_mask, error_message", |
| [ |
| ( |
| { |
| "is_paused": True, |
| }, |
| "update_mask=description", |
| "Only `is_paused` field can be updated through the REST API", |
| ), |
| ( |
| { |
| "is_paused": True, |
| }, |
| "update_mask=schedule_interval, description", |
| "Only `is_paused` field can be updated through the REST API", |
| ), |
| ], |
| ) |
| def test_should_respond_400_for_invalid_fields_in_update_mask(self, payload, update_mask, error_message): |
| dag_model = self._create_dag_model() |
| |
| response = self.client.patch( |
| f"/api/v1/dags/{dag_model.dag_id}?{update_mask}", |
| json=payload, |
| environ_overrides={"REMOTE_USER": "test"}, |
| ) |
| assert response.status_code == 400 |
| assert response.json["detail"] == error_message |
| |
| def test_should_respond_403_unauthorized(self): |
| dag_model = self._create_dag_model() |
| response = self.client.patch( |
| f"/api/v1/dags/{dag_model.dag_id}", |
| json={ |
| "is_paused": False, |
| }, |
| environ_overrides={"REMOTE_USER": "test_no_permissions"}, |
| ) |
| |
| assert response.status_code == 403 |
| |
| |
| class TestPatchDags(TestDagEndpoint): |
| @provide_session |
| def test_should_respond_200_on_patch_is_paused(self, session, url_safe_serializer): |
| file_token = url_safe_serializer.dumps("/tmp/dag_1.py") |
| file_token2 = url_safe_serializer.dumps("/tmp/dag_2.py") |
| self._create_dag_models(2) |
| self._create_deactivated_dag() |
| |
| dags_query = session.query(DagModel).filter(~DagModel.is_subdag) |
| assert len(dags_query.all()) == 3 |
| |
| response = self.client.patch( |
| "/api/v1/dags?dag_id_pattern=~", |
| json={ |
| "is_paused": False, |
| }, |
| environ_overrides={"REMOTE_USER": "test"}, |
| ) |
| |
| assert response.status_code == 200 |
| assert { |
| "dags": [ |
| { |
| "dag_id": "TEST_DAG_1", |
| "description": None, |
| "fileloc": "/tmp/dag_1.py", |
| "file_token": file_token, |
| "is_paused": False, |
| "is_active": True, |
| "is_subdag": False, |
| "owners": [], |
| "root_dag_id": None, |
| "schedule_interval": { |
| "__type": "CronExpression", |
| "value": "2 2 * * *", |
| }, |
| "tags": [], |
| "next_dagrun": None, |
| "has_task_concurrency_limits": True, |
| "next_dagrun_data_interval_start": None, |
| "next_dagrun_data_interval_end": None, |
| "max_active_runs": 16, |
| "next_dagrun_create_after": None, |
| "last_expired": None, |
| "max_active_tasks": 16, |
| "last_pickled": None, |
| "default_view": None, |
| "last_parsed_time": None, |
| "scheduler_lock": None, |
| "timetable_description": None, |
| "has_import_errors": False, |
| "pickle_id": None, |
| }, |
| { |
| "dag_id": "TEST_DAG_2", |
| "description": None, |
| "fileloc": "/tmp/dag_2.py", |
| "file_token": file_token2, |
| "is_paused": False, |
| "is_active": True, |
| "is_subdag": False, |
| "owners": [], |
| "root_dag_id": None, |
| "schedule_interval": { |
| "__type": "CronExpression", |
| "value": "2 2 * * *", |
| }, |
| "tags": [], |
| "next_dagrun": None, |
| "has_task_concurrency_limits": True, |
| "next_dagrun_data_interval_start": None, |
| "next_dagrun_data_interval_end": None, |
| "max_active_runs": 16, |
| "next_dagrun_create_after": None, |
| "last_expired": None, |
| "max_active_tasks": 16, |
| "last_pickled": None, |
| "default_view": None, |
| "last_parsed_time": None, |
| "scheduler_lock": None, |
| "timetable_description": None, |
| "has_import_errors": False, |
| "pickle_id": None, |
| }, |
| ], |
| "total_entries": 2, |
| } == response.json |
| |
| def test_only_active_true_returns_active_dags(self, url_safe_serializer): |
| file_token = url_safe_serializer.dumps("/tmp/dag_1.py") |
| self._create_dag_models(1) |
| self._create_deactivated_dag() |
| response = self.client.patch( |
| "/api/v1/dags?only_active=True&dag_id_pattern=~", |
| json={ |
| "is_paused": False, |
| }, |
| environ_overrides={"REMOTE_USER": "test"}, |
| ) |
| assert response.status_code == 200 |
| assert { |
| "dags": [ |
| { |
| "dag_id": "TEST_DAG_1", |
| "description": None, |
| "fileloc": "/tmp/dag_1.py", |
| "file_token": file_token, |
| "is_paused": False, |
| "is_active": True, |
| "is_subdag": False, |
| "owners": [], |
| "root_dag_id": None, |
| "schedule_interval": { |
| "__type": "CronExpression", |
| "value": "2 2 * * *", |
| }, |
| "tags": [], |
| "next_dagrun": None, |
| "has_task_concurrency_limits": True, |
| "next_dagrun_data_interval_start": None, |
| "next_dagrun_data_interval_end": None, |
| "max_active_runs": 16, |
| "next_dagrun_create_after": None, |
| "last_expired": None, |
| "max_active_tasks": 16, |
| "last_pickled": None, |
| "default_view": None, |
| "last_parsed_time": None, |
| "scheduler_lock": None, |
| "timetable_description": None, |
| "has_import_errors": False, |
| "pickle_id": None, |
| } |
| ], |
| "total_entries": 1, |
| } == response.json |
| |
| def test_only_active_false_returns_all_dags(self, url_safe_serializer): |
| file_token = url_safe_serializer.dumps("/tmp/dag_1.py") |
| self._create_dag_models(1) |
| self._create_deactivated_dag() |
| response = self.client.patch( |
| "/api/v1/dags?only_active=False&dag_id_pattern=~", |
| json={ |
| "is_paused": False, |
| }, |
| environ_overrides={"REMOTE_USER": "test"}, |
| ) |
| |
| file_token_2 = url_safe_serializer.dumps("/tmp/dag_del_1.py") |
| assert response.status_code == 200 |
| assert { |
| "dags": [ |
| { |
| "dag_id": "TEST_DAG_1", |
| "description": None, |
| "fileloc": "/tmp/dag_1.py", |
| "file_token": file_token, |
| "is_paused": False, |
| "is_active": True, |
| "is_subdag": False, |
| "owners": [], |
| "root_dag_id": None, |
| "schedule_interval": { |
| "__type": "CronExpression", |
| "value": "2 2 * * *", |
| }, |
| "tags": [], |
| "next_dagrun": None, |
| "has_task_concurrency_limits": True, |
| "next_dagrun_data_interval_start": None, |
| "next_dagrun_data_interval_end": None, |
| "max_active_runs": 16, |
| "next_dagrun_create_after": None, |
| "last_expired": None, |
| "max_active_tasks": 16, |
| "last_pickled": None, |
| "default_view": None, |
| "last_parsed_time": None, |
| "scheduler_lock": None, |
| "timetable_description": None, |
| "has_import_errors": False, |
| "pickle_id": None, |
| }, |
| { |
| "dag_id": "TEST_DAG_DELETED_1", |
| "description": None, |
| "fileloc": "/tmp/dag_del_1.py", |
| "file_token": file_token_2, |
| "is_paused": False, |
| "is_active": False, |
| "is_subdag": False, |
| "owners": [], |
| "root_dag_id": None, |
| "schedule_interval": { |
| "__type": "CronExpression", |
| "value": "2 2 * * *", |
| }, |
| "tags": [], |
| "next_dagrun": None, |
| "has_task_concurrency_limits": True, |
| "next_dagrun_data_interval_start": None, |
| "next_dagrun_data_interval_end": None, |
| "max_active_runs": 16, |
| "next_dagrun_create_after": None, |
| "last_expired": None, |
| "max_active_tasks": 16, |
| "last_pickled": None, |
| "default_view": None, |
| "last_parsed_time": None, |
| "scheduler_lock": None, |
| "timetable_description": None, |
| "has_import_errors": False, |
| "pickle_id": None, |
| }, |
| ], |
| "total_entries": 2, |
| } == response.json |
| |
| @pytest.mark.parametrize( |
| "url, expected_dag_ids", |
| [ |
| ("api/v1/dags?tags=t1&dag_id_pattern=~", ["TEST_DAG_1", "TEST_DAG_3"]), |
| ("api/v1/dags?tags=t2&dag_id_pattern=~", ["TEST_DAG_2", "TEST_DAG_3"]), |
| ("api/v1/dags?tags=t1,t2&dag_id_pattern=~", ["TEST_DAG_1", "TEST_DAG_2", "TEST_DAG_3"]), |
| ("api/v1/dags?dag_id_pattern=~", ["TEST_DAG_1", "TEST_DAG_2", "TEST_DAG_3", "TEST_DAG_4"]), |
| ], |
| ) |
| def test_filter_dags_by_tags_works(self, url, expected_dag_ids): |
| # test filter by tags |
| dag1 = DAG(dag_id="TEST_DAG_1", tags=["t1"]) |
| dag2 = DAG(dag_id="TEST_DAG_2", tags=["t2"]) |
| dag3 = DAG(dag_id="TEST_DAG_3", tags=["t1", "t2"]) |
| dag4 = DAG(dag_id="TEST_DAG_4") |
| dag1.sync_to_db() |
| dag2.sync_to_db() |
| dag3.sync_to_db() |
| dag4.sync_to_db() |
| response = self.client.patch( |
| url, |
| json={ |
| "is_paused": False, |
| }, |
| environ_overrides={"REMOTE_USER": "test"}, |
| ) |
| assert response.status_code == 200 |
| dag_ids = [dag["dag_id"] for dag in response.json["dags"]] |
| |
| assert expected_dag_ids == dag_ids |
| |
| @pytest.mark.parametrize( |
| "url, expected_dag_ids", |
| [ |
| ("api/v1/dags?dag_id_pattern=DAG_1", {"TEST_DAG_1", "SAMPLE_DAG_1"}), |
| ("api/v1/dags?dag_id_pattern=SAMPLE_DAG", {"SAMPLE_DAG_1", "SAMPLE_DAG_2"}), |
| ( |
| "api/v1/dags?dag_id_pattern=_DAG_", |
| {"TEST_DAG_1", "TEST_DAG_2", "SAMPLE_DAG_1", "SAMPLE_DAG_2"}, |
| ), |
| ], |
| ) |
| def test_filter_dags_by_dag_id_works(self, url, expected_dag_ids): |
| # test filter by tags |
| dag1 = DAG(dag_id="TEST_DAG_1") |
| dag2 = DAG(dag_id="TEST_DAG_2") |
| dag3 = DAG(dag_id="SAMPLE_DAG_1") |
| dag4 = DAG(dag_id="SAMPLE_DAG_2") |
| dag1.sync_to_db() |
| dag2.sync_to_db() |
| dag3.sync_to_db() |
| dag4.sync_to_db() |
| |
| response = self.client.patch( |
| url, |
| json={ |
| "is_paused": False, |
| }, |
| environ_overrides={"REMOTE_USER": "test"}, |
| ) |
| assert response.status_code == 200 |
| dag_ids = {dag["dag_id"] for dag in response.json["dags"]} |
| |
| assert expected_dag_ids == dag_ids |
| |
| def test_should_respond_200_with_granular_dag_access(self): |
| self._create_dag_models(3) |
| response = self.client.patch( |
| "api/v1/dags?dag_id_pattern=~", |
| json={ |
| "is_paused": False, |
| }, |
| environ_overrides={"REMOTE_USER": "test_granular_permissions"}, |
| ) |
| assert response.status_code == 200 |
| assert len(response.json["dags"]) == 1 |
| assert response.json["dags"][0]["dag_id"] == "TEST_DAG_1" |
| |
| @pytest.mark.parametrize( |
| "url, expected_dag_ids", |
| [ |
| ("api/v1/dags?limit=1&dag_id_pattern=~", ["TEST_DAG_1"]), |
| ("api/v1/dags?limit=2&dag_id_pattern=~", ["TEST_DAG_1", "TEST_DAG_10"]), |
| ( |
| "api/v1/dags?offset=5&dag_id_pattern=~", |
| ["TEST_DAG_5", "TEST_DAG_6", "TEST_DAG_7", "TEST_DAG_8", "TEST_DAG_9"], |
| ), |
| ( |
| "api/v1/dags?offset=0&dag_id_pattern=~", |
| [ |
| "TEST_DAG_1", |
| "TEST_DAG_10", |
| "TEST_DAG_2", |
| "TEST_DAG_3", |
| "TEST_DAG_4", |
| "TEST_DAG_5", |
| "TEST_DAG_6", |
| "TEST_DAG_7", |
| "TEST_DAG_8", |
| "TEST_DAG_9", |
| ], |
| ), |
| ("api/v1/dags?limit=1&offset=5&dag_id_pattern=~", ["TEST_DAG_5"]), |
| ("api/v1/dags?limit=1&offset=1&dag_id_pattern=~", ["TEST_DAG_10"]), |
| ("api/v1/dags?limit=2&offset=2&dag_id_pattern=~", ["TEST_DAG_2", "TEST_DAG_3"]), |
| ], |
| ) |
| def test_should_respond_200_and_handle_pagination(self, url, expected_dag_ids): |
| self._create_dag_models(10) |
| |
| response = self.client.patch( |
| url, |
| json={ |
| "is_paused": False, |
| }, |
| environ_overrides={"REMOTE_USER": "test"}, |
| ) |
| |
| assert response.status_code == 200 |
| |
| dag_ids = [dag["dag_id"] for dag in response.json["dags"]] |
| |
| assert expected_dag_ids == dag_ids |
| assert 10 == response.json["total_entries"] |
| |
| def test_should_respond_200_default_limit(self): |
| self._create_dag_models(101) |
| |
| response = self.client.patch( |
| "api/v1/dags?dag_id_pattern=~", |
| json={ |
| "is_paused": False, |
| }, |
| environ_overrides={"REMOTE_USER": "test"}, |
| ) |
| |
| assert response.status_code == 200 |
| |
| assert 100 == len(response.json["dags"]) |
| assert 101 == response.json["total_entries"] |
| |
| def test_should_raises_401_unauthenticated(self): |
| response = self.client.patch( |
| "api/v1/dags?dag_id_pattern=~", |
| json={ |
| "is_paused": False, |
| }, |
| ) |
| |
| assert_401(response) |
| |
| def test_should_respond_403_unauthorized(self): |
| self._create_dag_models(1) |
| response = self.client.patch( |
| "api/v1/dags?dag_id_pattern=~", |
| json={ |
| "is_paused": False, |
| }, |
| environ_overrides={"REMOTE_USER": "test_no_permissions"}, |
| ) |
| |
| assert response.status_code == 403 |
| |
| def test_should_respond_200_and_pause_dags(self, url_safe_serializer): |
| file_token = url_safe_serializer.dumps("/tmp/dag_1.py") |
| file_token2 = url_safe_serializer.dumps("/tmp/dag_2.py") |
| self._create_dag_models(2) |
| |
| response = self.client.patch( |
| "/api/v1/dags?dag_id_pattern=~", |
| json={ |
| "is_paused": True, |
| }, |
| environ_overrides={"REMOTE_USER": "test"}, |
| ) |
| |
| assert response.status_code == 200 |
| assert { |
| "dags": [ |
| { |
| "dag_id": "TEST_DAG_1", |
| "description": None, |
| "fileloc": "/tmp/dag_1.py", |
| "file_token": file_token, |
| "is_paused": True, |
| "is_active": True, |
| "is_subdag": False, |
| "owners": [], |
| "root_dag_id": None, |
| "schedule_interval": { |
| "__type": "CronExpression", |
| "value": "2 2 * * *", |
| }, |
| "tags": [], |
| "next_dagrun": None, |
| "has_task_concurrency_limits": True, |
| "next_dagrun_data_interval_start": None, |
| "next_dagrun_data_interval_end": None, |
| "max_active_runs": 16, |
| "next_dagrun_create_after": None, |
| "last_expired": None, |
| "max_active_tasks": 16, |
| "last_pickled": None, |
| "default_view": None, |
| "last_parsed_time": None, |
| "scheduler_lock": None, |
| "timetable_description": None, |
| "has_import_errors": False, |
| "pickle_id": None, |
| }, |
| { |
| "dag_id": "TEST_DAG_2", |
| "description": None, |
| "fileloc": "/tmp/dag_2.py", |
| "file_token": file_token2, |
| "is_paused": True, |
| "is_active": True, |
| "is_subdag": False, |
| "owners": [], |
| "root_dag_id": None, |
| "schedule_interval": { |
| "__type": "CronExpression", |
| "value": "2 2 * * *", |
| }, |
| "tags": [], |
| "next_dagrun": None, |
| "has_task_concurrency_limits": True, |
| "next_dagrun_data_interval_start": None, |
| "next_dagrun_data_interval_end": None, |
| "max_active_runs": 16, |
| "next_dagrun_create_after": None, |
| "last_expired": None, |
| "max_active_tasks": 16, |
| "last_pickled": None, |
| "default_view": None, |
| "last_parsed_time": None, |
| "scheduler_lock": None, |
| "timetable_description": None, |
| "has_import_errors": False, |
| "pickle_id": None, |
| }, |
| ], |
| "total_entries": 2, |
| } == response.json |
| |
| @provide_session |
| def test_should_respond_200_and_pause_dag_pattern(self, session, url_safe_serializer): |
| file_token = url_safe_serializer.dumps("/tmp/dag_1.py") |
| self._create_dag_models(10) |
| file_token10 = url_safe_serializer.dumps("/tmp/dag_10.py") |
| |
| response = self.client.patch( |
| "/api/v1/dags?dag_id_pattern=TEST_DAG_1", |
| json={ |
| "is_paused": True, |
| }, |
| environ_overrides={"REMOTE_USER": "test"}, |
| ) |
| |
| assert response.status_code == 200 |
| assert { |
| "dags": [ |
| { |
| "dag_id": "TEST_DAG_1", |
| "description": None, |
| "fileloc": "/tmp/dag_1.py", |
| "file_token": file_token, |
| "is_paused": True, |
| "is_active": True, |
| "is_subdag": False, |
| "owners": [], |
| "root_dag_id": None, |
| "schedule_interval": { |
| "__type": "CronExpression", |
| "value": "2 2 * * *", |
| }, |
| "tags": [], |
| "next_dagrun": None, |
| "has_task_concurrency_limits": True, |
| "next_dagrun_data_interval_start": None, |
| "next_dagrun_data_interval_end": None, |
| "max_active_runs": 16, |
| "next_dagrun_create_after": None, |
| "last_expired": None, |
| "max_active_tasks": 16, |
| "last_pickled": None, |
| "default_view": None, |
| "last_parsed_time": None, |
| "scheduler_lock": None, |
| "timetable_description": None, |
| "has_import_errors": False, |
| "pickle_id": None, |
| }, |
| { |
| "dag_id": "TEST_DAG_10", |
| "description": None, |
| "fileloc": "/tmp/dag_10.py", |
| "file_token": file_token10, |
| "is_paused": True, |
| "is_active": True, |
| "is_subdag": False, |
| "owners": [], |
| "root_dag_id": None, |
| "schedule_interval": { |
| "__type": "CronExpression", |
| "value": "2 2 * * *", |
| }, |
| "tags": [], |
| "next_dagrun": None, |
| "has_task_concurrency_limits": True, |
| "next_dagrun_data_interval_start": None, |
| "next_dagrun_data_interval_end": None, |
| "max_active_runs": 16, |
| "next_dagrun_create_after": None, |
| "last_expired": None, |
| "max_active_tasks": 16, |
| "last_pickled": None, |
| "default_view": None, |
| "last_parsed_time": None, |
| "scheduler_lock": None, |
| "timetable_description": None, |
| "has_import_errors": False, |
| "pickle_id": None, |
| }, |
| ], |
| "total_entries": 2, |
| } == response.json |
| |
| dags_not_updated = session.query(DagModel).filter(~DagModel.is_paused) |
| assert len(dags_not_updated.all()) == 8 |
| dags_updated = session.query(DagModel).filter(DagModel.is_paused) |
| assert len(dags_updated.all()) == 2 |
| |
| @provide_session |
| def test_should_respond_200_and_reverse_ordering(self, session, url_safe_serializer): |
| file_token = url_safe_serializer.dumps("/tmp/dag_1.py") |
| self._create_dag_models(2) |
| file_token10 = url_safe_serializer.dumps("/tmp/dag_2.py") |
| |
| response = self.client.get( |
| "/api/v1/dags?order_by=-dag_id", |
| environ_overrides={"REMOTE_USER": "test"}, |
| ) |
| |
| assert response.status_code == 200 |
| assert { |
| "dags": [ |
| { |
| "dag_id": "TEST_DAG_2", |
| "description": None, |
| "fileloc": "/tmp/dag_2.py", |
| "file_token": file_token10, |
| "is_paused": False, |
| "is_active": True, |
| "is_subdag": False, |
| "owners": [], |
| "root_dag_id": None, |
| "schedule_interval": { |
| "__type": "CronExpression", |
| "value": "2 2 * * *", |
| }, |
| "tags": [], |
| "next_dagrun": None, |
| "has_task_concurrency_limits": True, |
| "next_dagrun_data_interval_start": None, |
| "next_dagrun_data_interval_end": None, |
| "max_active_runs": 16, |
| "next_dagrun_create_after": None, |
| "last_expired": None, |
| "max_active_tasks": 16, |
| "last_pickled": None, |
| "default_view": None, |
| "last_parsed_time": None, |
| "scheduler_lock": None, |
| "timetable_description": None, |
| "has_import_errors": False, |
| "pickle_id": None, |
| }, |
| { |
| "dag_id": "TEST_DAG_1", |
| "description": None, |
| "fileloc": "/tmp/dag_1.py", |
| "file_token": file_token, |
| "is_paused": False, |
| "is_active": True, |
| "is_subdag": False, |
| "owners": [], |
| "root_dag_id": None, |
| "schedule_interval": { |
| "__type": "CronExpression", |
| "value": "2 2 * * *", |
| }, |
| "tags": [], |
| "next_dagrun": None, |
| "has_task_concurrency_limits": True, |
| "next_dagrun_data_interval_start": None, |
| "next_dagrun_data_interval_end": None, |
| "max_active_runs": 16, |
| "next_dagrun_create_after": None, |
| "last_expired": None, |
| "max_active_tasks": 16, |
| "last_pickled": None, |
| "default_view": None, |
| "last_parsed_time": None, |
| "scheduler_lock": None, |
| "timetable_description": None, |
| "has_import_errors": False, |
| "pickle_id": None, |
| }, |
| ], |
| "total_entries": 2, |
| } == response.json |
| |
| def test_should_respons_400_dag_id_pattern_missing(self): |
| self._create_dag_models(1) |
| response = self.client.patch( |
| "/api/v1/dags?only_active=True", |
| json={ |
| "is_paused": False, |
| }, |
| environ_overrides={"REMOTE_USER": "test"}, |
| ) |
| assert response.status_code == 400 |
