Merge branch 'master' of https://github.com/apache/airflow into #12401-…

…creating-duplicate-connection-in-ui

tests/www/test_views.py

@@ -228,281 +228,6 @@ def create_user_and_login(self, username, role_name, perms):
         self.login(username=username, password=username)
 
 
-class TestConnectionModelView(TestBase):
-    def setUp(self):
-        super().setUp()
-
-        self.connection = {
-            'conn_id': 'test_conn',
-            'conn_type': 'http',
-            'description': 'description',
-            'host': 'localhost',
-            'port': 8080,
-            'username': 'root',
-            'password': 'admin',
-        }
-
-    def tearDown(self):
-        self.clear_table(Connection)
-        super().tearDown()
-
-    def test_create_connection(self):
-        init_views.init_connection_form()
-        resp = self.client.post('/connection/add', data=self.connection, follow_redirects=True)
-        self.check_content_in_response('Added Row', resp)
-
-    def test_prefill_form_null_extra(self):
-        mock_form = mock.Mock()
-        mock_form.data = {"conn_id": "test", "extra": None}
-
-        cmv = ConnectionModelView()
-        cmv.prefill_form(form=mock_form, pk=1)
-
-    def test_duplicate_connection(self):
-        """Test Duplicate multiple connection with suffix"""
-        conn1 = Connection(
-            conn_id='test_duplicate_gcp_connection',
-            conn_type='Google Cloud',
-            description='Google Cloud Connection',
-        )
-
-        conn2 = Connection(
-            conn_id='test_duplicate_mysql_connection',
-            conn_type='FTP',
-            description='MongoDB2',
-            host='localhost',
-            schema='airflow',
-            port=3306,
-        )
-
-        conn3 = Connection(
-            conn_id='test_duplicate_postgres_connection_copy1',
-            conn_type='FTP',
-            description='Postgres',
-            host='localhost',
-            schema='airflow',
-            port=3306,
-        )
-
-        self.clear_table(Connection)
-        self.session.add_all([conn1, conn2, conn3])
-        self.session.commit()
-
-        mock_form = mock.Mock()
-        mock_form.data = {"action": "mulduplicate", "rowid": [conn1.id, conn3.id]}
-        resp = self.client.post('/connection/action_post', data=mock_form.data, follow_redirects=True)
-
-        expected_result = {
-            'test_duplicate_gcp_connection',
-            'test_duplicate_gcp_connection_copy1',
-            'test_duplicate_mysql_connection',
-            'test_duplicate_postgres_connection_copy1',
-            'test_duplicate_postgres_connection_copy2',
-        }
-        response = {conn[0] for conn in self.session.query(Connection.conn_id).all()}
-
-        assert resp.status_code == 200
-        assert expected_result == response
-
-
-class TestVariableModelView(TestBase):
-    def setUp(self):
-        super().setUp()
-        self.variable = {
-            'key': 'test_key',
-            'val': 'text_val',
-            'description': 'test_description',
-            'is_encrypted': True,
-        }
-
-    def tearDown(self):
-        self.clear_table(models.Variable)
-        super().tearDown()
-
-    def test_can_handle_error_on_decrypt(self):
-
-        # create valid variable
-        self.client.post('/variable/add', data=self.variable, follow_redirects=True)
-
-        # update the variable with a wrong value, given that is encrypted
-        Var = models.Variable  # pylint: disable=invalid-name
-        (
-            self.session.query(Var)
-            .filter(Var.key == self.variable['key'])
-            .update({'val': 'failed_value_not_encrypted'}, synchronize_session=False)
-        )
-        self.session.commit()
-
-        # retrieve Variables page, should not fail and contain the Invalid
-        # label for the variable
-        resp = self.client.get('/variable/list', follow_redirects=True)
-        self.check_content_in_response('<span class="label label-danger">Invalid</span>', resp)
-
-    def test_xss_prevention(self):
-        xss = "/variable/list/<img%20src=''%20onerror='alert(1);'>"
-
-        resp = self.client.get(
-            xss,
-            follow_redirects=True,
-        )
-        assert resp.status_code == 404
-        assert "<img src='' onerror='alert(1);'>" not in resp.data.decode("utf-8")
-
-    def test_import_variables_no_file(self):
-        resp = self.client.post('/variable/varimport', follow_redirects=True)
-        self.check_content_in_response('Missing file or syntax error.', resp)
-
-    def test_import_variables_failed(self):
-        content = '{"str_key": "str_value"}'
-
-        with mock.patch('airflow.models.Variable.set') as set_mock:
-            set_mock.side_effect = UnicodeEncodeError
-            assert self.session.query(models.Variable).count() == 0
-
-            bytes_content = io.BytesIO(bytes(content, encoding='utf-8'))
-
-            resp = self.client.post(
-                '/variable/varimport', data={'file': (bytes_content, 'test.json')}, follow_redirects=True
-            )
-            self.check_content_in_response('1 variable(s) failed to be updated.', resp)
-
-    def test_import_variables_success(self):
-        assert self.session.query(models.Variable).count() == 0
-
-        content = (
-            '{"str_key": "str_value", "int_key": 60, "list_key": [1, 2], "dict_key": {"k_a": 2, "k_b": 3}}'
-        )
-        bytes_content = io.BytesIO(bytes(content, encoding='utf-8'))
-
-        resp = self.client.post(
-            '/variable/varimport', data={'file': (bytes_content, 'test.json')}, follow_redirects=True
-        )
-        self.check_content_in_response('4 variable(s) successfully updated.', resp)
-
-    def test_description_retrieval(self):
-        # create valid variable
-        self.client.post('/variable/add', data=self.variable, follow_redirects=True)
-
-        row = self.session.query(models.Variable.key, models.Variable.description).first()
-        assert row.key == 'test_key' and row.description == 'test_description'
-
-
-class PluginOperator(BaseOperator):
-    pass
-
-
-class TestPluginView(TestBase):
-    def test_should_list_plugins_on_page_with_details(self):
-        resp = self.client.get('/plugin')
-        self.check_content_in_response("test_plugin", resp)
-        self.check_content_in_response("Airflow Plugins", resp)
-        self.check_content_in_response("source", resp)
-        self.check_content_in_response("<em>$PLUGINS_FOLDER/</em>test_plugin.py", resp)
-
-    def test_should_list_entrypoint_plugins_on_page_with_details(self):
-
-        mock_plugin = AirflowPlugin()
-        mock_plugin.name = "test_plugin"
-        mock_plugin.source = EntryPointSource(
-            mock.Mock(), mock.Mock(version='1.0.0', metadata={'name': 'test-entrypoint-testpluginview'})
-        )
-        with mock_plugin_manager(plugins=[mock_plugin]):
-            resp = self.client.get('/plugin')
-
-        self.check_content_in_response("test_plugin", resp)
-        self.check_content_in_response("Airflow Plugins", resp)
-        self.check_content_in_response("source", resp)
-        self.check_content_in_response("<em>test-entrypoint-testpluginview==1.0.0:</em> <Mock id=", resp)
-
-    def test_endpoint_should_not_be_unauthenticated(self):
-        self.logout()
-        resp = self.client.get('/plugin', follow_redirects=True)
-        self.check_content_not_in_response("test_plugin", resp)
-        self.check_content_in_response("Sign In - Airflow", resp)
-
-
-class TestPoolModelView(TestBase):
-    def setUp(self):
-        super().setUp()
-        self.pool = {
-            'pool': 'test-pool',
-            'slots': 777,
-            'description': 'test-pool-description',
-        }
-
-    def tearDown(self):
-        self.clear_table(models.Pool)
-        super().tearDown()
-
-    def test_create_pool_with_same_name(self):
-        # create test pool
-        resp = self.client.post('/pool/add', data=self.pool, follow_redirects=True)
-        self.check_content_in_response('Added Row', resp)
-
-        # create pool with the same name
-        resp = self.client.post('/pool/add', data=self.pool, follow_redirects=True)
-        self.check_content_in_response('Already exists.', resp)
-
-    def test_create_pool_with_empty_name(self):
-        self.pool['pool'] = ''
-        resp = self.client.post('/pool/add', data=self.pool, follow_redirects=True)
-        self.check_content_in_response('This field is required.', resp)
-
-    def test_odd_name(self):
-        self.pool['pool'] = 'test-pool<script></script>'
-        self.session.add(models.Pool(**self.pool))
-        self.session.commit()
-        resp = self.client.get('/pool/list/')
-        self.check_content_in_response('test-pool&lt;script&gt;', resp)
-        self.check_content_not_in_response('test-pool<script>', resp)
-
-    def test_list(self):
-        self.pool['pool'] = 'test-pool'
-        self.session.add(models.Pool(**self.pool))
-        self.session.commit()
-        resp = self.client.get('/pool/list/')
-        # We should see this link
-        with self.app.test_request_context():
-            url = url_for('TaskInstanceModelView.list', _flt_3_pool='test-pool', _flt_3_state='running')
-            used_tag = Markup("<a href='{url}'>{slots}</a>").format(url=url, slots=0)
-
-            url = url_for('TaskInstanceModelView.list', _flt_3_pool='test-pool', _flt_3_state='queued')
-            queued_tag = Markup("<a href='{url}'>{slots}</a>").format(url=url, slots=0)
-        self.check_content_in_response(used_tag, resp)
-        self.check_content_in_response(queued_tag, resp)
-
-
-class TestMountPoint(unittest.TestCase):
-    @classmethod
-    @conf_vars({("webserver", "base_url"): "http://localhost/test"})
-    def setUpClass(cls):
-        application.app = None
-        application.appbuilder = None
-        app = application.create_app(testing=True)
-        app.config['WTF_CSRF_ENABLED'] = False
-        cls.client = Client(app, BaseResponse)
-
-    @classmethod
-    def tearDownClass(cls):
-        application.app = None
-        application.appbuilder = None
-
-    def test_mount(self):
-        # Test an endpoint that doesn't need auth!
-        resp = self.client.get('/test/health')
-        assert resp.status_code == 200
-        assert b"healthy" in resp.data
-
-    def test_not_found(self):
-        resp = self.client.get('/', follow_redirects=True)
-        assert resp.status_code == 404
-
-    def test_index(self):
-        resp = self.client.get('/test/')
-        assert resp.status_code == 302
-        assert resp.headers['Location'] == 'http://localhost/test/home'
-
-
 class TestAirflowBaseViews(TestBase):
     EXAMPLE_DAG_DEFAULT_DATE = dates.days_ago(2)