KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use log_source method in robotframework-appiumlibrary
Best Python code snippet using robotframework-appiumlibrary_python
build_logging_test.py
Source:build_logging_test.py 
1"""Test for the anvil logging components.2"""3__author__ = 'joshharrison@google.com'4import re5import unittest26from mock import call7from mock import patch8from mock import MagicMock9from anvil import build_logging10from anvil import enums11from anvil import util12def _message(log_level, time, name, message):13 return {14 'log_level': log_level,15 'time': time,16 'name': name,17 'message': message18 }19class WorkUnitTest(unittest2.TestCase):20 def testRecordWorkUnits(self):21 work_unit = build_logging.WorkUnit('test')22 self.assertEquals(enums.Status.WAITING, work_unit.get_status())23 work_unit.total = 1024 self.assertEquals(10, work_unit.total)25 work_unit.complete = 1026 self.assertEquals(10, work_unit.complete)27 def testAddChildUnit(self):28 child = build_logging.WorkUnit('child')29 parent = build_logging.WorkUnit('parent')30 parent.add_child(child)31 self.assertEquals(parent, child.parent)32 self.assertEquals(1, len(parent.children))33 self.assertEquals(child, parent.children[0])34 def testAssumeChildCounts(self):35 child = build_logging.WorkUnit('child')36 parent = build_logging.WorkUnit('parent')37 parent.total = 10038 child.total = 5039 parent.add_child(child)40 self.assertEquals(150, parent.total)41 child.total = 25042 self.assertEquals(350, parent.total)43 # Test idempotence44 child.total = 25045 self.assertEquals(350, parent.total)46 parent.total = 20047 self.assertEquals(450, parent.total)48 sibling = build_logging.WorkUnit('sibling')49 grandchild = build_logging.WorkUnit('grandchild')50 grandchild.total = 1051 sibling.add_child(grandchild)52 self.assertEquals(10, sibling.total)53 parent.add_child(sibling)54 self.assertEquals(460, parent.total)55 grandchild.total = 10056 sibling.total = 20057 self.assertEquals(750, parent.total)58 def testInvalidCountsThrowsError(self):59 work_unit = build_logging.WorkUnit('test')60 work_unit.complete = 10061 with self.assertRaises(ValueError):62 work_unit.total = 5063 def testGetChildStatuses(self):64 child = build_logging.WorkUnit('child')65 parent = build_logging.WorkUnit('parent')66 parent.add_child(child)67 self.assertEquals(None, child.start_time)68 self.assertEquals(None, parent.start_time)69 self.assertEquals(enums.Status.WAITING, child.get_status())70 self.assertEquals(enums.Status.WAITING, parent.get_status())71 child.total = 2072 self.assertEquals(enums.Status.RUNNING, child.get_status())73 self.assertEquals(enums.Status.RUNNING, parent.get_status())74 parent.total = 3075 child.total = 076 child.complete = 077 self.assertEquals(enums.Status.SKIPPED, child.get_status())78 self.assertEquals(enums.Status.RUNNING, parent.get_status())79 parent.complete = 3080 self.assertEquals(enums.Status.SUCCEEDED, parent.get_status())81 child.exception = ValueError()82 self.assertEquals(enums.Status.FAILED, child.get_status())83 self.assertEquals(enums.Status.FAILED, parent.get_status())84 self.assertEquals(child.exception, parent.exception)85 def testChangeListenersCalled(self):86 mock_listener = MagicMock(name='listener')87 # If should_listen returns True, the listener will not be added to the88 # WorkUnit, and will therefor not have its handler triggered on89 # updates.90 mock_listener.should_listen.return_value = False91 test = build_logging.WorkUnit('test')92 test.add_change_listener(mock_listener)93 # Listeners should be triggered when updates occur anywhere along a WorkUnit94 # tree.95 child = build_logging.WorkUnit('child')96 parent = build_logging.WorkUnit('parent')97 parent.add_child(child)98 mock_child_listener = MagicMock(name='child_listener')99 mock_child_listener.should_listen.return_value = True100 mock_parent_listener = MagicMock(name='parent_listener')101 mock_parent_listener.should_listen.return_value = True102 child.add_change_listener(mock_child_listener)103 parent.add_change_listener(mock_parent_listener)104 child.total = 100105 parent.total = 100106 parent.complete = 100107 expected_child_calls = [108 call.should_listen(child),109 call.update(child, 'total', 100)110 ]111 expected_parent_calls = [112 call.should_listen(parent),113 call.update(parent, 'total', 100),114 call.update(parent, 'total', 200),115 call.update(parent, 'complete', 100)116 ]117 self.assertTrue(expected_child_calls == mock_child_listener.mock_calls)118 self.assertTrue(expected_parent_calls == mock_parent_listener.mock_calls)119class LogSourceTest(unittest2.TestCase):120 def testVerbosity(self):121 child_source = build_logging.LogSource(enums.Verbosity.VERBOSE)122 self.assertEquals(enums.Verbosity.VERBOSE, child_source.verbosity)123 child_source = build_logging.LogSource()124 self.assertEquals(enums.Verbosity.NORMAL, child_source.verbosity)125 child_source = build_logging.LogSource()126 parent_source = build_logging.LogSource(enums.Verbosity.VERBOSE)127 parent_source.add_child(child_source)128 self.assertEquals(enums.Verbosity.VERBOSE, child_source.verbosity)129 self.assertEquals(enums.Verbosity.VERBOSE, parent_source.verbosity)130 def testLogBasedOnVerbosity(self):131 log_source = build_logging.LogSource()132 log_source.verbosity = enums.Verbosity.SILENT133 with patch('anvil.util.timer') as mock_timer:134 mock_timer.side_effect = [1]135 log_source.log_debug('debug')136 log_source.log_info('info')137 log_source.log_warning('warning')138 log_source.log_error('error')139 expected = [140 (enums.LogLevel.ERROR, 1, None, '[ERROR] error')141 ]142 self.assertListEqual(expected, log_source.buffered_messages)143 log_source = build_logging.LogSource()144 log_source.verbosity = enums.Verbosity.NORMAL145 with patch('anvil.util.timer') as mock_timer:146 mock_timer.side_effect = [1, 2, 3]147 log_source.log_debug('debug')148 log_source.log_info('info')149 log_source.log_warning('warning')150 log_source.log_error('error')151 expected = [152 (enums.LogLevel.INFO, 1, None, '[INFO] info'),153 (enums.LogLevel.WARNING, 2, None, '[WARNING] warning'),154 (enums.LogLevel.ERROR, 3, None, '[ERROR] error')155 ]156 self.assertListEqual(expected, log_source.buffered_messages)157 log_source = build_logging.LogSource()158 log_source.verbosity = enums.Verbosity.VERBOSE159 with patch('anvil.util.timer') as mock_timer:160 mock_timer.side_effect = [1, 2, 3, 4]161 log_source.log_debug('debug')162 log_source.log_info('info', 'test')163 log_source.log_warning('warning', 'test')164 log_source.log_error('error')165 expected = [166 (enums.LogLevel.DEBUG, 1, None, '[DEBUG] debug'),167 (enums.LogLevel.INFO, 2, 'test', '[INFO] info'),168 (enums.LogLevel.WARNING, 3, 'test', '[WARNING] warning'),169 (enums.LogLevel.ERROR, 4, None, '[ERROR] error')170 ]171 self.assertListEqual(expected, log_source.buffered_messages)172 log_source = build_logging.LogSource()173 # Inherit should default to normal of no parent exists.174 log_source.verbosity = enums.Verbosity.INHERIT175 with patch('anvil.util.timer') as mock_timer:176 mock_timer.side_effect = [1, 2, 3]177 log_source.log_debug('debug')178 log_source.log_info('info')179 log_source.log_warning('warning')180 log_source.log_error('error')181 expected = [182 (enums.LogLevel.INFO, 1, None, '[INFO] info'),183 (enums.LogLevel.WARNING, 2, None, '[WARNING] warning'),184 (enums.LogLevel.ERROR, 3, None, '[ERROR] error')185 ]186 self.assertListEqual(expected, log_source.buffered_messages)187 def testNoDuplicateLogSinks(self):188 log_sink = MagicMock()189 log_source = build_logging.LogSource()190 log_source.add_log_sink(log_sink)191 log_source.add_log_sink(log_sink)192 self.assertEquals(1, len(log_source.log_sinks))193 def testMessagesSentToLogSink(self):194 log_source = build_logging.LogSource(enums.Verbosity.VERBOSE)195 with patch('anvil.util.timer') as mock_timer:196 mock_timer.side_effect = [1, 2, 3, 4]197 log_source.log_debug('debug', 'bar')198 log_source.log_info('info', 'bar')199 log_source.log_warning('warning', 'foo')200 log_source.log_error('error', 'foo')201 log_sink = MagicMock()202 log_source.add_log_sink(log_sink)203 expected = [204 call.log(_message(enums.LogLevel.DEBUG, 1, 'bar', '[DEBUG] debug')),205 call.log(_message(enums.LogLevel.INFO, 2, 'bar', '[INFO] info')),206 call.log(_message(enums.LogLevel.WARNING, 3, 'foo', '[WARNING] warning')),207 call.log(_message(enums.LogLevel.ERROR, 4, 'foo', '[ERROR] error'))208 ]209 self.assertEquals(expected, log_sink.mock_calls)210 log_sink.mock_calls = []211 with patch('anvil.util.timer') as mock_timer:212 mock_timer.side_effect = [5]213 log_source.log_debug('debug', 'bar')214 expected = [215 call.log(_message(enums.LogLevel.DEBUG, 5, 'bar', '[DEBUG] debug'))216 ]217 self.assertEquals(expected, log_sink.mock_calls)218 def testChildMessagesDetected(self):219 child_source = build_logging.LogSource(enums.Verbosity.INHERIT)220 parent_source = build_logging.LogSource(enums.Verbosity.VERBOSE)221 log_sink = MagicMock()222 parent_source.add_log_sink(log_sink)223 parent_source.add_child(child_source)224 with patch('anvil.util.timer') as mock_timer:225 mock_timer.side_effect = [1]226 child_source.log_debug('debug', 'foo')227 expected = [228 call.log(_message(enums.LogLevel.DEBUG, 1, 'foo', '[DEBUG] debug'))229 ]230 self.assertEquals(expected, log_sink.mock_calls)231 232class WorkUnitLogSourceTest(unittest2.TestCase):233 def setUp(self):234 self.log_source = ls = build_logging.WorkUnitLogSource(235 enums.Verbosity.VERBOSE)236 def testAddingListenerGrabsStatus(self):237 work_unit = build_logging.WorkUnit(name='test')238 work_unit.add_change_listener(self.log_source)239 msgs = self.log_source.buffered_messages240 self.assertEquals(2, len(self.log_source.buffered_messages))241 self.assertEquals(enums.LogLevel.DEBUG, msgs[0][0])242 self.assertEquals('test', msgs[0][2])243 self.assertEquals(enums.LogLevel.INFO, msgs[1][0])244 self.assertEquals('test', msgs[1][2])245 def testWorkUnitLogging(self):246 work_unit = build_logging.WorkUnit(name='test')247 work_unit.add_change_listener(self.log_source)248 self.log_source.buffered_messages = []249 msgs = self.log_source.buffered_messages250 work_unit.total = 200251 self.assertEquals(2, len(self.log_source.buffered_messages))252 self.assertEquals(enums.LogLevel.DEBUG, msgs[0][0])253 self.assertEquals('test', msgs[0][2])254 self.assertEquals(enums.LogLevel.INFO, msgs[1][0])255 self.assertRegexpMatches(msgs[1][3], re.compile('0 of 200'))256 self.assertRegexpMatches(msgs[1][3], re.compile('RUNNING'))257 self.assertEquals('test', msgs[1][2])258 work_unit.complete = 10259 self.assertEquals(4, len(self.log_source.buffered_messages))260 self.assertEquals(enums.LogLevel.DEBUG, msgs[2][0])261 self.assertEquals('test', msgs[2][2])262 self.assertEquals(enums.LogLevel.INFO, msgs[3][0])263 self.assertRegexpMatches(msgs[3][3], re.compile('10 of 200'))264 self.assertRegexpMatches(msgs[3][3], re.compile('RUNNING'))265 self.assertEquals('test', msgs[3][2])266 work_unit.complete = 200267 self.assertEquals(6, len(self.log_source.buffered_messages))268 self.assertEquals(enums.LogLevel.DEBUG, msgs[4][0])269 self.assertEquals('test', msgs[4][2])270 self.assertEquals(enums.LogLevel.INFO, msgs[5][0])271 self.assertRegexpMatches(msgs[5][3], re.compile('SUCCEEDED'))272 self.assertEquals('test', msgs[5][2])273 work_unit.complete = 0274 work_unit.total = 0275 self.assertEquals(10, len(self.log_source.buffered_messages))276 self.assertEquals(enums.LogLevel.DEBUG, msgs[8][0])277 self.assertEquals('test', msgs[8][2])278 self.assertEquals(enums.LogLevel.INFO, msgs[9][0])279 self.assertRegexpMatches(msgs[9][3], re.compile('SKIPPED'))280 self.assertEquals('test', msgs[9][2])281 282if __name__ == '__main__':...
test_logger.py
Source:test_logger.py
1# Copyright (c) Twisted Matrix Laboratories.2# See LICENSE for details.3"""4Test cases for L{twisted.logger._logger}.5"""6from twisted.trial import unittest7from .._levels import InvalidLogLevelError8from .._levels import LogLevel9from .._format import formatEvent10from .._logger import Logger11from .._global import globalLogPublisher12class TestLogger(Logger):13 """14 L{Logger} with an overriden C{emit} method that keeps track of received15 events.16 """17 def emit(self, level, format=None, **kwargs):18 def observer(event):19 self.event = event20 globalLogPublisher.addObserver(observer)21 try:22 Logger.emit(self, level, format, **kwargs)23 finally:24 globalLogPublisher.removeObserver(observer)25 self.emitted = {26 "level": level,27 "format": format,28 "kwargs": kwargs,29 }30class LogComposedObject(object):31 """32 A regular object, with a logger attached.33 """34 log = TestLogger()35 def __init__(self, state=None):36 self.state = state37 def __str__(self):38 return "<LogComposedObject {state}>".format(state=self.state)39class LoggerTests(unittest.TestCase):40 """41 Tests for L{Logger}.42 """43 def test_repr(self):44 """45 repr() on Logger46 """47 namespace = "bleargh"48 log = Logger(namespace)49 self.assertEqual(repr(log), "<Logger {0}>".format(repr(namespace)))50 def test_namespaceDefault(self):51 """52 Default namespace is module name.53 """54 log = Logger()55 self.assertEqual(log.namespace, __name__)56 def test_namespaceAttribute(self):57 """58 Default namespace for classes using L{Logger} as a descriptor is the59 class name they were retrieved from.60 """61 obj = LogComposedObject()62 expectedNamespace = "{0}.{1}".format(63 obj.__module__,64 obj.__class__.__name__,65 )66 self.assertEqual(obj.log.namespace, expectedNamespace)67 self.assertEqual(LogComposedObject.log.namespace, expectedNamespace)68 self.assertIs(LogComposedObject.log.source, LogComposedObject)69 self.assertIs(obj.log.source, obj)70 self.assertIs(Logger().source, None)71 def test_descriptorObserver(self):72 """73 When used as a descriptor, the observer is propagated.74 """75 observed = []76 class MyObject(object):77 log = Logger(observer=observed.append)78 MyObject.log.info("hello")79 self.assertEqual(len(observed), 1)80 self.assertEqual(observed[0]['log_format'], "hello")81 def test_sourceAvailableForFormatting(self):82 """83 On instances that have a L{Logger} class attribute, the C{log_source}84 key is available to format strings.85 """86 obj = LogComposedObject("hello")87 log = obj.log88 log.error("Hello, {log_source}.")89 self.assertIn("log_source", log.event)90 self.assertEqual(log.event["log_source"], obj)91 stuff = formatEvent(log.event)92 self.assertIn("Hello, <LogComposedObject hello>.", stuff)93 def test_basicLogger(self):94 """95 Test that log levels and messages are emitted correctly for96 Logger.97 """98 log = TestLogger()99 for level in LogLevel.iterconstants():100 format = "This is a {level_name} message"101 message = format.format(level_name=level.name)102 logMethod = getattr(log, level.name)103 logMethod(format, junk=message, level_name=level.name)104 # Ensure that test_emit got called with expected arguments105 self.assertEqual(log.emitted["level"], level)106 self.assertEqual(log.emitted["format"], format)107 self.assertEqual(log.emitted["kwargs"]["junk"], message)108 self.assertTrue(hasattr(log, "event"), "No event observed.")109 self.assertEqual(log.event["log_format"], format)110 self.assertEqual(log.event["log_level"], level)111 self.assertEqual(log.event["log_namespace"], __name__)112 self.assertEqual(log.event["log_source"], None)113 self.assertEqual(log.event["junk"], message)114 self.assertEqual(formatEvent(log.event), message)115 def test_sourceOnClass(self):116 """117 C{log_source} event key refers to the class.118 """119 def observer(event):120 self.assertEqual(event["log_source"], Thingo)121 class Thingo(object):122 log = TestLogger(observer=observer)123 Thingo.log.info()124 def test_sourceOnInstance(self):125 """126 C{log_source} event key refers to the instance.127 """128 def observer(event):129 self.assertEqual(event["log_source"], thingo)130 class Thingo(object):131 log = TestLogger(observer=observer)132 thingo = Thingo()133 thingo.log.info()134 def test_sourceUnbound(self):135 """136 C{log_source} event key is C{None}.137 """138 def observer(event):139 self.assertEqual(event["log_source"], None)140 log = TestLogger(observer=observer)141 log.info()142 def test_defaultFailure(self):143 """144 Test that log.failure() emits the right data.145 """146 log = TestLogger()147 try:148 raise RuntimeError("baloney!")149 except RuntimeError:150 log.failure("Whoops")151 errors = self.flushLoggedErrors(RuntimeError)152 self.assertEqual(len(errors), 1)153 self.assertEqual(log.emitted["level"], LogLevel.critical)154 self.assertEqual(log.emitted["format"], "Whoops")155 def test_conflictingKwargs(self):156 """157 Make sure that kwargs conflicting with args don't pass through.158 """159 log = TestLogger()160 log.warn(161 u"*",162 log_format="#",163 log_level=LogLevel.error,164 log_namespace="*namespace*",165 log_source="*source*",166 )167 self.assertEqual(log.event["log_format"], u"*")168 self.assertEqual(log.event["log_level"], LogLevel.warn)169 self.assertEqual(log.event["log_namespace"], log.namespace)170 self.assertEqual(log.event["log_source"], None)171 def test_logInvalidLogLevel(self):172 """173 Test passing in a bogus log level to C{emit()}.174 """175 log = TestLogger()176 log.emit("*bogus*")177 errors = self.flushLoggedErrors(InvalidLogLevelError)178 self.assertEqual(len(errors), 1)179 def test_trace(self):180 """181 Tracing keeps track of forwarding to the publisher.182 """183 def publisher(event):184 observer(event)185 def observer(event):186 self.assertEqual(event["log_trace"], [(log, publisher)])187 log = TestLogger(observer=publisher)...
Automation Testing Tutorials
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run robotframework-appiumlibrary automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
