<commit_before>#
# 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.
import datetime
import json
import urllib.parse

import pytest

from airflow.models import DagModel
from airflow.security import permissions
from airflow.utils import timezone
from airflow.utils.session import create_session
from airflow.utils.state import State
from airflow.utils.types import DagRunType
from airflow.www.views import FILTER_STATUS_COOKIE
from tests.test_utils.api_connexion_utils import create_user_scope
from tests.test_utils.db import clear_db_runs
from tests.test_utils.www import check_content_in_response, check_content_not_in_response, client_with_login

NEXT_YEAR = datetime.datetime.now().year + 1
DEFAULT_DATE = timezone.datetime(NEXT_YEAR, 6, 1)
DEFAULT_RUN_ID = "TEST_RUN_ID"
USER_DATA = {
    "dag_tester": (
        "dag_acl_tester",
        {
            "first_name": 'dag_test',
            "last_name": 'dag_test',
            "email": '<EMAIL>',
            "password": '<PASSWORD>',
        },
    ),
    "dag_faker": (  # User without permission.
        "dag_acl_faker",
        {
            "first_name": 'dag_faker',
            "last_name": 'dag_faker',
            "email": '<EMAIL>',
            "password": '<PASSWORD>',
        },
    ),
    "dag_read_only": (  # User with only read permission.
        "dag_acl_read_only",
        {
            "first_name": 'dag_read_only',
            "last_name": 'dag_read_only',
            "email": '<EMAIL>',
            "password": '<PASSWORD>',
        },
    ),
    "all_dag_user": (  # User has all dag access.
        "all_dag_role",
        {
            "first_name": 'all_dag_user',
            "last_name": 'all_dag_user',
            "email": '<EMAIL>',
            "password": '<PASSWORD>',
        },
    ),
}


@pytest.fixture(scope="module")
def acl_app(app):
    security_manager = app.appbuilder.sm
    for username, (role_name, kwargs) in USER_DATA.items():
        if not security_manager.find_user(username=username):
            role = security_manager.add_role(role_name)
            security_manager.add_user(
                role=role,
                username=username,
                **kwargs,
            )

    # FIXME: Clean up this block of code.....

    website_permission = security_manager.get_permission(
        permissions.ACTION_CAN_READ, permissions.RESOURCE_WEBSITE
    )

    dag_tester_role = security_manager.find_role('dag_acl_tester')
    edit_perm_on_dag = security_manager.get_permission(
        permissions.ACTION_CAN_EDIT, 'DAG:example_bash_operator'
    )
    security_manager.add_permission_to_role(dag_tester_role, edit_perm_on_dag)
    read_perm_on_dag = security_manager.get_permission(
        permissions.ACTION_CAN_READ, 'DAG:example_bash_operator'
    )
    security_manager.add_permission_to_role(dag_tester_role, read_perm_on_dag)
    security_manager.add_permission_to_role(dag_tester_role, website_permission)

    all_dag_role = security_manager.find_role('all_dag_role')
    edit_perm_on_all_dag = security_manager.get_permission(
        permissions.ACTION_CAN_EDIT, permissions.RESOURCE_DAG
    )
    security_manager.add_permission_to_role(all_dag_role, edit_perm_on_all_dag)
    read_perm_on_all_dag = security_manager.get_permission(
        permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG
    )
    security_manager.add_permission_to_role(all_dag_role, read_perm_on_all_dag)
    read_perm_on_task_instance = security_manager.get_permission(
        permissions.ACTION_CAN_READ, permissions.RESOURCE_TASK_INSTANCE
    )
    security_manager.add_permission_to_role(all_dag_role, read_perm_on_task_instance)
    security_manager.add_permission_to_role(all_dag_role, website_permission)

    role_user = security_manager.find_role('User')
    security_manager.add_permission_to_role(role_user, read_perm_on_all_dag)
    security_manager.add_permission_to_role(role_user, edit_perm_on_all_dag)
    security_manager.add_permission_to_role(role_user, website_permission)

    read_only_perm_on_dag = security_manager.get_permission(
        permissions.ACTION_CAN_READ, 'DAG:example_bash_operator'
    )
    dag_read_only_role = security_manager.find_role('dag_acl_read_only')
    security_manager.add_permission_to_role(dag_read_only_role, read_only_perm_on_dag)
    security_manager.add_permission_to_role(dag_read_only_role, website_permission)

    dag_acl_faker_role = security_manager.find_role('dag_acl_faker')
    security_manager.add_permission_to_role(dag_acl_faker_role, website_permission)

    yield app

    for username, _ in USER_DATA.items():
        user = security_manager.find_user(username=username)
        if user:
            security_manager.del_register_user(user)
<commit_msg>Remove code duplication in the test suite test_views_acl.py (#20887)<commit_after>#
# 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.
import datetime
import json
import urllib.parse

import pytest

from airflow.models import DagModel
from airflow.security import permissions
from airflow.utils import timezone
from airflow.utils.session import create_session
from airflow.utils.state import State
from airflow.utils.types import DagRunType
from airflow.www.views import FILTER_STATUS_COOKIE
from tests.test_utils.api_connexion_utils import create_user_scope
from tests.test_utils.db import clear_db_runs
from tests.test_utils.www import check_content_in_response, check_content_not_in_response, client_with_login

NEXT_YEAR = datetime.datetime.now().year + 1
DEFAULT_DATE = timezone.datetime(NEXT_YEAR, 6, 1)
DEFAULT_RUN_ID = "TEST_RUN_ID"
USER_DATA = {
    "dag_tester": (
        "dag_acl_tester",
        {
            "first_name": 'dag_test',
            "last_name": 'dag_test',
            "email": '<EMAIL>',
            "password": '<PASSWORD>',
        },
    ),
    "dag_faker": (  # User without permission.
        "dag_acl_faker",
        {
            "first_name": 'dag_faker',
            "last_name": 'dag_faker',
            "email": '<EMAIL>',
            "password": '<PASSWORD>',
        },
    ),
    "dag_read_only": (  # User with only read permission.
        "dag_acl_read_only",
        {
            "first_name": 'dag_read_only',
            "last_name": 'dag_read_only',
            "email": '<EMAIL>',
            "password": '<PASSWORD>',
        },
    ),
    "all_dag_user": (  # User has all dag access.
        "all_dag_role",
        {
            "first_name": 'all_dag_user',
            "last_name": 'all_dag_user',
            "email": '<EMAIL>',
            "password": '<PASSWORD>',
        },
    ),
}


@pytest.fixture(scope="module")
def acl_app(app):
    security_manager = app.appbuilder.sm
    for username, (role_name, kwargs) in USER_DATA.items():
        if not security_manager.find_user(username=username):
            role = security_manager.add_role(role_name)
            security_manager.add_user(
                role=role,
                username=username,
                **kwargs,
            )

    role_permissions = {
        'dag_acl_tester': [
            (permissions.ACTION_CAN_READ, permissions.RESOURCE_WEBSITE),
            (permissions.ACTION_CAN_EDIT, 'DAG:example_bash_operator'),
            (permissions.ACTION_CAN_READ, 'DAG:example_bash_operator'),
        ],
        'all_dag_role': [
            (permissions.ACTION_CAN_EDIT, permissions.RESOURCE_DAG),
            (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG),
            (permissions.ACTION_CAN_READ, permissions.RESOURCE_TASK_INSTANCE),
            (permissions.ACTION_CAN_READ, permissions.RESOURCE_WEBSITE),
        ],
        'User': [
            (permissions.ACTION_CAN_EDIT, permissions.RESOURCE_DAG),
            (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG),
            (permissions.ACTION_CAN_READ, permissions.RESOURCE_WEBSITE),
        ],
        'dag_acl_read_only': [
            (permissions.ACTION_CAN_READ, 'DAG:example_bash_operator'),
            (permissions.ACTION_CAN_READ, permissions.RESOURCE_WEBSITE),
        ],
        'dag_acl_faker': [(permissions.ACTION_CAN_READ, permissions.RESOURCE_WEBSITE)],
    }

    for _role, _permissions in role_permissions.items():
        role = security_manager.find_role(_role)
        for _action, _perm in _permissions:
            perm = security_manager.get_permission(_action, _perm)
            security_manager.add_permission_to_role(role, perm)

    yield app

    for username, _ in USER_DATA.items():
        user = security_manager.find_user(username=username)
        if user:
            security_manager.del_register_user(user)

{'fraction_non_alphanumeric': 0.089255541198688, 'fraction_numerical': 0.0018884802703508599, 'mean_word_length': 3.289002557544757, 'pattern_counts': {'":': 40, '<': 19, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 6, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '107375', 'n_tokens_mistral': 3054, 'n_tokens_neox': 3036, 'n_words': 642}