Source Code for Module pike.test.multichannel

  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  #        multichannel.py 
 29  # 
 30  # Abstract: 
 31  # 
 32  #        Multichannel tests 
 33  # 
 34  # Authors: Brian Koropoff (brian.koropoff@emc.com) 
 35  # 
 36   
 37  import pike.model 
 38  import pike.smb2 
 39  import pike.test 
 40  import pike.ntstatus 
 41   
 42  @pike.test.RequireDialect(pike.smb2.DIALECT_SMB3_0) 
 43  @pike.test.RequireCapabilities(pike.smb2.SMB2_GLOBAL_CAP_MULTI_CHANNEL) 
 44 -class MultiChannelTest(pike.test.PikeTest): 
 45 -    def session_bind(self, chan): 
 46          return chan.connection.client.connect(self.server).negotiate().session_setup(self.creds, bind=chan.session) 
 47   
 48      # Open a file, binding session to second channel, and close it there 
 49 -    def test_open_bind_close(self): 
 50          chan, tree = self.tree_connect() 
 51          handle = chan.create(tree, 'hello.txt').result() 
 52   
 53          # Open a second channel and bind the session 
 54          chan2 = self.session_bind(chan) 
 55       
 56          # Close the handle on the second channel 
 57          chan2.close(handle) 
 58   
 59      # Simulate a channel failover during a write and confirm that a 
 60      # stale write sent on the original channel is rejected 
 61 -    def test_write_fence_reject_stale(self): 
 62          data_stale = 'stale' 
 63          data_fresh = 'fresh' 
 64          chan, tree = self.tree_connect() 
 65          client = chan.connection.client 
 66           
 67          # Open a file 
 68          handle = chan.create(tree, 'hello.txt', disposition=pike.smb2.FILE_SUPERSEDE).result() 
 69   
 70          # Open a second channel 
 71          chan2 = self.session_bind(chan) 
 72   
 73          # Pretend channel failover occured 
 74          client.channel_sequence = 1 
 75   
 76          # Send write replay 
 77          with chan2.let(flags=pike.smb2.SMB2_FLAGS_REPLAY_OPERATION): 
 78              chan2.write(handle, 0, data_stale) 
 79   
 80          # Send normal write 
 81          chan2.write(handle, 0, data_fresh) 
 82   
 83          # Send stale write on original channel, which should fail. 
 84          # Spec is ambiguous as to status code, but Windows Server 2012 
 85          # seems to return STATUS_FILE_NOT_AVAILABLE 
 86          with self.assert_error(pike.ntstatus.STATUS_FILE_NOT_AVAILABLE): 
 87              with chan.let(channel_sequence=0): 
 88                  chan.write(handle, 0, data_stale) 
 89   
 90          # Read back data to ensure it is not stale value 
 91          result = chan2.read(handle, len(data_fresh), 0).tostring() 
 92          self.assertEquals(result, data_fresh) 
 93   
 94          # Close handle 
 95          chan2.close(handle) 
 96   
 97          # Close second connection 
 98          chan2.connection.close() 
 99