Skip to content
Projects
Groups
Snippets
Help
Loading...
Help
Support
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in / Register
Toggle navigation
P
Pyston
Project overview
Project overview
Details
Activity
Releases
Repository
Repository
Files
Commits
Branches
Tags
Contributors
Graph
Compare
Issues
0
Issues
0
List
Boards
Labels
Milestones
Merge Requests
0
Merge Requests
0
Analytics
Analytics
Repository
Value Stream
Wiki
Wiki
Members
Members
Collapse sidebar
Close sidebar
Activity
Graph
Create a new issue
Commits
Issue Boards
Open sidebar
Boxiang Sun
Pyston
Commits
5576cae7
Commit
5576cae7
authored
Sep 10, 2015
by
Kevin Modzelewski
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Enable the mmap module
parent
0f8114cf
Changes
4
Hide whitespace changes
Inline
Side-by-side
Showing
4 changed files
with
730 additions
and
1 deletion
+730
-1
from_cpython/CMakeLists.txt
from_cpython/CMakeLists.txt
+2
-0
from_cpython/Lib/test/test_mmap.py
from_cpython/Lib/test/test_mmap.py
+718
-0
from_cpython/setup.py
from_cpython/setup.py
+9
-1
test/cpython/test_mmap.py
test/cpython/test_mmap.py
+1
-0
No files found.
from_cpython/CMakeLists.txt
View file @
5576cae7
...
...
@@ -125,6 +125,7 @@ add_custom_command(OUTPUT
${
CMAKE_BINARY_DIR
}
/lib_pyston/grp.pyston.so
${
CMAKE_BINARY_DIR
}
/lib_pyston/termios.pyston.so
${
CMAKE_BINARY_DIR
}
/lib_pyston/_curses.pyston.so
${
CMAKE_BINARY_DIR
}
/lib_pyston/mmap.pyston.so
COMMAND
${
CMAKE_BINARY_DIR
}
/pyston setup.py build --build-lib
${
CMAKE_BINARY_DIR
}
/lib_pyston
DEPENDS
pyston
...
...
@@ -150,5 +151,6 @@ add_custom_command(OUTPUT
Modules/grpmodule.c
Modules/termios.c
Modules/_cursesmodule.c
Modules/mmapmodule.c
WORKING_DIRECTORY
${
CMAKE_CURRENT_SOURCE_DIR
}
)
add_custom_target
(
sharedmods DEPENDS
${
CMAKE_BINARY_DIR
}
/lib_pyston/_multiprocessing.pyston.so
)
from_cpython/Lib/test/test_mmap.py
0 → 100644
View file @
5576cae7
from
test.test_support
import
(
TESTFN
,
run_unittest
,
import_module
,
unlink
,
requires
,
_2G
,
_4G
)
import
unittest
import
os
,
re
,
itertools
,
socket
,
sys
mmap
=
import_module
(
'mmap'
)
PAGESIZE
=
mmap
.
PAGESIZE
class
MmapTests
(
unittest
.
TestCase
):
def
setUp
(
self
):
if
os
.
path
.
exists
(
TESTFN
):
os
.
unlink
(
TESTFN
)
def
tearDown
(
self
):
try
:
os
.
unlink
(
TESTFN
)
except
OSError
:
pass
def
test_basic
(
self
):
# Test mmap module on Unix systems and Windows
# Create a file to be mmap'ed.
f
=
open
(
TESTFN
,
'w+'
)
try
:
# Write 2 pages worth of data to the file
f
.
write
(
'
\
0
'
*
PAGESIZE
)
f
.
write
(
'foo'
)
f
.
write
(
'
\
0
'
*
(
PAGESIZE
-
3
)
)
f
.
flush
()
m
=
mmap
.
mmap
(
f
.
fileno
(),
2
*
PAGESIZE
)
f
.
close
()
# Simple sanity checks
tp
=
str
(
type
(
m
))
# SF bug 128713: segfaulted on Linux
self
.
assertEqual
(
m
.
find
(
'foo'
),
PAGESIZE
)
self
.
assertEqual
(
len
(
m
),
2
*
PAGESIZE
)
self
.
assertEqual
(
m
[
0
],
'
\
0
'
)
self
.
assertEqual
(
m
[
0
:
3
],
'
\
0
\
0
\
0
'
)
# Shouldn't crash on boundary (Issue #5292)
self
.
assertRaises
(
IndexError
,
m
.
__getitem__
,
len
(
m
))
self
.
assertRaises
(
IndexError
,
m
.
__setitem__
,
len
(
m
),
'
\
0
'
)
# Modify the file's content
m
[
0
]
=
'3'
m
[
PAGESIZE
+
3
:
PAGESIZE
+
3
+
3
]
=
'bar'
# Check that the modification worked
self
.
assertEqual
(
m
[
0
],
'3'
)
self
.
assertEqual
(
m
[
0
:
3
],
'3
\
0
\
0
'
)
self
.
assertEqual
(
m
[
PAGESIZE
-
1
:
PAGESIZE
+
7
],
'
\
0
foobar
\
0
'
)
m
.
flush
()
# Test doing a regular expression match in an mmap'ed file
match
=
re
.
search
(
'[A-Za-z]+'
,
m
)
if
match
is
None
:
self
.
fail
(
'regex match on mmap failed!'
)
else
:
start
,
end
=
match
.
span
(
0
)
length
=
end
-
start
self
.
assertEqual
(
start
,
PAGESIZE
)
self
.
assertEqual
(
end
,
PAGESIZE
+
6
)
# test seeking around (try to overflow the seek implementation)
m
.
seek
(
0
,
0
)
self
.
assertEqual
(
m
.
tell
(),
0
)
m
.
seek
(
42
,
1
)
self
.
assertEqual
(
m
.
tell
(),
42
)
m
.
seek
(
0
,
2
)
self
.
assertEqual
(
m
.
tell
(),
len
(
m
))
# Try to seek to negative position...
self
.
assertRaises
(
ValueError
,
m
.
seek
,
-
1
)
# Try to seek beyond end of mmap...
self
.
assertRaises
(
ValueError
,
m
.
seek
,
1
,
2
)
# Try to seek to negative position...
self
.
assertRaises
(
ValueError
,
m
.
seek
,
-
len
(
m
)
-
1
,
2
)
# Try resizing map
try
:
m
.
resize
(
512
)
except
SystemError
:
# resize() not supported
# No messages are printed, since the output of this test suite
# would then be different across platforms.
pass
else
:
# resize() is supported
self
.
assertEqual
(
len
(
m
),
512
)
# Check that we can no longer seek beyond the new size.
self
.
assertRaises
(
ValueError
,
m
.
seek
,
513
,
0
)
# Check that the underlying file is truncated too
# (bug #728515)
f
=
open
(
TESTFN
)
f
.
seek
(
0
,
2
)
self
.
assertEqual
(
f
.
tell
(),
512
)
f
.
close
()
self
.
assertEqual
(
m
.
size
(),
512
)
m
.
close
()
finally
:
try
:
f
.
close
()
except
OSError
:
pass
def
test_access_parameter
(
self
):
# Test for "access" keyword parameter
mapsize
=
10
open
(
TESTFN
,
"wb"
).
write
(
"a"
*
mapsize
)
f
=
open
(
TESTFN
,
"rb"
)
m
=
mmap
.
mmap
(
f
.
fileno
(),
mapsize
,
access
=
mmap
.
ACCESS_READ
)
self
.
assertEqual
(
m
[:],
'a'
*
mapsize
,
"Readonly memory map data incorrect."
)
# Ensuring that readonly mmap can't be slice assigned
try
:
m
[:]
=
'b'
*
mapsize
except
TypeError
:
pass
else
:
self
.
fail
(
"Able to write to readonly memory map"
)
# Ensuring that readonly mmap can't be item assigned
try
:
m
[
0
]
=
'b'
except
TypeError
:
pass
else
:
self
.
fail
(
"Able to write to readonly memory map"
)
# Ensuring that readonly mmap can't be write() to
try
:
m
.
seek
(
0
,
0
)
m
.
write
(
'abc'
)
except
TypeError
:
pass
else
:
self
.
fail
(
"Able to write to readonly memory map"
)
# Ensuring that readonly mmap can't be write_byte() to
try
:
m
.
seek
(
0
,
0
)
m
.
write_byte
(
'd'
)
except
TypeError
:
pass
else
:
self
.
fail
(
"Able to write to readonly memory map"
)
# Ensuring that readonly mmap can't be resized
try
:
m
.
resize
(
2
*
mapsize
)
except
SystemError
:
# resize is not universally supported
pass
except
TypeError
:
pass
else
:
self
.
fail
(
"Able to resize readonly memory map"
)
f
.
close
()
del
m
,
f
self
.
assertEqual
(
open
(
TESTFN
,
"rb"
).
read
(),
'a'
*
mapsize
,
"Readonly memory map data file was modified"
)
# Opening mmap with size too big
import
sys
f
=
open
(
TESTFN
,
"r+b"
)
try
:
m
=
mmap
.
mmap
(
f
.
fileno
(),
mapsize
+
1
)
except
ValueError
:
# we do not expect a ValueError on Windows
# CAUTION: This also changes the size of the file on disk, and
# later tests assume that the length hasn't changed. We need to
# repair that.
if
sys
.
platform
.
startswith
(
'win'
):
self
.
fail
(
"Opening mmap with size+1 should work on Windows."
)
else
:
# we expect a ValueError on Unix, but not on Windows
if
not
sys
.
platform
.
startswith
(
'win'
):
self
.
fail
(
"Opening mmap with size+1 should raise ValueError."
)
m
.
close
()
f
.
close
()
if
sys
.
platform
.
startswith
(
'win'
):
# Repair damage from the resizing test.
f
=
open
(
TESTFN
,
'r+b'
)
f
.
truncate
(
mapsize
)
f
.
close
()
# Opening mmap with access=ACCESS_WRITE
f
=
open
(
TESTFN
,
"r+b"
)
m
=
mmap
.
mmap
(
f
.
fileno
(),
mapsize
,
access
=
mmap
.
ACCESS_WRITE
)
# Modifying write-through memory map
m
[:]
=
'c'
*
mapsize
self
.
assertEqual
(
m
[:],
'c'
*
mapsize
,
"Write-through memory map memory not updated properly."
)
m
.
flush
()
m
.
close
()
f
.
close
()
f
=
open
(
TESTFN
,
'rb'
)
stuff
=
f
.
read
()
f
.
close
()
self
.
assertEqual
(
stuff
,
'c'
*
mapsize
,
"Write-through memory map data file not updated properly."
)
# Opening mmap with access=ACCESS_COPY
f
=
open
(
TESTFN
,
"r+b"
)
m
=
mmap
.
mmap
(
f
.
fileno
(),
mapsize
,
access
=
mmap
.
ACCESS_COPY
)
# Modifying copy-on-write memory map
m
[:]
=
'd'
*
mapsize
self
.
assertEqual
(
m
[:],
'd'
*
mapsize
,
"Copy-on-write memory map data not written correctly."
)
m
.
flush
()
self
.
assertEqual
(
open
(
TESTFN
,
"rb"
).
read
(),
'c'
*
mapsize
,
"Copy-on-write test data file should not be modified."
)
# Ensuring copy-on-write maps cannot be resized
self
.
assertRaises
(
TypeError
,
m
.
resize
,
2
*
mapsize
)
f
.
close
()
del
m
,
f
# Ensuring invalid access parameter raises exception
f
=
open
(
TESTFN
,
"r+b"
)
self
.
assertRaises
(
ValueError
,
mmap
.
mmap
,
f
.
fileno
(),
mapsize
,
access
=
4
)
f
.
close
()
if
os
.
name
==
"posix"
:
# Try incompatible flags, prot and access parameters.
f
=
open
(
TESTFN
,
"r+b"
)
self
.
assertRaises
(
ValueError
,
mmap
.
mmap
,
f
.
fileno
(),
mapsize
,
flags
=
mmap
.
MAP_PRIVATE
,
prot
=
mmap
.
PROT_READ
,
access
=
mmap
.
ACCESS_WRITE
)
f
.
close
()
# Try writing with PROT_EXEC and without PROT_WRITE
prot
=
mmap
.
PROT_READ
|
getattr
(
mmap
,
'PROT_EXEC'
,
0
)
with
open
(
TESTFN
,
"r+b"
)
as
f
:
m
=
mmap
.
mmap
(
f
.
fileno
(),
mapsize
,
prot
=
prot
)
self
.
assertRaises
(
TypeError
,
m
.
write
,
b"abcdef"
)
self
.
assertRaises
(
TypeError
,
m
.
write_byte
,
0
)
m
.
close
()
def
test_bad_file_desc
(
self
):
# Try opening a bad file descriptor...
self
.
assertRaises
(
mmap
.
error
,
mmap
.
mmap
,
-
2
,
4096
)
def
test_tougher_find
(
self
):
# Do a tougher .find() test. SF bug 515943 pointed out that, in 2.2,
# searching for data with embedded \0 bytes didn't work.
f
=
open
(
TESTFN
,
'w+'
)
data
=
'aabaac
\
x00
deef
\
x00
\
x00
aa
\
x00
'
n
=
len
(
data
)
f
.
write
(
data
)
f
.
flush
()
m
=
mmap
.
mmap
(
f
.
fileno
(),
n
)
f
.
close
()
for
start
in
range
(
n
+
1
):
for
finish
in
range
(
start
,
n
+
1
):
slice
=
data
[
start
:
finish
]
self
.
assertEqual
(
m
.
find
(
slice
),
data
.
find
(
slice
))
self
.
assertEqual
(
m
.
find
(
slice
+
'x'
),
-
1
)
m
.
close
()
def
test_find_end
(
self
):
# test the new 'end' parameter works as expected
f
=
open
(
TESTFN
,
'w+'
)
data
=
'one two ones'
n
=
len
(
data
)
f
.
write
(
data
)
f
.
flush
()
m
=
mmap
.
mmap
(
f
.
fileno
(),
n
)
f
.
close
()
self
.
assertEqual
(
m
.
find
(
'one'
),
0
)
self
.
assertEqual
(
m
.
find
(
'ones'
),
8
)
self
.
assertEqual
(
m
.
find
(
'one'
,
0
,
-
1
),
0
)
self
.
assertEqual
(
m
.
find
(
'one'
,
1
),
8
)
self
.
assertEqual
(
m
.
find
(
'one'
,
1
,
-
1
),
8
)
self
.
assertEqual
(
m
.
find
(
'one'
,
1
,
-
2
),
-
1
)
def
test_rfind
(
self
):
# test the new 'end' parameter works as expected
f
=
open
(
TESTFN
,
'w+'
)
data
=
'one two ones'
n
=
len
(
data
)
f
.
write
(
data
)
f
.
flush
()
m
=
mmap
.
mmap
(
f
.
fileno
(),
n
)
f
.
close
()
self
.
assertEqual
(
m
.
rfind
(
'one'
),
8
)
self
.
assertEqual
(
m
.
rfind
(
'one '
),
0
)
self
.
assertEqual
(
m
.
rfind
(
'one'
,
0
,
-
1
),
8
)
self
.
assertEqual
(
m
.
rfind
(
'one'
,
0
,
-
2
),
0
)
self
.
assertEqual
(
m
.
rfind
(
'one'
,
1
,
-
1
),
8
)
self
.
assertEqual
(
m
.
rfind
(
'one'
,
1
,
-
2
),
-
1
)
def
test_double_close
(
self
):
# make sure a double close doesn't crash on Solaris (Bug# 665913)
f
=
open
(
TESTFN
,
'w+'
)
f
.
write
(
2
**
16
*
'a'
)
# Arbitrary character
f
.
close
()
f
=
open
(
TESTFN
)
mf
=
mmap
.
mmap
(
f
.
fileno
(),
2
**
16
,
access
=
mmap
.
ACCESS_READ
)
mf
.
close
()
mf
.
close
()
f
.
close
()
@
unittest
.
skipUnless
(
hasattr
(
os
,
"stat"
),
"needs os.stat()"
)
def
test_entire_file
(
self
):
# test mapping of entire file by passing 0 for map length
f
=
open
(
TESTFN
,
"w+"
)
f
.
write
(
2
**
16
*
'm'
)
# Arbitrary character
f
.
close
()
f
=
open
(
TESTFN
,
"rb+"
)
mf
=
mmap
.
mmap
(
f
.
fileno
(),
0
)
self
.
assertEqual
(
len
(
mf
),
2
**
16
,
"Map size should equal file size."
)
self
.
assertEqual
(
mf
.
read
(
2
**
16
),
2
**
16
*
"m"
)
mf
.
close
()
f
.
close
()
@
unittest
.
skipUnless
(
hasattr
(
os
,
"stat"
),
"needs os.stat()"
)
def
test_length_0_offset
(
self
):
# Issue #10916: test mapping of remainder of file by passing 0 for
# map length with an offset doesn't cause a segfault.
# NOTE: allocation granularity is currently 65536 under Win64,
# and therefore the minimum offset alignment.
with
open
(
TESTFN
,
"wb"
)
as
f
:
f
.
write
((
65536
*
2
)
*
b'm'
)
# Arbitrary character
with
open
(
TESTFN
,
"rb"
)
as
f
:
mf
=
mmap
.
mmap
(
f
.
fileno
(),
0
,
offset
=
65536
,
access
=
mmap
.
ACCESS_READ
)
try
:
self
.
assertRaises
(
IndexError
,
mf
.
__getitem__
,
80000
)
finally
:
mf
.
close
()
@
unittest
.
skipUnless
(
hasattr
(
os
,
"stat"
),
"needs os.stat()"
)
def
test_length_0_large_offset
(
self
):
# Issue #10959: test mapping of a file by passing 0 for
# map length with a large offset doesn't cause a segfault.
with
open
(
TESTFN
,
"wb"
)
as
f
:
f
.
write
(
115699
*
b'm'
)
# Arbitrary character
with
open
(
TESTFN
,
"w+b"
)
as
f
:
self
.
assertRaises
(
ValueError
,
mmap
.
mmap
,
f
.
fileno
(),
0
,
offset
=
2147418112
)
def
test_move
(
self
):
# make move works everywhere (64-bit format problem earlier)
f
=
open
(
TESTFN
,
'w+'
)
f
.
write
(
"ABCDEabcde"
)
# Arbitrary character
f
.
flush
()
mf
=
mmap
.
mmap
(
f
.
fileno
(),
10
)
mf
.
move
(
5
,
0
,
5
)
self
.
assertEqual
(
mf
[:],
"ABCDEABCDE"
,
"Map move should have duplicated front 5"
)
mf
.
close
()
f
.
close
()
# more excessive test
data
=
"0123456789"
for
dest
in
range
(
len
(
data
)):
for
src
in
range
(
len
(
data
)):
for
count
in
range
(
len
(
data
)
-
max
(
dest
,
src
)):
expected
=
data
[:
dest
]
+
data
[
src
:
src
+
count
]
+
data
[
dest
+
count
:]
m
=
mmap
.
mmap
(
-
1
,
len
(
data
))
m
[:]
=
data
m
.
move
(
dest
,
src
,
count
)
self
.
assertEqual
(
m
[:],
expected
)
m
.
close
()
# segfault test (Issue 5387)
m
=
mmap
.
mmap
(
-
1
,
100
)
offsets
=
[
-
100
,
-
1
,
0
,
1
,
100
]
for
source
,
dest
,
size
in
itertools
.
product
(
offsets
,
offsets
,
offsets
):
try
:
m
.
move
(
source
,
dest
,
size
)
except
ValueError
:
pass
offsets
=
[(
-
1
,
-
1
,
-
1
),
(
-
1
,
-
1
,
0
),
(
-
1
,
0
,
-
1
),
(
0
,
-
1
,
-
1
),
(
-
1
,
0
,
0
),
(
0
,
-
1
,
0
),
(
0
,
0
,
-
1
)]
for
source
,
dest
,
size
in
offsets
:
self
.
assertRaises
(
ValueError
,
m
.
move
,
source
,
dest
,
size
)
m
.
close
()
m
=
mmap
.
mmap
(
-
1
,
1
)
# single byte
self
.
assertRaises
(
ValueError
,
m
.
move
,
0
,
0
,
2
)
self
.
assertRaises
(
ValueError
,
m
.
move
,
1
,
0
,
1
)
self
.
assertRaises
(
ValueError
,
m
.
move
,
0
,
1
,
1
)
m
.
move
(
0
,
0
,
1
)
m
.
move
(
0
,
0
,
0
)
def
test_anonymous
(
self
):
# anonymous mmap.mmap(-1, PAGE)
m
=
mmap
.
mmap
(
-
1
,
PAGESIZE
)
for
x
in
xrange
(
PAGESIZE
):
self
.
assertEqual
(
m
[
x
],
'
\
0
'
,
"anonymously mmap'ed contents should be zero"
)
for
x
in
xrange
(
PAGESIZE
):
m
[
x
]
=
ch
=
chr
(
x
&
255
)
self
.
assertEqual
(
m
[
x
],
ch
)
def
test_extended_getslice
(
self
):
# Test extended slicing by comparing with list slicing.
s
=
""
.
join
(
chr
(
c
)
for
c
in
reversed
(
range
(
256
)))
m
=
mmap
.
mmap
(
-
1
,
len
(
s
))
m
[:]
=
s
self
.
assertEqual
(
m
[:],
s
)
indices
=
(
0
,
None
,
1
,
3
,
19
,
300
,
-
1
,
-
2
,
-
31
,
-
300
)
for
start
in
indices
:
for
stop
in
indices
:
# Skip step 0 (invalid)
for
step
in
indices
[
1
:]:
self
.
assertEqual
(
m
[
start
:
stop
:
step
],
s
[
start
:
stop
:
step
])
def
test_extended_set_del_slice
(
self
):
# Test extended slicing by comparing with list slicing.
s
=
""
.
join
(
chr
(
c
)
for
c
in
reversed
(
range
(
256
)))
m
=
mmap
.
mmap
(
-
1
,
len
(
s
))
indices
=
(
0
,
None
,
1
,
3
,
19
,
300
,
-
1
,
-
2
,
-
31
,
-
300
)
for
start
in
indices
:
for
stop
in
indices
:
# Skip invalid step 0
for
step
in
indices
[
1
:]:
m
[:]
=
s
self
.
assertEqual
(
m
[:],
s
)
L
=
list
(
s
)
# Make sure we have a slice of exactly the right length,
# but with different data.
data
=
L
[
start
:
stop
:
step
]
data
=
""
.
join
(
reversed
(
data
))
L
[
start
:
stop
:
step
]
=
data
m
[
start
:
stop
:
step
]
=
data
self
.
assertEqual
(
m
[:],
""
.
join
(
L
))
def
make_mmap_file
(
self
,
f
,
halfsize
):
# Write 2 pages worth of data to the file
f
.
write
(
'
\
0
'
*
halfsize
)
f
.
write
(
'foo'
)
f
.
write
(
'
\
0
'
*
(
halfsize
-
3
))
f
.
flush
()
return
mmap
.
mmap
(
f
.
fileno
(),
0
)
def
test_empty_file
(
self
):
f
=
open
(
TESTFN
,
'w+b'
)
f
.
close
()
with
open
(
TESTFN
,
"rb"
)
as
f
:
self
.
assertRaisesRegexp
(
ValueError
,
"cannot mmap an empty file"
,
mmap
.
mmap
,
f
.
fileno
(),
0
,
access
=
mmap
.
ACCESS_READ
)
def
test_offset
(
self
):
f
=
open
(
TESTFN
,
'w+b'
)
try
:
# unlink TESTFN no matter what
halfsize
=
mmap
.
ALLOCATIONGRANULARITY
m
=
self
.
make_mmap_file
(
f
,
halfsize
)
m
.
close
()
f
.
close
()
mapsize
=
halfsize
*
2
# Try invalid offset
f
=
open
(
TESTFN
,
"r+b"
)
for
offset
in
[
-
2
,
-
1
,
None
]:
try
:
m
=
mmap
.
mmap
(
f
.
fileno
(),
mapsize
,
offset
=
offset
)
self
.
assertEqual
(
0
,
1
)
except
(
ValueError
,
TypeError
,
OverflowError
):
pass
else
:
self
.
assertEqual
(
0
,
0
)
f
.
close
()
# Try valid offset, hopefully 8192 works on all OSes
f
=
open
(
TESTFN
,
"r+b"
)
m
=
mmap
.
mmap
(
f
.
fileno
(),
mapsize
-
halfsize
,
offset
=
halfsize
)
self
.
assertEqual
(
m
[
0
:
3
],
'foo'
)
f
.
close
()
# Try resizing map
try
:
m
.
resize
(
512
)
except
SystemError
:
pass
else
:
# resize() is supported
self
.
assertEqual
(
len
(
m
),
512
)
# Check that we can no longer seek beyond the new size.
self
.
assertRaises
(
ValueError
,
m
.
seek
,
513
,
0
)
# Check that the content is not changed
self
.
assertEqual
(
m
[
0
:
3
],
'foo'
)
# Check that the underlying file is truncated too
f
=
open
(
TESTFN
)
f
.
seek
(
0
,
2
)
self
.
assertEqual
(
f
.
tell
(),
halfsize
+
512
)
f
.
close
()
self
.
assertEqual
(
m
.
size
(),
halfsize
+
512
)
m
.
close
()
finally
:
f
.
close
()
try
:
os
.
unlink
(
TESTFN
)
except
OSError
:
pass
def
test_subclass
(
self
):
class
anon_mmap
(
mmap
.
mmap
):
def
__new__
(
klass
,
*
args
,
**
kwargs
):
return
mmap
.
mmap
.
__new__
(
klass
,
-
1
,
*
args
,
**
kwargs
)
anon_mmap
(
PAGESIZE
)
@
unittest
.
skipUnless
(
hasattr
(
mmap
,
'PROT_READ'
),
"needs mmap.PROT_READ"
)
def
test_prot_readonly
(
self
):
mapsize
=
10
open
(
TESTFN
,
"wb"
).
write
(
"a"
*
mapsize
)
f
=
open
(
TESTFN
,
"rb"
)
m
=
mmap
.
mmap
(
f
.
fileno
(),
mapsize
,
prot
=
mmap
.
PROT_READ
)
self
.
assertRaises
(
TypeError
,
m
.
write
,
"foo"
)
f
.
close
()
def
test_error
(
self
):
self
.
assertTrue
(
issubclass
(
mmap
.
error
,
EnvironmentError
))
self
.
assertIn
(
"mmap.error"
,
str
(
mmap
.
error
))
def
test_io_methods
(
self
):
data
=
"0123456789"
open
(
TESTFN
,
"wb"
).
write
(
"x"
*
len
(
data
))
f
=
open
(
TESTFN
,
"r+b"
)
m
=
mmap
.
mmap
(
f
.
fileno
(),
len
(
data
))
f
.
close
()
# Test write_byte()
for
i
in
xrange
(
len
(
data
)):
self
.
assertEqual
(
m
.
tell
(),
i
)
m
.
write_byte
(
data
[
i
])
self
.
assertEqual
(
m
.
tell
(),
i
+
1
)
self
.
assertRaises
(
ValueError
,
m
.
write_byte
,
"x"
)
self
.
assertEqual
(
m
[:],
data
)
# Test read_byte()
m
.
seek
(
0
)
for
i
in
xrange
(
len
(
data
)):
self
.
assertEqual
(
m
.
tell
(),
i
)
self
.
assertEqual
(
m
.
read_byte
(),
data
[
i
])
self
.
assertEqual
(
m
.
tell
(),
i
+
1
)
self
.
assertRaises
(
ValueError
,
m
.
read_byte
)
# Test read()
m
.
seek
(
3
)
self
.
assertEqual
(
m
.
read
(
3
),
"345"
)
self
.
assertEqual
(
m
.
tell
(),
6
)
# Test write()
m
.
seek
(
3
)
m
.
write
(
"bar"
)
self
.
assertEqual
(
m
.
tell
(),
6
)
self
.
assertEqual
(
m
[:],
"012bar6789"
)
m
.
seek
(
8
)
self
.
assertRaises
(
ValueError
,
m
.
write
,
"bar"
)
@
unittest
.
skipUnless
(
os
.
name
==
'nt'
,
'requires Windows'
)
def
test_tagname
(
self
):
data1
=
"0123456789"
data2
=
"abcdefghij"
assert
len
(
data1
)
==
len
(
data2
)
# Test same tag
m1
=
mmap
.
mmap
(
-
1
,
len
(
data1
),
tagname
=
"foo"
)
m1
[:]
=
data1
m2
=
mmap
.
mmap
(
-
1
,
len
(
data2
),
tagname
=
"foo"
)
m2
[:]
=
data2
self
.
assertEqual
(
m1
[:],
data2
)
self
.
assertEqual
(
m2
[:],
data2
)
m2
.
close
()
m1
.
close
()
# Test different tag
m1
=
mmap
.
mmap
(
-
1
,
len
(
data1
),
tagname
=
"foo"
)
m1
[:]
=
data1
m2
=
mmap
.
mmap
(
-
1
,
len
(
data2
),
tagname
=
"boo"
)
m2
[:]
=
data2
self
.
assertEqual
(
m1
[:],
data1
)
self
.
assertEqual
(
m2
[:],
data2
)
m2
.
close
()
m1
.
close
()
@
unittest
.
skipUnless
(
os
.
name
==
'nt'
,
'requires Windows'
)
def
test_crasher_on_windows
(
self
):
# Should not crash (Issue 1733986)
m
=
mmap
.
mmap
(
-
1
,
1000
,
tagname
=
"foo"
)
try
:
mmap
.
mmap
(
-
1
,
5000
,
tagname
=
"foo"
)[:]
# same tagname, but larger size
except
:
pass
m
.
close
()
# Should not crash (Issue 5385)
open
(
TESTFN
,
"wb"
).
write
(
"x"
*
10
)
f
=
open
(
TESTFN
,
"r+b"
)
m
=
mmap
.
mmap
(
f
.
fileno
(),
0
)
f
.
close
()
try
:
m
.
resize
(
0
)
# will raise WindowsError
except
:
pass
try
:
m
[:]
except
:
pass
m
.
close
()
@
unittest
.
skipUnless
(
os
.
name
==
'nt'
,
'requires Windows'
)
def
test_invalid_descriptor
(
self
):
# socket file descriptors are valid, but out of range
# for _get_osfhandle, causing a crash when validating the
# parameters to _get_osfhandle.
s
=
socket
.
socket
()
try
:
with
self
.
assertRaises
(
mmap
.
error
):
m
=
mmap
.
mmap
(
s
.
fileno
(),
10
)
finally
:
s
.
close
()
class
LargeMmapTests
(
unittest
.
TestCase
):
def
setUp
(
self
):
unlink
(
TESTFN
)
def
tearDown
(
self
):
unlink
(
TESTFN
)
def
_make_test_file
(
self
,
num_zeroes
,
tail
):
if
sys
.
platform
[:
3
]
==
'win'
or
sys
.
platform
==
'darwin'
:
requires
(
'largefile'
,
'test requires %s bytes and a long time to run'
%
str
(
0x180000000
))
f
=
open
(
TESTFN
,
'w+b'
)
try
:
f
.
seek
(
num_zeroes
)
f
.
write
(
tail
)
f
.
flush
()
except
(
IOError
,
OverflowError
):
f
.
close
()
raise
unittest
.
SkipTest
(
"filesystem does not have largefile support"
)
return
f
def
test_large_offset
(
self
):
with
self
.
_make_test_file
(
0x14FFFFFFF
,
b" "
)
as
f
:
m
=
mmap
.
mmap
(
f
.
fileno
(),
0
,
offset
=
0x140000000
,
access
=
mmap
.
ACCESS_READ
)
try
:
self
.
assertEqual
(
m
[
0xFFFFFFF
],
b" "
)
finally
:
m
.
close
()
def
test_large_filesize
(
self
):
with
self
.
_make_test_file
(
0x17FFFFFFF
,
b" "
)
as
f
:
if
sys
.
maxsize
<
0x180000000
:
# On 32 bit platforms the file is larger than sys.maxsize so
# mapping the whole file should fail -- Issue #16743
with
self
.
assertRaises
(
OverflowError
):
mmap
.
mmap
(
f
.
fileno
(),
0x180000000
,
access
=
mmap
.
ACCESS_READ
)
with
self
.
assertRaises
(
ValueError
):
mmap
.
mmap
(
f
.
fileno
(),
0
,
access
=
mmap
.
ACCESS_READ
)
m
=
mmap
.
mmap
(
f
.
fileno
(),
0x10000
,
access
=
mmap
.
ACCESS_READ
)
try
:
self
.
assertEqual
(
m
.
size
(),
0x180000000
)
finally
:
m
.
close
()
# Issue 11277: mmap() with large (~4GB) sparse files crashes on OS X.
def
_test_around_boundary
(
self
,
boundary
):
tail
=
b' DEARdear '
start
=
boundary
-
len
(
tail
)
//
2
end
=
start
+
len
(
tail
)
with
self
.
_make_test_file
(
start
,
tail
)
as
f
:
m
=
mmap
.
mmap
(
f
.
fileno
(),
0
,
access
=
mmap
.
ACCESS_READ
)
try
:
self
.
assertEqual
(
m
[
start
:
end
],
tail
)
finally
:
m
.
close
()
@
unittest
.
skipUnless
(
sys
.
maxsize
>
_4G
,
"test cannot run on 32-bit systems"
)
def
test_around_2GB
(
self
):
self
.
_test_around_boundary
(
_2G
)
@
unittest
.
skipUnless
(
sys
.
maxsize
>
_4G
,
"test cannot run on 32-bit systems"
)
def
test_around_4GB
(
self
):
self
.
_test_around_boundary
(
_4G
)
def
test_main
():
run_unittest
(
MmapTests
,
LargeMmapTests
)
if
__name__
==
'__main__'
:
test_main
()
from_cpython/setup.py
View file @
5576cae7
...
...
@@ -88,6 +88,12 @@ def termios_ext():
"Modules/termios.c"
,
]))
@
unique
def
mmap_ext
():
return
Extension
(
"mmap"
,
sources
=
map
(
relpath
,
[
"Modules/mmapmodule.c"
,
]))
@
unique
def
pyexpat_ext
():
define_macros
=
[(
'HAVE_EXPAT_CONFIG_H'
,
'1'
),]
...
...
@@ -136,7 +142,9 @@ ext_modules = [future_builtins_ext(),
ctypes_test_ext
(),
grp_ext
(),
curses_ext
(),
termios_ext
()]
termios_ext
(),
mmap_ext
(),
]
builtin_headers
=
map
(
relpath
,
glob
.
glob
(
"Include/*.h"
))
...
...
test/cpython/test_mmap.py
0 → 120000
View file @
5576cae7
..
/
..
/
from_cpython
/
Lib
/
test
/
test_mmap
.
py
\ No newline at end of file
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment