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
|
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License.
import unittest
from unittest import mock
from azure.monitor.opentelemetry.exporter._quickpulse._processor import (
_QuickpulseLogRecordProcessor,
_QuickpulseSpanProcessor,
)
from azure.monitor.opentelemetry.exporter._quickpulse._manager import _QuickpulseManager
class TestQuickpulseLogRecordProcessor(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.qpm = mock.Mock()
_QuickpulseManager._instances[_QuickpulseManager] = cls.qpm
@classmethod
def tearDownClass(cls) -> None:
# Reset singleton state - only clear QuickpulseManager instances
if _QuickpulseManager in _QuickpulseManager._instances:
del _QuickpulseManager._instances[_QuickpulseManager]
@mock.patch("azure.monitor.opentelemetry.exporter._quickpulse._processor.get_quickpulse_manager")
def test_emit(self, mock_get_manager):
mock_manager = mock.Mock()
mock_get_manager.return_value = mock_manager
processor = _QuickpulseLogRecordProcessor()
log_data = mock.Mock()
processor.on_emit(log_data)
mock_get_manager.assert_called_once()
mock_manager._record_log_record.assert_called_once_with(log_data)
class TestQuickpulseSpanProcessor(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.qpm = mock.Mock()
_QuickpulseManager._instances[_QuickpulseManager] = cls.qpm
@classmethod
def tearDownClass(cls) -> None:
# Reset singleton state - only clear QuickpulseManager instances
if _QuickpulseManager in _QuickpulseManager._instances:
del _QuickpulseManager._instances[_QuickpulseManager]
@mock.patch("azure.monitor.opentelemetry.exporter._quickpulse._processor.get_quickpulse_manager")
def test_on_end(self, mock_get_manager):
mock_manager = mock.Mock()
mock_get_manager.return_value = mock_manager
processor = _QuickpulseSpanProcessor()
span = mock.Mock()
processor.on_end(span)
mock_manager._record_span.assert_called_once_with(span)
|
