tests/api_connexion/endpoints/test_config_endpoint.py

# 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 textwrap
from unittest.mock import patch

import pytest

from airflow.security import permissions
from tests.test_utils.api_connexion_utils import assert_401, create_user, delete_user
from tests.test_utils.config import conf_vars

MOCK_CONF = {
    "core": {
        "parallelism": "1024",
    },
    "smtp": {
        "smtp_host": "localhost",
        "smtp_mail_from": "airflow@example.com",
    },
}


@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_CONFIG)],  # type: ignore
    )
    create_user(app, username="test_no_permissions", role_name="TestNoPermissions")  # type: ignore

    with conf_vars({("webserver", "expose_config"): "True"}):
        yield minimal_app_for_api

    delete_user(app, username="test")  # type: ignore
    delete_user(app, username="test_no_permissions")  # type: ignore


class TestGetConfig:
    @pytest.fixture(autouse=True)
    def setup_attrs(self, configured_app) -> None:
        self.app = configured_app
        self.client = self.app.test_client()  # type:ignore

    @patch("airflow.api_connexion.endpoints.config_endpoint.conf.as_dict", return_value=MOCK_CONF)
    def test_should_respond_200_text_plain(self, mock_as_dict):
        response = self.client.get(
            "/api/v1/config", headers={"Accept": "text/plain"}, environ_overrides={"REMOTE_USER": "test"}
        )
        assert response.status_code == 200
        expected = textwrap.dedent(
            """\
        [core]
        parallelism = 1024

        [smtp]
        smtp_host = localhost
        smtp_mail_from = airflow@example.com
        """
        )
        assert expected == response.data.decode()

    @patch("airflow.api_connexion.endpoints.config_endpoint.conf.as_dict", return_value=MOCK_CONF)
    def test_should_respond_200_application_json(self, mock_as_dict):
        response = self.client.get(
            "/api/v1/config",
            headers={"Accept": "application/json"},
            environ_overrides={"REMOTE_USER": "test"},
        )
        assert response.status_code == 200
        expected = {
            "sections": [
                {
                    "name": "core",
                    "options": [
                        {"key": "parallelism", "value": "1024"},
                    ],
                },
                {
                    "name": "smtp",
                    "options": [
                        {"key": "smtp_host", "value": "localhost"},
                        {"key": "smtp_mail_from", "value": "airflow@example.com"},
                    ],
                },
            ]
        }
        assert expected == response.json

    @patch("airflow.api_connexion.endpoints.config_endpoint.conf.as_dict", return_value=MOCK_CONF)
    def test_should_respond_200_single_section_as_text_plain(self, mock_as_dict):
        response = self.client.get(
            "/api/v1/config?section=smtp",
            headers={"Accept": "text/plain"},
            environ_overrides={"REMOTE_USER": "test"},
        )
        assert response.status_code == 200
        expected = textwrap.dedent(
            """\
        [smtp]
        smtp_host = localhost
        smtp_mail_from = airflow@example.com
        """
        )
        assert expected == response.data.decode()

    @patch("airflow.api_connexion.endpoints.config_endpoint.conf.as_dict", return_value=MOCK_CONF)
    def test_should_respond_200_single_section_as_json(self, mock_as_dict):
        response = self.client.get(
            "/api/v1/config?section=smtp",
            headers={"Accept": "application/json"},
            environ_overrides={"REMOTE_USER": "test"},
        )
        assert response.status_code == 200
        expected = {
            "sections": [
                {
                    "name": "smtp",
                    "options": [
                        {"key": "smtp_host", "value": "localhost"},
                        {"key": "smtp_mail_from", "value": "airflow@example.com"},
                    ],
                },
            ]
        }
        assert expected == response.json

    @patch("airflow.api_connexion.endpoints.config_endpoint.conf.as_dict", return_value=MOCK_CONF)
    def test_should_respond_404_when_section_not_exist(self, mock_as_dict):
        response = self.client.get(
            "/api/v1/config?section=smtp1",
            headers={"Accept": "application/json"},
            environ_overrides={"REMOTE_USER": "test"},
        )

        assert response.status_code == 404
        assert "section=smtp1 not found." in response.json["detail"]

    @patch("airflow.api_connexion.endpoints.config_endpoint.conf.as_dict", return_value=MOCK_CONF)
    def test_should_respond_406(self, mock_as_dict):
        response = self.client.get(
            "/api/v1/config",
            headers={"Accept": "application/octet-stream"},
            environ_overrides={"REMOTE_USER": "test"},
        )
        assert response.status_code == 406

    def test_should_raises_401_unauthenticated(self):
        response = self.client.get("/api/v1/config", headers={"Accept": "application/json"})

        assert_401(response)

    def test_should_raises_403_unauthorized(self):
        response = self.client.get(
            "/api/v1/config",
            headers={"Accept": "application/json"},
            environ_overrides={"REMOTE_USER": "test_no_permissions"},
        )

        assert response.status_code == 403

    @conf_vars({("webserver", "expose_config"): "False"})
    def test_should_respond_403_when_expose_config_off(self):
        response = self.client.get(
            "/api/v1/config",
            headers={"Accept": "application/json"},
            environ_overrides={"REMOTE_USER": "test"},
        )
        assert response.status_code == 403
        assert "chose not to expose" in response.json["detail"]


class TestGetValue:
    @pytest.fixture(autouse=True)
    def setup_attrs(self, configured_app) -> None:
        self.app = configured_app
        self.client = self.app.test_client()  # type:ignore

    @patch("airflow.api_connexion.endpoints.config_endpoint.conf.as_dict", return_value=MOCK_CONF)
    def test_should_respond_200_text_plain(self, mock_as_dict):
        response = self.client.get(
            "/api/v1/config/section/smtp/option/smtp_mail_from",
            headers={"Accept": "text/plain"},
            environ_overrides={"REMOTE_USER": "test"},
        )
        assert response.status_code == 200
        expected = textwrap.dedent(
            """\
        [smtp]
        smtp_mail_from = airflow@example.com
        """
        )
        assert expected == response.data.decode()

    @patch("airflow.api_connexion.endpoints.config_endpoint.conf.as_dict", return_value=MOCK_CONF)
    def test_should_respond_200_application_json(self, mock_as_dict):
        response = self.client.get(
            "/api/v1/config/section/smtp/option/smtp_mail_from",
            headers={"Accept": "application/json"},
            environ_overrides={"REMOTE_USER": "test"},
        )
        assert response.status_code == 200
        expected = {
            "sections": [
                {
                    "name": "smtp",
                    "options": [
                        {"key": "smtp_mail_from", "value": "airflow@example.com"},
                    ],
                },
            ]
        }
        assert expected == response.json

    @patch("airflow.api_connexion.endpoints.config_endpoint.conf.as_dict", return_value=MOCK_CONF)
    def test_should_respond_404_when_option_not_exist(self, mock_as_dict):
        response = self.client.get(
            "/api/v1/config/section/smtp/option/smtp_mail_from1",
            headers={"Accept": "application/json"},
            environ_overrides={"REMOTE_USER": "test"},
        )

        assert response.status_code == 404
        assert "The option [smtp/smtp_mail_from1] is not found in config." in response.json["detail"]

    @patch("airflow.api_connexion.endpoints.config_endpoint.conf.as_dict", return_value=MOCK_CONF)
    def test_should_respond_406(self, mock_as_dict):
        response = self.client.get(
            "/api/v1/config/section/smtp/option/smtp_mail_from",
            headers={"Accept": "application/octet-stream"},
            environ_overrides={"REMOTE_USER": "test"},
        )
        assert response.status_code == 406

    def test_should_raises_401_unauthenticated(self):
        response = self.client.get(
            "/api/v1/config/section/smtp/option/smtp_mail_from", headers={"Accept": "application/json"}
        )

        assert_401(response)

    def test_should_raises_403_unauthorized(self):
        response = self.client.get(
            "/api/v1/config/section/smtp/option/smtp_mail_from",
            headers={"Accept": "application/json"},
            environ_overrides={"REMOTE_USER": "test_no_permissions"},
        )

        assert response.status_code == 403

    @conf_vars({("webserver", "expose_config"): "False"})
    def test_should_respond_403_when_expose_config_off(self):
        response = self.client.get(
            "/api/v1/config/section/smtp/option/smtp_mail_from",
            headers={"Accept": "application/json"},
            environ_overrides={"REMOTE_USER": "test"},
        )
        assert response.status_code == 403
        assert "chose not to expose" in response.json["detail"]