mik-laj commented on a change in pull request #9153:
URL: https://github.com/apache/airflow/pull/9153#discussion_r438013490
##########
File path: tests/api_connexion/endpoints/test_dag_run_endpoint.py
##########
@@ -15,41 +15,347 @@
# specific language governing permissions and limitations
# under the License.
import unittest
+from datetime import timedelta
import pytest
+from parameterized import parameterized
+from airflow.models import DagRun
+from airflow.utils import timezone
+from airflow.utils.session import provide_session
+from airflow.utils.types import DagRunType
from airflow.www import app
+from tests.test_utils.db import clear_db_runs
class TestDagRunEndpoint(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
super().setUpClass()
+
cls.app = app.create_app(testing=True) # type:ignore
def setUp(self) -> None:
self.client = self.app.test_client() # type:ignore
+ self.now = '2020-06-11T18:12:50.773601+00:00'
+ self.now2 = '2020-06-12T18:12:50.773601+00:00'
+ clear_db_runs()
+
+ def tearDown(self) -> None:
+ clear_db_runs()
+
+ def _create_test_dag_run(self, state='running'):
+ dagrun_model_1 = DagRun(
+ dag_id='TEST_DAG_ID',
+ run_id='TEST_DAG_RUN_ID_1',
+ run_type=DagRunType.MANUAL.value,
+ execution_date=timezone.parse(self.now),
+ start_date=timezone.parse(self.now),
+ external_trigger=True,
+ state=state
+ )
+ dagrun_model_2 = DagRun(
+ dag_id='TEST_DAG_ID',
+ run_id='TEST_DAG_RUN_ID_2',
+ run_type=DagRunType.MANUAL.value,
+ execution_date=timezone.parse(self.now2),
+ start_date=timezone.parse(self.now),
+ external_trigger=True,
+ )
+ return [dagrun_model_1, dagrun_model_2]
class TestDeleteDagRun(TestDagRunEndpoint):
@pytest.mark.skip(reason="Not implemented yet")
def test_should_response_200(self):
- response =
self.client.delete("/dags/TEST_DAG_ID}/dagRuns/TEST_DAG_RUN_ID")
+ response =
self.client.delete("api/v1/dags/TEST_DAG_ID}/dagRuns/TEST_DAG_RUN_ID")
assert response.status_code == 204
class TestGetDagRun(TestDagRunEndpoint):
- @pytest.mark.skip(reason="Not implemented yet")
- def test_should_response_200(self):
- response = self.client.get("/dags/TEST_DAG_ID/dagRuns/TEST_DAG_RUN_ID")
+
+ @provide_session
+ def test_should_response_200(self, session):
+ dagrun_model = DagRun(
+ dag_id='TEST_DAG_ID',
+ run_id='TEST_DAG_RUN_ID',
+ run_type=DagRunType.MANUAL.value,
+ execution_date=timezone.parse(self.now),
+ start_date=timezone.parse(self.now),
+ external_trigger=True,
+ )
+ session.add(dagrun_model)
+ session.commit()
+ result = session.query(DagRun).all()
+ assert len(result) == 1
+ response =
self.client.get("api/v1/dags/TEST_DAG_ID/dagRuns/TEST_DAG_RUN_ID")
assert response.status_code == 200
+ self.assertEqual(
+ response.json,
+ {
+ 'dag_id': 'TEST_DAG_ID',
+ 'dag_run_id': 'TEST_DAG_RUN_ID',
+ 'end_date': None,
+ 'state': 'running',
+ 'execution_date': self.now,
+ 'external_trigger': True,
+ 'start_date': self.now,
+ 'conf': {},
+ }
+ )
+
+ def test_should_response_404(self):
+ response = self.client.get("api/v1/dags/invalid-id/dagRuns/invalid-id")
+ assert response.status_code == 404
+ self.assertEqual(
+ {
+ 'detail': None,
+ 'status': 404,
+ 'title': 'DAGRun not found',
+ 'type': 'about:blank'
+ },
+ response.json
+ )
class TestGetDagRuns(TestDagRunEndpoint):
- @pytest.mark.skip(reason="Not implemented yet")
- def test_should_response_200(self):
- response = self.client.get("/dags/TEST_DAG_ID/dagRuns/")
+
+ @provide_session
+ def test_should_response_200(self, session):
+ dagruns = self._create_test_dag_run()
+ session.add_all(dagruns)
+ session.commit()
+ result = session.query(DagRun).all()
+ assert len(result) == 2
+ response = self.client.get("api/v1/dags/TEST_DAG_ID/dagRuns")
+ assert response.status_code == 200
+ self.assertEqual(
+ response.json,
+ {
+ "dag_runs": [
+ {
+ 'dag_id': 'TEST_DAG_ID',
+ 'dag_run_id': 'TEST_DAG_RUN_ID_1',
+ 'end_date': None,
+ 'state': 'running',
+ 'execution_date': self.now,
+ 'external_trigger': True,
+ 'start_date': self.now,
+ 'conf': {},
+ },
+ {
+ 'dag_id': 'TEST_DAG_ID',
+ 'dag_run_id': 'TEST_DAG_RUN_ID_2',
+ 'end_date': None,
+ 'state': 'running',
+ 'execution_date': self.now2,
+ 'external_trigger': True,
+ 'start_date': self.now,
+ 'conf': {},
+ }
+ ],
+ "total_entries": 2
+ }
+ )
+
+ @provide_session
+ def test_should_return_all_with_tilde_as_dag_id(self, session):
+ dagruns = self._create_test_dag_run()
+ session.add_all(dagruns)
Review comment:
~ means that you can fetch any DAG ID. Your assertion should check if
you have two DAGs e.g. DAG A and DAG B, you can fetch them. For now, you only
have one DAG, which does not allow you to check the behavior of this view.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
