|
13 | 13 | # limitations under the License. |
14 | 14 |
|
15 | 15 | # type: ignore |
16 | | - |
| 16 | +import io |
17 | 17 | import unittest |
18 | 18 | from typing import MutableSequence |
19 | 19 |
|
| 20 | +# Optional Django imports for testing |
| 21 | +try: |
| 22 | + from django.conf import settings |
| 23 | + from django.core.handlers.wsgi import WSGIRequest |
| 24 | + |
| 25 | + _HAS_DJANGO = True |
| 26 | +except ImportError: |
| 27 | + WSGIRequest = None # type: ignore |
| 28 | + settings = None # type: ignore |
| 29 | + _HAS_DJANGO = False |
| 30 | + |
20 | 31 | from opentelemetry.attributes import ( |
21 | 32 | BoundedAttributes, |
22 | 33 | _clean_attribute, |
@@ -181,71 +192,70 @@ def test_mapping(self): |
181 | 192 | self.assertEqual( |
182 | 193 | _clean_extended_attribute("headers", mapping, None), expected |
183 | 194 | ) |
184 | | - |
| 195 | + |
185 | 196 | def test_wsgi_request_attribute(self): |
186 | 197 | # Test WSGIRequest type validation |
187 | 198 | try: |
188 | | - from django.core.handlers.wsgi import WSGIRequest |
189 | | - from django.conf import settings |
190 | | - import io |
191 | | - |
192 | 199 | # Configure Django settings if not already configured |
193 | 200 | if not settings.configured: |
194 | 201 | settings.configure( |
195 | 202 | DEBUG=True, |
196 | | - SECRET_KEY='test-secret-key', |
| 203 | + SECRET_KEY="test-secret-key", |
197 | 204 | USE_TZ=True, |
198 | 205 | ROOT_URLCONF=[], |
199 | 206 | MIDDLEWARE=[], |
200 | 207 | ) |
201 | | - |
| 208 | + |
202 | 209 | # Create a minimal WSGI environ dict |
203 | 210 | environ = { |
204 | | - 'REQUEST_METHOD': 'GET', |
205 | | - 'PATH_INFO': '/test', |
206 | | - 'QUERY_STRING': '', |
207 | | - 'CONTENT_TYPE': '', |
208 | | - 'CONTENT_LENGTH': '', |
209 | | - 'HTTP_HOST': 'testserver', |
210 | | - 'wsgi.version': (1, 0), |
211 | | - 'wsgi.url_scheme': 'http', |
212 | | - 'wsgi.input': io.StringIO(), |
213 | | - 'wsgi.errors': io.StringIO(), |
214 | | - 'wsgi.multithread': False, |
215 | | - 'wsgi.multiprocess': False, |
216 | | - 'wsgi.run_once': False, |
217 | | - 'SERVER_NAME': 'testserver', |
218 | | - 'SERVER_PORT': '80', |
| 211 | + "REQUEST_METHOD": "GET", |
| 212 | + "PATH_INFO": "/test", |
| 213 | + "QUERY_STRING": "", |
| 214 | + "CONTENT_TYPE": "", |
| 215 | + "CONTENT_LENGTH": "", |
| 216 | + "HTTP_HOST": "testserver", |
| 217 | + "wsgi.version": (1, 0), |
| 218 | + "wsgi.url_scheme": "http", |
| 219 | + "wsgi.input": io.StringIO(), |
| 220 | + "wsgi.errors": io.StringIO(), |
| 221 | + "wsgi.multithread": False, |
| 222 | + "wsgi.multiprocess": False, |
| 223 | + "wsgi.run_once": False, |
| 224 | + "SERVER_NAME": "testserver", |
| 225 | + "SERVER_PORT": "80", |
219 | 226 | } |
220 | | - |
| 227 | + |
221 | 228 | # Create a WSGIRequest object |
222 | 229 | wsgi_request = WSGIRequest(environ) |
223 | | - |
| 230 | + |
224 | 231 | # Test that WSGIRequest gets cleaned to a dictionary format |
225 | 232 | expected_cleaned = { |
226 | | - 'method': 'GET', |
227 | | - 'path': '/test', |
228 | | - 'path_info': '/test', |
229 | | - 'content_type': '' |
| 233 | + "method": "GET", |
| 234 | + "path": "/test", |
| 235 | + "path_info": "/test", |
| 236 | + "content_type": "", |
230 | 237 | } |
231 | | - |
232 | | - # Test WSGIRequest cleaning directly |
233 | | - from opentelemetry.attributes import _clean_extended_attribute |
234 | | - cleaned_value = _clean_extended_attribute("request", wsgi_request, None) |
| 238 | + |
| 239 | + cleaned_value = _clean_extended_attribute( |
| 240 | + "request", wsgi_request, None |
| 241 | + ) |
235 | 242 | self.assertEqual(cleaned_value, expected_cleaned) |
236 | | - |
237 | | - # Test WSGIRequest in sequences - should be cleaned to dict |
238 | | - cleaned_sequence = _clean_extended_attribute("requests", [wsgi_request], None) |
| 243 | + |
| 244 | + cleaned_sequence = _clean_extended_attribute( |
| 245 | + "requests", [wsgi_request], None |
| 246 | + ) |
239 | 247 | self.assertEqual(cleaned_sequence, (expected_cleaned,)) |
240 | | - |
241 | | - # Test WSGIRequest in mappings - should be cleaned to dict |
242 | | - cleaned_mapping = _clean_extended_attribute("data", {"request": wsgi_request}, None) |
| 248 | + |
| 249 | + cleaned_mapping = _clean_extended_attribute( |
| 250 | + "data", {"request": wsgi_request}, None |
| 251 | + ) |
243 | 252 | self.assertEqual(cleaned_mapping, {"request": expected_cleaned}) |
244 | | - |
| 253 | + |
245 | 254 | except ImportError: |
246 | 255 | # Skip test if django is not available |
247 | 256 | self.skipTest("Django not available") |
248 | 257 |
|
| 258 | + |
249 | 259 | class TestBoundedAttributes(unittest.TestCase): |
250 | 260 | # pylint: disable=consider-using-dict-items |
251 | 261 | base = { |
|
0 commit comments