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
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
|
#
# Copyright (C) 2006 Chris Halls <halls@debian.org>
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of version 2.1 of the GNU Lesser General Public
# License as published by the Free Software Foundation.
#
# This library 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
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this library; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
"""Unit test for cache.py"""
import os, time, shutil, copy
from twisted.trial import unittest
from twisted.internet import reactor, defer
from StringIO import StringIO
from apt_proxy.apt_proxy_conf import apConfig
from apt_proxy.test.test_apt_proxy import apTestHelper
from apt_proxy.cache import CacheEntry, findFileType, filetypes
from apt_proxy.apt_proxy import Factory
from apt_proxy.misc import log
from apt_proxy.fetchers import DownloadQueue
class DummyRequest:
def __init__(self):
self.finished = False
self.streamed = 0
self.if_modified_since = None
def finishCode(self, code, reason):
self.finished = True
def start_streaming(self, file_size, file_mtime):
self.streamed = self.streamed + 1
def getFileno(self):
return 0
class CacheEntryTest(apTestHelper):
def setUp(self):
"""
Make a configuration with a single backend
[files]
backends=file:///<path to test packages directory>
"""
DownloadQueue.closeTimeout = 0 # Close fetcher immediately
apTestHelper.setUp(self)
packagedir = self.cache_dir+"/packages"
filedir = os.path.normpath(os.getcwd()+"/../test_data/packages")
config = (self.config +
"[files]\n" +
"backends=file://" + filedir)
#print "config: " + config
self.c = apConfig(StringIO(config))
self.factory = Factory(self.c)
self.factory.createBackends()
self.backend = self.factory.getBackend("files")
self.entry = self.backend.get_cache_entry("testdir/testfile.deb")
self.request = DummyRequest()
def tearDown(self):
del(self.factory)
apTestHelper.tearDown(self)
def testInit(self):
entry = self.entry
self.assertEquals(entry.backend, self.backend, "CacheEntry did not initialise backend")
self.assertEquals(entry.factory, self.factory, "CacheEntry did not initialise factory")
self.assertEquals(entry.path, "testdir/testfile.deb")
self.assertEquals(entry.file_path, self.cache_dir+"/files/testdir/testfile.deb")
self.assertEquals(entry.filedir, self.cache_dir+"/files/testdir")
self.assertEquals(entry.filetype.contype, "application/dpkg")
self.assertEquals(entry.filename, "testfile.deb")
self.assertEquals(entry.filebase, "testfile")
self.assertEquals(entry.fileext, ".deb")
self.assertEquals(len(entry.requests), 0)
def testAddClient(self):
self.entry.add_request(self.request)
self.assertEquals(len(self.entry.requests), 1)
def testAddDuplicate(self):
self.entry.add_request(self.request)
self.assertRaises(RuntimeError, self.entry.add_request, self.request)
def testRemove(self):
self.entry.add_request(self.request)
self.entry.remove_request(self.request)
self.assertEquals(len(self.entry.requests), 0)
def testStartDownload(self):
def start_download(entry):
# This test function replaces the normal
# Backend.start_download so we can see that
# it was called without starting the download
entry.entry_download_triggered = True
self.backend.start_download = start_download
self.entry.add_request(self.request)
# Check that our special function was called
self.failUnless(self.entry.entry_download_triggered)
def testCachedFile(self):
"""
CacheEntry starts streaming a text file
"""
def start_download(entry):
# This test function replaces the normal
# Backend.start_download so we can see that
# it was called without starting the download
entry.test_download = True
self.backend.start_download = start_download
entry = CacheEntry(self.backend, "testdir/test.txt")
entry.test_download = False
entry.create_directory()
f = open(entry.file_path, 'w')
f.write('12345')
f.close()
entry.add_request(self.request)
while not entry.test_download and not self.request.streamed:
#print "iterate.."
reactor.iterate(0.1)
# Check that our special function was not called
self.failIf(entry.test_download)
self.failUnless(self.request.streamed)
def testVerifyFail(self):
"""
Create a bogus .deb and check that CacheEntry starts
a download
"""
self.testResult = defer.Deferred()
class VerifySizeError:
pass
class VerifyMtimeError:
pass
class StreamedError:
pass
class UnknownError:
pass
def start_download(entry):
# This test function replaces the normal
# Backend.start_download so we can see that
# it was called without starting the download
if entry.file_mtime is not None:
self.testResult.errback(failure.Failure(VerifyMtimeError()))
if entry.file_size is not None:
self.testResult.errback(failure.Failure(VerifySizeError()))
if self.request.streamed:
self.testResult.errback(failure.Failure(StreamedError()))
self.VerifyFailCleanup.callback(None)
self.backend.start_download = start_download
entry = CacheEntry(self.backend, "testdir/test.deb")
entry.test_download = False
entry.create_directory()
f = open(entry.file_path, 'w')
f.write('this is not a real .deb')
f.close()
entry.add_request(self.request)
self.VerifyFailCleanup = defer.Deferred()
self.VerifyFailCleanup.addCallback(self.VerifyFail2)
self.VerifyFailCleanup.addErrback(lambda x: self.testResult.errback(self.UnknownError))
return self.testResult
testVerifyFail.timeout = 2
def VerifyFail2(self, x):
#reactor.iterate(0.1) # Allow process to be reaped
self.testResult.callback(None)
def testCheckAgeImmutable(self):
# testfile.deb is immutable
self.entry.file_mtime = 0
self.failUnless(self.entry.check_age())
self.entry.file_mtime = time.time()+1000
self.failUnless(self.entry.check_age())
def immutableEntry(self):
# pretend that testfile.deb is immutable, i.e.
# it will be updated like Packages, Release
self.entry.filetype = copy.deepcopy(self.entry.filetype) # Take a copy of the filetype object
self.entry.filetype.mutable = True
def testCheckAgeMutable(self):
self.immutableEntry()
self.entry.file_mtime = 0
self.failIf(self.entry.check_age())
self.entry.file_mtime = time.time()+1000
self.failUnless(self.entry.check_age())
def testMinRefresh(self):
self.immutableEntry()
min_refresh_delay = 60
self.backend.config.min_refresh_delay = min_refresh_delay
self.entry.file_mtime = 0
self.failIf(self.entry.check_age())
self.entry.file_mtime = time.time() - min_refresh_delay - 60
self.failIf(self.entry.check_age())
self.entry.file_mtime = time.time()
self.failUnless(self.entry.check_age())
def testCreateDirectory(self):
dirname = self.cache_dir+"/files/testdir"
self.assertRaises(OSError, os.stat, dirname) # Will return exception if directory does not exist
self.entry.create_directory()
os.stat(dirname) # Will return exception if directory does not exist
def testStatFile(self):
filename = self.cache_dir+"/files/testdir/testfile.deb"
self.entry.create_directory()
f = open(filename, 'w')
f.write('12345')
f.close()
close_time = time.time()
self.entry.stat_file()
self.assertApproximates(self.entry.file_mtime, close_time, 3)
self.assertEquals(self.entry.file_size, 5)
class FileTypeTest(unittest.TestCase):
def testUnknownFiletype(self):
self.assertEquals(findFileType('unknownfile.xxx'), None)
def testFileTypes(self):
# Test filname recognition
# First entry - filename to test
# Second entry - mime type
# Third entry - mutable (can this file be changed in the archive?)
tests = [ ('test.deb', 'application/dpkg', False),
('test2.udeb', 'application/dpkg', False),
('Release.dsc', 'text/plain', False),
('file.diff.gz', 'x-gzip', False),
('Packages.gz', 'text/plain', True),
('Packages.bz2', 'text/plain', True),
('Sources.bz2', 'text/plain', True),
('dists/sid/main/binary-i386/Packages.diff/Index', 'text/plain', True),
('dists/sid/main/binary-i386/Packages.diff/2006-06-05-1427.58.gz', 'text/plain', False),
('dists/sid/main/source/Sources.diff/Index', 'text/plain', True),
('dists/sid/main/source/Sources.diff/2006-06-05-1427.58.gz', 'text/plain', False),
('dists/sid/Contents-i386', 'text/plain', True),
('dists/sid/Contents-i386.gz', 'text/plain', True),
('dists/sid/Contents-i386.diff/Index', 'text/plain', True),
('dists/sid/Contents-i386.diff/2006-06-02-1349.52.gz', 'text/plain', False),
('dists/sid/main/i18n/Translation-de', 'text/plain', True),
('dists/sid/main/i18n/Translation-de.gz', 'text/plain', True),
('dists/sid/main/i18n/Translation-de.bz2', 'text/plain', True)
]
for name,mimetype,mutable in tests:
log.debug('Testing filetype, name=%s mimetype=%s mutable=%s' % (name, mimetype, mutable))
result = findFileType(name)
self.assertNotEquals(result, None)
log.debug(' -> result contype=%s mutable=%s' % (result.contype, result.mutable))
self.assertEquals(mimetype, result.contype)
self.assertEquals(mutable, result.mutable)
|
