Source Code for Module pike.test.readwrite

  1  # 
  2  # Copyright (c) 2013, EMC Corporation 
  3  # All rights reserved. 
  4  # 
  5  # Redistribution and use in source and binary forms, with or without 
  6  # modification, are permitted provided that the following conditions are met: 
  7  # 
  8  # 1. Redistributions of source code must retain the above copyright notice, 
  9  # this list of conditions and the following disclaimer. 
 10  # 2. Redistributions in binary form must reproduce the above copyright notice, 
 11  # this list of conditions and the following disclaimer in the documentation 
 12  # and/or other materials provided with the distribution. 
 13  # 
 14  # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" 
 15  # AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE 
 16  # IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE 
 17  # ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE 
 18  # LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR 
 19  # CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF 
 20  # SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS 
 21  # INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN 
 22  # CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) 
 23  # ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE 
 24  # POSSIBILITY OF SUCH DAMAGE. 
 25  # 
 26  # Module Name: 
 27  # 
 28  #        readwrite.py 
 29  # 
 30  # Abstract: 
 31  # 
 32  #        Read and write tests 
 33  # 
 34  # Authors: Arlene Berry (arlene.berry@emc.com) 
 35  # 
 36   
 37  import pike.model 
 38  import pike.smb2 
 39  import pike.test 
 40  import pike.ntstatus 
 41  import array 
 42  import random 
 43   
 44 -class ReadWriteTest(pike.test.PikeTest): 
 45      # Test that we can write to a file 
 46 -    def test_write(self): 
 47          chan, tree = self.tree_connect() 
 48          buffer = "testing123" 
 49   
 50          share_all = pike.smb2.FILE_SHARE_READ | pike.smb2.FILE_SHARE_WRITE | pike.smb2.FILE_SHARE_DELETE 
 51   
 52          file = chan.create(tree, 
 53                             'write.txt', 
 54                             access=pike.smb2.FILE_READ_DATA | pike.smb2.FILE_WRITE_DATA | pike.smb2.DELETE, 
 55                             share=share_all, 
 56                             disposition=pike.smb2.FILE_SUPERSEDE, 
 57                             options=pike.smb2.FILE_DELETE_ON_CLOSE, 
 58                             oplock_level=pike.smb2.SMB2_OPLOCK_LEVEL_EXCLUSIVE).result() 
 59          
 60          bytes_written = chan.write(file, 
 61                                     0, 
 62                                     buffer) 
 63          self.assertEqual(bytes_written, len(buffer)) 
 64           
 65          chan.close(file) 
 66   
 67      # Test that a 0-byte write succeeds 
 68 -    def test_write_none(self): 
 69          chan, tree = self.tree_connect() 
 70          buffer = None 
 71   
 72          share_all = pike.smb2.FILE_SHARE_READ | pike.smb2.FILE_SHARE_WRITE | pike.smb2.FILE_SHARE_DELETE 
 73   
 74          file = chan.create(tree, 
 75                             'write.txt', 
 76                             access=pike.smb2.FILE_READ_DATA | pike.smb2.FILE_WRITE_DATA | pike.smb2.DELETE, 
 77                             share=share_all, 
 78                             disposition=pike.smb2.FILE_SUPERSEDE, 
 79                             options=pike.smb2.FILE_DELETE_ON_CLOSE, 
 80                             oplock_level=pike.smb2.SMB2_OPLOCK_LEVEL_EXCLUSIVE).result() 
 81          
 82          bytes_written = chan.write(file, 
 83                                     0, 
 84                                     buffer) 
 85          self.assertEqual(bytes_written, 0) 
 86           
 87          chan.close(file) 
 88   
 89      # Test that a 0-byte write succeeds 
 90 -    def test_write_none(self): 
 91          chan, tree = self.tree_connect() 
 92          buffer = None 
 93   
 94          share_all = pike.smb2.FILE_SHARE_READ | pike.smb2.FILE_SHARE_WRITE | pike.smb2.FILE_SHARE_DELETE 
 95   
 96          file = chan.create(tree, 
 97                             'write.txt', 
 98                             access=pike.smb2.FILE_READ_DATA | pike.smb2.FILE_WRITE_DATA | pike.smb2.DELETE, 
 99                             share=share_all, 
100                             disposition=pike.smb2.FILE_SUPERSEDE, 
101                             options=pike.smb2.FILE_DELETE_ON_CLOSE, 
102                             oplock_level=pike.smb2.SMB2_OPLOCK_LEVEL_EXCLUSIVE).result() 
103          
104          bytes_written = chan.write(file, 
105                                     0, 
106                                     buffer) 
107          self.assertEqual(bytes_written, 0) 
108           
109          chan.close(file) 
110   
111      # Test that a 0-byte write triggers access checks 
112      # (and fails since we do not have write access) 
113 -    def test_write_none_access(self): 
114          chan, tree = self.tree_connect() 
115   
116          share_all = pike.smb2.FILE_SHARE_READ | pike.smb2.FILE_SHARE_WRITE | pike.smb2.FILE_SHARE_DELETE 
117   
118          file = chan.create(tree, 
119                             'write.txt', 
120                             access=pike.smb2.FILE_READ_DATA | pike.smb2.DELETE, 
121                             share=share_all, 
122                             disposition=pike.smb2.FILE_SUPERSEDE, 
123                             oplock_level=pike.smb2.SMB2_OPLOCK_LEVEL_EXCLUSIVE).result() 
124          
125          with self.assert_error(pike.ntstatus.STATUS_ACCESS_DENIED): 
126              chan.write(file, 0, None) 
127           
128          chan.close(file) 
129   
130      # Test that 0-byte write does not cause an oplock break 
131 -    def test_write_none_oplock(self): 
132          chan, tree = self.tree_connect() 
133   
134          share_all = pike.smb2.FILE_SHARE_READ | pike.smb2.FILE_SHARE_WRITE | pike.smb2.FILE_SHARE_DELETE 
135   
136          file = chan.create(tree, 
137                             'write.txt', 
138                             access=pike.smb2.FILE_READ_DATA | pike.smb2.FILE_WRITE_DATA | pike.smb2.DELETE, 
139                             share=share_all, 
140                             disposition=pike.smb2.FILE_SUPERSEDE, 
141                             options=pike.smb2.FILE_DELETE_ON_CLOSE, 
142                             oplock_level=pike.smb2.SMB2_OPLOCK_LEVEL_NONE).result() 
143   
144          file2 = chan.create(tree, 
145                              'write.txt', 
146                              access=pike.smb2.FILE_READ_DATA | pike.smb2.FILE_WRITE_DATA | pike.smb2.DELETE, 
147                              share=share_all, 
148                              disposition=pike.smb2.FILE_SUPERSEDE, 
149                              options=pike.smb2.FILE_DELETE_ON_CLOSE, 
150                              oplock_level=pike.smb2.SMB2_OPLOCK_LEVEL_II).result() 
151          self.assertEqual(file2.oplock_level, pike.smb2.SMB2_OPLOCK_LEVEL_II) 
152   
153          bytes_written = chan.write(file, 
154                                     0, 
155                                     None) 
156          self.assertEqual(bytes_written, 0) 
157   
158          # We should not receive an oplock break 
159          with self.assertRaises(pike.model.TimeoutError): 
160              file2.oplock_future.wait(timeout=0.1) 
161   
162          chan.close(file) 
163          chan.close(file2) 
164   
165      # Test that 0-byte write does not cause a lease break 
166      @pike.test.RequireDialect(pike.smb2.DIALECT_SMB2_1) 
167      @pike.test.RequireCapabilities(pike.smb2.SMB2_GLOBAL_CAP_LEASING) 
168 -    def test_write_none_lease(self): 
169          chan, tree = self.tree_connect() 
170          lease1 = array.array('B',map(random.randint, [0]*16, [255]*16)) 
171   
172          share_all = pike.smb2.FILE_SHARE_READ | pike.smb2.FILE_SHARE_WRITE | pike.smb2.FILE_SHARE_DELETE 
173   
174          file = chan.create(tree, 
175                             'write.txt', 
176                             access=pike.smb2.FILE_READ_DATA | pike.smb2.FILE_WRITE_DATA | pike.smb2.DELETE, 
177                             share=share_all, 
178                             disposition=pike.smb2.FILE_SUPERSEDE, 
179                             oplock_level=pike.smb2.SMB2_OPLOCK_LEVEL_NONE).result() 
180   
181          file2 = chan.create(tree, 
182                              'write.txt', 
183                              access=pike.smb2.FILE_READ_DATA | pike.smb2.FILE_WRITE_DATA | pike.smb2.DELETE, 
184                              share=share_all, 
185                              disposition=pike.smb2.FILE_SUPERSEDE, 
186                              oplock_level=pike.smb2.SMB2_OPLOCK_LEVEL_LEASE, 
187                              lease_key=lease1, 
188                              lease_state=pike.smb2.SMB2_LEASE_READ_CACHING).result() 
189   
190          bytes_written = chan.write(file, 64, None) 
191          self.assertEqual(bytes_written, 0) 
192   
193          # We should not receive a lease break 
194          with self.assertRaises(pike.model.TimeoutError): 
195              file2.lease.future.wait(timeout=0.1) 
196   
197          chan.close(file) 
198          chan.close(file2) 
199