forked from lightspeed-core/lightspeed-stack
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_query_request.py
More file actions
168 lines (146 loc) · 6.28 KB
/
Copy pathtest_query_request.py
File metadata and controls
168 lines (146 loc) · 6.28 KB
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
"""Unit tests for QueryRequest model."""
from logging import Logger
from pytest_mock import MockerFixture
import pytest
from models.requests import QueryRequest, Attachment
class TestQueryRequest:
"""Test cases for the QueryRequest model."""
def test_constructor(self) -> None:
"""Test the QueryRequest constructor."""
qr = QueryRequest(query="Tell me about Kubernetes")
assert qr.query == "Tell me about Kubernetes"
assert qr.conversation_id is None
assert qr.provider is None
assert qr.model is None
assert qr.system_prompt is None
assert qr.attachments is None
def test_constructor_wrong_conversation_id(self) -> None:
"""Test the QueryRequest constructor with wrong conversation_id."""
with pytest.raises(ValueError, match="Improper conversation ID 'xyzzy'"):
_ = QueryRequest(query="Tell me about Kubernetes", conversation_id="xyzzy")
def test_with_attachments(self) -> None:
"""Test the QueryRequest with attachments."""
attachments = [
Attachment(
attachment_type="log",
content_type="text/plain",
content="this is attachment",
),
Attachment(
attachment_type="configuration",
content_type="application/yaml",
content="kind: Pod\n metadata:\n name: private-reg",
),
]
qr = QueryRequest(
query="Tell me about Kubernetes",
attachments=attachments,
)
assert qr.attachments is not None
assert len(qr.attachments) == 2
# the following warning is false positive
# pylint: disable=unsubscriptable-object
assert qr.attachments[0].attachment_type == "log"
assert qr.attachments[0].content_type == "text/plain"
assert qr.attachments[0].content == "this is attachment"
assert qr.attachments[1].attachment_type == "configuration"
assert qr.attachments[1].content_type == "application/yaml"
assert (
qr.attachments[1].content == "kind: Pod\n metadata:\n name: private-reg"
)
def test_with_optional_fields(self) -> None:
"""Test the QueryRequest with optional fields."""
qr = QueryRequest(
query="Tell me about Kubernetes",
conversation_id="123e4567-e89b-12d3-a456-426614174000",
provider="OpenAI",
model="gpt-3.5-turbo",
system_prompt="You are a helpful assistant",
)
assert qr.query == "Tell me about Kubernetes"
assert qr.conversation_id == "123e4567-e89b-12d3-a456-426614174000"
assert qr.provider == "OpenAI"
assert qr.model == "gpt-3.5-turbo"
assert qr.system_prompt == "You are a helpful assistant"
assert qr.attachments is None
def test_get_documents(self) -> None:
"""Test the get_documents method."""
attachments = [
Attachment(
attachment_type="log",
content_type="text/plain",
content="this is attachment",
),
Attachment(
attachment_type="configuration",
content_type="application/yaml",
content="kind: Pod\n metadata:\n name: private-reg",
),
]
qr = QueryRequest(
query="Tell me about Kubernetes",
attachments=attachments,
)
documents = qr.get_documents()
assert len(documents) == 2
assert documents[0]["content"] == "this is attachment"
assert documents[0]["mime_type"] == "text/plain"
assert documents[1]["content"] == "kind: Pod\n metadata:\n name: private-reg"
assert documents[1]["mime_type"] == "application/yaml"
def test_get_documents_no_attachments(self) -> None:
"""Test the get_documents method."""
attachments: list[Attachment] = []
qr = QueryRequest(
query="Tell me about Kubernetes",
attachments=attachments,
)
documents = qr.get_documents()
assert len(documents) == 0
def test_validate_provider_and_model(self) -> None:
"""Test the validate_provider_and_model method."""
qr = QueryRequest(
query="Tell me about Kubernetes",
provider="OpenAI",
model="gpt-3.5-turbo",
)
assert qr is not None
validated_qr = qr.validate_provider_and_model()
assert validated_qr.provider == "OpenAI"
assert validated_qr.model == "gpt-3.5-turbo"
# Test with missing provider
with pytest.raises(
ValueError, match="Provider must be specified if model is specified"
):
QueryRequest(query="Tell me about Kubernetes", model="gpt-3.5-turbo")
# Test with missing model
with pytest.raises(
ValueError, match="Model must be specified if provider is specified"
):
QueryRequest(query="Tell me about Kubernetes", provider="OpenAI")
def test_validate_media_type(self, mocker: MockerFixture) -> None:
"""Test the validate_media_type method."""
# Mock the logger
mock_logger = mocker.Mock(spec=Logger)
mocker.patch("models.requests.logger", mock_logger)
qr = QueryRequest(
query="Tell me about Kubernetes",
provider="OpenAI",
model="gpt-3.5-turbo",
media_type="text/plain",
)
assert qr is not None
assert qr.provider == "OpenAI"
assert qr.model == "gpt-3.5-turbo"
assert qr.media_type == "text/plain"
# Media type is now fully supported, no warning expected
mock_logger.warning.assert_not_called()
def test_generate_topic_summary_explicit_false(self) -> None:
"""Test that generate_topic_summary can be explicitly set to False."""
qr = QueryRequest(
query="Tell me about Kubernetes", generate_topic_summary=False
)
assert qr.generate_topic_summary is False
def test_generate_topic_summary_explicit_true(self) -> None:
"""Test that generate_topic_summary can be explicitly set to True."""
qr = QueryRequest(query="Tell me about Kubernetes", generate_topic_summary=True)
assert qr.generate_topic_summary is True