# 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 pytest

from airflow.api_connexion.exceptions import EXCEPTIONS_LINK_MAP
from airflow.models import Log
from airflow.security import permissions
from airflow.utils import timezone
from tests.test_utils.api_connexion_utils import assert_401, create_user, delete_user
from tests.test_utils.config import conf_vars
from tests.test_utils.db import clear_db_logs


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

    yield app

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


@pytest.fixture
def task_instance(session, create_task_instance, request):
    return create_task_instance(
        session=session,
        dag_id="TEST_DAG_ID",
        task_id="TEST_TASK_ID",
        execution_date=request.instance.default_time,
    )


@pytest.fixture()
def log_model(create_log_model, request):
    return create_log_model(
        event="TEST_EVENT",
        when=request.instance.default_time,
    )


@pytest.fixture
def create_log_model(create_task_instance, task_instance, session, request):
    def maker(event, when, **kwargs):
        log_model = Log(
            event=event,
            task_instance=task_instance,
            **kwargs,
        )
        log_model.dttm = when

        session.add(log_model)
        session.flush()
        return log_model

    return maker


class TestEventLogEndpoint:
    @pytest.fixture(autouse=True)
    def setup_attrs(self, configured_app) -> None:
        self.app = configured_app
        self.client = self.app.test_client()  # type:ignore
        clear_db_logs()
        self.default_time = timezone.parse("2020-06-10T20:00:00+00:00")
        self.default_time_2 = timezone.parse("2020-06-11T07:00:00+00:00")

    def teardown_method(self) -> None:
        clear_db_logs()


class TestGetEventLog(TestEventLogEndpoint):
    def test_should_respond_200(self, log_model):
        event_log_id = log_model.id
        response = self.client.get(
            f"/api/v1/eventLogs/{event_log_id}", environ_overrides={"REMOTE_USER": "test"}
        )
        assert response.status_code == 200
        assert response.json == {
            "event_log_id": event_log_id,
            "event": "TEST_EVENT",
            "dag_id": "TEST_DAG_ID",
            "task_id": "TEST_TASK_ID",
            "execution_date": self.default_time.isoformat(),
            "owner": "airflow",
            "when": self.default_time.isoformat(),
            "extra": None,
        }

    def test_should_respond_404(self):
        response = self.client.get("/api/v1/eventLogs/1", environ_overrides={"REMOTE_USER": "test"})
        assert response.status_code == 404
        assert {
            "detail": None,
            "status": 404,
            "title": "Event Log not found",
            "type": EXCEPTIONS_LINK_MAP[404],
        } == response.json

    def test_should_raises_401_unauthenticated(self, log_model):
        event_log_id = log_model.id

        response = self.client.get(f"/api/v1/eventLogs/{event_log_id}")

        assert_401(response)

    def test_should_raise_403_forbidden(self):
        response = self.client.get(
            "/api/v1/eventLogs", environ_overrides={"REMOTE_USER": "test_no_permissions"}
        )
        assert response.status_code == 403


class TestGetEventLogs(TestEventLogEndpoint):
    def test_should_respond_200(self, session, create_log_model):
        log_model_1 = create_log_model(event="TEST_EVENT_1", when=self.default_time)
        log_model_2 = create_log_model(event="TEST_EVENT_2", when=self.default_time_2)
        log_model_3 = Log(event="cli_scheduler", owner="root", extra='{"host_name": "e24b454f002a"}')
        log_model_3.dttm = self.default_time_2

        session.add(log_model_3)
        session.flush()
        response = self.client.get("/api/v1/eventLogs", environ_overrides={"REMOTE_USER": "test"})
        assert response.status_code == 200
        assert response.json == {
            "event_logs": [
                {
                    "event_log_id": log_model_1.id,
                    "event": "TEST_EVENT_1",
                    "dag_id": "TEST_DAG_ID",
                    "task_id": "TEST_TASK_ID",
                    "execution_date": self.default_time.isoformat(),
                    "owner": "airflow",
                    "when": self.default_time.isoformat(),
                    "extra": None,
                },
                {
                    "event_log_id": log_model_2.id,
                    "event": "TEST_EVENT_2",
                    "dag_id": "TEST_DAG_ID",
                    "task_id": "TEST_TASK_ID",
                    "execution_date": self.default_time.isoformat(),
                    "owner": "airflow",
                    "when": self.default_time_2.isoformat(),
                    "extra": None,
                },
                {
                    "event_log_id": log_model_3.id,
                    "event": "cli_scheduler",
                    "dag_id": None,
                    "task_id": None,
                    "execution_date": None,
                    "owner": "root",
                    "when": self.default_time_2.isoformat(),
                    "extra": '{"host_name": "e24b454f002a"}',
                },
            ],
            "total_entries": 3,
        }

    def test_order_eventlogs_by_owner(self, create_log_model, session):
        log_model_1 = create_log_model(event="TEST_EVENT_1", when=self.default_time)
        log_model_2 = create_log_model(event="TEST_EVENT_2", when=self.default_time_2, owner="zsh")
        log_model_3 = Log(event="cli_scheduler", owner="root", extra='{"host_name": "e24b454f002a"}')
        log_model_3.dttm = self.default_time_2
        session.add(log_model_3)
        session.flush()
        response = self.client.get(
            "/api/v1/eventLogs?order_by=-owner", environ_overrides={"REMOTE_USER": "test"}
        )
        assert response.status_code == 200
        assert response.json == {
            "event_logs": [
                {
                    "event_log_id": log_model_2.id,
                    "event": "TEST_EVENT_2",
                    "dag_id": "TEST_DAG_ID",
                    "task_id": "TEST_TASK_ID",
                    "execution_date": self.default_time.isoformat(),
                    "owner": "zsh",  # Order by name, sort order is descending(-)
                    "when": self.default_time_2.isoformat(),
                    "extra": None,
                },
                {
                    "event_log_id": log_model_3.id,
                    "event": "cli_scheduler",
                    "dag_id": None,
                    "task_id": None,
                    "execution_date": None,
                    "owner": "root",
                    "when": self.default_time_2.isoformat(),
                    "extra": '{"host_name": "e24b454f002a"}',
                },
                {
                    "event_log_id": log_model_1.id,
                    "event": "TEST_EVENT_1",
                    "dag_id": "TEST_DAG_ID",
                    "task_id": "TEST_TASK_ID",
                    "execution_date": self.default_time.isoformat(),
                    "owner": "airflow",
                    "when": self.default_time.isoformat(),
                    "extra": None,
                },
            ],
            "total_entries": 3,
        }

    def test_should_raises_401_unauthenticated(self, log_model):
        response = self.client.get("/api/v1/eventLogs")

        assert_401(response)


class TestGetEventLogPagination(TestEventLogEndpoint):
    @pytest.mark.parametrize(
        ("url", "expected_events"),
        [
            ("api/v1/eventLogs?limit=1", ["TEST_EVENT_1"]),
            ("api/v1/eventLogs?limit=2", ["TEST_EVENT_1", "TEST_EVENT_2"]),
            (
                "api/v1/eventLogs?offset=5",
                [
                    "TEST_EVENT_6",
                    "TEST_EVENT_7",
                    "TEST_EVENT_8",
                    "TEST_EVENT_9",
                    "TEST_EVENT_10",
                ],
            ),
            (
                "api/v1/eventLogs?offset=0",
                [
                    "TEST_EVENT_1",
                    "TEST_EVENT_2",
                    "TEST_EVENT_3",
                    "TEST_EVENT_4",
                    "TEST_EVENT_5",
                    "TEST_EVENT_6",
                    "TEST_EVENT_7",
                    "TEST_EVENT_8",
                    "TEST_EVENT_9",
                    "TEST_EVENT_10",
                ],
            ),
            ("api/v1/eventLogs?limit=1&offset=5", ["TEST_EVENT_6"]),
            ("api/v1/eventLogs?limit=1&offset=1", ["TEST_EVENT_2"]),
            (
                "api/v1/eventLogs?limit=2&offset=2",
                ["TEST_EVENT_3", "TEST_EVENT_4"],
            ),
        ],
    )
    def test_handle_limit_and_offset(self, url, expected_events, task_instance, session):
        log_models = self._create_event_logs(task_instance, 10)
        session.add_all(log_models)
        session.commit()

        response = self.client.get(url, environ_overrides={"REMOTE_USER": "test"})
        assert response.status_code == 200

        assert response.json["total_entries"] == 10
        events = [event_log["event"] for event_log in response.json["event_logs"]]
        assert events == expected_events

    def test_should_respect_page_size_limit_default(self, task_instance, session):
        log_models = self._create_event_logs(task_instance, 200)
        session.add_all(log_models)
        session.flush()

        response = self.client.get("/api/v1/eventLogs", environ_overrides={"REMOTE_USER": "test"})
        assert response.status_code == 200

        assert response.json["total_entries"] == 200
        assert len(response.json["event_logs"]) == 100  # default 100

    def test_should_raise_400_for_invalid_order_by_name(self, task_instance, session):
        log_models = self._create_event_logs(task_instance, 200)
        session.add_all(log_models)
        session.flush()

        response = self.client.get(
            "/api/v1/eventLogs?order_by=invalid", environ_overrides={"REMOTE_USER": "test"}
        )
        assert response.status_code == 400
        msg = "Ordering with 'invalid' is disallowed or the attribute does not exist on the model"
        assert response.json["detail"] == msg

    @conf_vars({("api", "maximum_page_limit"): "150"})
    def test_should_return_conf_max_if_req_max_above_conf(self, task_instance, session):
        log_models = self._create_event_logs(task_instance, 200)
        session.add_all(log_models)
        session.flush()

        response = self.client.get("/api/v1/eventLogs?limit=180", environ_overrides={"REMOTE_USER": "test"})
        assert response.status_code == 200
        assert len(response.json["event_logs"]) == 150

    def _create_event_logs(self, task_instance, count):
        return [Log(event=f"TEST_EVENT_{i}", task_instance=task_instance) for i in range(1, count + 1)]
