mirror of
https://github.com/langgenius/dify.git
synced 2026-10-16 15:35:32 +08:00
refactor(api): enable strict typing for inner API auth wrapper tests (#43817)
This commit is contained in:
@@ -2,6 +2,7 @@
|
||||
Unit tests for inner_api auth decorators
|
||||
"""
|
||||
|
||||
from collections.abc import Callable
|
||||
from uuid import NAMESPACE_URL, uuid5
|
||||
|
||||
import pytest
|
||||
@@ -21,7 +22,7 @@ from models.model import EndUser
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _inner_api_config(config_overrides) -> None:
|
||||
def _inner_api_config(config_overrides: Callable[..., None]) -> None:
|
||||
config_overrides(
|
||||
INNER_API=True,
|
||||
INNER_API_KEY="valid_key",
|
||||
@@ -37,12 +38,12 @@ def _stable_uuid(value: str) -> str:
|
||||
class TestEnterpriseInnerApiOnly:
|
||||
"""Test enterprise_inner_api_only decorator"""
|
||||
|
||||
def test_should_allow_when_inner_api_enabled_and_valid_key(self, app: Flask):
|
||||
def test_should_allow_when_inner_api_enabled_and_valid_key(self, app: Flask) -> None:
|
||||
"""Test that valid API key allows access when INNER_API is enabled"""
|
||||
|
||||
# Arrange
|
||||
@enterprise_inner_api_only
|
||||
def protected_view():
|
||||
def protected_view() -> str:
|
||||
return "success"
|
||||
|
||||
# Act
|
||||
@@ -52,12 +53,12 @@ class TestEnterpriseInnerApiOnly:
|
||||
# Assert
|
||||
assert result == "success"
|
||||
|
||||
def test_should_return_404_when_inner_api_disabled(self, app: Flask, config_overrides):
|
||||
def test_should_return_404_when_inner_api_disabled(self, app: Flask, config_overrides: Callable[..., None]) -> None:
|
||||
"""Test that 404 is returned when INNER_API is disabled"""
|
||||
|
||||
# Arrange
|
||||
@enterprise_inner_api_only
|
||||
def protected_view():
|
||||
def protected_view() -> str:
|
||||
return "success"
|
||||
|
||||
# Act & Assert
|
||||
@@ -67,12 +68,12 @@ class TestEnterpriseInnerApiOnly:
|
||||
protected_view()
|
||||
assert exc_info.value.code == 404
|
||||
|
||||
def test_should_return_401_when_api_key_missing(self, app: Flask):
|
||||
def test_should_return_401_when_api_key_missing(self, app: Flask) -> None:
|
||||
"""Test that 401 is returned when X-Inner-Api-Key header is missing"""
|
||||
|
||||
# Arrange
|
||||
@enterprise_inner_api_only
|
||||
def protected_view():
|
||||
def protected_view() -> str:
|
||||
return "success"
|
||||
|
||||
# Act & Assert
|
||||
@@ -81,12 +82,12 @@ class TestEnterpriseInnerApiOnly:
|
||||
protected_view()
|
||||
assert exc_info.value.code == 401
|
||||
|
||||
def test_should_return_401_when_api_key_invalid(self, app: Flask):
|
||||
def test_should_return_401_when_api_key_invalid(self, app: Flask) -> None:
|
||||
"""Test that 401 is returned when X-Inner-Api-Key header is invalid"""
|
||||
|
||||
# Arrange
|
||||
@enterprise_inner_api_only
|
||||
def protected_view():
|
||||
def protected_view() -> str:
|
||||
return "success"
|
||||
|
||||
# Act & Assert
|
||||
@@ -99,9 +100,9 @@ class TestEnterpriseInnerApiOnly:
|
||||
class TestInnerApiOnly:
|
||||
"""Test inner_api_only decorator."""
|
||||
|
||||
def test_should_allow_when_inner_api_enabled_and_valid_key(self, app: Flask):
|
||||
def test_should_allow_when_inner_api_enabled_and_valid_key(self, app: Flask) -> None:
|
||||
@inner_api_only
|
||||
def protected_view():
|
||||
def protected_view() -> str:
|
||||
return "success"
|
||||
|
||||
with app.test_request_context(headers={"X-Inner-Api-Key": "valid_key"}):
|
||||
@@ -109,9 +110,9 @@ class TestInnerApiOnly:
|
||||
|
||||
assert result == "success"
|
||||
|
||||
def test_should_return_404_when_inner_api_disabled(self, app: Flask, config_overrides):
|
||||
def test_should_return_404_when_inner_api_disabled(self, app: Flask, config_overrides: Callable[..., None]) -> None:
|
||||
@inner_api_only
|
||||
def protected_view():
|
||||
def protected_view() -> str:
|
||||
return "success"
|
||||
|
||||
config_overrides(INNER_API=False)
|
||||
@@ -120,9 +121,9 @@ class TestInnerApiOnly:
|
||||
protected_view()
|
||||
assert exc_info.value.code == 404
|
||||
|
||||
def test_should_return_401_when_api_key_missing(self, app: Flask):
|
||||
def test_should_return_401_when_api_key_missing(self, app: Flask) -> None:
|
||||
@inner_api_only
|
||||
def protected_view():
|
||||
def protected_view() -> str:
|
||||
return "success"
|
||||
|
||||
with app.test_request_context(headers={}):
|
||||
@@ -130,9 +131,9 @@ class TestInnerApiOnly:
|
||||
protected_view()
|
||||
assert exc_info.value.code == 401
|
||||
|
||||
def test_should_return_401_when_api_key_invalid(self, app: Flask):
|
||||
def test_should_return_401_when_api_key_invalid(self, app: Flask) -> None:
|
||||
@inner_api_only
|
||||
def protected_view():
|
||||
def protected_view() -> str:
|
||||
return "success"
|
||||
|
||||
with app.test_request_context(headers={"X-Inner-Api-Key": "invalid_key"}):
|
||||
@@ -144,12 +145,14 @@ class TestInnerApiOnly:
|
||||
class TestEnterpriseInnerApiUserAuth:
|
||||
"""Test enterprise_inner_api_user_auth decorator for HMAC-based user authentication"""
|
||||
|
||||
def test_should_pass_through_when_inner_api_disabled(self, app: Flask, config_overrides):
|
||||
def test_should_pass_through_when_inner_api_disabled(
|
||||
self, app: Flask, config_overrides: Callable[..., None]
|
||||
) -> None:
|
||||
"""Test that request passes through when INNER_API is disabled"""
|
||||
|
||||
# Arrange
|
||||
@enterprise_inner_api_user_auth
|
||||
def protected_view(**kwargs):
|
||||
def protected_view(**kwargs: object) -> object:
|
||||
return kwargs.get("user", "no_user")
|
||||
|
||||
# Act
|
||||
@@ -160,12 +163,12 @@ class TestEnterpriseInnerApiUserAuth:
|
||||
# Assert
|
||||
assert result == "no_user"
|
||||
|
||||
def test_should_pass_through_when_authorization_header_missing(self, app: Flask):
|
||||
def test_should_pass_through_when_authorization_header_missing(self, app: Flask) -> None:
|
||||
"""Test that request passes through when Authorization header is missing"""
|
||||
|
||||
# Arrange
|
||||
@enterprise_inner_api_user_auth
|
||||
def protected_view(**kwargs):
|
||||
def protected_view(**kwargs: object) -> object:
|
||||
return kwargs.get("user", "no_user")
|
||||
|
||||
# Act
|
||||
@@ -175,12 +178,12 @@ class TestEnterpriseInnerApiUserAuth:
|
||||
# Assert
|
||||
assert result == "no_user"
|
||||
|
||||
def test_should_pass_through_when_authorization_format_invalid(self, app: Flask):
|
||||
def test_should_pass_through_when_authorization_format_invalid(self, app: Flask) -> None:
|
||||
"""Test that request passes through when Authorization format is invalid (no colon)"""
|
||||
|
||||
# Arrange
|
||||
@enterprise_inner_api_user_auth
|
||||
def protected_view(**kwargs):
|
||||
def protected_view(**kwargs: object) -> object:
|
||||
return kwargs.get("user", "no_user")
|
||||
|
||||
# Act
|
||||
@@ -190,15 +193,15 @@ class TestEnterpriseInnerApiUserAuth:
|
||||
# Assert
|
||||
assert result == "no_user"
|
||||
|
||||
def test_should_pass_through_when_hmac_signature_invalid(self, app: Flask, sqlite_engine: Engine):
|
||||
def test_should_pass_through_when_hmac_signature_invalid(self, app: Flask, sqlite_engine: Engine) -> None:
|
||||
"""Invalid HMAC auth passes through without opening a database session."""
|
||||
|
||||
# Arrange
|
||||
@enterprise_inner_api_user_auth
|
||||
def protected_view(**kwargs):
|
||||
def protected_view(**kwargs: object) -> object:
|
||||
return kwargs.get("user", "no_user")
|
||||
|
||||
def fail_on_query(*_args, **_kwargs):
|
||||
def fail_on_query(*_args: object, **_kwargs: object) -> None:
|
||||
pytest.fail("invalid HMAC must not access the database")
|
||||
|
||||
event.listen(sqlite_engine, "before_cursor_execute", fail_on_query)
|
||||
@@ -213,7 +216,7 @@ class TestEnterpriseInnerApiUserAuth:
|
||||
assert result == "no_user"
|
||||
|
||||
@pytest.mark.parametrize("sqlite_session", [(EndUser,)], indirect=True)
|
||||
def test_should_inject_user_when_hmac_signature_valid(self, app: Flask, sqlite_session: Session):
|
||||
def test_should_inject_user_when_hmac_signature_valid(self, app: Flask, sqlite_session: Session) -> None:
|
||||
"""Test that user is injected when HMAC signature is valid"""
|
||||
# Arrange
|
||||
from base64 import b64encode
|
||||
@@ -221,7 +224,7 @@ class TestEnterpriseInnerApiUserAuth:
|
||||
from hmac import new as hmac_new
|
||||
|
||||
@enterprise_inner_api_user_auth
|
||||
def protected_view(**kwargs):
|
||||
def protected_view(**kwargs: object) -> object:
|
||||
return kwargs.get("user")
|
||||
|
||||
# Calculate valid HMAC signature
|
||||
@@ -257,12 +260,12 @@ class TestEnterpriseInnerApiUserAuth:
|
||||
class TestPluginInnerApiOnly:
|
||||
"""Test plugin_inner_api_only decorator"""
|
||||
|
||||
def test_should_allow_when_plugin_daemon_key_set_and_valid_key(self, app: Flask):
|
||||
def test_should_allow_when_plugin_daemon_key_set_and_valid_key(self, app: Flask) -> None:
|
||||
"""Test that valid API key allows access when PLUGIN_DAEMON_KEY is set"""
|
||||
|
||||
# Arrange
|
||||
@plugin_inner_api_only
|
||||
def protected_view():
|
||||
def protected_view() -> str:
|
||||
return "success"
|
||||
|
||||
# Act
|
||||
@@ -272,12 +275,14 @@ class TestPluginInnerApiOnly:
|
||||
# Assert
|
||||
assert result == "success"
|
||||
|
||||
def test_should_return_404_when_plugin_daemon_key_not_set(self, app: Flask, config_overrides):
|
||||
def test_should_return_404_when_plugin_daemon_key_not_set(
|
||||
self, app: Flask, config_overrides: Callable[..., None]
|
||||
) -> None:
|
||||
"""Test that 404 is returned when PLUGIN_DAEMON_KEY is not set"""
|
||||
|
||||
# Arrange
|
||||
@plugin_inner_api_only
|
||||
def protected_view():
|
||||
def protected_view() -> str:
|
||||
return "success"
|
||||
|
||||
# Act & Assert
|
||||
@@ -287,12 +292,12 @@ class TestPluginInnerApiOnly:
|
||||
protected_view()
|
||||
assert exc_info.value.code == 404
|
||||
|
||||
def test_should_return_404_when_api_key_invalid(self, app: Flask):
|
||||
def test_should_return_404_when_api_key_invalid(self, app: Flask) -> None:
|
||||
"""Test that 404 is returned when X-Inner-Api-Key header is invalid (note: returns 404, not 401)"""
|
||||
|
||||
# Arrange
|
||||
@plugin_inner_api_only
|
||||
def protected_view():
|
||||
def protected_view() -> str:
|
||||
return "success"
|
||||
|
||||
# Act & Assert
|
||||
|
||||
@@ -135,7 +135,6 @@ project-excludes = [
|
||||
"controllers/inner_api/plugin/test_agent_config.py",
|
||||
"controllers/inner_api/plugin/test_plugin.py",
|
||||
"controllers/inner_api/plugin/test_plugin_wraps.py",
|
||||
"controllers/inner_api/test_auth_wraps.py",
|
||||
"controllers/inner_api/test_knowledge_retrieval.py",
|
||||
"controllers/inner_api/test_mail.py",
|
||||
"controllers/inner_api/test_runtime_credentials.py",
|
||||
|
||||
Reference in New Issue
Block a user