This is an automated email from the ASF dual-hosted git repository.
ferruzzi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 802f398731f Start a real login in the AWS auth manager system test
(#74431)
802f398731f is described below
commit 802f398731fd30482c157b877f753cd6f236a3ea
Author: D. Ferruzzi <[email protected]>
AuthorDate: Fri Oct 9 09:56:49 2026 -0700
Start a real login in the AWS auth manager system test (#74431)
* Start a real login in the AWS auth manager system test
Start the login first and echo back the RelayState it issued. The SAML
auth mock now returns real values for login() and get_last_request_id(), which
_start_login needs as a str and as JSON respectively, and api.ssl_cert is
pinned empty so the login-state cookie is not secure-only over
http://testserver.
---
.../amazon/aws/tests/test_aws_auth_manager.py | 65 ++++++++++++----------
1 file changed, 36 insertions(+), 29 deletions(-)
diff --git
a/providers/amazon/tests/system/amazon/aws/tests/test_aws_auth_manager.py
b/providers/amazon/tests/system/amazon/aws/tests/test_aws_auth_manager.py
index c5faac50971..835a59a64f9 100644
--- a/providers/amazon/tests/system/amazon/aws/tests/test_aws_auth_manager.py
+++ b/providers/amazon/tests/system/amazon/aws/tests/test_aws_auth_manager.py
@@ -63,6 +63,8 @@ permit (
);
"""
+TEST_REQUEST_ID = "ONELOGIN_authn_request_id"
+
def create_avp_policy_store(env_id):
description = f"Created by system test TestAwsAuthManager: {env_id}"
@@ -125,6 +127,7 @@ def base_app(region_name, avp_policy_store_id):
("aws_auth_manager", "region_name"): region_name,
("aws_auth_manager", "saml_metadata_url"): SAML_METADATA_URL,
("aws_auth_manager", "avp_policy_store_id"): avp_policy_store_id,
+ ("api", "ssl_cert"): "",
}
):
with (
@@ -137,30 +140,42 @@ def base_app(region_name, avp_policy_store_id):
yield mock_init_saml_auth
[email protected]
-def client_no_permissions(base_app):
+def _saml_auth_mock(nameid: str, attributes: dict) -> Mock:
auth = Mock()
auth.is_authenticated.return_value = True
- auth.get_nameid.return_value = "user_no_permissions"
- auth.get_attributes.return_value = {
- "id": ["user_no_permissions"],
- "groups": [],
- "email": ["email"],
- }
- base_app.return_value = auth
+ auth.get_nameid.return_value = nameid
+ auth.get_attributes.return_value = attributes
+ auth.login.return_value = "https://idp.example.com/sso";
+ auth.get_last_request_id.return_value = TEST_REQUEST_ID
+ return auth
+
+
+def _complete_login(client: TestClient, saml_auth: Mock, login_path: str):
+ """Make the whole login round trip and return the callback response."""
+ client.get(AUTH_MANAGER_FASTAPI_APP_PREFIX + login_path,
follow_redirects=False)
+ relay_state = saml_auth.login.call_args[0][0]
+ return client.post(
+ AUTH_MANAGER_FASTAPI_APP_PREFIX + "/login_callback",
+ follow_redirects=False,
+ data={"RelayState": relay_state},
+ )
+
+
[email protected]
+def client_no_permissions(base_app):
+ base_app.return_value = _saml_auth_mock(
+ "user_no_permissions",
+ {"id": ["user_no_permissions"], "groups": [], "email": ["email"]},
+ )
return TestClient(create_app())
@pytest.fixture
def client_admin_permissions(base_app):
- auth = Mock()
- auth.is_authenticated.return_value = True
- auth.get_nameid.return_value = "user_admin_permissions"
- auth.get_attributes.return_value = {
- "id": ["user_admin_permissions"],
- "groups": ["Admin"],
- }
- base_app.return_value = auth
+ base_app.return_value = _saml_auth_mock(
+ "user_admin_permissions",
+ {"id": ["user_admin_permissions"], "groups": ["Admin"]},
+ )
return TestClient(create_app())
@@ -191,23 +206,15 @@ class TestAwsAuthManager:
for policy_store_id in policy_store_ids:
client.delete_policy_store(policyStoreId=policy_store_id)
- def test_login_admin_redirect(self, client_admin_permissions):
- response = client_admin_permissions.post(
- AUTH_MANAGER_FASTAPI_APP_PREFIX + "/login_callback",
- follow_redirects=False,
- data={"RelayState": "login-redirect"},
- )
+ def test_login_admin_redirect(self, base_app, client_admin_permissions):
+ response = _complete_login(client_admin_permissions,
base_app.return_value, "/login")
token = response.cookies.get(COOKIE_NAME_JWT_TOKEN)
assert response.status_code == 303
assert "location" in response.headers
assert response.headers["location"] == "/"
assert token is not None
- def test_login_admin_token(self, client_admin_permissions):
- response = client_admin_permissions.post(
- AUTH_MANAGER_FASTAPI_APP_PREFIX + "/login_callback",
- follow_redirects=False,
- data={"RelayState": "login-token"},
- )
+ def test_login_admin_token(self, base_app, client_admin_permissions):
+ response = _complete_login(client_admin_permissions,
base_app.return_value, "/login/token")
assert response.status_code == 200
assert response.json()["access_token"]
