Author: Matti Picus <[email protected]>
Branch: py3.5
Changeset: r93728:c2a7f649293c
Date: 2018-02-01 00:17 +0200
http://bitbucket.org/pypy/pypy/changeset/c2a7f649293c/
Log: remove old tests of deleted module attributes, fix most tests
diff --git a/pypy/module/_multiprocessing/test/test_memory.py
b/pypy/module/_multiprocessing/test/test_memory.py
deleted file mode 100644
--- a/pypy/module/_multiprocessing/test/test_memory.py
+++ /dev/null
@@ -1,40 +0,0 @@
-import sys
-
-class AppTestMemory:
- spaceconfig = dict(usemodules=('_multiprocessing', 'mmap',
- '_rawffi',
- 'signal', 'select', 'fcntl',
- 'binascii', '_posixsubprocess'))
-
- def test_address_of(self):
- import _multiprocessing
- raises(TypeError, _multiprocessing.address_of_buffer, None)
- raises(TypeError, _multiprocessing.address_of_buffer, "a")
-
- def test_mmap_address(self):
- import mmap
- import _multiprocessing
-
- # This is a bit faster than importing ctypes
- import _ctypes
- class c_double(_ctypes._SimpleCData):
- _type_ = "d"
- sizeof_double = _ctypes.sizeof(c_double)
-
- buf = mmap.mmap(-1, 300)
- buf[0:300] = b'\0' * 300
-
- # Get the address of shared memory
- address, length = _multiprocessing.address_of_buffer(buf)
- assert length == 300
-
- # build a ctypes object from it
- var = c_double.from_address(address)
- assert buf[0:sizeof_double] == b'\0' * sizeof_double
- assert var.value == 0
-
- # check that both objects share the same memory
- var.value = 123.456
- assert buf[0:sizeof_double] != b'\0' * sizeof_double
- buf[0:sizeof_double] = b'\0' * sizeof_double
- assert var.value == 0
diff --git a/pypy/module/_multiprocessing/test/test_semaphore.py
b/pypy/module/_multiprocessing/test/test_semaphore.py
--- a/pypy/module/_multiprocessing/test/test_semaphore.py
+++ b/pypy/module/_multiprocessing/test/test_semaphore.py
@@ -102,7 +102,13 @@
assert res == False
def test_semaphore_rebuild(self):
- from _multiprocessing import SemLock, sem_unlink
+ import sys
+ if sys.platform == 'win32':
+ from _multiprocessing import SemLock
+ def sem_unlink(*args):
+ pass
+ else:
+ from _multiprocessing import SemLock, sem_unlink
kind = self.SEMAPHORE
value = 1
maxvalue = 1
diff --git a/pypy/module/_multiprocessing/test/test_win32.py
b/pypy/module/_multiprocessing/test/test_win32.py
--- a/pypy/module/_multiprocessing/test/test_win32.py
+++ b/pypy/module/_multiprocessing/test/test_win32.py
@@ -13,34 +13,3 @@
from _multiprocessing import closesocket
raises(WindowsError, closesocket, -1)
- def test_pipe(self):
- import _winapi as win32
- import os
- address = r'\\.\pipe\pypy-test-%s' % (os.getpid())
- openmode = win32.PIPE_ACCESS_INBOUND
- access = win32.GENERIC_WRITE
- obsize, ibsize = 0, 8192
- readhandle = win32.CreateNamedPipe(
- address, openmode,
- win32.PIPE_TYPE_MESSAGE | win32.PIPE_READMODE_MESSAGE |
- win32.PIPE_WAIT,
- 1, obsize, ibsize, win32.NMPWAIT_WAIT_FOREVER, win32.NULL
- )
- writehandle = win32.CreateFile(
- address, access, 0, win32.NULL, win32.OPEN_EXISTING, 0, win32.NULL
- )
- win32.SetNamedPipeHandleState(
- writehandle, win32.PIPE_READMODE_MESSAGE, None, None)
-
- try:
- win32.ConnectNamedPipe(readhandle, win32.NULL)
- except WindowsError as e:
- if e.args[0] != win32.ERROR_PIPE_CONNECTED:
- raise
-
- timeout = 100
- exc = raises(WindowsError, win32.WaitNamedPipe, address, timeout)
- assert exc.value.winerror == 121 # ERROR_SEM_TIMEOUT
-
- win32.CloseHandle(readhandle)
- win32.CloseHandle(writehandle)
_______________________________________________
pypy-commit mailing list
[email protected]
https://mail.python.org/mailman/listinfo/pypy-commit
