Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions tests/test_utils/www.py
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,13 @@ def client_without_login(app):
return client


def client_without_login_as_admin(app):
# Anonymous users as Admin if set AUTH_ROLE_PUBLIC=Admin
app.config["AUTH_ROLE_PUBLIC"] = "Admin"
client = app.test_client()
return client


def check_content_in_response(text, resp, resp_code=200):
resp_html = resp.data.decode("utf-8")
assert resp_code == resp.status_code
Expand Down
7 changes: 6 additions & 1 deletion tests/www/views/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@
from tests.test_utils.api_connexion_utils import delete_user
from tests.test_utils.config import conf_vars
from tests.test_utils.decorators import dont_initialize_flask_app_submodules
from tests.test_utils.www import client_with_login, client_without_login
from tests.test_utils.www import client_with_login, client_without_login, client_without_login_as_admin


@pytest.fixture(autouse=True, scope="module")
Expand Down Expand Up @@ -130,6 +130,11 @@ def anonymous_client(app):
return client_without_login(app)


@pytest.fixture()
def anonymous_client_as_admin(app):
return client_without_login_as_admin(app)


class _TemplateWithContext(NamedTuple):
template: jinja2.environment.Template
context: dict[str, Any]
Expand Down
63 changes: 63 additions & 0 deletions tests/www/views/test_anonymous_as_admin_role.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,63 @@
#
# 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.models import Pool
from airflow.utils.session import create_session
from tests.test_utils.www import check_content_in_response

pytestmark = pytest.mark.db_test

POOL = {
"pool": "test-pool",
"slots": 777,
"description": "test-pool-description",
"include_deferred": False,
}


@pytest.fixture(autouse=True)
def clear_pools():
with create_session() as session:
session.query(Pool).delete()


@pytest.fixture()
def pool_factory(session):
def factory(**values):
pool = Pool(**{**POOL, **values}) # Passed in values override defaults.
session.add(pool)
session.commit()
return pool

return factory


def test_delete_pool_anonymous_user_no_role(anonymous_client, pool_factory):
pool = pool_factory()
resp = anonymous_client.post(f"pool/delete/{pool.id}")
assert 302 == resp.status_code
assert "/login/" == resp.headers["Location"]


def test_delete_pool_anonymous_user_as_admin(anonymous_client_as_admin, pool_factory):
pool = pool_factory()
resp = anonymous_client_as_admin.post(f"pool/delete/{pool.id}", follow_redirects=True)
check_content_in_response("Deleted Row", resp)