# test_multiprocessing_spawn, etc) in parallel.
return prefix + str(os.getpid())
+ def test_shared_memory_name_with_embedded_null(self):
+ name_tsmb = self._new_shm_name('test01_null')
+ sms = shared_memory.SharedMemory(name_tsmb, create=True, size=512)
+ self.addCleanup(sms.unlink)
+ with self.assertRaises(ValueError):
+ shared_memory.SharedMemory(name_tsmb + '\0a', create=False, size=512)
+ if shared_memory._USE_POSIX:
+ orig_name = sms._name
+ try:
+ sms._name = orig_name + '\0a'
+ with self.assertRaises(ValueError):
+ sms.unlink()
+ finally:
+ sms._name = orig_name
+
def test_shared_memory_basics(self):
name_tsmb = self._new_shm_name('test01_tsmb')
sms = shared_memory.SharedMemory(name_tsmb, create=True, size=512)
self.addCleanup(shm2.unlink)
self.assertEqual(shm2._name, names[1])
- def test_invalid_shared_memory_cration(self):
+ def test_invalid_shared_memory_creation(self):
# Test creating a shared memory segment with negative size
with self.assertRaises(ValueError):
sms_invalid = shared_memory.SharedMemory(create=True, size=-1)
{
int fd;
int async_err = 0;
- const char *name = PyUnicode_AsUTF8(path);
+ Py_ssize_t name_size;
+ const char *name = PyUnicode_AsUTF8AndSize(path, &name_size);
if (name == NULL) {
return -1;
}
+ if (strlen(name) != (size_t)name_size) {
+ PyErr_SetString(PyExc_ValueError, "embedded null character");
+ return -1;
+ }
do {
Py_BEGIN_ALLOW_THREADS
fd = shm_open(name, flags, mode);
{
int rv;
int async_err = 0;
- const char *name = PyUnicode_AsUTF8(path);
+ Py_ssize_t name_size;
+ const char *name = PyUnicode_AsUTF8AndSize(path, &name_size);
if (name == NULL) {
return NULL;
}
+ if (strlen(name) != (size_t)name_size) {
+ PyErr_SetString(PyExc_ValueError, "embedded null character");
+ return NULL;
+ }
do {
Py_BEGIN_ALLOW_THREADS
rv = shm_unlink(name);