1from test.support import import_helper, threading_helper
2syslog = import_helper.import_module("syslog") #skip if not supported
3from test import support
4import sys
5import threading
6import time
7import unittest
8
9# XXX(nnorwitz): This test sucks.  I don't know of a platform independent way
10# to verify that the messages were really logged.
11# The only purpose of this test is to verify the code doesn't crash or leak.
12
13class Test(unittest.TestCase):
14
15    def tearDown(self):
16        syslog.closelog()
17
18    def test_openlog(self):
19        syslog.openlog('python')
20        # Issue #6697.
21        self.assertRaises(UnicodeEncodeError, syslog.openlog, '\uD800')
22
23    def test_syslog(self):
24        syslog.openlog('python')
25        syslog.syslog('test message from python test_syslog')
26        syslog.syslog(syslog.LOG_ERR, 'test error from python test_syslog')
27
28    def test_syslog_implicit_open(self):
29        syslog.closelog() # Make sure log is closed
30        syslog.syslog('test message from python test_syslog')
31        syslog.syslog(syslog.LOG_ERR, 'test error from python test_syslog')
32
33    def test_closelog(self):
34        syslog.openlog('python')
35        syslog.closelog()
36        syslog.closelog()  # idempotent operation
37
38    def test_setlogmask(self):
39        mask = syslog.LOG_UPTO(syslog.LOG_WARNING)
40        oldmask = syslog.setlogmask(mask)
41        self.assertEqual(syslog.setlogmask(0), mask)
42        self.assertEqual(syslog.setlogmask(oldmask), mask)
43
44    def test_log_mask(self):
45        mask = syslog.LOG_UPTO(syslog.LOG_WARNING)
46        self.assertTrue(mask & syslog.LOG_MASK(syslog.LOG_WARNING))
47        self.assertTrue(mask & syslog.LOG_MASK(syslog.LOG_ERR))
48        self.assertFalse(mask & syslog.LOG_MASK(syslog.LOG_INFO))
49
50    def test_openlog_noargs(self):
51        syslog.openlog()
52        syslog.syslog('test message from python test_syslog')
53
54    @threading_helper.requires_working_threading()
55    def test_syslog_threaded(self):
56        start = threading.Event()
57        stop = False
58        def opener():
59            start.wait(10)
60            i = 1
61            while not stop:
62                syslog.openlog(f'python-test-{i}')  # new string object
63                i += 1
64        def logger():
65            start.wait(10)
66            while not stop:
67                syslog.syslog('test message from python test_syslog')
68
69        orig_si = sys.getswitchinterval()
70        support.setswitchinterval(1e-9)
71        try:
72            threads = [threading.Thread(target=opener)]
73            threads += [threading.Thread(target=logger) for k in range(10)]
74            with threading_helper.start_threads(threads):
75                start.set()
76                time.sleep(0.1)
77                stop = True
78        finally:
79            sys.setswitchinterval(orig_si)
80
81
82if __name__ == "__main__":
83    unittest.main()
84