-
Notifications
You must be signed in to change notification settings - Fork 490
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add test for maximum SMB valid pwrite range
We need to validate we can write above 16 TiB offset and that we get properly stopped writing about 64 TiB offset.
- Loading branch information
Showing
1 changed file
with
60 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,60 @@ | ||
import pytest | ||
import random | ||
|
||
from middlewared.test.integration.assets.account import user, group | ||
from middlewared.test.integration.assets.pool import dataset | ||
|
||
from protocols import smb_connection | ||
|
||
from samba import ntstatus | ||
from samba import NTSTATUSError | ||
|
||
SHARE_NAME = 'offset_test' | ||
|
||
LARGE_OFFSET = 0x0000140000000000 | ||
INVALID_OFFSET = 0x0000410000000000 | ||
|
||
|
||
@pytest.fixture(scope='module') | ||
def setup_smb_tests(): | ||
with dataset('smbclient-testing', data={'share_type': 'SMB'}) as ds: | ||
with user({ | ||
'username': 'smbuser', | ||
'full_name': 'smbuser', | ||
'group_create': True, | ||
'password': 'Abcd1234' | ||
}) as u: | ||
with smb_share(os.path.join('/mnt', ds), SHARE_NAME) as s: | ||
try: | ||
call('service.start', 'cifs') | ||
yield {'dataset': ds, 'share': s, 'user': u} | ||
finally: | ||
call('service.stop', 'cifs') | ||
|
||
|
||
def test_valid_offset(setup_smb_tests): | ||
ds, share, smb_user = setup_smb_tests | ||
with smb_connection( | ||
share=SHARE_NAME, | ||
username=smb_user['username'], | ||
password='Abcd1234', | ||
smb1=False | ||
) as c: | ||
fd = c.create_file('file_valid_offset', 'w') | ||
buf = random.randbytes(1024) | ||
|
||
c.write(fd, offset=LARGE_OFFSET, data=buf) | ||
out = c.read(fd, offset=LARGE_OFFSET, cnt=1024) | ||
assert buf == out | ||
|
||
|
||
def test_invalid_offset(setup_smb_tests): | ||
ds, share, smb_user = setup_smb_tests | ||
with smb_connection( | ||
share=SHARE_NAME, | ||
username=smb_user['username'], | ||
password='Abcd1234', | ||
smb1=False | ||
) as c: | ||
fd = c.create_file('file_valid_offset', 'w') | ||
c.write(fd, offset=INVALID_OFFSET, data=b'CANARY') |