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()