+class TestTopLevel(unittest.TestCase):
+ """Unit tests for the HTTP Server."""
+
+ client = None
+ pending_calls = []
+ torrent_hash = '\xca \xb8\x0c\x00\xe7\x07\xf8~])+\x9d\xe5_B\xff\x1a\xc4!'
+ torrent = 'abcdefghij0123456789\xca\xec\xb8\x0c\x00\xe7\x07\xf8~])\x8f\x9d\xe5_B\xff\x1a\xc4!'
+ file_hash = '\xf8~])+\x9d\xe5_B\xff\x1a\xc4!\xca \xb8\x0c\x00\xe7\x07'
+
+ def setUp(self):
+ self.client = TopLevel(FilePath('/boot'), self, None, 0)
+
+ def lookupHash(self, hash):
+ if hash == self.torrent_hash:
+ return [{'pieces': self.torrent}]
+ elif hash == self.file_hash:
+ return [{'path': FilePath('/boot/initrd')}]
+ else:
+ return []
+
+ def create_request(self, host, path):
+ req = server.Request(None, 'GET', path, (1,1), 0, http_headers.Headers())
+ class addr:
+ host = ''
+ port = 0
+ req.remoteAddr = addr()
+ req.remoteAddr.host = host
+ req.remoteAddr.port = 23456
+ server.Request._parseURL(req)
+ return req
+
+ def test_unauthorized(self):
+ req = self.create_request('128.0.0.1', '/foo/bar')
+ self.failUnlessRaises(http.HTTPError, req._getChild, None, self.client, req.postpath)
+
+ def test_Packages_diff(self):
+ req = self.create_request('127.0.0.1',
+ '/ftp.us.debian.org/debian/dists/unstable/main/binary-i386/Packages.diff/Index')
+ self.failUnlessRaises(http.HTTPError, req._getChild, None, self.client, req.postpath)
+
+ def test_Statistics(self):
+ req = self.create_request('127.0.0.1', '/')
+ res = req._getChild(None, self.client, req.postpath)
+ self.failIfEqual(res, None)
+ df = defer.maybeDeferred(res.renderHTTP, req)
+ df.addCallback(self.check_resp, 200)
+ return df
+
+ def test_apt_download(self):
+ req = self.create_request('127.0.0.1',
+ '/ftp.us.debian.org/debian/dists/stable/Release')
+ res = req._getChild(None, self.client, req.postpath)
+ self.failIfEqual(res, None)
+ self.failUnless(isinstance(res, FileDownloader))
+ df = defer.maybeDeferred(res.renderHTTP, req)
+ df.addCallback(self.check_resp, 404)
+ return df
+
+ def test_torrent_upload(self):
+ req = self.create_request('123.45.67.89',
+ '/~/' + quote_plus(self.torrent_hash))
+ res = req._getChild(None, self.client, req.postpath)
+ self.failIfEqual(res, None)
+ self.failUnless(isinstance(res, static.Data))
+ df = defer.maybeDeferred(res.renderHTTP, req)
+ df.addCallback(self.check_resp, 200)
+ return df
+
+ def test_file_upload(self):
+ req = self.create_request('123.45.67.89',
+ '/~/' + quote_plus(self.file_hash))
+ res = req._getChild(None, self.client, req.postpath)
+ self.failIfEqual(res, None)
+ self.failUnless(isinstance(res, FileUploader))
+ df = defer.maybeDeferred(res.renderHTTP, req)
+ df.addCallback(self.check_resp, 200)
+ return df
+
+ def test_missing_hash(self):
+ req = self.create_request('123.45.67.89',
+ '/~/' + quote_plus('foobar'))
+ self.failUnlessRaises(http.HTTPError, req._getChild, None, self.client, req.postpath)
+
+ def check_resp(self, resp, code):
+ self.failUnlessEqual(resp.code, code)
+ return resp
+
+ def tearDown(self):
+ for p in self.pending_calls:
+ if p.active():
+ p.cancel()
+ self.pending_calls = []
+ if self.client:
+ self.client = None
+