File: test_unit_message_handling.py

package info (click to toggle)
python-azure 20251118%2Bgit-1
  • links: PTS, VCS
  • area: main
  • in suites: sid
  • size: 783,356 kB
  • sloc: python: 6,474,533; ansic: 804; javascript: 287; sh: 205; makefile: 198; xml: 109
file content (453 lines) | stat: -rw-r--r-- 18,117 bytes parent folder | download | duplicates (2)
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
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
# coding=utf-8
# --------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License.
# --------------------------------------------------------------------------

import base64
from typing import List

from azure.ai.voicelive.models import (
    # Message Content Parts
    MessageContentPart,
    InputAudioContentPart,
    InputTextContentPart,
    OutputTextContentPart,
    # Message Items
    MessageItem,
    UserMessageItem,
    AssistantMessageItem,
    SystemMessageItem,
    # Enums
    MessageRole,
    ContentPartType,
    ItemParamStatus,
    ItemType,
    # Supporting Models
    ConversationRequestItem,
    InputAudioFormat,
)


class TestMessageContentPartHierarchy:
    """Test MessageContentPart hierarchy and inheritance."""

    def test_message_content_part_base_class(self):
        """Test MessageContentPart as base class."""
        # MessageContentPart is an abstract base, so we test through subclasses
        text_part = InputTextContentPart(text="Hello")
        audio_part = InputAudioContentPart(audio=b"audio data")
        output_part = OutputTextContentPart(text="Response")

        # All should inherit from MessageContentPart
        assert isinstance(text_part, MessageContentPart)
        assert isinstance(audio_part, MessageContentPart)
        assert isinstance(output_part, MessageContentPart)

    def test_content_part_type_discrimination(self):
        """Test that content parts are properly discriminated by type."""
        text_part = InputTextContentPart(text="Hello")
        audio_part = InputAudioContentPart(audio=b"audio")
        output_part = OutputTextContentPart(text="Output")

        assert text_part.type == "input_text"
        assert audio_part.type == "input_audio"
        assert output_part.type == "text"

    def test_content_part_polymorphism(self):
        """Test polymorphic usage of content parts."""
        parts: List[MessageContentPart] = [
            InputTextContentPart(text="Text input"),
            InputAudioContentPart(audio=b"audio input"),
            OutputTextContentPart(text="Text output"),
        ]

        # All items should be MessageContentPart instances
        for part in parts:
            assert isinstance(part, MessageContentPart)
            assert hasattr(part, "type")

        # Types should be different
        types = [part.type for part in parts]
        assert "input_text" in types
        assert "input_audio" in types
        assert "text" in types


class TestInputTextContentPart:
    """Test InputTextContentPart model."""

    def test_create_input_text_content_part(self):
        """Test creating input text content part."""
        text = "Hello, how can I help you today?"
        part = InputTextContentPart(text=text)

        assert part.type == "input_text"
        assert part.text == text
        assert isinstance(part, MessageContentPart)

    def test_input_text_content_part_empty_string(self):
        """Test input text content part with empty string."""
        part = InputTextContentPart(text="")

        assert part.type == "input_text"
        assert part.text == ""

    def test_input_text_content_part_long_text(self):
        """Test input text content part with long text."""
        long_text = "This is a very long text that represents a substantial user input. " * 100
        part = InputTextContentPart(text=long_text)

        assert part.type == "input_text"
        assert part.text == long_text
        assert len(part.text) > 1000

    def test_input_text_content_part_unicode(self):
        """Test input text content part with unicode characters."""
        unicode_text = "Hello 世界! 🌍 Café naïve résumé"
        part = InputTextContentPart(text=unicode_text)

        assert part.type == "input_text"
        assert part.text == unicode_text


class TestInputAudioContentPart:
    """Test InputAudioContentPart model."""

    def test_create_input_audio_content_part(self):
        """Test creating input audio content part."""
        audio_data = b"fake audio binary data"
        part = InputAudioContentPart(audio=audio_data)

        assert part.type == "input_audio"
        assert part.audio == base64.b64encode(audio_data).decode("ascii")
        assert isinstance(part, MessageContentPart)

    def test_input_audio_content_part_empty_data(self):
        """Test input audio content part with empty data."""
        part = InputAudioContentPart(audio=b"")

        assert part.type == "input_audio"
        assert part.audio == ""

    def test_input_audio_content_part_large_data(self):
        """Test input audio content part with large audio data."""
        large_audio = b"audio data chunk " * 1000
        part = InputAudioContentPart(audio=large_audio)

        assert part.type == "input_audio"
        assert part.audio == base64.b64encode(large_audio).decode("ascii")
        assert len(part.audio) > 10000


class TestOutputTextContentPart:
    """Test OutputTextContentPart model."""

    def test_create_output_text_content_part(self):
        """Test creating output text content part."""
        text = "I'm happy to help you with that!"
        part = OutputTextContentPart(text=text)

        assert part.type == "text"
        assert part.text == text
        assert isinstance(part, MessageContentPart)

    def test_output_text_content_part_inheritance(self):
        """Test that OutputTextContentPart inherits from MessageContentPart."""
        part = OutputTextContentPart(text="Test output")

        # Should inherit from MessageContentPart (this is the recent change)
        assert isinstance(part, MessageContentPart)
        assert hasattr(part, "type")

    def test_output_text_content_part_discriminator(self):
        """Test OutputTextContentPart type discriminator."""
        part = OutputTextContentPart(text="Response text")

        # Type should be "text" for output text content
        assert part.type == "text"
        assert part.type == ContentPartType.TEXT


class TestMessageItemsWithContentParts:
    """Test message items using MessageContentPart."""

    def test_user_message_with_text_content(self):
        """Test user message with text content."""
        content = [InputTextContentPart(text="What's the weather like?")]
        message = UserMessageItem(content=content)

        assert message.role == MessageRole.USER
        assert message.type == ItemType.MESSAGE
        assert len(message.content) == 1
        assert isinstance(message.content[0], MessageContentPart)
        assert message.content[0].text == "What's the weather like?"

    def test_user_message_with_audio_content(self):
        """Test user message with audio content."""
        audio_data = b"spoken question audio data"
        content = [InputAudioContentPart(audio=audio_data)]
        message = UserMessageItem(content=content)

        assert message.role == MessageRole.USER
        assert len(message.content) == 1
        assert isinstance(message.content[0], MessageContentPart)
        assert message.content[0].audio == base64.b64encode(audio_data).decode("ascii")

    def test_user_message_with_mixed_content(self):
        """Test user message with mixed content types."""
        text_part = InputTextContentPart(text="Here's my question:")
        audio_part = InputAudioContentPart(audio=b"spoken question")
        content = [text_part, audio_part]

        message = UserMessageItem(content=content)

        assert message.role == MessageRole.USER
        assert len(message.content) == 2
        assert all(isinstance(part, MessageContentPart) for part in message.content)
        assert message.content[0].type == "input_text"
        assert message.content[1].type == "input_audio"

    def test_assistant_message_with_output_content(self):
        """Test assistant message with output content."""
        content = [OutputTextContentPart(text="The weather is sunny today!")]
        message = AssistantMessageItem(content=content)

        assert message.role == MessageRole.ASSISTANT
        assert message.type == ItemType.MESSAGE
        assert len(message.content) == 1
        assert isinstance(message.content[0], MessageContentPart)
        assert message.content[0].text == "The weather is sunny today!"

    def test_system_message_with_text_content(self):
        """Test system message with text content."""
        content = [InputTextContentPart(text="You are a helpful weather assistant.")]
        message = SystemMessageItem(content=content)

        assert message.role == MessageRole.SYSTEM
        assert message.type == ItemType.MESSAGE
        assert len(message.content) == 1
        assert isinstance(message.content[0], MessageContentPart)
        assert message.content[0].text == "You are a helpful weather assistant."


class TestMessageItemPolymorphism:
    """Test message item polymorphism and inheritance."""

    def test_message_item_inheritance(self):
        """Test that all message items inherit from MessageItem."""
        user_content = [InputTextContentPart(text="User message")]
        assistant_content = [OutputTextContentPart(text="Assistant response")]
        system_content = [InputTextContentPart(text="System instruction")]

        user_msg = UserMessageItem(content=user_content)
        assistant_msg = AssistantMessageItem(content=assistant_content)
        system_msg = SystemMessageItem(content=system_content)

        # All should inherit from MessageItem
        assert isinstance(user_msg, MessageItem)
        assert isinstance(assistant_msg, MessageItem)
        assert isinstance(system_msg, MessageItem)

        # All should inherit from ConversationRequestItem
        assert isinstance(user_msg, ConversationRequestItem)
        assert isinstance(assistant_msg, ConversationRequestItem)
        assert isinstance(system_msg, ConversationRequestItem)

    def test_message_item_discrimination(self):
        """Test message item role discrimination."""
        user_content = [InputTextContentPart(text="User")]
        assistant_content = [OutputTextContentPart(text="Assistant")]
        system_content = [InputTextContentPart(text="System")]

        messages = [
            UserMessageItem(content=user_content),
            AssistantMessageItem(content=assistant_content),
            SystemMessageItem(content=system_content),
        ]

        roles = [msg.role for msg in messages]
        expected_roles = [MessageRole.USER, MessageRole.ASSISTANT, MessageRole.SYSTEM]

        assert roles == expected_roles

    def test_message_collection_polymorphism(self):
        """Test using message items in collections."""
        messages: List[MessageItem] = [
            SystemMessageItem(content=[InputTextContentPart(text="You are helpful")]),
            UserMessageItem(content=[InputTextContentPart(text="Hello")]),
            AssistantMessageItem(content=[OutputTextContentPart(text="Hi there!")]),
            UserMessageItem(
                content=[InputTextContentPart(text="Can you help me?"), InputAudioContentPart(audio=b"audio question")]
            ),
        ]

        # All should be MessageItem instances
        for message in messages:
            assert isinstance(message, MessageItem)
            assert hasattr(message, "role")
            assert hasattr(message, "content")
            assert hasattr(message, "type")
            assert message.type == ItemType.MESSAGE

        # Verify mixed content types work in collection
        last_message = messages[-1]
        assert len(last_message.content) == 2
        assert last_message.content[0].type == "input_text"
        assert last_message.content[1].type == "input_audio"


class TestMessageWithOptionalFields:
    """Test message items with optional fields."""

    def test_message_with_id_and_status(self):
        """Test message with optional ID and status."""
        content = [InputTextContentPart(text="Message with metadata")]
        message = UserMessageItem(content=content, id="msg-123", status=ItemParamStatus.COMPLETED)

        assert message.id == "msg-123"
        assert message.status == ItemParamStatus.COMPLETED
        assert message.role == MessageRole.USER

    def test_message_with_incomplete_status(self):
        """Test message with incomplete status."""
        content = [InputTextContentPart(text="Incomplete message")]
        message = AssistantMessageItem(content=content, status=ItemParamStatus.INCOMPLETE)

        assert message.status == ItemParamStatus.INCOMPLETE
        assert message.role == MessageRole.ASSISTANT

    def test_message_without_optional_fields(self):
        """Test message without optional fields."""
        content = [InputTextContentPart(text="Simple message")]
        message = UserMessageItem(content=content)

        assert message.id is None
        assert message.status is None
        assert message.role == MessageRole.USER


class TestContentPartValidation:
    """Test content part validation and edge cases."""

    def test_empty_content_list(self):
        """Test message with empty content list."""
        # This might be invalid depending on the model validation
        try:
            message = UserMessageItem(content=[])
            # If it succeeds, verify it's at least structured correctly
            assert message.role == MessageRole.USER
            assert len(message.content) == 0
        except (ValueError, TypeError):
            # If validation prevents empty content, that's also valid behavior
            pass

    def test_content_part_type_consistency(self):
        """Test that content part types are consistent."""
        text_parts = [InputTextContentPart(text="First text"), InputTextContentPart(text="Second text")]

        message = UserMessageItem(content=text_parts)

        # All parts should have the same type
        types = [part.type for part in message.content]
        assert all(t == "input_text" for t in types)

    def test_content_part_mixed_types(self):
        """Test content parts with mixed types."""
        mixed_content = [
            InputTextContentPart(text="Text part"),
            InputAudioContentPart(audio=b"Audio part"),
            InputTextContentPart(text="Another text part"),
        ]

        message = UserMessageItem(content=mixed_content)

        types = [part.type for part in message.content]
        assert types == ["input_text", "input_audio", "input_text"]


class TestMessageHandlingIntegration:
    """Integration tests for message handling."""

    def test_conversation_flow(self):
        """Test a complete conversation flow with proper content parts."""
        # System message
        system_msg = SystemMessageItem(content=[InputTextContentPart(text="You are a helpful assistant.")], id="sys-1")

        # User message with text
        user_msg_1 = UserMessageItem(content=[InputTextContentPart(text="Hello, how are you?")], id="user-1")

        # Assistant response
        assistant_msg_1 = AssistantMessageItem(
            content=[OutputTextContentPart(text="I'm doing well, thank you!")],
            id="assistant-1",
            status=ItemParamStatus.COMPLETED,
        )

        # User message with audio
        user_msg_2 = UserMessageItem(
            content=[
                InputTextContentPart(text="I have a question:"),
                InputAudioContentPart(audio=b"spoken question audio"),
            ],
            id="user-2",
        )

        # Verify conversation structure
        conversation = [system_msg, user_msg_1, assistant_msg_1, user_msg_2]

        for message in conversation:
            assert isinstance(message, MessageItem)
            assert all(isinstance(part, MessageContentPart) for part in message.content)

        # Verify roles
        roles = [msg.role for msg in conversation]
        expected_roles = [MessageRole.SYSTEM, MessageRole.USER, MessageRole.ASSISTANT, MessageRole.USER]
        assert roles == expected_roles

    def test_message_content_serialization_ready(self):
        """Test that messages and content parts are ready for serialization."""
        content_parts = [
            InputTextContentPart(text="Serializable text"),
            InputAudioContentPart(audio=b"serializable audio"),
            OutputTextContentPart(text="Serializable output"),
        ]

        messages = [
            UserMessageItem(content=[content_parts[0]], id="user-ser"),
            UserMessageItem(content=[content_parts[1]], id="user-audio"),
            AssistantMessageItem(content=[content_parts[2]], id="assistant-ser"),
        ]

        for message in messages:
            # All messages should have required serialization attributes
            assert hasattr(message, "__dict__")
            assert hasattr(message, "role")
            assert hasattr(message, "content")
            assert hasattr(message, "type")

            for part in message.content:
                assert hasattr(part, "__dict__")
                assert hasattr(part, "type")

    def test_backwards_compatibility(self):
        """Test backwards compatibility with content part changes."""
        # Verify that the rename from UserContentPart to MessageContentPart
        # doesn't break existing functionality

        # These should all work with MessageContentPart
        text_part = InputTextContentPart(text="Compatibility test")
        audio_part = InputAudioContentPart(audio=b"compatibility audio")
        output_part = OutputTextContentPart(text="Compatibility output")

        # All should be MessageContentPart instances
        assert isinstance(text_part, MessageContentPart)
        assert isinstance(audio_part, MessageContentPart)
        assert isinstance(output_part, MessageContentPart)

        # Should work in messages
        user_msg = UserMessageItem(content=[text_part, audio_part])
        assistant_msg = AssistantMessageItem(content=[output_part])

        assert len(user_msg.content) == 2
        assert len(assistant_msg.content) == 1