File size: 11,455 Bytes
6510698
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
"""
Unit tests for interface components.
"""

import pytest
from unittest.mock import MagicMock, patch
import sys
from pathlib import Path

# Add project root to path
project_root = Path(__file__).parent.parent.parent
sys.path.insert(0, str(project_root))

from app.interface import ExpertClassifier, ResponseGenerator, GradioInterface
from app.model_loader import ModelLoader


class TestExpertClassifier:
    """Test ExpertClassifier functionality."""
    
    def test_classify_code_query(self):
        """Test classification of coding-related queries."""
        queries = [
            "How do I implement a binary search algorithm in Python?",
            "What's the best way to debug JavaScript code?",
            "Can you help me with this API integration?",
            "I need help with my Java class implementation"
        ]
        
        for query in queries:
            result = ExpertClassifier.classify_query(query)
            # Should be Code due to keywords like "algorithm", "Python", "JavaScript", "API", "Java"
            assert result == "Code", f"Query '{query}' should be classified as Code, got {result}"
    
    def test_classify_math_query(self):
        """Test classification of math-related queries."""
        queries = [
            "What is the derivative of x² + 3x + 1?",
            "How do I solve this calculus problem?",
            "Calculate the probability of this event",
            "What's the integral of sin(x)?"
        ]
        
        for query in queries:
            result = ExpertClassifier.classify_query(query)
            assert result == "Math"
    
    def test_classify_reasoning_query(self):
        """Test classification of reasoning-related queries."""
        queries = [
            "Explain the logical reasoning behind the Monty Hall problem",
            "Why does this argument make sense?",
            "How should I analyze this situation?",
            "What's the critical thinking approach here?"
        ]
        
        for query in queries:
            result = ExpertClassifier.classify_query(query)
            assert result == "Reasoning"
    
    def test_classify_multilingual_query(self):
        """Test classification of multilingual queries."""
        queries = [
            "Translate 'Hello, how are you?' to Spanish",
            "What does this French phrase mean?",
            "Help me learn German vocabulary",
            "How do you say 'thank you' in Japanese?"
        ]
        
        for query in queries:
            result = ExpertClassifier.classify_query(query)
            assert result == "Multilingual"
    
    def test_classify_general_query(self):
        """Test classification of general queries."""
        queries = [
            "What are the benefits of renewable energy?",
            "Tell me about the history of computers",
            "Hello, how can you help me?",
            "What's the weather like?"
        ]
        
        for query in queries:
            result = ExpertClassifier.classify_query(query)
            assert result == "General"
    
    def test_classify_empty_query(self):
        """Test classification of empty query."""
        result = ExpertClassifier.classify_query("")
        assert result == "General"
    
    def test_classify_none_query(self):
        """Test classification of None query."""
        result = ExpertClassifier.classify_query(None)
        assert result == "General"
    
    def test_classify_ambiguous_query(self):
        """Test classification of ambiguous query with multiple keywords."""
        # Query with both code and math keywords
        query = "How do I calculate the algorithm complexity in Python?"
        result = ExpertClassifier.classify_query(query)
        # Should pick the one with higher score (Code has 'algorithm' and 'Python')
        assert result in ["Code", "Math"]  # Either is acceptable for ambiguous cases


class TestResponseGenerator:
    """Test ResponseGenerator functionality."""
    
    def setup_method(self):
        """Setup test fixtures."""
        self.mock_model_loader = MagicMock(spec=ModelLoader)
        self.response_generator = ResponseGenerator(self.mock_model_loader)
    
    def test_generate_fallback_response_code(self):
        """Test fallback response generation for code queries."""
        query = "How do I implement a function?"
        expert_type = "Code"
        
        response = self.response_generator.generate_fallback_response(query, expert_type)
        
        assert "Code Expert" in response
        assert query in response
        assert "code examples" in response
    
    def test_generate_fallback_response_math(self):
        """Test fallback response generation for math queries."""
        query = "What is the derivative of x²?"
        expert_type = "Math"
        
        response = self.response_generator.generate_fallback_response(query, expert_type)
        
        assert "Math Expert" in response
        assert query in response
        assert "step-by-step" in response
    
    def test_generate_fallback_response_unknown_expert(self):
        """Test fallback response for unknown expert type."""
        query = "Test query"
        expert_type = "Unknown"
        
        response = self.response_generator.generate_fallback_response(query, expert_type)
        
        # Should default to General expert response
        assert "General Expert" in response
    
    def test_generate_response_model_not_loaded(self):
        """Test response generation when model is not loaded."""
        self.mock_model_loader.is_loaded = False
        
        query = "Test query"
        response = self.response_generator.generate_response(query)
        
        assert "Expert Type:" in response
        assert "model is currently unavailable" in response
    
    def test_generate_response_model_loaded_success(self):
        """Test successful response generation when model is loaded."""
        # Setup mock model loader
        self.mock_model_loader.is_loaded = True
        self.mock_model_loader.pipeline = MagicMock()
        self.mock_model_loader.tokenizer = MagicMock()
        self.mock_model_loader.tokenizer.eos_token_id = 2
        
        # Mock pipeline response
        mock_response = [{
            'generated_text': 'System message\nUser: Test query\nAssistant: This is a test response'
        }]
        self.mock_model_loader.pipeline.return_value = mock_response
        
        query = "Test query"
        response = self.response_generator.generate_response(query)
        
        assert "Expert Type:" in response
        assert "This is a test response" in response
        
        # Verify pipeline was called with correct parameters
        self.mock_model_loader.pipeline.assert_called_once()
        call_args = self.mock_model_loader.pipeline.call_args
        assert call_args[1]['max_new_tokens'] == 500
        assert call_args[1]['temperature'] == 0.7
        assert call_args[1]['do_sample'] is True
    
    def test_generate_response_model_loaded_custom_params(self):
        """Test response generation with custom parameters."""
        self.mock_model_loader.is_loaded = True
        self.mock_model_loader.pipeline = MagicMock()
        self.mock_model_loader.tokenizer = MagicMock()
        self.mock_model_loader.tokenizer.eos_token_id = 2
        
        mock_response = [{'generated_text': 'Test response'}]
        self.mock_model_loader.pipeline.return_value = mock_response
        
        query = "Test query"
        response = self.response_generator.generate_response(
            query, 
            max_tokens=200, 
            temperature=0.5
        )
        
        # Verify custom parameters were used
        call_args = self.mock_model_loader.pipeline.call_args
        assert call_args[1]['max_new_tokens'] == 200
        assert call_args[1]['temperature'] == 0.5
    
    def test_generate_response_pipeline_error(self):
        """Test response generation when pipeline raises an error."""
        self.mock_model_loader.is_loaded = True
        self.mock_model_loader.pipeline = MagicMock()
        self.mock_model_loader.pipeline.side_effect = Exception("Pipeline error")
        
        query = "Test query"
        response = self.response_generator.generate_response(query)
        
        assert "Error generating response" in response
        assert "Pipeline error" in response


class TestGradioInterface:
    """Test GradioInterface functionality."""
    
    def setup_method(self):
        """Setup test fixtures."""
        self.mock_model_loader = MagicMock(spec=ModelLoader)
        self.mock_model_loader.get_model_info.return_value = {
            "status": "loaded",
            "model_id": "test/model",
            "revision": "abc123"
        }
    
    def test_gradio_interface_creation(self):
        """Test GradioInterface creation."""
        interface = GradioInterface(self.mock_model_loader)
        
        assert interface.model_loader == self.mock_model_loader
        assert interface.response_generator is not None
        assert interface.demo is None
    
    @patch('gradio.Blocks')
    def test_create_interface_model_loaded(self, mock_blocks):
        """Test interface creation when model is loaded."""
        self.mock_model_loader.is_loaded = True
        
        interface = GradioInterface(self.mock_model_loader)
        demo = interface.create_interface()
        
        # Verify Blocks was called
        mock_blocks.assert_called_once()
        
        # Verify model info was requested
        self.mock_model_loader.get_model_info.assert_called()
    
    @patch('gradio.Blocks')
    def test_create_interface_model_not_loaded(self, mock_blocks):
        """Test interface creation when model is not loaded."""
        self.mock_model_loader.is_loaded = False
        self.mock_model_loader.get_model_info.return_value = {"status": "not_loaded"}
        
        interface = GradioInterface(self.mock_model_loader)
        demo = interface.create_interface()
        
        # Verify Blocks was called
        mock_blocks.assert_called_once()
    
    @patch('gradio.Blocks')
    def test_launch_creates_interface_if_needed(self, mock_blocks):
        """Test that launch creates interface if it doesn't exist."""
        mock_demo = MagicMock()
        mock_blocks.return_value.__enter__.return_value = mock_demo
        
        interface = GradioInterface(self.mock_model_loader)
        
        # Mock the launch method to avoid actual server startup
        with patch.object(mock_demo, 'launch'):
            interface.launch()
        
        # Verify interface was created
        assert interface.demo is not None
        mock_demo.launch.assert_called_once()
    
    def test_launch_uses_existing_interface(self):
        """Test that launch uses existing interface if available."""
        interface = GradioInterface(self.mock_model_loader)
        mock_demo = MagicMock()
        interface.demo = mock_demo
        
        # Mock the launch method
        with patch.object(mock_demo, 'launch'):
            interface.launch(server_name="127.0.0.1", server_port=8080)
        
        # Verify existing demo was used
        mock_demo.launch.assert_called_once_with(
            server_name="127.0.0.1",
            server_port=8080
        )


if __name__ == "__main__":
    pytest.main([__file__])