s3:pylibsmb: only use poll_mt backend if multi_threaded=True is specified
[amitay/samba.git] / python / samba / tests / libsmb_samba_internal.py
1 # Unix SMB/CIFS implementation.
2 # Copyright Volker Lendecke <vl@samba.org> 2012
3 #
4 # This program is free software; you can redistribute it and/or modify
5 # it under the terms of the GNU General Public License as published by
6 # the Free Software Foundation; either version 3 of the License, or
7 # (at your option) any later version.
8 #
9 # This program is distributed in the hope that it will be useful,
10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
12 # GNU General Public License for more details.
13 #
14 # You should have received a copy of the GNU General Public License
15 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
16 #
17
18 """Tests for samba.samba3.libsmb_samba_internal."""
19
20 from samba.samba3 import libsmb_samba_internal
21 from samba.dcerpc import security
22 from samba.samba3 import param as s3param
23 from samba import credentials
24 import samba.tests
25 import threading
26 import sys
27 import os
28
29
30 class LibsmbTestCase(samba.tests.TestCase):
31
32     class OpenClose(threading.Thread):
33
34         def __init__(self, conn, filename, num_ops):
35             threading.Thread.__init__(self)
36             self.conn = conn
37             self.filename = filename
38             self.num_ops = num_ops
39             self.exc = False
40
41         def run(self):
42             c = self.conn
43             try:
44                 for i in range(self.num_ops):
45                     f = c.create(self.filename, CreateDisposition=3,
46                                  DesiredAccess=security.SEC_STD_DELETE)
47                     c.delete_on_close(f, True)
48                     c.close(f)
49             except Exception:
50                 self.exc = sys.exc_info()
51
52     def test_OpenClose(self):
53
54         lp = s3param.get_context()
55         lp.load(os.getenv("SMB_CONF_PATH"))
56
57         creds = credentials.Credentials()
58         creds.guess(lp)
59         creds.set_username(os.getenv("USERNAME"))
60         creds.set_password(os.getenv("PASSWORD"))
61
62         c = libsmb_samba_internal.Conn(os.getenv("SERVER_IP"), "tmp",
63                                        creds, multi_threaded=True)
64
65         mythreads = []
66
67         for i in range(3):
68             t = LibsmbTestCase.OpenClose(c, "test" + str(i), 10)
69             mythreads.append(t)
70
71         for t in mythreads:
72             t.start()
73
74         for t in mythreads:
75             t.join()
76             if t.exc:
77                 raise t.exc[0](t.exc[1])
78
79
80 if __name__ == "__main__":
81     import unittest
82     unittest.main()