# Copyright (C) 2019 Red Hat, Inc. # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation; either version 2 of the License, or # (at your option) any later version. """ Test sanlock python binding with sanlock daemon. """ from __future__ import absolute_import import errno import io import os import time from contextlib import contextmanager import pytest import sanlock from . import constants from . import util from . units import KiB, MiB, GiB, TiB # Largest file size on ext4 is 16TiB, and on xfs 500 TiB. Use 1 TiB as it is # large enough to test large offsets, and less likely to fail on developer # machine or CI slave. # See https://access.redhat.com/articles/rhel-limits LARGE_FILE_SIZE = TiB LOCKSPACE_SIZE = MiB MIN_RES_SIZE = MiB ALIGNMENT_1M = 1 * MiB ALIGNMENT_2M = 2 * MiB SECTOR_SIZE_512 = 512 SECTOR_SIZE_4K = 4 * KiB FILE_NAMES = [ # name, encoding ("ascii", None), (u"ascii", None), (u"\u05d0", None), (u"\u05d0", "utf-8"), ] NAME = b"\xd7\x90" # Unicode Hebrew Alef using utf-8 encoding. @pytest.mark.parametrize("filename, encoding", FILE_NAMES) @pytest.mark.parametrize("size, offset", [ # Smallest offset. (LOCKSPACE_SIZE, 0), # Large offset. (LARGE_FILE_SIZE, LARGE_FILE_SIZE - LOCKSPACE_SIZE), ]) def test_write_lockspace( tmpdir, sanlock_daemon, filename, encoding, size, offset): path = util.generate_path(tmpdir, filename, encoding) util.create_file(path, size) # Test read and write with default alignment and sector size values. sanlock.write_lockspace(b"ls_name", path, offset=offset, iotimeout=1) ls = sanlock.read_lockspace(path, offset=offset) assert ls == {"iotimeout": 1, "lockspace": b"ls_name"} # Test read and write with explicit alignment and sector size values. sanlock.write_lockspace( b"ls_name", path, offset=offset, iotimeout=1, align=ALIGNMENT_1M, sector=SECTOR_SIZE_512) ls = sanlock.read_lockspace( path, offset=offset, align=ALIGNMENT_1M, sector=SECTOR_SIZE_512) assert ls == {"iotimeout": 1, "lockspace": b"ls_name"} acquired = sanlock.inq_lockspace( b"ls_name", 1, path, offset=offset, wait=False) assert acquired is False magic = util.read_magic(path, offset) assert magic == constants.DELTA_DISK_MAGIC # TODO: check more stuff here... util.check_guard(path, size) @pytest.mark.parametrize("align", sanlock.ALIGN_SIZE) def test_write_lockspace_4k(user_4k_path, sanlock_daemon, align): # Poison lockspace area, ensuring that previous tests will not break this # test, and sanlock does not write beyond the lockspace area. with io.open(user_4k_path, "rb+") as f: f.write(align * b"x") util.write_guard(user_4k_path, align) sanlock.write_lockspace( b"ls_name", user_4k_path, iotimeout=1, align=align, sector=SECTOR_SIZE_4K) ls = sanlock.read_lockspace( user_4k_path, align=align, sector=SECTOR_SIZE_4K) assert ls == {"iotimeout": 1, "lockspace": b"ls_name"} acquired = sanlock.inq_lockspace(b"ls_name", 1, user_4k_path, wait=False) assert acquired is False # Verify that lockspace was written. magic = util.read_magic(user_4k_path) assert magic == constants.DELTA_DISK_MAGIC # Check that sanlock did not write beyond the lockspace area. util.check_guard(user_4k_path, align) def test_write_lockspace_4k_invalid_sector_size(sanlock_daemon, user_4k_path): with pytest.raises(sanlock.SanlockException) as e: sanlock.write_lockspace( b"ls_name", user_4k_path, iotimeout=1, sector=SECTOR_SIZE_512) assert e.value.errno == errno.EINVAL def test_read_lockspace_4k_invalid_sector_size(sanlock_daemon, user_4k_path): sanlock.write_lockspace( b"ls_name", user_4k_path, iotimeout=1, sector=SECTOR_SIZE_4K) with pytest.raises(sanlock.SanlockException) as e: sanlock.read_lockspace(user_4k_path, sector=SECTOR_SIZE_512) assert e.value.errno == errno.EINVAL @pytest.mark.parametrize("filename,encoding", FILE_NAMES) @pytest.mark.parametrize("size,offset", [ # Smallest offset. (MIN_RES_SIZE, 0), # Large offset. (LARGE_FILE_SIZE, LARGE_FILE_SIZE - MIN_RES_SIZE), ]) def test_write_resource( tmpdir, sanlock_daemon, filename, encoding, size, offset): path = util.generate_path(tmpdir, filename, encoding) util.create_file(path, size) disks = [(path, offset)] # Test read and write with default alignment and sector size values. sanlock.write_resource(b"ls_name", b"res_name", disks) res = sanlock.read_resource(path, offset=offset) assert res == { "lockspace": b"ls_name", "resource": b"res_name", "version": 0 } # Test read and write with explicit alignment and sector size values. sanlock.write_resource( b"ls_name", b"res_name", disks, align=ALIGNMENT_1M, sector=SECTOR_SIZE_512) res = sanlock.read_resource( path, offset=offset, align=ALIGNMENT_1M, sector=SECTOR_SIZE_512) assert res == { "lockspace": b"ls_name", "resource": b"res_name", "version": 0 } owners = sanlock.read_resource_owners(b"ls_name", b"res_name", disks) assert owners == [] magic = util.read_magic(path, offset) assert magic == constants.PAXOS_DISK_MAGIC util.check_guard(path, size) @pytest.mark.parametrize("align", sanlock.ALIGN_SIZE) def test_write_resource_4k(sanlock_daemon, user_4k_path, align): disks = [(user_4k_path, 0)] # Poison resource area, ensuring that previous tests will not break this # test, and sanlock does not write beyond the lockspace area. with io.open(user_4k_path, "rb+") as f: f.write(align * b"x") util.write_guard(user_4k_path, align) sanlock.write_resource( b"ls_name", b"res_name", disks, align=align, sector=SECTOR_SIZE_4K) res = sanlock.read_resource( user_4k_path, align=align, sector=SECTOR_SIZE_4K) assert res == { "lockspace": b"ls_name", "resource": b"res_name", "version": 0 } owners = sanlock.read_resource_owners( b"ls_name", b"res_name", disks, align=align, sector=SECTOR_SIZE_4K) assert owners == [] # Verify that resource was written. magic = util.read_magic(user_4k_path) assert magic == constants.PAXOS_DISK_MAGIC # Check that sanlock did not write beyond the lockspace area. util.check_guard(user_4k_path, align) @pytest.mark.xfail(reason="need to investigate why the call succeed") def test_write_resource_4k_invalid_sector_size(sanlock_daemon, user_4k_path): disks = [(user_4k_path, 0)] with pytest.raises(sanlock.SanlockException) as e: sanlock.write_resource( b"ls_name", b"res_name", disks, sector=SECTOR_SIZE_512) assert e.value.errno == errno.EINVAL def test_clear_resource(tmpdir, sanlock_daemon): path = util.generate_path(tmpdir, "clear_test") util.create_file(path, MiB) disks = [(path, 0)] sanlock.write_resource(b"ls_name", b"res_name", disks) sanlock.write_resource(b"ls_name", b"res_name", disks, clear=True) with pytest.raises(sanlock.SanlockException) as e: sanlock.read_resource(path) assert e.value.errno == constants.SANLK_LEADER_MAGIC magic = util.read_magic(path) assert magic == constants.PAXOS_DISK_CLEAR util.check_guard(path, MiB) # run clear on already cleared resource sanlock.write_resource(b"ls_name", b"res_name", disks, clear=True) magic = util.read_magic(path) assert magic == constants.PAXOS_DISK_CLEAR def test_clear_empty_lockspace_resource(tmpdir, sanlock_daemon): path = util.generate_path(tmpdir, "clear_test") util.create_file(path, MiB) disks = [(path, 0)] sanlock.write_resource(b"ls_name", b"res_name", disks) # Clear with empty lockspace and resource - should succeed sanlock.write_resource(b"", b"", disks, clear=True) magic = util.read_magic(path) assert magic == constants.PAXOS_DISK_CLEAR def test_clear_empty_storage(tmpdir, sanlock_daemon): path = util.generate_path(tmpdir, "clear_test") util.create_file(path, MiB) disks = [(path, 0)] # Clear area without any resource written - should succeed sanlock.write_resource(b"ls_name", b"inval_res_name", disks, clear=True) magic = util.read_magic(path) assert magic == constants.PAXOS_DISK_CLEAR def test_read_resource_4k_invalid_sector_size(sanlock_daemon, user_4k_path): disks = [(user_4k_path, 0)] sanlock.write_resource( b"ls_name", b"res_name", disks, align=ALIGNMENT_1M, sector=SECTOR_SIZE_4K) with pytest.raises(sanlock.SanlockException) as e: sanlock.read_resource(user_4k_path, sector=SECTOR_SIZE_512) assert e.value.errno == errno.EINVAL def test_read_resource_owners_4k_invalid_sector_size( sanlock_daemon, user_4k_path): disks = [(user_4k_path, 0)] sanlock.write_resource( b"ls_name", b"res_name", disks, align=ALIGNMENT_1M, sector=SECTOR_SIZE_4K) with pytest.raises(sanlock.SanlockException) as e: sanlock.read_resource_owners( b"ls_name", b"res_name", disks, sector=SECTOR_SIZE_512) assert e.value.errno == errno.EINVAL def test_read_resource_owners_invalid_align_size(tmpdir, sanlock_daemon): path = str(tmpdir.join("path")) util.create_file(path, GiB) disks = [(path, 0)] sanlock.write_resource( b"ls_name", b"res_name", disks, align=ALIGNMENT_1M, sector=SECTOR_SIZE_512) with pytest.raises(sanlock.SanlockException) as e: sanlock.read_resource_owners( b"ls_name", b"res_name", disks, align=ALIGNMENT_2M, sector=SECTOR_SIZE_512) assert e.value.errno == errno.EINVAL @pytest.mark.parametrize("size,offset", [ # Smallest offset. (MIN_RES_SIZE, 0), # Large offset. (LARGE_FILE_SIZE, LARGE_FILE_SIZE - MIN_RES_SIZE), ]) def test_add_rem_lockspace(tmpdir, sanlock_daemon, size, offset): path = str(tmpdir.join("ls_name")) util.create_file(path, size) sanlock.write_lockspace(b"ls_name", path, offset=offset, iotimeout=1) # Since the lockspace is not acquired, we expect to get False. acquired = sanlock.inq_lockspace( b"ls_name", 1, path, offset=offset, wait=False) assert acquired is False sanlock.add_lockspace(b"ls_name", 1, path, offset=offset, iotimeout=1) # Once the lockspace is acquired, we expect to get True. acquired = sanlock.inq_lockspace( b"ls_name", 1, path, offset=offset, wait=False) assert acquired is True lockspaces = sanlock.get_lockspaces() assert lockspaces == [{ 'flags': 0, 'host_id': 1, 'lockspace': b'ls_name', 'offset': offset, 'path': path }] sanlock.rem_lockspace(b"ls_name", 1, path, offset=offset) # Once the lockspace is released, we expect to get False. acquired = sanlock.inq_lockspace( b"ls_name", 1, path, offset=offset, wait=False) assert acquired is False lockspaces = sanlock.get_lockspaces() assert lockspaces == [] def test_add_rem_lockspace_async(tmpdir, sanlock_daemon): path = str(tmpdir.join("ls_name")) util.create_file(path, MiB) sanlock.write_lockspace(b"ls_name", path, iotimeout=1) acquired = sanlock.inq_lockspace(b"ls_name", 1, path, wait=False) assert acquired is False # This will take 3 seconds. sanlock.add_lockspace(b"ls_name", 1, path, iotimeout=1, wait=False) # While the lockspace is being acquired, we expect to get None. time.sleep(1) acquired = sanlock.inq_lockspace(b"ls_name", 1, path, wait=False) assert acquired is None # Once the lockspace is acquired, we expect to get True. acquired = sanlock.inq_lockspace(b"ls_name", 1, path, wait=True) assert acquired is True # This will take about 3 seconds. sanlock.rem_lockspace(b"ls_name", 1, path, wait=False) # Wait until the lockspace change state from True to None. while sanlock.inq_lockspace(b"ls_name", 1, path, wait=False): time.sleep(1) # While the lockspace is being released, we expect to get None. acquired = sanlock.inq_lockspace(b"ls_name", 1, path, wait=False) assert acquired is None # Once the lockspace was released, we expect to get False. acquired = sanlock.inq_lockspace(b"ls_name", 1, path, wait=True) assert acquired is False @pytest.mark.parametrize("size,offset", [ # Smallest offset. (MIN_RES_SIZE, 0), # Large offset. (LARGE_FILE_SIZE, LARGE_FILE_SIZE - MIN_RES_SIZE), ]) def test_acquire_release_resource(tmpdir, sanlock_daemon, size, offset): ls_path = str(tmpdir.join("ls_name")) util.create_file(ls_path, size) res_path = str(tmpdir.join("res_name")) util.create_file(res_path, size) sanlock.write_lockspace(b"ls_name", ls_path, offset=offset, iotimeout=1) sanlock.add_lockspace(b"ls_name", 1, ls_path, offset=offset, iotimeout=1) # Host status is not available until the first renewal. with pytest.raises(sanlock.SanlockException) as e: sanlock.get_hosts(b"ls_name", 1) assert e.value.errno == errno.EAGAIN time.sleep(1) host = sanlock.get_hosts(b"ls_name", 1)[0] assert host["flags"] == sanlock.HOST_LIVE disks = [(res_path, offset)] sanlock.write_resource(b"ls_name", b"res_name", disks) res = sanlock.read_resource(res_path, offset=offset) assert res == { "lockspace": b"ls_name", "resource": b"res_name", "version": 0 } owners = sanlock.read_resource_owners(b"ls_name", b"res_name", disks) assert owners == [] fd = sanlock.register() sanlock.acquire(b"ls_name", b"res_name", disks, slkfd=fd) res = sanlock.read_resource(res_path, offset=offset) assert res == { "lockspace": b"ls_name", "resource": b"res_name", "version": 1 } owner = sanlock.read_resource_owners(b"ls_name", b"res_name", disks)[0] assert owner["host_id"] == 1 assert owner["flags"] == 0 assert owner["generation"] == 1 assert owner["io_timeout"] == 0 # Why 0? # TODO: check timestamp. host = sanlock.get_hosts(b"ls_name", 1)[0] assert host["flags"] == sanlock.HOST_LIVE assert host["generation"] == owner["generation"] sanlock.release(b"ls_name", b"res_name", disks, slkfd=fd) res = sanlock.read_resource(res_path, offset=offset) assert res == { "lockspace": b"ls_name", "resource": b"res_name", "version": 1 } owners = sanlock.read_resource_owners(b"ls_name", b"res_name", disks) assert owners == [] @pytest.mark.parametrize("res_name", [ "ascii", "\u05d0", # Hebrew Alef ]) def test_inquire(tmpdir, sanlock_daemon, res_name): ls_path = str(tmpdir.join("ls_name")) util.create_file(ls_path, MiB) res_path = str(tmpdir.join(res_name)) util.create_file(res_path, 10 * MiB) fd = sanlock.register() # No lockspace yet. assert sanlock.inquire(slkfd=fd) == [] sanlock.write_lockspace(b"ls_name", ls_path, offset=0, iotimeout=1) sanlock.add_lockspace(b"ls_name", 1, ls_path, offset=0, iotimeout=1) # No resources created yet. assert sanlock.inquire(slkfd=fd) == [] resources = [ # name, offset, acquire (b"res-0", 0 * MiB, True), (b"res-1", 1 * MiB, False), (b"res-2", 2 * MiB, True), (b"res-8", 8 * MiB, False), (b"res-9", 9 * MiB, True), ] for res_name, res_offset, acquire in resources: sanlock.write_resource(b"ls_name", res_name, [(res_path, res_offset)]) # No resource acquired yet. assert sanlock.inquire(slkfd=fd) == [] # Acquire resources. for res_name, res_offset, acquire in resources: if acquire: sanlock.acquire( b"ls_name", res_name, [(res_path, res_offset)], slkfd=fd) time.sleep(1) expected = [ { "lockspace": b"ls_name", "resource": b"res-0", "flags": sanlock.RES_LVER, "version": 1, "disks": [(res_path, 0 * MiB)], }, { "lockspace": b"ls_name", "resource": b"res-2", "flags": sanlock.RES_LVER, "version": 1, "disks": [(res_path, 2 * MiB)], }, { "lockspace": b"ls_name", "resource": b"res-9", "flags": sanlock.RES_LVER, "version": 1, "disks": [(res_path, 9 * MiB)], }, ] # Check acquired resources using snlkfd. assert sanlock.inquire(slkfd=fd) == expected # Check acquired resources using pid. assert sanlock.inquire(pid=os.getpid()) == expected for res_name, res_offset, acquire in resources: if acquire: sanlock.release( b"ls_name", res_name, [(res_path, res_offset)], slkfd=fd) # All resource released. assert sanlock.inquire(slkfd=fd) == [] @pytest.mark.parametrize("align, sector", [ # Invalid alignment (KiB, sanlock.SECTOR_SIZE[0]), # Invalid sector size (sanlock.ALIGN_SIZE[0], 8 * KiB), ]) def test_write_lockspace_invalid_align_sector( tmpdir, sanlock_daemon, align, sector): path = str(tmpdir.join("lockspace")) util.create_file(path, LOCKSPACE_SIZE) with pytest.raises(ValueError): sanlock.write_lockspace(b"ls_name", path, align=align, sector=sector) @pytest.mark.parametrize("align, sector", [ # Invalid alignment (KiB, sanlock.SECTOR_SIZE[0]), # Invalid sector size (sanlock.ALIGN_SIZE[0], 8 * KiB), ]) def test_write_resource_invalid_align_sector( tmpdir, sanlock_daemon, align, sector): path = str(tmpdir.join("resources")) util.create_file(path, MIN_RES_SIZE) disks = [(path, 0)] with pytest.raises(ValueError): sanlock.write_resource( b"ls_name", b"res_name", disks, align=align, sector=sector) @pytest.mark.parametrize("disk", [ # Not a tuple - unicode and bytes: "not a tuple", b"not a tuple", u'\u05e9\u05dc\u05d5\u05dd', b"\xd7\x90", # Tuple with incorrect length: (), ("path",), ("path", 0, "extra"), # Tuple with invalid content: (0, "path"), ("path", "not an offset"), ]) def test_write_resource_invalid_disk(tmpdir, sanlock_daemon, disk): # Test parsing disks list with invalid content. disks = [disk] with pytest.raises(ValueError) as e: sanlock.write_resource(b"ls_name", b"res_name", disks) assert repr(disk) in str(e.value) @pytest.mark.parametrize("filename,encoding", FILE_NAMES) def test_killpath(tmpdir, sanlock_daemon, filename, encoding): cmd_path = util.generate_path(tmpdir, filename, encoding) fd = sanlock.register() sanlock.killpath(cmd_path, [cmd_path], fd) @contextmanager def raises_sanlock_errno(expected_errno=errno.ECONNREFUSED): with pytest.raises(sanlock.SanlockException) as e: yield assert e.value.errno == expected_errno @pytest.mark.parametrize("filename,encoding", FILE_NAMES) def test_rem_lockspace_parse_args(no_sanlock_daemon, filename, encoding): path = util.generate_path("/tmp/", filename, encoding) with raises_sanlock_errno(): sanlock.rem_lockspace(NAME, 1, path, 0, wait=False) @pytest.mark.parametrize("filename,encoding", FILE_NAMES) def test_add_lockspace_parse_args(no_sanlock_daemon, filename, encoding): path = util.generate_path("/tmp/", filename, encoding) with raises_sanlock_errno(): sanlock.add_lockspace(NAME, 1, path, 0, wait=False) @pytest.mark.parametrize("filename,encoding", FILE_NAMES) def test_write_lockspace_parse_args(no_sanlock_daemon, filename, encoding): path = util.generate_path("/tmp/", filename, encoding) with raises_sanlock_errno(): sanlock.write_lockspace(NAME, path) @pytest.mark.parametrize("filename,encoding", FILE_NAMES) def test_write_resource_parse_args(no_sanlock_daemon, filename, encoding): path = util.generate_path("/tmp/", filename, encoding) disks = [(path, 0)] with raises_sanlock_errno(): sanlock.write_resource(NAME, b"res_name", disks) with raises_sanlock_errno(): sanlock.write_resource(b"ls_name", NAME, disks) def test_write_resource_path_length(no_sanlock_daemon): path = "x" * constants.SANLK_PATH_LEN with pytest.raises(ValueError): sanlock.write_resource(b"ls_name", b"res_name", [(path, 0)]) path = "x" * (constants.SANLK_PATH_LEN - 1) with raises_sanlock_errno(): sanlock.write_resource(b"ls_name", b"res_name", [(path, 0)]) @pytest.mark.parametrize("filename,encoding", FILE_NAMES) def test_release_resource_parse_args(no_sanlock_daemon, filename, encoding): path = util.generate_path("/tmp/", filename, encoding) disks = [(path, 0)] with raises_sanlock_errno(): sanlock.release(NAME, b"res_name", disks) with raises_sanlock_errno(): sanlock.release(b"ls_name", NAME, disks) def test_release_resource_path_length(no_sanlock_daemon): path = "x" * constants.SANLK_PATH_LEN with pytest.raises(ValueError): sanlock.release(b"ls_name", b"res_name", [(path, 0)]) path = "x" * (constants.SANLK_PATH_LEN - 1) with raises_sanlock_errno(): sanlock.release(b"ls_name", b"res_name", [(path, 0)]) @pytest.mark.parametrize("filename,encoding", FILE_NAMES) def test_read_resource_owners_parse_args( no_sanlock_daemon, filename, encoding): path = util.generate_path("/tmp/", filename, encoding) disks = [(path, 0)] with raises_sanlock_errno(): sanlock.read_resource_owners(NAME, b"res_name", disks) with raises_sanlock_errno(): sanlock.read_resource_owners(b"ls_name", NAME, disks) def test_read_resource_owners_path_length(no_sanlock_daemon): path = "x" * constants.SANLK_PATH_LEN with pytest.raises(ValueError): sanlock.read_resource_owners(b"ls_name", b"res_name", [(path, 0)]) path = "x" * (constants.SANLK_PATH_LEN - 1) with raises_sanlock_errno(): sanlock.read_resource_owners(b"ls_name", b"res_name", [(path, 0)]) def test_get_hosts_parse_args(no_sanlock_daemon): with raises_sanlock_errno(): sanlock.get_hosts(NAME, 1) @pytest.mark.parametrize("filename,encoding", FILE_NAMES) def test_inq_lockspace_parse_args(no_sanlock_daemon, filename, encoding): path = util.generate_path("/tmp/", filename, encoding) with raises_sanlock_errno(): sanlock.inq_lockspace(NAME, 1, path, wait=False) def test_reg_event_parse_args(no_sanlock_daemon): with raises_sanlock_errno(): sanlock.reg_event(NAME) def test_end_event_parse_args(no_sanlock_daemon): with raises_sanlock_errno(errno.EALREADY): sanlock.end_event(-1, NAME) def test_set_event_parse_args(no_sanlock_daemon): with raises_sanlock_errno(): sanlock.set_event(NAME, 1, 1, 1) @pytest.mark.parametrize("filename,encoding", FILE_NAMES) def test_get_alignment_parse_args(no_sanlock_daemon, filename, encoding): path = util.generate_path("/tmp/", filename, encoding) with raises_sanlock_errno(errno.ENOENT): sanlock.get_alignment(path) @pytest.mark.parametrize("filename,encoding", FILE_NAMES) def test_read_lockspace_parse_args(no_sanlock_daemon, filename, encoding): path = util.generate_path("/tmp/", filename, encoding) with raises_sanlock_errno(): sanlock.read_lockspace(path) @pytest.mark.parametrize("filename,encoding", FILE_NAMES) def test_read_resource_parse_args(no_sanlock_daemon, filename, encoding): path = util.generate_path("/tmp/", filename, encoding) with raises_sanlock_errno(): sanlock.read_resource(path) def test_read_resource_path_length(no_sanlock_daemon): path = "x" * constants.SANLK_PATH_LEN with pytest.raises(ValueError): sanlock.read_resource(path) path = "x" * (constants.SANLK_PATH_LEN - 1) with raises_sanlock_errno(): sanlock.read_resource(path) @pytest.mark.parametrize("filename,encoding", FILE_NAMES) def test_request_parse_args(no_sanlock_daemon, filename, encoding): path = util.generate_path("/tmp/", filename, encoding) disks = [(path, 0)] with raises_sanlock_errno(): sanlock.request(b"ls_name", NAME, disks) with raises_sanlock_errno(): sanlock.request(NAME, b"res_name", disks) def test_request_path_length(no_sanlock_daemon): path = "x" * constants.SANLK_PATH_LEN with pytest.raises(ValueError): sanlock.request(b"ls_name", b"res_name", [(path, 0)]) path = "x" * (constants.SANLK_PATH_LEN - 1) with raises_sanlock_errno(): sanlock.request(b"ls_name", b"res_name", [(path, 0)]) @pytest.mark.parametrize("filename,encoding", FILE_NAMES) def test_acquire_parse_args(no_sanlock_daemon, filename, encoding): path = util.generate_path("/tmp/", filename, encoding) disks = [(path, 0)] with raises_sanlock_errno(): sanlock.acquire(b"ls_name", NAME, disks, pid=os.getpid()) with raises_sanlock_errno(): sanlock.acquire(NAME, b"res_name", disks, pid=os.getpid()) def test_acquire_path_length(no_sanlock_daemon): path = "x" * constants.SANLK_PATH_LEN with pytest.raises(ValueError): sanlock.acquire(b"ls_name", b"res_name", [(path, 0)], pid=os.getpid()) path = "x" * (constants.SANLK_PATH_LEN - 1) with raises_sanlock_errno(): sanlock.acquire(b"ls_name", b"res_name", [(path, 0)], pid=os.getpid()) def test_lvb(tmpdir, sanlock_daemon): ls_path = str(tmpdir.join("ls_name")) util.create_file(ls_path, MiB) res_path = str(tmpdir.join("res_name")) util.create_file(res_path, MiB) sanlock.write_lockspace(b"ls_name", ls_path, offset=0, iotimeout=1) sanlock.add_lockspace(b"ls_name", 1, ls_path, offset=0, iotimeout=1) disks = [(res_path, 0)] sanlock.write_resource(b"ls_name", b"res_name", disks) fd = sanlock.register() # Test reading unset lvb. sanlock.acquire(b"ls_name", b"res_name", disks, slkfd=fd, lvb=True) result = sanlock.get_lvb(b"ls_name", b"res_name", disks, 512) sanlock.release(b"ls_name", b"res_name", disks, slkfd=fd) assert result == b"\0" * 512 lvb_data = b"first\0second" lvb_sector = lvb_data.ljust(512, b"\0") sanlock.acquire(b"ls_name", b"res_name", disks, slkfd=fd, lvb=True) sanlock.set_lvb(b"ls_name", b"res_name", disks, lvb_sector) sanlock.release(b"ls_name", b"res_name", disks, slkfd=fd) # Test reading complete sector (typical usage). sanlock.acquire(b"ls_name", b"res_name", disks, slkfd=fd, lvb=True) result = sanlock.get_lvb(b"ls_name", b"res_name", disks, len(lvb_sector)) sanlock.release(b"ls_name", b"res_name", disks, slkfd=fd) assert result == lvb_sector # Test reading less than sector size. sanlock.acquire(b"ls_name", b"res_name", disks, slkfd=fd, lvb=True) result = sanlock.get_lvb(b"ls_name", b"res_name", disks, len(lvb_data)) sanlock.release(b"ls_name", b"res_name", disks, slkfd=fd) assert result == lvb_data # Test failure to write more than sector size. sanlock.acquire(b"ls_name", b"res_name", disks, slkfd=fd, lvb=True) with raises_sanlock_errno(errno.E2BIG): sanlock.set_lvb(b"ls_name", b"res_name", disks, lvb_sector + b"x") def test_lvb_invalid_value(): disks = [("/no/such/path", 0)] with pytest.raises(ValueError): sanlock.get_lvb(b"ls_name", b"res_name", disks, 0) with pytest.raises(ValueError): sanlock.get_lvb(b"ls_name", b"res_name", disks, 4097)