Repository: apache/airflow
File: airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_connections.py (Python)

Code before:
        )
        assert response.status_code == 403

    @skip_if_force_lowest_dependencies_marker
    @mock.patch.dict(os.environ, {"AIRFLOW__CORE__TEST_CONNECTION": "Enabled"})
    @pytest.mark.parametrize(
        "body",
        [
            {"connection_id": TEST_CONN_ID, "conn_type": "sqlite"},
            {"connection_id": TEST_CONN_ID, "conn_type": "ftp"},
        ],
    )
    def test_connection_env_is_cleaned_after_run(self, test_client, body):
        test_client.post("/connections/test", json=body)
        assert not any([key.startswith(CONN_ENV_PREFIX) for key in os.environ.keys()])

    @pytest.mark.parametrize(
        "body",
        [
            {"connection_id": TEST_CONN_ID, "conn_type": "sqlite"},
            {"connection_id": TEST_CONN_ID, "conn_type": "ftp"},
        ],
    )
    @mock.patch("airflow.api_fastapi.core_api.routes.public.connections.conf.get")
    def test_should_respond_403_by_default(self, mock_conf_get, test_client, body):
        mock_conf_get.return_value = "Disabled"
        response = test_client.post("/connections/test", json=body)
        assert response.status_code == 403
        assert response.json() == {
            "detail": "Testing connections is disabled in Airflow configuration. "
            "Contact your deployment admin to enable it."
        }

    @mock.patch.dict(os.environ, {"AIRFLOW__CORE__TEST_CONNECTION": "Enabled"})
    def test_should_merge_password_with_existing_connection(self, test_client, session):
        connection = Connection(
            conn_id=TEST_CONN_ID,
            conn_type="sqlite",
            password="existing_password",
        )
        session.add(connection)
        session.commit()

        body = {
            "connection_id": TEST_CONN_ID,
            "conn_type": "sqlite",
            "password": "***",
        }
        response = test_client.post("/connections/test", json=body)
        assert response.status_code == 200
        assert response.json()["status"] is True

Review comment (suggestion):
Why is that needed. By default this is "Disabled", you shouldn't need that.

Probably a local conf or something interfering with the tests, your change doesn't seem to impact this.

Code after:
        response = unauthorized_test_client.post(
            "/connections/test", json={"connection_id": TEST_CONN_ID, "conn_type": "sqlite"}
        )
        assert response.status_code == 403

    @skip_if_force_lowest_dependencies_marker
    @mock.patch.dict(os.environ, {"AIRFLOW__CORE__TEST_CONNECTION": "Enabled"})
    @pytest.mark.parametrize(
        "body",
        [
            {"connection_id": TEST_CONN_ID, "conn_type": "sqlite"},
            {"connection_id": TEST_CONN_ID, "conn_type": "ftp"},
        ],
    )
    def test_connection_env_is_cleaned_after_run(self, test_client, body):
        test_client.post("/connections/test", json=body)
        assert not any([key.startswith(CONN_ENV_PREFIX) for key in os.environ.keys()])

    @pytest.mark.parametrize(
        "body",
        [
            {"connection_id": TEST_CONN_ID, "conn_type": "sqlite"},
            {"connection_id": TEST_CONN_ID, "conn_type": "ftp"},
        ],
    )
    def test_should_respond_403_by_default(self, test_client, body):
        response = test_client.post("/connections/test", json=body)
        assert response.status_code == 403
        assert response.json() == {
            "detail": "Testing connections is disabled in Airflow configuration. "
            "Contact your deployment admin to enable it."
        }

    @mock.patch.dict(os.environ, {"AIRFLOW__CORE__TEST_CONNECTION": "Enabled"})
    def test_should_merge_password_with_existing_connection(self, test_client, session):
        connection = Connection(
            conn_id=TEST_CONN_ID,
            conn_type="sqlite",
            password="existing_password",
        )
        session.add(connection)
        session.commit()
        initial_count = session.scalar(select(func.count()).select_from(Connection))

        captured_value = {}

        def mock_test_connection(self):
            captured_value["password"] = self.password
            captured_value["conn_type"] = self.conn_type
            return True, "mocked"