FazBrowse GitHub Viewer
|
Trending
|
URL:
|
Home
Tools:
[Download Repo ZIP]
[View Raw Code]
[Original HTTPS Page]
sqlmap/tests/test_bigarray.py at master · sqlmapproject/sqlmap · GitHub
Uh oh!
There was an error while loading.
Please reload this page
.
sqlmapproject
/
sqlmap
Public
Uh oh!
There was an error while loading.
Please reload this page
.
Notifications
You must be signed in to change notification settings
Fork
6.3k
Star
38.3k
Code
Issues
25
Pull requests
7
Actions
Projects
Wiki
Security and quality
0
Insights
Additional navigation options
Code
Issues
Pull requests
Actions
Projects
Wiki
Security and quality
Insights
Expand file tree
Breadcrumbs
sqlmap
/
tests
/
test_bigarray.py
Copy path
More file actions
More file actions
Latest commit
History
History
History
166 lines (132 loc) · 6.04 KB
Breadcrumbs
sqlmap
/
tests
/
test_bigarray.py
Copy path
File metadata and controls
166 lines (132 loc) · 6.04 KB
Raw
Copy raw file
Download raw file
Open symbols panel
Edit and raw actions
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
#!/usr/bin/env python
"""
Copyright (c) 2006-2026 sqlmap developers (https://sqlmap.org)
See the file 'LICENSE' for copying permission
BigArray disk-spill semantics (lib/core/bigarray.py).
BigArray is the structure that lets sqlmap dump tables far larger than RAM: once
the in-memory chunk exceeds chunk_size it is pickled to a temp file and a new
chunk starts. The tricky, easy-to-break part is that indexing / iteration /
pop / pickling must stay correct ACROSS the in-memory<->on-disk boundary.
These force a spill with a tiny chunk_size and assert the data survives intact.
"""
import
os
import
pickle
import
sys
import
unittest
sys
.
path
.
insert
(
0
,
os
.
path
.
dirname
(
os
.
path
.
abspath
(
__file__
)))
from
_testutils
import
bootstrap
bootstrap
()
from
lib
.
core
.
bigarray
import
BigArray
N
=
5000
_SPILLED
=
[]
def
_make_spilled
():
# tiny chunk_size guarantees many on-disk chunks for N items
ba
=
BigArray
(
chunk_size
=
1024
)
for
i
in
range
(
N
):
ba
.
append
(
"item-%d"
%
i
)
_SPILLED
.
append
(
ba
)
# tracked so tearDownModule closes it (release the on-disk chunk files)
return
ba
def
tearDownModule
():
for
ba
in
_SPILLED
:
try
:
ba
.
close
()
except
Exception
:
pass
del
_SPILLED
[:]
class
TestSpill
(
unittest
.
TestCase
):
def
test_actually_spilled_to_disk
(
self
):
ba
=
_make_spilled
()
self
.
assertGreater
(
len
(
ba
.
chunks
),
1
,
msg
=
"expected multiple chunks (a disk spill)"
)
# stronger than "more than one chunk": at least one chunk must be a real on-disk file
# (spilled chunks are stored as filenames). Otherwise this could pass while everything
# stayed in RAM.
disk_chunks
=
[
c
for
c
in
ba
.
chunks
if
isinstance
(
c
,
str
)]
self
.
assertTrue
(
disk_chunks
,
msg
=
"no chunk was spilled to disk"
)
self
.
assertTrue
(
os
.
path
.
exists
(
disk_chunks
[
0
]),
msg
=
"spilled chunk file missing on disk"
)
def
test_len
(
self
):
self
.
assertEqual
(
len
(
_make_spilled
()),
N
)
def
test_random_access_across_boundary
(
self
):
ba
=
_make_spilled
()
for
i
in
(
0
,
1
,
499
,
500
,
2500
,
N
-
1
):
self
.
assertEqual
(
ba
[
i
],
"item-%d"
%
i
,
msg
=
"ba[%d]"
%
i
)
def
test_negative_index
(
self
):
ba
=
_make_spilled
()
self
.
assertEqual
(
ba
[
-
1
],
"item-%d"
%
(
N
-
1
))
def
test_iteration_order_preserved
(
self
):
ba
=
_make_spilled
()
for
idx
,
value
in
enumerate
(
ba
):
if
value
!=
"item-%d"
%
idx
:
self
.
fail
(
"iteration order broke at %d: %r"
%
(
idx
,
value
))
self
.
assertEqual
(
idx
,
N
-
1
)
def
test_pop_from_end
(
self
):
ba
=
_make_spilled
()
self
.
assertEqual
(
ba
.
pop
(),
"item-%d"
%
(
N
-
1
))
self
.
assertEqual
(
len
(
ba
),
N
-
1
)
def
test_pickle_roundtrip_across_spill
(
self
):
ba
=
_make_spilled
()
restored
=
pickle
.
loads
(
pickle
.
dumps
(
ba
))
self
.
assertIsInstance
(
restored
,
BigArray
)
self
.
assertEqual
(
len
(
restored
),
N
)
self
.
assertEqual
(
restored
[
0
],
"item-0"
)
self
.
assertEqual
(
restored
[
N
-
1
],
"item-%d"
%
(
N
-
1
))
class
TestCacheConsistency
(
unittest
.
TestCase
):
"""The on-disk chunk is served through a single-slot cache (read caching plus
dirty write-back). These check that the cache never serves stale data."""
def
test_setitem_writeback_across_chunks
(
self
):
ba
=
_make_spilled
()
ref
=
[
"item-%d"
%
i
for
i
in
range
(
N
)]
# mutate elements spread across several different on-disk chunks
for
i
in
(
0
,
1
,
499
,
500
,
2500
,
N
-
1
):
ba
[
i
]
=
ref
[
i
]
=
"EDIT-%d"
%
i
try
:
for
i
in
(
0
,
1
,
499
,
500
,
2500
,
N
-
1
):
self
.
assertEqual
(
ba
[
i
],
ref
[
i
],
msg
=
"readback ba[%d]"
%
i
)
self
.
assertEqual
(
list
(
ba
),
ref
)
# full independent traversal agrees
finally
:
ba
.
close
()
def
test_dirty_edit_survives_pickle
(
self
):
ba
=
_make_spilled
()
ba
[
10
]
=
"EDITED-LOW"
ba
[
N
-
10
]
=
"EDITED-HIGH"
restored
=
pickle
.
loads
(
pickle
.
dumps
(
ba
))
try
:
self
.
assertEqual
(
restored
[
10
],
"EDITED-LOW"
)
self
.
assertEqual
(
restored
[
N
-
10
],
"EDITED-HIGH"
)
finally
:
restored
.
close
()
ba
.
close
()
def
test_pop_then_append_then_direct_read
(
self
):
# Regression: pop() reloads the last on-disk chunk into memory and deletes its
# file, but a non-dirty cache entry still pointing at that chunk index was left
# in place. A later append that re-dumps the chunk index then made the stale
# cache serve outdated data on a direct __getitem__ (silent data corruption).
ref
=
[
"item-%d"
%
i
for
i
in
range
(
N
)]
ba
=
_make_spilled
()
try
:
cl
=
ba
.
chunk_length
last
=
len
(
ba
.
chunks
)
-
2
# last on-disk chunk (tail is the in-memory list)
base
=
last
*
cl
ba
[
base
]
# populate cache at idx=last, NOT dirty
while
len
(
ba
)
>
base
+
1
:
# pop() reloads chunk 'last' from disk, removes its file
ba
.
pop
()
ref
.
pop
()
for
i
in
range
(
cl
):
# re-dump chunk 'last' to a brand new temp file
value
=
"NEW-%d"
%
i
ba
.
append
(
value
)
ref
.
append
(
value
)
# direct access to the re-dumped chunk, with no prior read to refresh the cache
for
off
in
range
(
cl
):
self
.
assertEqual
(
ba
[
base
+
off
],
ref
[
base
+
off
],
msg
=
"offset %d"
%
off
)
finally
:
ba
.
close
()
class
TestInMemorySmall
(
unittest
.
TestCase
):
def
test_no_spill_for_small
(
self
):
ba
=
BigArray
([
1
,
2
,
3
])
self
.
assertEqual
(
len
(
ba
),
3
)
self
.
assertEqual
(
list
(
ba
), [
1
,
2
,
3
])
# the actual point of this test (the name promised it): a tiny array stays in ONE
# in-memory chunk and never touches disk
self
.
assertEqual
(
len
(
ba
.
chunks
),
1
,
msg
=
"small array unexpectedly spilled: %r"
%
(
ba
.
chunks
,))
self
.
assertFalse
(
any
(
isinstance
(
c
,
str
)
for
c
in
ba
.
chunks
),
msg
=
"small array wrote a disk chunk"
)
if
__name__
==
"__main__"
:
unittest
.
main
(
verbosity
=
2
)
Back
|
FazBrowse Home
|
New Git URL