pike.test.copychunk

64

65 - def setUp(self):

66 self.chan, self.tree = self.tree_connect()

67

68 - def tearDown(self):

69 self.chan.tree_disconnect(self.tree) 70 self.chan.logoff()

71

72 - def _create_and_write(self, filename, content):

73 """ create / overwrite filename with content """ 74 fh1 = self.chan.create(self.tree, 75 filename, 76 access=access_rwd, 77 share=share_all, 78 disposition=pike.smb2.FILE_SUPERSEDE).result() 79 80 bytes_written = self.chan.write(fh1, 0, content) 81 self.chan.close(fh1)

82

83 - def _open_src_dst(self, src_filename, dst_filename, 84 src_access=None, 85 src_disp=None, 86 src_options=None, 87 dst_access=None, 88 dst_disp=None, 89 dst_options=None):

90 if src_access is None: 91 src_access = pike.smb2.FILE_READ_DATA | pike.smb2.DELETE 92 if dst_access is None: 93 dst_access = access_rwd 94 if src_disp is None: 95 src_disp = pike.smb2.FILE_OPEN 96 if dst_disp is None: 97 dst_disp = pike.smb2.FILE_SUPERSEDE 98 if src_options is None: 99 src_options = pike.smb2.FILE_DELETE_ON_CLOSE 100 if dst_options is None: 101 dst_options = pike.smb2.FILE_DELETE_ON_CLOSE 102 103 if src_filename == dst_filename: 104 src_options = 0 # if we're opening the same file, only delete one 105 fh_src = self.chan.create(self.tree, 106 src_filename, 107 access=src_access, 108 share=share_all, 109 disposition=src_disp, 110 options=src_options).result() 111 112 fh_dst = self.chan.create(self.tree, 113 dst_filename, 114 access=dst_access, 115 share=share_all, 116 disposition=dst_disp, 117 options=dst_options).result() 118 return (fh_src, fh_dst)

119

120 - def generic_ssc_test_case(self, block, number_of_chunks, total_offset=0):

121 """ 122 copy block in number_of_chunks, offset the destination copy by total_offset 123 """ 124 src_filename = "src_copy_chunk_offset.txt" 125 dst_filename = "dst_copy_chunk_offset.txt" 126 self._create_and_write(src_filename, block) 127 128 total_len = len(block) 129 chunk_sz = (total_len / number_of_chunks) + 1 130 this_offset = 0 131 132 chunks = [] 133 while this_offset < total_len: 134 offset = this_offset 135 if this_offset + chunk_sz < total_len: 136 length = chunk_sz 137 else: 138 length = total_len - this_offset 139 chunks.append((offset, offset+total_offset, length)) 140 this_offset += chunk_sz 141 142 fh_src, fh_dst = self._open_src_dst(src_filename, dst_filename) 143 144 result = self.chan.copychunk(fh_src, fh_dst, chunks) 145 self.assertEqual(result[0][0].chunks_written, number_of_chunks) 146 self.assertEqual(result[0][0].total_bytes_written, total_len) 147 148 # read each file and verify the result 149 src_buf = self.chan.read(fh_src, total_len, 0).tostring() 150 self.assertBufferEqual(src_buf, block) 151 dst_buf = self.chan.read(fh_dst, total_len, total_offset).tostring() 152 self.assertBufferEqual(dst_buf, block) 153 154 self.chan.close(fh_src) 155 self.chan.close(fh_dst)

156

157 - def test_copy_small_file(self):

158 block = "Hello" 159 num_of_chunks = 1 160 self.generic_ssc_test_case(block, num_of_chunks)

161

162 - def test_copy_big_file(self):

163 block = _gen_test_buffer(65535) 164 num_of_chunks = 1 165 self.generic_ssc_test_case(block, num_of_chunks)

166

167 - def test_copy_multiple_chunks(self):

168 block = _gen_test_buffer(65535) 169 num_of_chunks = 10 170 self.generic_ssc_test_case(block, num_of_chunks)

171

172 - def test_copy_max_chunks(self):

173 block = _gen_test_buffer(65535) 174 num_of_chunks = 16 175 self.generic_ssc_test_case(block, num_of_chunks)

176

177 - def test_offset_copy_small_file(self):

178 block = "Hello" 179 num_of_chunks = 1 180 offset = 64 181 self.generic_ssc_test_case(block, num_of_chunks, offset)

182

183 - def test_offset_copy_big_file(self):

184 block = _gen_test_buffer(65535) 185 num_of_chunks = 1 186 offset = 64 187 self.generic_ssc_test_case(block, num_of_chunks, offset)

188

189 - def test_offset_copy_multiple_chunks(self):

190 block = _gen_test_buffer(65535) 191 num_of_chunks = 10 192 offset = 64 193 self.generic_ssc_test_case(block, num_of_chunks, offset)

194

195 - def generic_ssc_same_file_test_case(self, block, number_of_chunks, total_offset=0):

196 """ 197 duplicate block in number_of_chunks to the same file, 198 the copy will be total_offset from the current end of file 199 """ 200 filename = "src_copy_chunk_same.txt" 201 self._create_and_write(filename, block) 202 203 total_len = len(block) 204 chunk_sz = (total_len / number_of_chunks) + 1 205 this_offset = 0 206 207 chunks = [] 208 while this_offset < total_len: 209 offset = this_offset 210 if this_offset + chunk_sz < total_len: 211 length = chunk_sz 212 else: 213 length = total_len - this_offset 214 chunks.append((offset, offset+total_len+total_offset, length)) 215 this_offset += chunk_sz 216 217 fh_src, fh_dst = self._open_src_dst(filename, filename, 218 dst_disp=pike.smb2.FILE_OPEN_IF) 219 220 result = self.chan.copychunk(fh_src, fh_dst, chunks) 221 self.assertEqual(result[0][0].chunks_written, number_of_chunks) 222 self.assertEqual(result[0][0].total_bytes_written, total_len) 223 224 # read the file and verify the result 225 src_buf = self.chan.read(fh_src, total_len, 0).tostring() 226 self.assertBufferEqual(src_buf, block) 227 if total_offset > 0: 228 offs_buf = self.chan.read(fh_src, total_offset, total_len).tostring() 229 self.assertBufferEqual(offs_buf, "\x00"*total_offset) 230 dst_buf = self.chan.read(fh_src, total_len, total_len+total_offset).tostring() 231 self.assertBufferEqual(dst_buf, block) 232 233 self.chan.close(fh_src) 234 self.chan.close(fh_dst)

235

236 - def test_same_small_file(self):

237 block = "Hello" 238 num_of_chunks = 1 239 self.generic_ssc_same_file_test_case(block, num_of_chunks)

240

241 - def test_same_big_file(self):

242 block = _gen_test_buffer(65535) 243 num_of_chunks = 1 244 self.generic_ssc_same_file_test_case(block, num_of_chunks)

245

246 - def test_same_multiple_chunks(self):

247 block = _gen_test_buffer(65535) 248 num_of_chunks = 10 249 self.generic_ssc_same_file_test_case(block, num_of_chunks)

250

251 - def test_same_offset_small_file(self):

252 block = "Hello" 253 num_of_chunks = 1 254 offset = 64 255 self.generic_ssc_same_file_test_case(block, num_of_chunks, offset)

256

257 - def test_same_offset_big_file(self):

258 block = _gen_test_buffer(65535) 259 num_of_chunks = 1 260 offset = 64 261 self.generic_ssc_same_file_test_case(block, num_of_chunks, offset)

262

263 - def test_same_offset_multiple_chunks(self):

264 block = _gen_test_buffer(65535) 265 num_of_chunks = 10 266 offset = 64 267 self.generic_ssc_same_file_test_case(block, num_of_chunks, offset)

268

269 - def generic_ssc_overlap_test_case(self, block, number_of_chunks, overlap_each_block=0):

270 """ 271 copy block in number_of_chunks, each destination block offset will be overlapped 272 over the previous block by overlap_each_block bytes 273 """ 274 src_filename = "src_copy_chunk_overlap.txt" 275 dst_filename = "dst_copy_chunk_overlap.txt" 276 self._create_and_write(src_filename, block) 277 278 total_len = len(block) 279 chunk_sz = (total_len / number_of_chunks) + 1 280 this_offset = 0 281 282 chunks = [] 283 src_block = list(block) 284 dst_block = list(block) 285 while this_offset < total_len: 286 offset = dst_offset = this_offset 287 if offset - overlap_each_block > 0: 288 dst_offset = offset - overlap_each_block 289 if this_offset + chunk_sz < total_len: 290 length = chunk_sz 291 else: 292 length = total_len - this_offset 293 chunks.append((offset, dst_offset, length)) 294 dst_block[dst_offset:dst_offset+length] = \ 295 src_block[offset:offset+length] 296 this_offset += chunk_sz 297 dst_len = dst_offset+length 298 dst_block = "".join(dst_block[:dst_len]) 299 300 fh_src, fh_dst = self._open_src_dst(src_filename, dst_filename) 301 302 result = self.chan.copychunk(fh_src, fh_dst, chunks) 303 self.assertEqual(result[0][0].chunks_written, number_of_chunks) 304 self.assertEqual(result[0][0].total_bytes_written, total_len) 305 306 # read each file and verify the result 307 src_buf = self.chan.read(fh_src, total_len, 0).tostring() 308 self.assertBufferEqual(src_buf, block) 309 dst_buf = self.chan.read(fh_dst, dst_len, 0).tostring() 310 self.assertBufferEqual(dst_buf, dst_block) 311 312 self.chan.close(fh_src) 313 self.chan.close(fh_dst)

314

315 - def test_overlap_multiple_chunks(self):

316 block = _gen_test_buffer(65535) 317 num_of_chunks = 10 318 overlap = 64 319 self.generic_ssc_overlap_test_case(block, num_of_chunks, overlap)

320

321 - def test_overlap_16_chunks_1024_overlap(self):

322 block = _gen_test_buffer(65535) 323 num_of_chunks = 16 324 overlap = 1024 325 self.generic_ssc_overlap_test_case(block, num_of_chunks, overlap)

326

327 - def test_overlap_15_chunks_4096_overlap(self):

328 block = _gen_test_buffer(65535) 329 num_of_chunks = 15 330 overlap = 4096 331 self.generic_ssc_overlap_test_case(block, num_of_chunks, overlap)

332

333 - def generic_ssc_negative_test_case(self, src_access=None, dst_access=None, 334 src_disp=None, dst_disp=None, 335 src_options=None, dst_options=None, 336 src_brl=None, dst_brl=None, 337 exp_error=None):

338 block = "Hello" 339 total_len = len(block) 340 src_filename = "src_negative.txt" 341 dst_filename = "dst_negative.txt" 342 self._create_and_write(src_filename, block) 343 344 fh_src, fh_dst = self._open_src_dst(src_filename, dst_filename, 345 src_access=src_access, 346 src_disp=src_disp, 347 src_options=src_options, 348 dst_access=dst_access, 349 dst_disp=dst_disp, 350 dst_options=dst_options) 351 close_handles = [] 352 if src_brl or dst_brl: 353 chan2, tree2 = self.tree_connect() 354 if src_brl: 355 fh_src_other = chan2.create(tree2, 356 src_filename, 357 access=access_rwd, 358 share=share_all).result() 359 chan2.lock(fh_src_other, 360 [( 0, 2, pike.smb2.SMB2_LOCKFLAG_EXCLUSIVE_LOCK )]).result() 361 close_handles.append((chan2, fh_src_other)) 362 if dst_brl: 363 fh_dst_other = chan2.create(tree2, 364 dst_filename, 365 access=access_rwd, 366 share=share_all).result() 367 chan2.lock(fh_dst_other, 368 [( 3, 2, pike.smb2.SMB2_LOCKFLAG_EXCLUSIVE_LOCK )]).result() 369 close_handles.append((chan2, fh_dst_other)) 370 if exp_error is None: 371 exp_error = pike.ntstatus.STATUS_SUCCESS 372 try: 373 with self.assert_error(exp_error): 374 result = self.chan.copychunk(fh_src, fh_dst, [(0,0,5)]) 375 finally: 376 for chan, fh in close_handles: 377 chan.close(fh) 378 self.chan.close(fh_src) 379 self.chan.close(fh_dst)

380 381 @pike.test.RequireDialect(pike.smb2.DIALECT_SMB3_0)

382 - def test_neg_src_exc_brl(self):

383 """ 384 Initiate copychunk when another handle has an exclusive BRL on the 385 source file (win 8 / 2012) 386 """ 387 self.generic_ssc_negative_test_case(src_brl=True, 388 exp_error=pike.ntstatus.STATUS_FILE_LOCK_CONFLICT)

389 390 @pike.test.RequireDialect(pike.smb2.DIALECT_SMB3_0)

391 - def test_neg_dst_exc_brl(self):

392 """ 393 Initiate copychunk when another handle has an exclusive BRL on the 394 destination file (win 8 / 2012) 395 """ 396 self.generic_ssc_negative_test_case(dst_brl=True, 397 exp_error=pike.ntstatus.STATUS_FILE_LOCK_CONFLICT)

398

399 - def test_neg_src_no_read(self):

400 """ 401 Try to copychunk with no read access on the source file 402 """ 403 self.generic_ssc_negative_test_case(src_access=pike.smb2.FILE_WRITE_DATA | \ 404 pike.smb2.DELETE, 405 exp_error=pike.ntstatus.STATUS_ACCESS_DENIED)

406

407 - def test_neg_dst_no_read(self):

408 """ 409 Try to copychunk with no read access on the destination file 410 """ 411 self.generic_ssc_negative_test_case(dst_access=pike.smb2.FILE_WRITE_DATA | \ 412 pike.smb2.DELETE, 413 exp_error=pike.ntstatus.STATUS_ACCESS_DENIED)

414

415 - def test_neg_dst_no_write(self):

416 """ 417 Try to copychunk with no write access on the destination file 418 """ 419 self.generic_ssc_negative_test_case(dst_access=pike.smb2.FILE_READ_DATA | \ 420 pike.smb2.DELETE, 421 exp_error=pike.ntstatus.STATUS_ACCESS_DENIED)

422

423 - def test_neg_dst_is_a_dir(self):

424 """ 425 Try to copychunk with destination file being a directory 426 """ 427 self.generic_ssc_negative_test_case(dst_options=pike.smb2.FILE_DIRECTORY_FILE | \ 428 pike.smb2.FILE_DELETE_ON_CLOSE, 429 dst_disp=pike.smb2.FILE_OPEN_IF, 430 exp_error=pike.ntstatus.STATUS_INVALID_DEVICE_REQUEST)

Source Code for Module pike.test.copychunk