1
0

test_log_context.py 1.4 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243
  1. from twisted.internet import defer
  2. from twisted.internet import reactor
  3. from .. import unittest
  4. from synapse.util.async import sleep
  5. from synapse.util.logcontext import LoggingContext
  6. class LoggingContextTestCase(unittest.TestCase):
  7. def _check_test_key(self, value):
  8. self.assertEquals(
  9. LoggingContext.current_context().test_key, value
  10. )
  11. def test_with_context(self):
  12. with LoggingContext() as context_one:
  13. context_one.test_key = "test"
  14. self._check_test_key("test")
  15. def test_chaining(self):
  16. with LoggingContext() as context_one:
  17. context_one.test_key = "one"
  18. with LoggingContext() as context_two:
  19. self._check_test_key("one")
  20. context_two.test_key = "two"
  21. self._check_test_key("two")
  22. self._check_test_key("one")
  23. @defer.inlineCallbacks
  24. def test_sleep(self):
  25. @defer.inlineCallbacks
  26. def competing_callback():
  27. with LoggingContext() as competing_context:
  28. competing_context.test_key = "competing"
  29. yield sleep(0)
  30. self._check_test_key("competing")
  31. reactor.callLater(0, competing_callback)
  32. with LoggingContext() as context_one:
  33. context_one.test_key = "one"
  34. yield sleep(0)
  35. self._check_test_key("one")