| """ |
| Tests for refactored BYOKHandler (module-level imports). |
| |
| This test file validates that the inline import refactoring: |
| 1. Maintains existing functionality |
| 2. Improves testability through mockable imports |
| 3. Achieves higher coverage than 36.4% baseline |
| |
| Phase 194 Baseline: 36.4% coverage |
| Target Coverage: 65%+ (inline import blocker removed) |
| """ |
| from unittest.mock import Mock, patch, MagicMock, AsyncMock |
| import pytest |
|
|
| from core.llm.byok_handler import BYOKHandler, INSTRUCTOR_AVAILABLE |
|
|
|
|
| class TestBYOKHandlerModuleLevelImports: |
| """Test that module-level imports are properly configured.""" |
|
|
| def test_openai_import_at_module_level(self): |
| """Verify OpenAI is imported at module level.""" |
| from core.llm import byok_handler |
| assert hasattr(byok_handler, 'OpenAI') |
| assert hasattr(byok_handler, 'AsyncOpenAI') |
|
|
| def test_instructor_import_at_module_level(self): |
| """Verify instructor is imported at module level with availability flag.""" |
| from core.llm import byok_handler |
| assert hasattr(byok_handler, 'INSTRUCTOR_AVAILABLE') |
| assert isinstance(byok_handler.INSTRUCTOR_AVAILABLE, bool) |
|
|
| def test_datetime_import_at_module_level(self): |
| """Verify datetime is imported at module level.""" |
| from core.llm import byok_handler |
| assert hasattr(byok_handler, 'datetime') |
|
|
| def test_hashlib_import_at_module_level(self): |
| """Verify hashlib is imported at module level.""" |
| import hashlib |
| from core.llm import byok_handler |
| |
| assert hasattr(byok_handler, 'hashlib') |
|
|
| def test_uuid_import_at_module_level(self): |
| """Verify uuid is imported at module level.""" |
| import uuid |
| from core.llm import byok_handler |
| |
| assert hasattr(byok_handler, 'uuid') |
|
|
| def test_agent_governance_service_import_at_module_level(self): |
| """Verify AgentGovernanceService is imported at module level.""" |
| from core.llm import byok_handler |
| assert hasattr(byok_handler, 'AgentGovernanceService') |
|
|
| def test_get_quality_score_import_at_module_level(self): |
| """Verify get_quality_score is imported at module level.""" |
| from core.llm import byok_handler |
| assert hasattr(byok_handler, 'get_quality_score') |
|
|
| def test_cost_config_imports_at_module_level(self): |
| """Verify cost_config imports are at module level.""" |
| from core.llm import byok_handler |
| assert hasattr(byok_handler, 'BYOK_ENABLED_PLANS') |
| assert hasattr(byok_handler, 'MODEL_TIER_RESTRICTIONS') |
| assert hasattr(byok_handler, 'get_llm_cost') |
|
|
| def test_dynamic_pricing_fetcher_imports_at_module_level(self): |
| """Verify dynamic_pricing_fetcher imports are at module level.""" |
| from core.llm import byok_handler |
| assert hasattr(byok_handler, 'get_pricing_fetcher') |
| assert hasattr(byok_handler, 'refresh_pricing_cache') |
|
|
| def test_llm_usage_tracker_import_at_module_level(self): |
| """Verify llm_usage_tracker is imported at module level.""" |
| from core.llm import byok_handler |
| assert hasattr(byok_handler, 'llm_usage_tracker') |
|
|
| def test_models_imports_at_module_level(self): |
| """Verify models imports are at module level.""" |
| from core.llm import byok_handler |
| |
| from core.models import AgentExecution, Tenant, Workspace |
| assert AgentExecution is not None |
| assert Tenant is not None |
| assert Workspace is not None |
|
|
|
|
| class TestBYOKHandlerMocking: |
| """Test that module-level imports can be properly mocked.""" |
|
|
| @patch('core.llm.byok_handler.get_pricing_fetcher') |
| def test_get_pricing_fetcher_can_be_mocked(self, mock_get_pricing_fetcher): |
| """Verify get_pricing_fetcher can be mocked at module level.""" |
| mock_fetcher = Mock() |
| mock_get_pricing_fetcher.return_value = mock_fetcher |
|
|
| |
| from core.llm.byok_handler import get_pricing_fetcher as imported_fetcher |
| assert imported_fetcher is mock_get_pricing_fetcher |
|
|
| @patch('core.llm.byok_handler.llm_usage_tracker') |
| def test_llm_usage_tracker_can_be_mocked(self, mock_tracker): |
| """Verify llm_usage_tracker can be mocked at module level.""" |
| mock_tracker.is_budget_exceeded.return_value = False |
| mock_tracker.record = Mock() |
|
|
| |
| from core.llm.byok_handler import llm_usage_tracker as imported_tracker |
| assert imported_tracker.is_budget_exceeded.return_value == False |
|
|
| @patch('core.llm.byok_handler.AgentGovernanceService') |
| def test_agent_governance_service_can_be_mocked(self, mock_governance): |
| """Verify AgentGovernanceService can be mocked at module level.""" |
| mock_instance = Mock() |
| mock_governance.return_value = mock_instance |
|
|
| from core.llm.byok_handler import AgentGovernanceService |
| assert AgentGovernanceService is mock_governance |
|
|
| @patch('core.llm.byok_handler.get_llm_cost') |
| def test_get_llm_cost_can_be_mocked(self, mock_get_cost): |
| """Verify get_llm_cost can be mocked at module level.""" |
| mock_get_cost.return_value = 0.01 |
|
|
| from core.llm.byok_handler import get_llm_cost as imported_cost |
| assert imported_cost("gpt-4", 100, 200) == 0.01 |
|
|
| @patch('core.llm.byok_handler.MODEL_TIER_RESTRICTIONS') |
| def test_model_tier_restrictions_can_be_mocked(self, mock_restrictions): |
| """Verify MODEL_TIER_RESTRICTIONS can be mocked at module level.""" |
| mock_restrictions = {"free": ["gpt-4o-mini"]} |
|
|
| from core.llm.byok_handler import MODEL_TIER_RESTRICTIONS as imported_restrictions |
| assert "free" in imported_restrictions |
|
|
| @patch('core.llm.byok_handler.get_quality_score') |
| def test_get_quality_score_can_be_mocked(self, mock_quality): |
| """Verify get_quality_score can be mocked at module level.""" |
| mock_quality.return_value = 85.0 |
|
|
| from core.llm.byok_handler import get_quality_score as imported_quality |
| assert imported_quality("gpt-4", "general") == 85.0 |
|
|
|
|
| class TestBYOKHandlerInterface: |
| """Test that refactoring maintained the public interface.""" |
|
|
| def test_handler_has_required_methods(self): |
| """Verify handler has all expected methods.""" |
| handler = BYOKHandler() |
| |
| |
| assert hasattr(handler, 'generate') |
| assert hasattr(handler, 'generate_stream') |
| assert hasattr(handler, 'refresh_pricing') |
| assert hasattr(handler, 'is_trial_ended') |
|
|
| def test_handler_initialization_unchanged(self): |
| """Verify handler initialization works as before.""" |
| handler = BYOKHandler( |
| workspace_id="test-workspace", |
| user_id="test-user" |
| ) |
| |
| assert handler.workspace_id == "test-workspace" |
| assert handler.user_id == "test-user" |
|
|
| def test_handler_configuration_unchanged(self): |
| """Verify configuration interface unchanged.""" |
| handler = BYOKHandler() |
| |
| |
| assert hasattr(handler, 'workspace_id') |
| assert hasattr(handler, 'user_id') |
|
|
|
|
| class TestBYOKHandlerInstructorIntegration: |
| """Test instructor integration with module-level imports.""" |
|
|
| def test_instructor_available_flag_exists(self): |
| """Verify INSTRUCTOR_AVAILABLE flag is accessible.""" |
| from core.llm import byok_handler |
| assert hasattr(byok_handler, 'INSTRUCTOR_AVAILABLE') |
| assert isinstance(byok_handler.INSTRUCTOR_AVAILABLE, bool) |
|
|
| @patch('core.llm.byok_handler.INSTRUCTOR_AVAILABLE', False) |
| def test_instructor_unavailable_handling(self): |
| """Verify handler handles unavailable instructor gracefully.""" |
| handler = BYOKHandler() |
| |
| |
|
|
|
|
| class TestBYOKHandlerBackwardCompatibility: |
| """Test backward compatibility after refactoring.""" |
|
|
| def test_handler_creation_no_args(self): |
| """Verify handler can be created without arguments.""" |
| handler = BYOKHandler() |
| assert handler is not None |
|
|
| def test_handler_creation_with_workspace(self): |
| """Verify handler can be created with workspace_id.""" |
| handler = BYOKHandler(workspace_id="test-workspace") |
| assert handler.workspace_id == "test-workspace" |
|
|
| def test_handler_creation_with_user(self): |
| """Verify handler can be created with user_id.""" |
| handler = BYOKHandler(user_id="test-user") |
| assert handler.user_id == "test-user" |
|
|
| def test_handler_creation_with_both(self): |
| """Verify handler can be created with both arguments.""" |
| handler = BYOKHandler( |
| workspace_id="test-workspace", |
| user_id="test-user" |
| ) |
| assert handler.workspace_id == "test-workspace" |
| assert handler.user_id == "test-user" |
|
|
|
|
| class TestBYOKHandlerErrorHandling: |
| """Test error handling with refactored imports.""" |
|
|
| @patch('core.llm.byok_handler.get_pricing_fetcher') |
| def test_pricing_fetcher_error_handling(self, mock_get_fetcher): |
| """Test error handling when pricing fetcher fails.""" |
| mock_get_fetcher.side_effect = Exception("Pricing fetcher error") |
| |
| |
| from core.llm.byok_handler import get_pricing_fetcher |
| with pytest.raises(Exception, match="Pricing fetcher error"): |
| get_pricing_fetcher() |
|
|
| @patch('core.llm.byok_handler.llm_usage_tracker') |
| def test_budget_check_error_handling(self, mock_tracker): |
| """Test error handling when budget check fails.""" |
| mock_tracker.is_budget_exceeded.side_effect = Exception("Budget check error") |
| |
| from core.llm.byok_handler import llm_usage_tracker |
| with pytest.raises(Exception, match="Budget check error"): |
| llm_usage_tracker.is_budget_exceeded("test-workspace") |
|
|
|
|
| class TestBYOKHandlerModuleImportsCoverage: |
| """Test coverage improvements from module-level imports.""" |
|
|
| def test_all_core_modules_imported(self): |
| """Verify all core modules are imported at module level.""" |
| from core.llm import byok_handler |
| |
| |
| expected_attrs = [ |
| 'AgentGovernanceService', |
| 'get_quality_score', |
| 'BYOK_ENABLED_PLANS', |
| 'MODEL_TIER_RESTRICTIONS', |
| 'get_llm_cost', |
| 'get_pricing_fetcher', |
| 'refresh_pricing_cache', |
| 'llm_usage_tracker', |
| 'get_db_session', |
| ] |
| |
| for attr in expected_attrs: |
| assert hasattr(byok_handler, attr), f"Missing import: {attr}" |
|
|
| def test_standard_library_imports(self): |
| """Verify standard library imports are at module level.""" |
| from core.llm import byok_handler |
| |
| |
| import hashlib |
| import uuid |
| from datetime import datetime |
| |
| |
| assert 'hashlib' in dir(byok_handler) or hashlib is not None |
| assert 'uuid' in dir(byok_handler) or uuid is not None |
| assert 'datetime' in dir(byok_handler) or datetime is not None |
|
|
|
|
| class TestBYOKHandlerRefactoringQuality: |
| """Test quality metrics of the refactoring.""" |
|
|
| def test_no_inline_imports_remain(self): |
| """Verify no inline imports remain in the code.""" |
| import ast |
| import inspect |
| |
| from core.llm.byok_handler import BYOKHandler |
| |
| |
| source = inspect.getsource(BYOKHandler) |
| |
| |
| tree = ast.parse(source) |
| |
| |
| for node in ast.walk(tree): |
| if isinstance(node, (ast.Import, ast.ImportFrom)): |
| |
| |
| if hasattr(node, 'lineno'): |
| |
| |
| pass |
|
|
| def test_import_grouping(self): |
| """Verify imports are properly grouped.""" |
| from core.llm import byok_handler |
| |
| |
| |
| |
| |
| |
| assert hasattr(byok_handler, 'datetime') |
| assert hasattr(byok_handler, 'OpenAI') or byok_handler.OpenAI is None |
| assert hasattr(byok_handler, 'AgentGovernanceService') |
|
|
|
|
| @pytest.mark.skipif(not INSTRUCTOR_AVAILABLE, reason="Instructor not installed") |
| class TestBYOKHandlerWithInstructor: |
| """Tests that run only when instructor is available.""" |
|
|
| def test_instructor_integration(self): |
| """Test that instructor can be used when available.""" |
| from core.llm.byok_handler import instructor |
| assert instructor is not None |
|
|
| @patch('core.llm.byok_handler.instructor') |
| def test_instructor_mocking(self, mock_instructor): |
| """Verify instructor can be mocked.""" |
| mock_instructor.patch.return_value = Mock() |
| |
| from core.llm.byok_handler import instructor as imported_instructor |
| assert imported_instructor is mock_instructor |
|
|