From 1e3f0c9c8b241193d66bb335b7d53d935ff87437 Mon Sep 17 00:00:00 2001 From: Lucas Gomide Date: Tue, 17 Feb 2026 16:29:03 -0300 Subject: [PATCH] feat: enhance platform_integration_context with nullcontext support Improve the platform integration context manager to handle None/falsy tokens gracefully by returning nullcontext --- .../tests/test_platform_tools_misc.py | 45 +++ lib/crewai/src/crewai/context.py | 29 +- lib/crewai/tests/test_context.py | 308 +++++++----------- 3 files changed, 180 insertions(+), 202 deletions(-) create mode 100644 lib/crewai-tools/tests/test_platform_tools_misc.py diff --git a/lib/crewai-tools/tests/test_platform_tools_misc.py b/lib/crewai-tools/tests/test_platform_tools_misc.py new file mode 100644 index 000000000..d36c1ded2 --- /dev/null +++ b/lib/crewai-tools/tests/test_platform_tools_misc.py @@ -0,0 +1,45 @@ +"""Tests for platform tools misc functionality.""" + +import os +from unittest.mock import patch + +import pytest +from crewai.context import platform_integration_context +from crewai_tools.tools.crewai_platform_tools.misc import ( + get_platform_integration_token, +) + + + +class TestTokenRetrievalWithFallback: + """Test token retrieval logic with environment fallback.""" + + def test_context_token_takes_precedence(self, clean_context): + """Test that context token takes precedence over environment variable.""" + context_token = "context-token" + env_token = "env-token" + + with patch.dict(os.environ, {"CREWAI_PLATFORM_INTEGRATION_TOKEN": env_token}): + with platform_integration_context(context_token): + token = get_platform_integration_token() + assert token == context_token + + def test_environment_fallback_when_no_context(self, clean_context): + """Test fallback to environment variable when no context token.""" + env_token = "env-fallback-token" + + with patch.dict(os.environ, {"CREWAI_PLATFORM_INTEGRATION_TOKEN": env_token}): + token = get_platform_integration_token() + assert token == env_token + + @pytest.mark.parametrize("empty_value", ["", None]) + def test_missing_token_raises_error(self, clean_context, empty_value): + """Test that missing tokens raise appropriate errors.""" + env_dict = {"CREWAI_PLATFORM_INTEGRATION_TOKEN": empty_value} if empty_value is not None else {} + + with patch.dict(os.environ, env_dict, clear=True): + with pytest.raises(ValueError) as exc_info: + get_platform_integration_token() + + assert "No platform integration token found" in str(exc_info.value) + assert "platform_integration_context()" in str(exc_info.value) diff --git a/lib/crewai/src/crewai/context.py b/lib/crewai/src/crewai/context.py index 633eeae1a..149a85d42 100644 --- a/lib/crewai/src/crewai/context.py +++ b/lib/crewai/src/crewai/context.py @@ -1,8 +1,7 @@ from collections.abc import Generator -from contextlib import contextmanager +from contextlib import contextmanager, nullcontext import contextvars -import os -from typing import Any +from typing import Any, ContextManager _platform_integration_token: contextvars.ContextVar[str | None] = ( @@ -32,18 +31,28 @@ def get_platform_integration_token() -> str | None: return _platform_integration_token.get() -@contextmanager -def platform_integration_context(integration_token: str) -> Generator[None, Any, None]: +def platform_integration_context(integration_token: str | None) -> ContextManager[None]: """Context manager to temporarily set the platform integration token. Args: integration_token: The integration token to set within the context. + If None or falsy, returns nullcontext (no-op). + + Returns: + A context manager that either sets the token or does nothing. """ - token = set_platform_integration_token(integration_token) - try: - yield - finally: - reset_platform_integration_token(token) + if not integration_token: + return nullcontext() + + @contextmanager + def _token_context() -> Generator[None, Any, None]: + token = set_platform_integration_token(integration_token) + try: + yield + finally: + reset_platform_integration_token(token) + + return _token_context() _current_task_id: contextvars.ContextVar[str | None] = contextvars.ContextVar( "current_task_id", default=None diff --git a/lib/crewai/tests/test_context.py b/lib/crewai/tests/test_context.py index 99ce38dde..0320b2d57 100644 --- a/lib/crewai/tests/test_context.py +++ b/lib/crewai/tests/test_context.py @@ -7,215 +7,139 @@ import pytest from crewai.context import ( _platform_integration_token, get_platform_integration_token, - platform_context, + platform_integration_context, + reset_platform_integration_token, set_platform_integration_token, ) -class TestPlatformIntegrationToken: - def setup_method(self): - _platform_integration_token.set(None) +@pytest.fixture +def clean_context(): + """Fixture to ensure clean context state for each test.""" + _platform_integration_token.set(None) + yield + _platform_integration_token.set(None) - def teardown_method(self): - _platform_integration_token.set(None) - @patch.dict(os.environ, {}, clear=True) - def test_set_platform_integration_token(self): +class TestContextVariableCore: + """Test core context variable functionality (set/get/reset).""" + + def test_set_and_get_token(self, clean_context): + """Test basic token setting and retrieval.""" test_token = "test-token-123" assert get_platform_integration_token() is None - set_platform_integration_token(test_token) - + context_token = set_platform_integration_token(test_token) assert get_platform_integration_token() == test_token + assert context_token is not None - def test_get_platform_integration_token_from_context_var(self): - test_token = "context-var-token" - - _platform_integration_token.set(test_token) - - assert get_platform_integration_token() == test_token - - @patch.dict(os.environ, {"CREWAI_PLATFORM_INTEGRATION_TOKEN": "env-token-456"}) - def test_get_platform_integration_token_from_env_var(self): - assert _platform_integration_token.get() is None - - assert get_platform_integration_token() == "env-token-456" - - @patch.dict(os.environ, {"CREWAI_PLATFORM_INTEGRATION_TOKEN": "env-token"}) - def test_context_var_takes_precedence_over_env_var(self): - context_token = "context-token" - - set_platform_integration_token(context_token) - - assert get_platform_integration_token() == context_token - - @patch.dict(os.environ, {}, clear=True) - def test_get_platform_integration_token_returns_none_when_not_set(self): - assert _platform_integration_token.get() is None - - assert get_platform_integration_token() is None - - @patch.dict(os.environ, {}, clear=True) - def test_platform_context_manager_basic_usage(self): - test_token = "context-manager-token" - - assert get_platform_integration_token() is None - - with platform_context(test_token): - assert get_platform_integration_token() == test_token - - assert get_platform_integration_token() is None - - @patch.dict(os.environ, {}, clear=True) - def test_platform_context_manager_nested_contexts(self): - """Test nested platform_context context managers.""" - outer_token = "outer-token" - inner_token = "inner-token" - - assert get_platform_integration_token() is None - - with platform_context(outer_token): - assert get_platform_integration_token() == outer_token - - with platform_context(inner_token): - assert get_platform_integration_token() == inner_token - - assert get_platform_integration_token() == outer_token - - assert get_platform_integration_token() is None - - def test_platform_context_manager_preserves_existing_token(self): - """Test that platform_context preserves existing token when exiting.""" - initial_token = "initial-token" - context_token = "context-token" - - set_platform_integration_token(initial_token) - assert get_platform_integration_token() == initial_token - - with platform_context(context_token): - assert get_platform_integration_token() == context_token - - assert get_platform_integration_token() == initial_token - - def test_platform_context_manager_exception_handling(self): - """Test that platform_context properly resets token even when exception occurs.""" - initial_token = "initial-token" - context_token = "context-token" - - set_platform_integration_token(initial_token) - - with pytest.raises(ValueError): - with platform_context(context_token): - assert get_platform_integration_token() == context_token - raise ValueError("Test exception") - - assert get_platform_integration_token() == initial_token - - @patch.dict(os.environ, {}, clear=True) - def test_platform_context_manager_with_none_initial_state(self): - """Test platform_context when initial state is None.""" - context_token = "context-token" - - assert get_platform_integration_token() is None - - with pytest.raises(RuntimeError): - with platform_context(context_token): - assert get_platform_integration_token() == context_token - raise RuntimeError("Test exception") - - assert get_platform_integration_token() is None - - @patch.dict(os.environ, {"CREWAI_PLATFORM_INTEGRATION_TOKEN": "env-backup"}) - def test_platform_context_with_env_fallback(self): - """Test platform_context interaction with environment variable fallback.""" - context_token = "context-token" - - assert get_platform_integration_token() == "env-backup" - - with platform_context(context_token): - assert get_platform_integration_token() == context_token - - assert get_platform_integration_token() == "env-backup" - - @patch.dict(os.environ, {}, clear=True) - def test_multiple_sequential_context_managers(self): - """Test multiple sequential uses of platform_context.""" + def test_reset_token_restores_previous_state(self, clean_context): + """Test that reset properly restores previous context state.""" token1 = "token-1" token2 = "token-2" - token3 = "token-3" - with platform_context(token1): - assert get_platform_integration_token() == token1 + context_token1 = set_platform_integration_token(token1) + assert get_platform_integration_token() == token1 + + context_token2 = set_platform_integration_token(token2) + assert get_platform_integration_token() == token2 + + reset_platform_integration_token(context_token2) + assert get_platform_integration_token() == token1 + + reset_platform_integration_token(context_token1) + assert get_platform_integration_token() is None + + def test_nested_token_management(self, clean_context): + """Test proper token management with deeply nested contexts.""" + tokens = ["token-1", "token-2", "token-3"] + context_tokens = [] + + for token in tokens: + context_tokens.append(set_platform_integration_token(token)) + assert get_platform_integration_token() == token + + for i in range(len(tokens) - 1, 0, -1): + reset_platform_integration_token(context_tokens[i]) + assert get_platform_integration_token() == tokens[i - 1] + + reset_platform_integration_token(context_tokens[0]) + assert get_platform_integration_token() is None + + @patch.dict(os.environ, {"CREWAI_PLATFORM_INTEGRATION_TOKEN": "env-token"}) + def test_context_module_ignores_environment_variables(self, clean_context): + """Test that context module only returns context values, not env vars.""" + # Context module should not read environment variables + assert get_platform_integration_token() is None + + # Only context variable should be returned + set_platform_integration_token("context-token") + assert get_platform_integration_token() == "context-token" + + +class TestPlatformIntegrationContext: + """Test platform integration context manager behavior.""" + + def test_basic_context_manager_usage(self, clean_context): + """Test basic context manager functionality.""" + test_token = "context-token" assert get_platform_integration_token() is None - with platform_context(token2): - assert get_platform_integration_token() == token2 - - assert get_platform_integration_token() is None - - with platform_context(token3): - assert get_platform_integration_token() == token3 - - assert get_platform_integration_token() is None - - def test_empty_string_token(self): - empty_token = "" - - set_platform_integration_token(empty_token) - assert get_platform_integration_token() == "" - - with platform_context(empty_token): - assert get_platform_integration_token() == "" - - def test_special_characters_in_token(self): - special_token = "token-with-!@#$%^&*()_+-={}[]|\\:;\"'<>?,./" - - set_platform_integration_token(special_token) - assert get_platform_integration_token() == special_token - - with platform_context(special_token): - assert get_platform_integration_token() == special_token - - def test_very_long_token(self): - long_token = "a" * 10000 - - set_platform_integration_token(long_token) - assert get_platform_integration_token() == long_token - - with platform_context(long_token): - assert get_platform_integration_token() == long_token - - @patch.dict(os.environ, {"CREWAI_PLATFORM_INTEGRATION_TOKEN": ""}) - def test_empty_env_var(self): - assert _platform_integration_token.get() is None - assert get_platform_integration_token() == "" - - @patch("crewai.context.os.getenv") - def test_env_var_access_error_handling(self, mock_getenv): - mock_getenv.side_effect = OSError("Environment access error") - - with pytest.raises(OSError): - get_platform_integration_token() - - @patch.dict(os.environ, {}, clear=True) - def test_context_var_isolation_between_tests(self): - """Test that context variable changes don't leak between test methods.""" - test_token = "isolation-test-token" - - assert get_platform_integration_token() is None - - set_platform_integration_token(test_token) - assert get_platform_integration_token() == test_token - - def test_context_manager_return_value(self): - """Test that platform_context can be used in with statement with return value.""" - test_token = "return-value-token" - - with platform_context(test_token): + with platform_integration_context(test_token): assert get_platform_integration_token() == test_token - with platform_context(test_token) as ctx: - assert ctx is None - assert get_platform_integration_token() == test_token + assert get_platform_integration_token() is None + + @pytest.mark.parametrize("falsy_value", [None, "", False, 0]) + def test_falsy_values_return_nullcontext(self, clean_context, falsy_value): + """Test that falsy values return nullcontext (no-op).""" + # Set initial token to verify nullcontext doesn't affect it + initial_token = "initial-token" + initial_context_token = set_platform_integration_token(initial_token) + + try: + with platform_integration_context(falsy_value): + # Should preserve existing context (nullcontext behavior) + assert get_platform_integration_token() == initial_token + + # Should still have initial token after nullcontext + assert get_platform_integration_token() == initial_token + finally: + reset_platform_integration_token(initial_context_token) + + @pytest.mark.parametrize("truthy_value", ["token", "123", " ", "0"]) + def test_truthy_values_create_context(self, clean_context, truthy_value): + """Test that truthy values create proper context.""" + with platform_integration_context(truthy_value): + assert get_platform_integration_token() == truthy_value + + # Should be cleaned up + assert get_platform_integration_token() is None + + def test_context_preserves_existing_token(self, clean_context): + """Test that context manager preserves existing token when exiting.""" + existing_token = "existing-token" + context_token = "context-token" + + existing_context_token = set_platform_integration_token(existing_token) + + try: + with platform_integration_context(context_token): + assert get_platform_integration_token() == context_token + + assert get_platform_integration_token() == existing_token + finally: + reset_platform_integration_token(existing_context_token) + + def test_context_manager_return_type(self, clean_context): + """Test that context manager returns proper types for both cases.""" + # Both should be usable as context managers + valid_ctx = platform_integration_context("token") + none_ctx = platform_integration_context(None) + + assert hasattr(valid_ctx, '__enter__') + assert hasattr(valid_ctx, '__exit__') + assert hasattr(none_ctx, '__enter__') + assert hasattr(none_ctx, '__exit__')