airflow-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From maximebeauche...@apache.org
Subject [1/2] incubator-airflow git commit: [AIRFLOW-2267] Airflow DAG level access
Date Mon, 16 Jul 2018 20:13:49 GMT
Repository: incubator-airflow
Updated Branches:
  refs/heads/master 17ddbcff0 -> f3f2eb323


http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/f3f2eb32/tests/www_rbac/test_views.py
----------------------------------------------------------------------
diff --git a/tests/www_rbac/test_views.py b/tests/www_rbac/test_views.py
index 0c25972..b4a2f02 100644
--- a/tests/www_rbac/test_views.py
+++ b/tests/www_rbac/test_views.py
@@ -29,12 +29,11 @@ import unittest
 import urllib
 
 from flask._compat import PY2
-from flask_appbuilder.security.sqla.models import User as ab_user
 from urllib.parse import quote_plus
 from werkzeug.test import Client
 
 from airflow import configuration as conf
-from airflow import models
+from airflow import models, settings
 from airflow.config_templates.airflow_local_settings import DEFAULT_LOGGING_CONFIG
 from airflow.models import DAG, DagRun, TaskInstance
 from airflow.operators.dummy_operator import DummyOperator
@@ -48,17 +47,17 @@ from airflow.www_rbac import app as application
 class TestBase(unittest.TestCase):
     def setUp(self):
         conf.load_test_config()
-        self.app, self.appbuilder = application.create_app(testing=True)
+        self.app, self.appbuilder = application.create_app(session=Session, testing=True)
         self.app.config['WTF_CSRF_ENABLED'] = False
         self.client = self.app.test_client()
-        self.session = Session()
+        settings.configure_orm()
+        self.session = Session
         self.login()
 
     def login(self):
-        sm_session = self.appbuilder.sm.get_session()
-        self.user = sm_session.query(ab_user).first()
-        if not self.user:
-            role_admin = self.appbuilder.sm.find_role('Admin')
+        role_admin = self.appbuilder.sm.find_role('Admin')
+        tester = self.appbuilder.sm.find_user(username='test')
+        if not tester:
             self.appbuilder.sm.add_user(
                 username='test',
                 first_name='test',
@@ -88,6 +87,15 @@ class TestBase(unittest.TestCase):
         else:
             self.assertIn(text, resp_html)
 
+    def check_content_not_in_response(self, text, resp, resp_code=200):
+        resp_html = resp.data.decode('utf-8')
+        self.assertEqual(resp_code, resp.status_code)
+        if isinstance(text, list):
+            for kw in text:
+                self.assertNotIn(kw, resp_html)
+        else:
+            self.assertNotIn(text, resp_html)
+
     def percent_encode(self, obj):
         if PY2:
             return urllib.quote_plus(str(obj))
@@ -137,10 +145,6 @@ class TestVariableModelView(TestBase):
         resp = self.client.post('/variable/add',
                                 data=self.variable,
                                 follow_redirects=True)
-        self.assertEqual(resp.status_code, 200)
-        v = self.session.query(models.Variable).first()
-        self.assertEqual(v.key, 'test_key')
-        self.assertEqual(v.val, 'text_val')
 
         # update the variable with a wrong value, given that is encrypted
         Var = models.Variable
@@ -248,6 +252,8 @@ class TestAirflowBaseViews(TestBase):
 
     def setUp(self):
         super(TestAirflowBaseViews, self).setUp()
+        self.logout()
+        self.login()
         self.cleanup_dagruns()
         self.prepare_dagruns()
 
@@ -292,7 +298,7 @@ class TestAirflowBaseViews(TestBase):
         self.check_content_in_response('DAGs', resp)
 
     def test_health(self):
-        resp = self.client.get('health')
+        resp = self.client.get('health', follow_redirects=True)
         self.check_content_in_response('The server is healthy!', resp)
 
     def test_home(self):
@@ -361,7 +367,7 @@ class TestAirflowBaseViews(TestBase):
         self.check_content_in_response('example_bash_operator', resp)
 
     def test_landing_times(self):
-        url = 'landing_times?days=30&dag_id=test_example_bash_operator'
+        url = 'landing_times?days=30&dag_id=example_bash_operator'
         resp = self.client.get(url, follow_redirects=True)
         self.check_content_in_response('example_bash_operator', resp)
 
@@ -415,6 +421,8 @@ class TestAirflowBaseViews(TestBase):
 
 class TestConfigurationView(TestBase):
     def test_configuration(self):
+        self.logout()
+        self.login()
         resp = self.client.get('configuration', follow_redirects=True)
         self.check_content_in_response(
             ['Airflow Configuration', 'Running Configuration'], resp)
@@ -437,6 +445,7 @@ class TestLogView(TestBase):
         current_dir = os.path.dirname(os.path.abspath(__file__))
         logging_config['handlers']['task']['base_log_folder'] = os.path.normpath(
             os.path.join(current_dir, 'test_logs'))
+
         logging_config['handlers']['task']['filename_template'] = \
             '{{ ti.dag_id }}/{{ ti.task_id }}/' \
             '{{ ts | replace(":", ".") }}/{{ try_number }}.log'
@@ -450,11 +459,12 @@ class TestLogView(TestBase):
         sys.path.append(self.settings_folder)
         conf.set('core', 'logging_config_class', 'airflow_local_settings.LOGGING_CONFIG')
 
-        self.app, self.appbuilder = application.create_app(testing=True)
+        self.app, self.appbuilder = application.create_app(session=Session, testing=True)
         self.app.config['WTF_CSRF_ENABLED'] = False
         self.client = self.app.test_client()
+        settings.configure_orm()
+        self.session = Session
         self.login()
-        self.session = Session()
 
         from airflow.www_rbac.views import dagbag
         dag = DAG(self.DAG_ID, start_date=self.DEFAULT_DATE)
@@ -477,9 +487,9 @@ class TestLogView(TestBase):
 
     def test_get_file_task_log(self):
         response = self.client.get(
-            TestLogView.ENDPOINT,
-            follow_redirects=True,
-        )
+            TestLogView.ENDPOINT, data=dict(
+                username='test',
+                password='test'), follow_redirects=True)
         self.assertEqual(response.status_code, 200)
         self.assertIn('Log by attempts',
                       response.data.decode('utf-8'))
@@ -493,7 +503,10 @@ class TestLogView(TestBase):
                                                 self.TASK_ID,
                                                 quote_plus(self.DEFAULT_DATE.isoformat()),
                                                 1,
-                                                json.dumps({})), follow_redirects=True)
+                                                json.dumps({})), data=dict(
+                                                    username='test',
+                                                    password='test'),
+                            follow_redirects=True)
 
         self.assertIn('"message":', response.data.decode('utf-8'))
         self.assertIn('"metadata":', response.data.decode('utf-8'))
@@ -508,7 +521,10 @@ class TestLogView(TestBase):
             self.client.get(url_template.format(self.DAG_ID,
                                                 self.TASK_ID,
                                                 quote_plus(self.DEFAULT_DATE.isoformat()),
-                                                1), follow_redirects=True)
+                                                1), data=dict(
+                                                    username='test',
+                                                    password='test'),
+                            follow_redirects=True)
 
         self.assertIn('"message":', response.data.decode('utf-8'))
         self.assertIn('"metadata":', response.data.decode('utf-8'))
@@ -518,7 +534,10 @@ class TestLogView(TestBase):
 
 class TestVersionView(TestBase):
     def test_version(self):
-        resp = self.client.get('version', follow_redirects=True)
+        resp = self.client.get('version', data=dict(
+            username='test',
+            password='test'
+        ), follow_redirects=True)
         self.check_content_in_response('Version Info', resp)
 
 
@@ -582,8 +601,9 @@ class ViewWithDateTimeAndNumRunsAndDagRunsFormTester:
         Should set base date to current date (not asserted)
         """
         response = self.test.client.get(
-            self.endpoint
-        )
+            self.endpoint, data=dict(
+                username='test',
+                password='test'), follow_redirects=True)
         self.test.assertEqual(response.status_code, 200)
         data = response.data.decode('utf-8')
         self.test.assertIn('Base date:', data)
@@ -603,7 +623,11 @@ class ViewWithDateTimeAndNumRunsAndDagRunsFormTester:
         """
         response = self.test.client.get(
             self.endpoint + '&execution_date={}'.format(
-                self.runs[1].execution_date.isoformat())
+                self.runs[1].execution_date.isoformat()),
+            data=dict(
+                username='test',
+                password='test'
+            ), follow_redirects=True
         )
         self.test.assertEqual(response.status_code, 200)
         data = response.data.decode('utf-8')
@@ -626,7 +650,11 @@ class ViewWithDateTimeAndNumRunsAndDagRunsFormTester:
         """
         response = self.test.client.get(
             self.endpoint + '&base_date={}&num_runs=2'.format(
-                self.runs[1].execution_date.isoformat())
+                self.runs[1].execution_date.isoformat()),
+            data=dict(
+                username='test',
+                password='test'
+            ), follow_redirects=True
         )
         self.test.assertEqual(response.status_code, 200)
         data = response.data.decode('utf-8')
@@ -648,7 +676,11 @@ class ViewWithDateTimeAndNumRunsAndDagRunsFormTester:
         response = self.test.client.get(
             self.endpoint + '&base_date={}&num_runs=42&execution_date={}'.format(
                 self.runs[1].execution_date.isoformat(),
-                self.runs[0].execution_date.isoformat())
+                self.runs[0].execution_date.isoformat()),
+            data=dict(
+                username='test',
+                password='test'
+            ), follow_redirects=True
         )
         self.test.assertEqual(response.status_code, 200)
         data = response.data.decode('utf-8')
@@ -670,7 +702,11 @@ class ViewWithDateTimeAndNumRunsAndDagRunsFormTester:
         response = self.test.client.get(
             self.endpoint + '&base_date={}&num_runs=5&execution_date={}'.format(
                 self.runs[2].execution_date.isoformat(),
-                self.runs[3].execution_date.isoformat())
+                self.runs[3].execution_date.isoformat()),
+            data=dict(
+                username='test',
+                password='test'
+            ), follow_redirects=True
         )
         self.test.assertEqual(response.status_code, 200)
         data = response.data.decode('utf-8')
@@ -759,5 +795,505 @@ class TestGanttView(TestBase):
         self.tester.test_with_base_date_and_num_runs_and_execution_date_within()
 
 
+class TestDagACLView(TestBase):
+    """
+    Test Airflow DAG acl
+    """
+    default_date = timezone.datetime(2018, 6, 1)
+    run_id = "test_{}".format(models.DagRun.id_for_date(default_date))
+
+    @classmethod
+    def setUpClass(cls):
+        super(TestDagACLView, cls).setUpClass()
+
+    def cleanup_dagruns(self):
+        DR = models.DagRun
+        dag_ids = ['example_bash_operator',
+                   'example_subdag_operator']
+        (self.session
+             .query(DR)
+             .filter(DR.dag_id.in_(dag_ids))
+             .filter(DR.run_id == self.run_id)
+             .delete(synchronize_session='fetch'))
+        self.session.commit()
+
+    def prepare_dagruns(self):
+        dagbag = models.DagBag(include_examples=True)
+        self.bash_dag = dagbag.dags['example_bash_operator']
+        self.sub_dag = dagbag.dags['example_subdag_operator']
+
+        self.bash_dagrun = self.bash_dag.create_dagrun(
+            run_id=self.run_id,
+            execution_date=self.default_date,
+            start_date=timezone.utcnow(),
+            state=State.RUNNING)
+
+        self.sub_dagrun = self.sub_dag.create_dagrun(
+            run_id=self.run_id,
+            execution_date=self.default_date,
+            start_date=timezone.utcnow(),
+            state=State.RUNNING)
+
+    def setUp(self):
+        super(TestDagACLView, self).setUp()
+        self.cleanup_dagruns()
+        self.prepare_dagruns()
+        self.logout()
+        self.appbuilder.sm.sync_roles()
+        self.add_permission_for_role()
+
+    def login(self, username=None, password=None):
+        role_admin = self.appbuilder.sm.find_role('Admin')
+        tester = self.appbuilder.sm.find_user(username='test')
+        if not tester:
+            self.appbuilder.sm.add_user(
+                username='test',
+                first_name='test',
+                last_name='test',
+                email='test@fab.org',
+                role=role_admin,
+                password='test')
+
+        dag_acl_role = self.appbuilder.sm.add_role('dag_acl_tester')
+        dag_tester = self.appbuilder.sm.find_user(username='dag_tester')
+        if not dag_tester:
+            self.appbuilder.sm.add_user(
+                username='dag_tester',
+                first_name='dag_test',
+                last_name='dag_test',
+                email='dag_test@fab.org',
+                role=dag_acl_role,
+                password='dag_test')
+
+        # create an user without permission
+        dag_no_role = self.appbuilder.sm.add_role('dag_acl_faker')
+        dag_faker = self.appbuilder.sm.find_user(username='dag_faker')
+        if not dag_faker:
+            self.appbuilder.sm.add_user(
+                username='dag_faker',
+                first_name='dag_faker',
+                last_name='dag_faker',
+                email='dag_fake@fab.org',
+                role=dag_no_role,
+                password='dag_faker')
+
+        # create an user with only read permission
+        dag_read_only_role = self.appbuilder.sm.add_role('dag_acl_read_only')
+        dag_read_only = self.appbuilder.sm.find_user(username='dag_read_only')
+        if not dag_read_only:
+            self.appbuilder.sm.add_user(
+                username='dag_read_only',
+                first_name='dag_read_only',
+                last_name='dag_read_only',
+                email='dag_read_only@fab.org',
+                role=dag_read_only_role,
+                password='dag_read_only')
+
+        # create an user that has all dag access
+        all_dag_role = self.appbuilder.sm.add_role('all_dag_role')
+        all_dag_tester = self.appbuilder.sm.find_user(username='all_dag_user')
+        if not all_dag_tester:
+            self.appbuilder.sm.add_user(
+                username='all_dag_user',
+                first_name='all_dag_user',
+                last_name='all_dag_user',
+                email='all_dag_user@fab.org',
+                role=all_dag_role,
+                password='all_dag_user')
+
+        user = username if username else 'dag_tester'
+        passwd = password if password else 'dag_test'
+
+        return self.client.post('/login/', data=dict(
+            username=user,
+            password=passwd
+        ))
+
+    def logout(self):
+        return self.client.get('/logout/')
+
+    def add_permission_for_role(self):
+        self.logout()
+        self.login(username='test',
+                   password='test')
+        perm_on_dag = self.appbuilder.sm.\
+            find_permission_view_menu('can_dag_edit', 'example_bash_operator')
+        dag_tester_role = self.appbuilder.sm.find_role('dag_acl_tester')
+        self.appbuilder.sm.add_permission_role(dag_tester_role, perm_on_dag)
+
+        perm_on_all_dag = self.appbuilder.sm.\
+            find_permission_view_menu('can_dag_edit', 'all_dags')
+        all_dag_role = self.appbuilder.sm.find_role('all_dag_role')
+        self.appbuilder.sm.add_permission_role(all_dag_role, perm_on_all_dag)
+
+        read_only_perm_on_dag = self.appbuilder.sm.\
+            find_permission_view_menu('can_dag_read', 'example_bash_operator')
+        dag_read_only_role = self.appbuilder.sm.find_role('dag_acl_read_only')
+        self.appbuilder.sm.add_permission_role(dag_read_only_role, read_only_perm_on_dag)
+
+    def test_permission_exist(self):
+        self.logout()
+        self.login(username='test',
+                   password='test')
+        test_view_menu = self.appbuilder.sm.find_view_menu('example_bash_operator')
+        perms_views = self.appbuilder.sm.find_permissions_view_menu(test_view_menu)
+        self.assertEqual(len(perms_views), 2)
+        # each dag view will create one write, and one read permission
+        self.assertTrue(str(perms_views[0]).startswith('can dag'))
+        self.assertTrue(str(perms_views[1]).startswith('can dag'))
+
+    def test_role_permission_associate(self):
+        self.logout()
+        self.login(username='test',
+                   password='test')
+        test_role = self.appbuilder.sm.find_role('dag_acl_tester')
+        perms = set([str(perm) for perm in test_role.permissions])
+        self.assertIn('can dag edit on example_bash_operator', perms)
+        self.assertNotIn('can dag read on example_bash_operator', perms)
+
+    def test_index_success(self):
+        self.logout()
+        self.login()
+        resp = self.client.get('/', follow_redirects=True)
+        self.check_content_in_response('example_bash_operator', resp)
+
+    def test_index_failure(self):
+        self.logout()
+        self.login()
+        resp = self.client.get('/', follow_redirects=True)
+        # The user can only access/view example_bash_operator dag.
+        self.check_content_not_in_response('example_subdag_operator', resp)
+
+    def test_index_for_all_dag_user(self):
+        self.logout()
+        self.login(username='all_dag_user',
+                   password='all_dag_user')
+        resp = self.client.get('/', follow_redirects=True)
+        # The all dag user can access/view all dags.
+        self.check_content_in_response('example_subdag_operator', resp)
+        self.check_content_in_response('example_bash_operator', resp)
+
+    def test_dag_stats_success(self):
+        self.logout()
+        self.login()
+        resp = self.client.get('dag_stats', follow_redirects=True)
+        self.check_content_in_response('example_bash_operator', resp)
+
+    def test_dag_stats_failure(self):
+        self.logout()
+        self.login()
+        resp = self.client.get('dag_stats', follow_redirects=True)
+        self.check_content_not_in_response('example_subdag_operator', resp)
+
+    def test_dag_stats_success_for_all_dag_user(self):
+        self.logout()
+        self.login(username='all_dag_user',
+                   password='all_dag_user')
+        resp = self.client.get('dag_stats', follow_redirects=True)
+        self.check_content_in_response('example_subdag_operator', resp)
+        self.check_content_in_response('example_bash_operator', resp)
+
+    def test_task_stats_success(self):
+        self.logout()
+        self.login()
+        resp = self.client.get('task_stats', follow_redirects=True)
+        self.check_content_in_response('example_bash_operator', resp)
+
+    def test_task_stats_failure(self):
+        self.logout()
+        self.login()
+        resp = self.client.get('task_stats', follow_redirects=True)
+        self.check_content_not_in_response('example_subdag_operator', resp)
+
+    def test_task_stats_success_for_all_dag_user(self):
+        self.logout()
+        self.login(username='all_dag_user',
+                   password='all_dag_user')
+        resp = self.client.get('task_stats', follow_redirects=True)
+        self.check_content_in_response('example_bash_operator', resp)
+        self.check_content_in_response('example_subdag_operator', resp)
+
+    def test_code_success(self):
+        self.logout()
+        self.login()
+        url = 'code?dag_id=example_bash_operator'
+        resp = self.client.get(url, follow_redirects=True)
+        self.check_content_in_response('example_bash_operator', resp)
+
+    def test_code_failure(self):
+        self.logout()
+        self.login(username='dag_faker',
+                   password='dag_faker')
+        url = 'code?dag_id=example_bash_operator'
+        resp = self.client.get(url, follow_redirects=True)
+        self.check_content_not_in_response('example_bash_operator', resp)
+
+    def test_code_success_for_all_dag_user(self):
+        self.logout()
+        self.login(username='all_dag_user',
+                   password='all_dag_user')
+        url = 'code?dag_id=example_bash_operator'
+        resp = self.client.get(url, follow_redirects=True)
+        self.check_content_in_response('example_bash_operator', resp)
+
+        url = 'code?dag_id=example_subdag_operator'
+        resp = self.client.get(url, follow_redirects=True)
+        self.check_content_in_response('example_subdag_operator', resp)
+
+    def test_dag_details_success(self):
+        self.logout()
+        self.login()
+        url = 'dag_details?dag_id=example_bash_operator'
+        resp = self.client.get(url, follow_redirects=True)
+        self.check_content_in_response('DAG details', resp)
+
+    def test_dag_details_failure(self):
+        self.logout()
+        self.login(username='dag_faker',
+                   password='dag_faker')
+        url = 'dag_details?dag_id=example_bash_operator'
+        resp = self.client.get(url, follow_redirects=True)
+        self.check_content_not_in_response('DAG details', resp)
+
+    def test_dag_details_success_for_all_dag_user(self):
+        self.logout()
+        self.login(username='all_dag_user',
+                   password='all_dag_user')
+        url = 'dag_details?dag_id=example_bash_operator'
+        resp = self.client.get(url, follow_redirects=True)
+        self.check_content_in_response('example_bash_operator', resp)
+
+        url = 'dag_details?dag_id=example_subdag_operator'
+        resp = self.client.get(url, follow_redirects=True)
+        self.check_content_in_response('example_subdag_operator', resp)
+
+    def test_pickle_info_success(self):
+        self.logout()
+        self.login()
+        url = 'pickle_info?dag_id=example_bash_operator'
+        resp = self.client.get(url, follow_redirects=True)
+        self.assertEqual(resp.status_code, 200)
+
+    def test_rendered_success(self):
+        self.logout()
+        self.login()
+        url = ('rendered?task_id=runme_0&dag_id=example_bash_operator&execution_date={}'
+               .format(self.percent_encode(self.default_date)))
+        resp = self.client.get(url, follow_redirects=True)
+        self.check_content_in_response('Rendered Template', resp)
+
+    def test_rendered_failure(self):
+        self.logout()
+        self.login(username='dag_faker',
+                   password='dag_faker')
+        url = ('rendered?task_id=runme_0&dag_id=example_bash_operator&execution_date={}'
+               .format(self.percent_encode(self.default_date)))
+        resp = self.client.get(url, follow_redirects=True)
+        self.check_content_not_in_response('Rendered Template', resp)
+
+    def test_rendered_success_for_all_dag_user(self):
+        self.logout()
+        self.login(username='all_dag_user',
+                   password='all_dag_user')
+        url = ('rendered?task_id=runme_0&dag_id=example_bash_operator&execution_date={}'
+               .format(self.percent_encode(self.default_date)))
+        resp = self.client.get(url, follow_redirects=True)
+        self.check_content_in_response('Rendered Template', resp)
+
+    def test_task_success(self):
+        self.logout()
+        self.login()
+        url = ('task?task_id=runme_0&dag_id=example_bash_operator&execution_date={}'
+               .format(self.percent_encode(self.default_date)))
+        resp = self.client.get(url, follow_redirects=True)
+        self.check_content_in_response('Task Instance Details', resp)
+
+    def test_task_failure(self):
+        self.logout()
+        self.login(username='dag_faker',
+                   password='dag_faker')
+        url = ('task?task_id=runme_0&dag_id=example_bash_operator&execution_date={}'
+               .format(self.percent_encode(self.default_date)))
+        resp = self.client.get(url, follow_redirects=True)
+        self.check_content_not_in_response('Task Instance Details', resp)
+
+    def test_task_success_for_all_dag_user(self):
+        self.logout()
+        self.login(username='all_dag_user',
+                   password='all_dag_user')
+        url = ('task?task_id=runme_0&dag_id=example_bash_operator&execution_date={}'
+               .format(self.percent_encode(self.default_date)))
+        resp = self.client.get(url, follow_redirects=True)
+        self.check_content_in_response('Task Instance Details', resp)
+
+    def test_xcom_success(self):
+        self.logout()
+        self.login()
+        url = ('xcom?task_id=runme_0&dag_id=example_bash_operator&execution_date={}'
+               .format(self.percent_encode(self.default_date)))
+        resp = self.client.get(url, follow_redirects=True)
+        self.check_content_in_response('XCom', resp)
+
+    def test_xcom_failure(self):
+        self.logout()
+        self.login(username='dag_faker',
+                   password='dag_faker')
+        url = ('xcom?task_id=runme_0&dag_id=example_bash_operator&execution_date={}'
+               .format(self.percent_encode(self.default_date)))
+        resp = self.client.get(url, follow_redirects=True)
+        self.check_content_not_in_response('XCom', resp)
+
+    def test_xcom_success_for_all_dag_user(self):
+        self.logout()
+        self.login(username='all_dag_user',
+                   password='all_dag_user')
+        url = ('xcom?task_id=runme_0&dag_id=example_bash_operator&execution_date={}'
+               .format(self.percent_encode(self.default_date)))
+        resp = self.client.get(url, follow_redirects=True)
+        self.check_content_in_response('XCom', resp)
+
+    def test_run_success(self):
+        self.logout()
+        self.login()
+        url = ('run?task_id=runme_0&dag_id=example_bash_operator&ignore_all_deps=false&'
+               'ignore_ti_state=true&execution_date={}'
+               .format(self.percent_encode(self.default_date)))
+        resp = self.client.get(url)
+        self.check_content_in_response('', resp, resp_code=302)
+
+    def test_run_success_for_all_dag_user(self):
+        self.logout()
+        self.login(username='all_dag_user',
+                   password='all_dag_user')
+        url = ('run?task_id=runme_0&dag_id=example_bash_operator&ignore_all_deps=false&'
+               'ignore_ti_state=true&execution_date={}'
+               .format(self.percent_encode(self.default_date)))
+        resp = self.client.get(url)
+        self.check_content_in_response('', resp, resp_code=302)
+
+    def test_blocked_success(self):
+        url = 'blocked'
+        self.logout()
+        self.login()
+        resp = self.client.get(url, follow_redirects=True)
+        self.check_content_in_response('example_bash_operator', resp)
+
+    def test_blocked_success_for_all_dag_user(self):
+        url = 'blocked'
+        self.logout()
+        self.login(username='all_dag_user',
+                   password='all_dag_user')
+        resp = self.client.get(url, follow_redirects=True)
+        self.check_content_in_response('example_bash_operator', resp)
+        self.check_content_in_response('example_subdag_operator', resp)
+
+    def test_failed_success(self):
+        self.logout()
+        self.login()
+        url = ('failed?task_id=run_this_last&dag_id=example_bash_operator&'
+               'execution_date={}&upstream=false&downstream=false&future=false&past=false'
+               .format(self.percent_encode(self.default_date)))
+        resp = self.client.get(url)
+        self.check_content_in_response('Redirecting', resp, 302)
+
+    def test_duration_success(self):
+        url = 'duration?days=30&dag_id=example_bash_operator'
+        self.logout()
+        self.login()
+        resp = self.client.get(url, follow_redirects=True)
+        self.check_content_in_response('example_bash_operator', resp)
+
+    def test_duration_failure(self):
+        url = 'duration?days=30&dag_id=example_bash_operator'
+        self.logout()
+        # login as an user without permissions
+        self.login(username='dag_faker',
+                   password='dag_faker')
+        resp = self.client.get(url, follow_redirects=True)
+        self.check_content_not_in_response('example_bash_operator', resp)
+
+    def test_tries_success(self):
+        url = 'tries?days=30&dag_id=example_bash_operator'
+        self.logout()
+        self.login()
+        resp = self.client.get(url, follow_redirects=True)
+        self.check_content_in_response('example_bash_operator', resp)
+
+    def test_tries_failure(self):
+        url = 'tries?days=30&dag_id=example_bash_operator'
+        self.logout()
+        # login as an user without permissions
+        self.login(username='dag_faker',
+                   password='dag_faker')
+        resp = self.client.get(url, follow_redirects=True)
+        self.check_content_not_in_response('example_bash_operator', resp)
+
+    def test_landing_times_success(self):
+        url = 'landing_times?days=30&dag_id=example_bash_operator'
+        self.logout()
+        self.login()
+        resp = self.client.get(url, follow_redirects=True)
+        self.check_content_in_response('example_bash_operator', resp)
+
+    def test_landing_times_failure(self):
+        url = 'landing_times?days=30&dag_id=example_bash_operator'
+        self.logout()
+        self.login(username='dag_faker',
+                   password='dag_faker')
+        resp = self.client.get(url, follow_redirects=True)
+        self.check_content_not_in_response('example_bash_operator', resp)
+
+    def test_paused_success(self):
+        # post request failure won't test
+        url = 'paused?dag_id=example_bash_operator&is_paused=false'
+        self.logout()
+        self.login()
+        resp = self.client.post(url, follow_redirects=True)
+        self.check_content_in_response('OK', resp)
+
+    def test_refresh_success(self):
+        self.logout()
+        self.login()
+        resp = self.client.get('refresh?dag_id=example_bash_operator')
+        self.check_content_in_response('', resp, resp_code=302)
+
+    def test_gantt_success(self):
+        url = 'gantt?dag_id=example_bash_operator'
+        self.logout()
+        self.login()
+        resp = self.client.get(url, follow_redirects=True)
+        self.check_content_in_response('example_bash_operator', resp)
+
+    def test_gantt_failure(self):
+        url = 'gantt?dag_id=example_bash_operator'
+        self.logout()
+        self.login(username='dag_faker',
+                   password='dag_faker')
+        resp = self.client.get(url, follow_redirects=True)
+        self.check_content_not_in_response('example_bash_operator', resp)
+
+    def test_success_fail_for_read_only_role(self):
+        # succcess endpoint need can_dag_edit, which read only role can not access
+        self.logout()
+        self.login(username='dag_read_only',
+                   password='dag_read_only')
+
+        url = ('success?task_id=run_this_last&dag_id=example_bash_operator&'
+               'execution_date={}&upstream=false&downstream=false&future=false&past=false'
+               .format(self.percent_encode(self.default_date)))
+        resp = self.client.get(url)
+        self.check_content_not_in_response('Wait a minute', resp, resp_code=302)
+
+    def test_tree_success_for_read_only_role(self):
+        # tree view only allows can_dag_read, which read only role could access
+        self.logout()
+        self.login(username='dag_read_only',
+                   password='dag_read_only')
+
+        url = 'tree?dag_id=example_bash_operator'
+        resp = self.client.get(url, follow_redirects=True)
+        self.check_content_in_response('runme_1', resp)
+
+
 if __name__ == '__main__':
     unittest.main()


Mime
View raw message