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
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
|
# Copyright (C) 2009, 2010 Roman Zimbelmann <romanz@lavabit.com>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
if __name__ == '__main__': from __init__ import init; init()
import unittest
import os
from os.path import realpath, join, dirname
from time import time
from test import Fake
from ranger.shared import FileManagerAware, SettingsAware
from ranger.core.loader import Loader
from ranger.fsobject import Directory, File
from ranger.ext.openstruct import OpenStruct
TESTDIR = realpath(join(dirname(__file__), 'testdir'))
#TESTDIR = "/usr/sbin"
def raw_load_content(self):
"""
The method which is used in a Directory object to load stuff.
Keep this up to date!
"""
from os.path import join, isdir, basename
from os import listdir
import ranger.ext.mount_path
self.loading = True
self.load_if_outdated()
try:
if self.exists and self.runnable:
self.mount_path = ranger.ext.mount_path.mount_path(self.path)
filenames = []
for fname in listdir(self.path):
if not self.settings.show_hidden:
hfilter = self.settings.hidden_filter
if hfilter:
if isinstance(hfilter, str) and hfilter in fname:
continue
if hasattr(hfilter, 'search') and \
hfilter.search(fname):
continue
if isinstance(self.filter, str) and self.filter \
and self.filter not in fname:
continue
filenames.append(join(self.path, fname))
self.load_content_mtime = os.stat(self.path).st_mtime
marked_paths = [obj.path for obj in self.marked_items]
files = []
for name in filenames:
if isdir(name):
try:
item = self.fm.env.get_directory(name)
except:
item = Directory(name)
else:
item = File(name)
item.load_if_outdated()
files.append(item)
self.disk_usage = sum(f.size for f in files if f.is_file)
self.scroll_offset = 0
self.filenames = filenames
self.files = files
self._clear_marked_items()
for item in self.files:
if item.path in marked_paths:
self.mark_item(item, True)
else:
self.mark_item(item, False)
self.sort()
if len(self.files) > 0:
if self.pointed_obj is not None:
self.sync_index()
else:
self.move(to=0)
else:
self.filenames = None
self.files = None
self.cycle_list = None
self.content_loaded = True
self.determine_infostring()
self.last_update_time = time()
self.correct_pointer()
finally:
self.loading = False
def benchmark_load(n):
loader = Loader()
fm = OpenStruct(loader=loader)
SettingsAware.settings = Fake()
FileManagerAware.fm = fm
dir = Directory(TESTDIR)
t1 = time()
for _ in range(n):
dir.load_content(schedule=True)
while loader.has_work():
loader.work()
t2 = time()
return t2 - t1
def benchmark_raw_load(n):
SettingsAware.settings = Fake()
dir = Directory(TESTDIR)
generator = dir.load_bit_by_bit()
t1 = time()
for _ in range(n):
raw_load_content(dir)
t2 = time()
return t2 - t1
class Test1(unittest.TestCase):
def test_loader(self):
loader = Loader()
fm = OpenStruct(loader=loader)
SettingsAware.settings = Fake()
FileManagerAware.fm = fm
# initially, the loader has nothing to do
self.assertFalse(loader.has_work())
dir = Directory(TESTDIR)
self.assertEqual(None, dir.files)
self.assertFalse(loader.has_work())
# Calling load_content() will enqueue the loading operation.
# dir is not loaded yet, but the loader has work
dir.load_content(schedule=True)
self.assertEqual(None, dir.files)
self.assertTrue(loader.has_work())
iterations = 0
while loader.has_work():
iterations += 1
loader.work()
#print(iterations)
self.assertNotEqual(None, dir.files)
self.assertFalse(loader.has_work())
def test_get_overhead_of_loader(self):
N = 5
tloader = benchmark_load(N)
traw = benchmark_raw_load(N)
#traw1k = 250.0
#traw = traw1k * N / 1000.0
#print("Loader: {0}s".format(tloader))
#print("Raw: {0}s".format(traw))
self.assertTrue(tloader > traw)
overhead = tloader * 100 / traw - 100
self.assertTrue(overhead < 2, "overhead of loader too high: {0}" \
.format(overhead))
#print("Overhead: {0:.5}%".format(overhead))
if __name__ == '__main__':
unittest.main()
|