Backup_Repos/beets
1
0
Fork 0
mirror of https://github.com/beetbox/beets.git synced 2025-12-15 13:07:09 +01:00
Code Issues Projects Releases Wiki Activity

Fix concurrent logging testing

Browse source 
Set the verbosity in the config from the beginning.

Use self-made barriers (DummyPlugin.step1, DummyPlugin.step2) and
Thread.join() to ensure everything is going as expected.
This commit is contained in:
Bruno Cauet 2015-04-01 17:38:49 +02:00
parent 2c37602d66
commit 76be623031
1 changed files with 18 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

27
test/test_logging.py
View file

@ -3,7 +3,6 @@ from __future__ import (division, absolute_import, print_function,
unicode_literals) unicode_literals)
import sys import sys
import time
import threading import threading
import logging as log import logging as log
from StringIO import StringIO from StringIO import StringIO
@ -173,6 +172,7 @@ class ConcurrentEventsTest(TestCase, helper.TestHelper):
self.lock2 = threading.Lock() self.lock2 = threading.Lock()
self.test_case = test_case self.test_case = test_case
self.exc_info = None self.exc_info = None
self.t1_step = self.t2_step = 0
def log_all(self, name): def log_all(self, name):
self._log.debug('debug ' + name) self._log.debug('debug ' + name)
@ -182,8 +182,10 @@ class ConcurrentEventsTest(TestCase, helper.TestHelper):
def listener1(self): def listener1(self):
try: try:
self.test_case.assertEqual(self._log.level, log.INFO) self.test_case.assertEqual(self._log.level, log.INFO)
self.t1_step = 1
self.lock1.acquire() self.lock1.acquire()
self.test_case.assertEqual(self._log.level, log.INFO) self.test_case.assertEqual(self._log.level, log.INFO)
self.t1_step = 2
except Exception: except Exception:
import sys import sys
self.exc_info = sys.exc_info() self.exc_info = sys.exc_info()
@ -191,8 +193,10 @@ class ConcurrentEventsTest(TestCase, helper.TestHelper):
def listener2(self): def listener2(self):
try: try:
self.test_case.assertEqual(self._log.level, log.DEBUG) self.test_case.assertEqual(self._log.level, log.DEBUG)
self.t2_step = 1
self.lock2.acquire() self.lock2.acquire()
self.test_case.assertEqual(self._log.level, log.DEBUG) self.test_case.assertEqual(self._log.level, log.DEBUG)
self.t2_step = 2
except Exception: except Exception:
import sys import sys
self.exc_info = sys.exc_info() self.exc_info = sys.exc_info()
@ -215,29 +219,34 @@ class ConcurrentEventsTest(TestCase, helper.TestHelper):
dp.lock2.acquire() dp.lock2.acquire()
self.assertEqual(dp._log.level, log.NOTSET) self.assertEqual(dp._log.level, log.NOTSET)
self.config['verbose'] = 1
t1 = threading.Thread(target=dp.listeners['dummy_event1'][0]) t1 = threading.Thread(target=dp.listeners['dummy_event1'][0])
t1.start() # blocked. t1 tested its log level t1.start() # blocked. t1 tested its log level
check_dp_exc() while dp.t1_step != 1:
check_dp_exc()
self.assertTrue(t1.is_alive()) self.assertTrue(t1.is_alive())
self.assertEqual(dp._log.level, log.NOTSET) self.assertEqual(dp._log.level, log.NOTSET)
self.config['verbose'] = 2 self.config['verbose'] = 2
t2 = threading.Thread(target=dp.listeners['dummy_event2'][0]) t2 = threading.Thread(target=dp.listeners['dummy_event2'][0])
t2.start() # blocked. t2 tested its log level t2.start() # blocked. t2 tested its log level
check_dp_exc() while dp.t2_step != 1:
check_dp_exc()
self.assertTrue(t2.is_alive()) self.assertTrue(t2.is_alive())
self.assertEqual(dp._log.level, log.NOTSET) self.assertEqual(dp._log.level, log.NOTSET)
dp.lock1.release() dp.lock1.release() # dummy_event1 tests its log level + finishes
time.sleep(.1) # dummy_event1 tests its log level + finishes while dp.t1_step != 2:
check_dp_exc() check_dp_exc()
t1.join(.1)
self.assertFalse(t1.is_alive()) self.assertFalse(t1.is_alive())
self.assertTrue(t2.is_alive()) self.assertTrue(t2.is_alive())
self.assertEqual(dp._log.level, log.NOTSET) self.assertEqual(dp._log.level, log.NOTSET)
dp.lock2.release() dp.lock2.release() # dummy_event2 tests its log level + finishes
time.sleep(.1) # dummy_event2 tests its log level + finishes while dp.t2_step != 2:
check_dp_exc() check_dp_exc()
t2.join(.1)
self.assertFalse(t2.is_alive()) self.assertFalse(t2.is_alive())
except: except: