# Python imports import unittest import time import os # Project imports from .base import Base, make_key, sleep_past_granularity import sysv_ipc # Not tested -- # - mode seems to be settable and readable, but ignored by the OS # - address param of attach() # - attempt to write to segment attached with SHM_RDONLY gives a segfault under OS X and Linux. class SharedMemoryTestBase(Base): """base class for SharedMemory test classes""" # SIZE should be something that's not a power of 2 since that's more # likely to expose odd behavior. SIZE = 3333 def setUp(self): self.mem = sysv_ipc.SharedMemory(None, sysv_ipc.IPC_CREX, size=self.SIZE) def tearDown(self): if self.mem: if self.mem.attached: self.mem.detach() self.mem.remove() def assertWriteToReadOnlyPropertyFails(self, property_name, value): """test that writing to a readonly property raises TypeError""" Base.assertWriteToReadOnlyPropertyFails(self, self.mem, property_name, value) class TestSharedMemoryCreation(SharedMemoryTestBase): """Exercise stuff related to creating SharedMemory""" def get_block_size(self): """Return block size as reported by operating system""" return os.statvfs('.')[1] def test_no_flags(self): """tests that opening a SharedMemory with no flags opens the existing SharedMemory and doesn't create a new SharedMemory""" mem_copy = sysv_ipc.SharedMemory(self.mem.key) self.assertEqual(self.mem.key, mem_copy.key) def test_IPC_CREAT_existing(self): """tests sysv_ipc.IPC_CREAT to open an existing SharedMemory without IPC_EXCL""" mem_copy = sysv_ipc.SharedMemory(self.mem.key, sysv_ipc.IPC_CREAT) self.assertEqual(self.mem.key, mem_copy.key) def test_IPC_CREAT_new(self): """tests sysv_ipc.IPC_CREAT to create a new SharedMemory without IPC_EXCL""" # I can't pass None for the name unless I also pass IPC_EXCL. key = make_key() # Note: this method of finding an unused key is vulnerable to a race # condition. It's good enough for test, but don't copy it for use in # production code! key_is_available = False while not key_is_available: try: mem = sysv_ipc.SharedMemory(key) mem.detach() mem.remove() except sysv_ipc.ExistentialError: key_is_available = True else: key = make_key() mem = sysv_ipc.SharedMemory(key, sysv_ipc.IPC_CREAT, size=sysv_ipc.PAGE_SIZE) self.assertIsNotNone(mem) mem.detach() mem.remove() def test_IPC_EXCL(self): """tests IPC_CREAT | IPC_EXCL prevents opening an existing SharedMemory""" with self.assertRaises(sysv_ipc.ExistentialError): # I have to specify the size, otherwise I get ValueError: The size is invalid # which trumps the ExistentialError I'm looking for. sysv_ipc.SharedMemory(self.mem.key, sysv_ipc.IPC_CREX, size=self.mem.size) def test_randomly_generated_key(self): """tests that the randomly-generated key works""" # This is tested implicitly elsewhere but I want to test it explicitly mem = sysv_ipc.SharedMemory(None, sysv_ipc.IPC_CREX) self.assertIsNotNone(mem.key) self.assertGreaterEqual(mem.key, sysv_ipc.KEY_MIN) self.assertLessEqual(mem.key, sysv_ipc.KEY_MAX) mem.detach() mem.remove() # don't bother testing mode, it's ignored by the OS? def test_default_flags(self): """tests that the flag is 0 by default (==> open existing)""" mem = sysv_ipc.SharedMemory(self.mem.key) self.assertEqual(self.mem.id, mem.id) mem.detach() def test_size(self): """test that the size specified is (somewhat) respected""" # My experience with posix_ipc was that the Linuxes I tested respect the exact size # specified in the SharedMemory() ctor. # e.g. when self.SIZE = 3333, the mmapped file is # also 3333 bytes. # # OS X's mmapped files always have sizes that are mod 4096 which is probably block size. # # I haven't tested other operating systems, and I haven't run this experiment with sysv_ipc. # # AFAICT the specification doesn't demand that the size has to match # exactly, so this code accepts either value as correct. block_size = self.get_block_size() delta = self.SIZE % block_size if delta: # Round up to nearest block size crude_size = (self.SIZE - delta) + block_size else: crude_size = self.SIZE self.assertIn(self.mem.size, (self.SIZE, crude_size)) def test_default_init_character(self): """tests that the init_character defaulted to a blank""" self.assertEqual(self.mem.read(self.mem.size), b' ' * self.mem.size) def test_nondefault_init_character(self): """tests that the init_character can be something other than the default""" init_character = b'@' mem = sysv_ipc.SharedMemory(None, sysv_ipc.IPC_CREX, init_character=init_character) self.assertEqual(mem.read(mem.size), init_character * mem.size) mem.detach() mem.remove() def test_autoattach(self): """tests that attach() is performed as part of init""" self.assertTrue(self.mem.attached) def test_kwargs(self): """ensure init accepts keyword args as advertised""" # mode 0x180 = 0600. Octal is difficult to express in Python 2/3 compatible code. mem = sysv_ipc.SharedMemory(None, flags=sysv_ipc.IPC_CREX, mode=0x180, size=sysv_ipc.PAGE_SIZE, init_character=b'x') mem.detach() mem.remove() class TestSharedMemoryAttachDetach(SharedMemoryTestBase): """Exercise attach() and detach()""" def test_detach(self): """exercise detach()""" self.assertTrue(self.mem.attached) self.mem.detach() self.assertFalse(self.mem.attached) with self.assertRaises(sysv_ipc.NotAttachedError): self.mem.read(1) def test_attach(self): """exercise attach()""" self.assertTrue(self.mem.attached) self.mem.detach() self.assertFalse(self.mem.attached) self.mem.attach() # Should not raise an error self.mem.read(1) def test_attach_read_only(self): """exercise attach(SHM_RDONLY)""" self.assertTrue(self.mem.attached) self.mem.detach() self.assertFalse(self.mem.attached) self.mem.attach(None, sysv_ipc.SHM_RDONLY) with self.assertRaises(OSError): self.mem.write(' ') def test_attach_kwargs(self): """ensure attach() takes kwargs as advertised""" self.mem.detach() self.mem.attach(address=None, flags=0) class TestSharedMemoryReadWrite(SharedMemoryTestBase): """Exercise read() and write()""" def test_simple_read_write(self): test_string = b'abcdefg' self.assertEqual(self.mem.read(20), b' ' * 20) self.mem.write(test_string) self.assertEqual(self.mem.read(len(test_string)), test_string) def test_read_no_byte_count(self): """test the default return-all aspect of read()""" self.assertEqual(self.mem.read(), b' ' * self.mem.size) def test_read_byte_count(self): """test the byte_count param of read()""" self.assertEqual(len(self.mem.read(5)), 5) def test_read_offset(self): """test the offset param of read()""" test_string = 'abcdefg' self.mem.write(test_string) s = self.mem.read(5, 2) self.assertEqual(s, b'cdefg') def test_read_keywords(self): """ensure read() accepts kwargs""" test_string = 'abcdefg' self.mem.write(test_string) s = self.mem.read(byte_count=5, offset=2) self.assertEqual(s, b'cdefg') def test_read_past_end_of_segment(self): """ensure I don't crash if I try to read past the end of the segment""" self.assertEqual(self.mem.read(self.mem.size + 100), b' ' * self.mem.size) def test_read_past_end_of_segment_with_offset(self): """ensure I don't crash if I try to read past the end of the segment using an offset""" self.assertEqual(self.mem.read(self.mem.size + 100, 100), b' ' * (self.mem.size - 100)) def test_read_bad_offset(self): """Ensure ValueError is raised when I use a bad offset""" with self.assertRaises(ValueError): self.mem.read(offset=self.mem.size) with self.assertRaises(ValueError): self.mem.read(offset=self.mem.size + 1) with self.assertRaises(ValueError): self.mem.read(offset=-1) def test_write_offset(self): """test the offset param of write()""" test_string = 'abcdefg' self.mem.write(test_string, 3) self.assertEqual(self.mem.read(10), b' abcdefg') def test_write_keyword(self): """ensure write() accepts a keyword arg""" test_string = 'abcdefg' self.mem.write(test_string, offset=3) self.assertEqual(self.mem.read(10), b' abcdefg') def test_write_past_end_of_segment(self): """ensure ValueError is raised if I try to write past the end of the segment""" with self.assertRaises(ValueError): self.mem.write('x' * (self.mem.size + 100)) def test_write_past_end_of_segment_with_offset(self): """ensure ValueError is raised if I try to write past the end of the segment w/an offset""" with self.assertRaises(ValueError): self.mem.write('x' * (self.mem.size - 50), 100) def test_write_bad_offset(self): """ensure ValueError is raised if I try to write using a bad offset""" with self.assertRaises(ValueError): self.mem.write('x', -1) def test_ascii_null(self): """ensure I can write & read 0x00""" test_string = b'abc' + bytes(0) + b'def' self.mem.write(test_string) self.assertEqual(self.mem.read(len(test_string)), test_string) def test_utf8(self): """Test writing encoded Unicode""" test_string = 'G' + '\u00F6' + 'teborg' test_string = test_string.encode('utf-8') self.mem.write(test_string) self.assertEqual(self.mem.read(len(test_string)), test_string) def test_read_kwargs(self): """ensure read() accepts keyword args as advertised""" self.mem.read(byte_count=1, offset=0) def test_write_kwargs(self): """ensure write() accepts keyword args as advertised""" self.mem.write(b'x', offset=0) class TestSharedMemoryRemove(SharedMemoryTestBase): """Exercise mem.remove()""" def test_remove(self): """tests that mem.remove() works""" self.mem.detach() self.mem.remove() with self.assertRaises(sysv_ipc.ExistentialError): sysv_ipc.SharedMemory(self.mem.key) # Wipe this out so that self.tearDown() doesn't crash. self.mem = None class TestSharedMemoryPropertiesAndAttributes(SharedMemoryTestBase): """Exercise props and attrs""" def test_property_key(self): """exercise SharedMemory.key""" self.assertGreaterEqual(self.mem.key, sysv_ipc.KEY_MIN) self.assertLessEqual(self.mem.key, sysv_ipc.KEY_MAX) self.assertWriteToReadOnlyPropertyFails('key', 42) def test_property_id(self): """exercise SharedMemory.id""" self.assertGreaterEqual(self.mem.id, 0) self.assertWriteToReadOnlyPropertyFails('id', 42) def test_property_size(self): """exercise SharedMemory.size""" self.assertGreaterEqual(self.mem.size, 0) self.assertWriteToReadOnlyPropertyFails('size', 42) def test_property_address(self): """exercise SharedMemory.address""" self.assertGreaterEqual(self.mem.address, 0) self.assertWriteToReadOnlyPropertyFails('address', 42) def test_property_attached(self): """exercise SharedMemory.attached""" self.assertTrue(self.mem.attached) self.assertWriteToReadOnlyPropertyFails('attached', False) self.mem.detach() self.assertFalse(self.mem.attached) def test_property_last_attach_time(self): """exercise SharedMemory.last_attach_time""" self.mem.detach() original_last_attach_time = self.mem.last_attach_time sleep_past_granularity() # I can't record exactly when this attach() happens, but as long as it is within 5 seconds # of the assertion happening, this test will pass. self.mem.attach() self.assertLess(self.mem.last_attach_time - time.time(), 5) # Ensure the time actually changed. self.assertNotEqual(self.mem.last_attach_time, original_last_attach_time) self.assertWriteToReadOnlyPropertyFails('last_attach_time', 42) def test_property_last_detach_time(self): """exercise SharedMemory.last_detach_time""" original_last_detach_time = self.mem.last_detach_time sleep_past_granularity() # I can't record exactly when this detach() happens, but as long as it is within 5 seconds # of the assertion happening, this test will pass. self.mem.detach() self.assertLess(self.mem.last_detach_time - time.time(), 5) # Ensure the time actually changed. self.assertNotEqual(self.mem.last_detach_time, original_last_detach_time) self.assertWriteToReadOnlyPropertyFails('last_detach_time', 42) def test_property_last_change_time(self): """exercise SharedMemory.last_change_time""" original_last_change_time = self.mem.last_change_time sleep_past_granularity() # I can't record exactly when this last_change_time is set, but as long as it is within # 5 seconds of the assertion happening, this test will pass. # The statement below might seem like a no-op, but setting the UID to any value triggers # a call that should set last_change_time. self.mem.uid = self.mem.uid self.assertLess(self.mem.last_change_time - time.time(), 5) # Ensure the time actually changed. self.assertNotEqual(self.mem.last_change_time, original_last_change_time) self.assertWriteToReadOnlyPropertyFails('last_change_time', 42) def test_property_creator_pid(self): """exercise SharedMemory.creator_pid""" self.assertEqual(self.mem.creator_pid, os.getpid()) self.assertWriteToReadOnlyPropertyFails('creator_pid', 42) def test_property_last_pid(self): """exercise SharedMemory.last_pid""" self.assertEqual(self.mem.last_pid, os.getpid()) self.assertWriteToReadOnlyPropertyFails('last_pid', 42) def test_property_number_attached(self): """exercise SharedMemory.number_attached""" self.assertEqual(self.mem.number_attached, 1) self.mem.detach() self.assertEqual(self.mem.number_attached, 0) self.assertWriteToReadOnlyPropertyFails('number_attached', 42) def test_attribute_uid(self): """exercise SharedMemory.uid""" self.assertEqual(self.mem.uid, os.geteuid()) def test_attribute_gid(self): """exercise SharedMemory.gid""" self.assertEqual(self.mem.gid, os.getgid()) def test_attribute_cuid(self): """exercise SharedMemory.cuid""" self.assertEqual(self.mem.cuid, os.geteuid()) self.assertWriteToReadOnlyPropertyFails('cuid', 42) def test_attribute_cgid(self): """exercise SharedMemory.cgid""" self.assertEqual(self.mem.cgid, os.getgid()) self.assertWriteToReadOnlyPropertyFails('cgid', 42) class BufferProtocolTest(unittest.TestCase): '''Exercise buffer protocol implementation which allows creating memoryviews and bytearrays''' def setUp(self): # Create a shared memory segment and write the (English) alphabet to it. self.mem = sysv_ipc.SharedMemory(None, sysv_ipc.IPC_CREX, size=sysv_ipc.PAGE_SIZE) ASCII_A = 0x61 self.alphabet = ''.join([chr(ASCII_A + i) for i in range(26)]) self.alphabet = bytes(self.alphabet, 'ASCII') self.mem.write(self.alphabet) def tearDown(self): self.mem.detach() self.mem.remove() def test_bytearray(self): '''Exercise creating a writeable bytearray''' # Confirm that we're so far, so good. self.assertEqual(self.mem.read(26), self.alphabet) ba = bytearray(self.mem) self.assertEqual(len(ba), self.mem.size) # Test reading & writing. ba = ba.replace(b'c', b'x') self.assertEqual(ba[:4], b'abxd') def test_memoryview(self): '''Exercise creating a writeable memoryview''' # Confirm that we're so far, so good. self.assertEqual(self.mem.read(26), self.alphabet) mv = memoryview(self.mem) # Ensure the memoryview has the properties I expect. Python populates a lot of the # memoryview attributes (e.g. ndim, strides, etc.) based on what my sysv_ipc code reports # to it, so testing the memoryview attributes is how I test that my code isn't lying to # Python. self.assertIsInstance(mv, memoryview) self.assertEqual(mv.format, 'B') self.assertEqual(mv.itemsize, 1) self.assertEqual(mv.shape, (self.mem.size, )) self.assertEqual(mv.ndim, 1) self.assertEqual(mv.strides, (1, )) self.assertFalse(mv.readonly) self.assertEqual(len(mv), self.mem.size) # Test slicing self.assertEqual([chr(c) for c in mv[3:6]], ['d', 'e', 'f']) # Test writing to the memoryview mv[4] = ord('x') self.assertEqual([chr(c) for c in mv[3:6]], ['d', 'x', 'f']) # Ensure changes to the underlying segment are reflected in the memoryview self.mem.write(b'xxx') self.assertEqual([chr(c) for c in mv[:6]], ['x', 'x', 'x', 'd', 'x', 'f']) if __name__ == '__main__': unittest.main()