1 # Copyright (c) 2020 Deutsche Telekom.
2 # Licensed under the Apache License, Version 2.0 (the "License");
3 # you may not use this file except in compliance with the License.
4 # You may obtain a copy of the License at:
5 # http://www.apache.org/licenses/LICENSE-2.0
7 # Unless required by applicable law or agreed to in writing, software
8 # distributed under the License is distributed on an "AS IS" BASIS,
9 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 from collections import namedtuple
15 if sys.version_info[0] < 3:
16 from mock import patch
17 if sys.version_info[0] >= 3:
18 from unittest.mock import patch
22 from onaplogging.marker import BaseMarker, matchMarkerHelp, MarkerFactory, MarkerFilter, MarkerNotifyHandler
25 class TestRecordMixin(object):
27 Record = namedtuple("Record", "marker")
30 class TestNameMixin(object):
32 TEST_NAME = "test_base"
35 class TestBaseMarker(unittest.TestCase, TestNameMixin):
38 super(TestBaseMarker, self).setUp()
39 self.base_marker = BaseMarker(name=self.TEST_NAME)
41 def test_base_marker_name(self):
42 with pytest.raises(TypeError):
45 with pytest.raises(ValueError):
48 self.assertEqual(self.base_marker.getName(), self.TEST_NAME)
50 def test_base_marker_contains(self):
51 self.assertTrue(self.base_marker.contains(self.base_marker))
52 self.assertTrue(self.base_marker.contains(self.TEST_NAME))
54 def test_base_marker_compare(self):
55 self.assertNotEqual(self.base_marker, 3)
56 self.assertEqual(self.base_marker, self.base_marker)
57 other = BaseMarker("Other")
58 self.assertNotEqual(self.base_marker, other)
59 other = BaseMarker(self.TEST_NAME)
60 self.assertEqual(self.base_marker, other)
62 def test_base_marker_child(self):
63 self.assertListEqual(list(iter(self.base_marker)), [])
64 self.assertFalse(self.base_marker.contains(3))
65 with pytest.raises(TypeError):
66 self.base_marker.addChild(3)
67 with pytest.raises(TypeError):
68 self.base_marker.addChild("str")
69 with pytest.raises(TypeError):
70 self.base_marker.removeChild(3)
72 self.base_marker.addChild(self.base_marker)
73 self.assertListEqual(list(iter(self.base_marker)), [])
75 child1 = BaseMarker(name="child1")
76 self.assertFalse(self.base_marker.contains(child1))
77 self.base_marker.addChild(child1)
78 self.assertListEqual(list(iter(self.base_marker)), [child1])
79 self.assertTrue(self.base_marker.contains(child1))
80 self.base_marker.addChild(child1)
81 self.assertListEqual(list(iter(self.base_marker)), [child1])
83 self.base_marker.removeChild(child1)
84 self.assertListEqual(list(iter(self.base_marker)), [])
85 self.assertFalse(self.base_marker.contains(child1))
87 child2 = BaseMarker(name="child2")
88 self.assertFalse(self.base_marker.contains(child2))
90 with pytest.raises(TypeError):
91 self.base_marker.addChilds(None)
92 self.base_marker.addChilds((child1, child2,))
93 self.assertTrue(self.base_marker.contains(child1))
94 self.assertTrue(self.base_marker.contains(child2))
95 self.base_marker.removeChild(child1)
96 self.assertFalse(self.base_marker.contains(child1))
97 self.assertTrue(self.base_marker.contains(child2))
98 self.assertFalse(self.base_marker.contains("child1"))
99 self.assertTrue(self.base_marker.contains("child2"))
102 class TestMatchMarkerHelp(unittest.TestCase, TestRecordMixin, TestNameMixin):
105 def test_match_marker_help(self):
106 record = self.Record(None)
107 self.assertFalse(matchMarkerHelp(record, "anything"))
109 record = self.Record("not_marker_instance")
110 self.assertFalse(matchMarkerHelp(record, "not_marker_instance"))
112 marker = BaseMarker(self.TEST_NAME)
113 record = self.Record(marker)
114 self.assertFalse(matchMarkerHelp(record, "invalid_name"))
115 self.assertTrue(matchMarkerHelp(record, marker))
116 self.assertTrue(matchMarkerHelp(record, self.TEST_NAME))
118 child = BaseMarker(self.CHILD_NAME)
119 marker.addChild(child)
120 self.assertTrue(matchMarkerHelp(record, [self.TEST_NAME, self.CHILD_NAME]))
121 self.assertTrue(matchMarkerHelp(record, [marker, self.CHILD_NAME]))
122 self.assertTrue(matchMarkerHelp(record, [marker, child]))
123 self.assertTrue(matchMarkerHelp(record, [marker, "invalid"]))
126 class TestMarkerFactory(unittest.TestCase, TestNameMixin):
129 super(TestMarkerFactory, self).setUp()
130 self.marker_factory = MarkerFactory()
132 def test_get_marker(self):
133 with pytest.raises(ValueError):
134 self.marker_factory.getMarker()
135 self.assertEqual(len(self.marker_factory._marker_map), 0)
136 marker = self.marker_factory.getMarker(self.TEST_NAME)
137 self.assertEqual(marker.getName(), self.TEST_NAME)
138 self.assertEqual(len(self.marker_factory._marker_map), 1)
139 marker = self.marker_factory.getMarker(self.TEST_NAME)
140 self.assertEqual(marker.getName(), self.TEST_NAME)
141 self.assertEqual(len(self.marker_factory._marker_map), 1)
143 self.assertTrue(self.marker_factory.exist(marker.getName()))
145 self.assertTrue(self.marker_factory.deleteMarker(marker.getName()))
146 self.assertFalse(self.marker_factory.exist(marker.getName()))
147 self.assertEqual(len(self.marker_factory._marker_map), 0)
149 self.assertFalse(self.marker_factory.deleteMarker(marker.getName()))
152 class TestMarkerFilter(unittest.TestCase, TestRecordMixin, TestNameMixin):
154 def test_marker_filter(self):
155 marker_filter = MarkerFilter()
157 record = self.Record(BaseMarker(self.TEST_NAME))
158 self.assertFalse(marker_filter.filter(record))
160 marker_filter = MarkerFilter(markers=BaseMarker(self.TEST_NAME))
161 self.assertTrue(marker_filter.filter(record))
164 class TestMarkerNotifyHandler(unittest.TestCase, TestRecordMixin, TestNameMixin):
166 def test_marker_notify_handler(self):
167 record = self.Record(BaseMarker(self.TEST_NAME))
169 notify_handler = MarkerNotifyHandler("test_host", "fromaddr", "toaddrs", "subject")
170 self.assertIsNone(notify_handler.markers)
171 self.assertFalse(notify_handler.handle(record))
173 marker = BaseMarker(self.TEST_NAME)
174 notify_handler = MarkerNotifyHandler("test_host", "fromaddr", "toaddrs", "subject", markers=[marker])
175 with patch("onaplogging.marker.markerHandler.SMTPHandler.handle") as mock_smtp_handler_handle:
176 mock_smtp_handler_handle.return_value = True
177 self.assertTrue(notify_handler.handle(record))
178 record = self.Record(BaseMarker("other"))
179 self.assertFalse(notify_handler.handle(record))