source: ceda_http_fileserver/trunk/ceda_http_fileserver/ceda/server/wsgi/fileserver/test/test_fileserver.py @ 7043

Subversion URL: http://proj.badc.rl.ac.uk/svn/ndg/ceda_http_fileserver/trunk/ceda_http_fileserver/ceda/server/wsgi/fileserver/test/test_fileserver.py@7043
Revision 7043, 12.7 KB checked in by pjkersha, 9 years ago (diff)

Incomplete - task 9: Data Browser Replacement

  • Property svn:keywords set to Id
Line 
1"""CEDA (Centre for Environmental Data Archival) HTTP file server application
2unit test module
3
4CEDA HTTP File Server WSGI Application
5"""
6__author__ = "P J Kershaw"
7__date__ = "11/06/10"
8__copyright__ = "(C) 2010 Science and Technology Facilities Council"
9__license__ = """http://www.apache.org/licenses/LICENSE-2.0"""
10__contact__ = "Philip.Kershaw@stfc.ac.uk"
11__revision__ = '$Id$'
12import logging
13logging.basicConfig(level=logging.DEBUG)
14
15import unittest
16from os import path
17import os
18import socket
19import httplib
20import urllib
21import urllib2
22
23import paste.fixture
24from paste.deploy import loadapp
25
26from ceda.server.wsgi.fileserver.test import PasteDeployAppServer
27from ceda.server.wsgi.fileserver.app import FileServerApp
28
29
30class FileServerAppTestCaseBase(unittest.TestCase):
31    THIS_DIR = path.abspath(path.dirname(__file__))
32   
33    HTDOCS_DIRNAME = 'htdocs'
34    HTDOCS_DIR = path.join(THIS_DIR, HTDOCS_DIRNAME)
35   
36    HTDOCS_SUBDIR1 = 'sub+dir-1'
37    HTDOCS_SUBDIR1_URIPATH = '/' + HTDOCS_SUBDIR1
38   
39    HTDOCS_SUBDIR2 = path.join(HTDOCS_SUBDIR1, '.sub-dir2')
40    HTDOCS_SUBDIR3 = path.join(HTDOCS_SUBDIR1, 'SubDir 3')
41    HTDOCS_SUBDIR3_URIPATH = '/' + urllib.pathname2url(HTDOCS_SUBDIR3)
42   
43    PNG_REL_FILEPATH = path.join(HTDOCS_SUBDIR1,
44                                 'a png with uppercase suffix.PNG')
45    PNG_REL_URIPATH = '/' + urllib.pathname2url(PNG_REL_FILEPATH)
46   
47    JPEG_REL_FILEPATH = 'my test jpeg.jpg'
48    JPEG_REL_URIPATH = '/' + urllib.pathname2url(JPEG_REL_FILEPATH)
49   
50    PLAIN_TEXT_REL_FILEPATH = path.join(HTDOCS_SUBDIR2, 'plain-text.file.txt')
51    PLAIN_TEXT_REL_URIPATH = '/' + urllib.pathname2url(PLAIN_TEXT_REL_FILEPATH)
52   
53    HTML_REL_FILEPATH = 'my file.html'
54    HTML_REL_URIPATH = '/' + urllib.pathname2url(HTML_REL_FILEPATH)
55   
56    PDF_FILENAME = 'a test PDF.pdf'
57    PDF_REL_FILEPATH = path.join(HTDOCS_SUBDIR3, PDF_FILENAME)
58    PDF_REL_URIPATH = '/' + urllib.pathname2url(PDF_REL_FILEPATH)   
59     
60     
61class FileServerAppTestCase(FileServerAppTestCaseBase):
62    """Test ceda.server.wsgi.fileserver.app.FileServerApp with Paste Fixture
63    """
64   
65    def __init__(self, *args, **kwargs):
66        self.fileServerApp = FileServerApp(self.__class__.HTDOCS_DIR)
67       
68        # Small block size to test WSGI iterator for writing response
69        self.fileServerApp.writeBlkSize = 256
70        self.app = paste.fixture.TestApp(self.fileServerApp)
71         
72        FileServerAppTestCaseBase.__init__(self, *args, **kwargs)
73
74    def test01Assert(self):
75        response = self.app.get('/', status=200)
76        print(response.body)
77        self.assert_(response.body)
78        self.assert_(self.__class__.HTML_REL_FILEPATH in response.body)
79
80    def test02CheckBlkSizeValidation(self):
81        def _set(x):
82            self.fileServerApp.readBlkSize = x
83           
84        self.assertRaises(ValueError, _set, 'my block size') 
85        self.assertRaises(ValueError, _set, -10)
86       
87    def test03CheckHttpMethodValidation(self):
88        def _set():
89            self.fileServerApp.httpMethodMap = None
90           
91        self.assertRaises(TypeError, _set)
92       
93    def test04TrapHttpMethodNotAllowed(self):
94        self.fileServerApp.httpMethodMap = {}
95        response = self.app.get('/', status=httplib.METHOD_NOT_ALLOWED)
96
97    def test05GetJPEG(self):
98        # Test jpeg is returned with correct content type
99        response = self.app.get(self.__class__.JPEG_REL_URIPATH, 
100                                status=httplib.OK)
101        print(response.headers)
102        print(response.body)
103        self.assert_(response.header_dict['content-type'] == 'image/jpeg')
104
105    def test06GetPngWithUppercaseSuffix(self):
106        # Test PNG is returned with correct content type
107        response = self.app.get(self.__class__.PNG_REL_URIPATH, 
108                                status=httplib.OK)
109        print(response.headers)
110        print(response.body)
111        self.assert_(response.header_dict['content-type'] == 'image/png')
112
113    def test07WriteBlocks(self):
114        # Test writing output from server in blocks
115        self.fileServerApp.writeBlkSize = 8
116       
117        localFilePath = path.join(self.__class__.HTDOCS_DIR, 
118                                  self.__class__.PNG_REL_FILEPATH)
119        fileContent = open(localFilePath, 'rb').read()
120       
121        response = self.app.get(self.__class__.PNG_REL_URIPATH, 
122                                status=httplib.OK)
123       
124        print(response.headers)
125        print(response.body)
126        self.assert_(response.header_dict['content-type'] == 'image/png')
127        self.assert_(response.body == fileContent)
128
129    def test08FileNotFound(self):
130        response = self.app.get('/abc', status=httplib.NOT_FOUND)
131        print(response.headers)
132        print(response.body)
133
134    def test09ListSubDir(self):
135        # list content of a sub-directory
136        response = self.app.get(self.__class__.HTDOCS_SUBDIR3_URIPATH, 
137                                status=httplib.OK)
138        print(response.body)
139        self.assert_(self.__class__.PDF_FILENAME in response.body)
140
141    def test10PathFilter(self):
142        # Test filter which filters out path matching a given regular expression
143        self.fileServerApp.fileFilterRegEx = '.*\.(PNG|png)'
144       
145        # Test direct access to the JPEG URI
146        response1 = self.app.get(self.__class__.PNG_REL_URIPATH, 
147                                 status=httplib.NOT_FOUND)
148       
149        print(response1.headers)
150        print(response1.body)
151       
152        # Now test the list for the PNG file's directory
153        response2 = self.app.get(self.__class__.HTDOCS_SUBDIR1_URIPATH, 
154                                 status=httplib.OK)
155       
156        print(response2.headers)
157        print(response2.body)
158       
159        self.assert_(self.__class__.PNG_REL_FILEPATH not in response2.body)
160       
161    def test11AccessForbidden(self):
162        # Set a file on the file system with permissions so that it can be read
163        # and ensure that the server returns a 403 Forbidden response for a
164        # request for this file
165        jpegFilePath = path.join(self.__class__.HTDOCS_DIR,
166                                 self.__class__.JPEG_REL_FILEPATH)
167        try:
168            # Get file permissions for the file
169            mode = os.stat(jpegFilePath).st_mode
170           
171            # Make file inaccessible
172            os.chmod(jpegFilePath, 0000)
173
174            response = self.app.get(self.__class__.JPEG_REL_URIPATH, 
175                                    status=httplib.FORBIDDEN)
176            print(response.headers)
177            print(response.body)
178           
179        finally:
180            # Restore the original file permissions
181            os.chmod(jpegFilePath, mode)
182   
183    def test12WithAltMountPoint(self):
184        # Test with alternative mount point to the default '/'
185        self.fileServerApp.mountPoint = '/file-server'
186        requestPath = '/file-server' + self.__class__.HTDOCS_SUBDIR3_URIPATH
187       
188        response = self.app.get(requestPath, status=httplib.OK)
189        print(response.body)
190        self.assert_(self.__class__.PDF_FILENAME in response.body)
191       
192    def test13WithEnvironScriptName(self):
193        # Test with the application mounted at environ['SCRIPT_NAME']
194        extra_environ = {'SCRIPT_NAME': '/myfileserver'}
195        requestPath = '/myfileserver' + self.__class__.HTDOCS_SUBDIR3_URIPATH
196       
197        app = paste.fixture.TestApp(self.fileServerApp,
198                                    extra_environ=extra_environ)
199       
200        response = app.get(requestPath,
201           extra_environ = {'PATH_INFO': self.__class__.HTDOCS_SUBDIR3_URIPATH},
202           status=httplib.OK)
203        print(response.body)
204        self.assert_(self.__class__.PDF_FILENAME in response.body)
205       
206    def test14CheckResponseTypeForSubDir(self):
207        # Check the response type for sub-directory page listing retrieval
208        # It must be text/html or application/xhtml+xml for WGet recursive
209        # retrieval.
210        response = self.app.get(self.__class__.HTDOCS_SUBDIR3_URIPATH, 
211                                status=httplib.OK)
212        print(response.headers)
213        self.assert_('text/html' in response.header_dict['content-type'])
214       
215    def test15RetrieveRange(self):
216        # Specify a range of bytes for retrieval
217        headers = {'Range': 'bytes=10-609'}
218        response = self.app.get(self.__class__.PDF_REL_URIPATH,
219                                headers=headers, 
220                                status=httplib.PARTIAL_CONTENT)
221        self.assert_('content-range' in response.header_dict)
222        print('Content-range: %s' % response.header_dict['content-range'])
223        self.assert_(len(response.body) == 600)
224       
225    def test16RetrieveFirst10Bytes(self):
226        # Specify a range of bytes for retrieval
227        headers = {'Range': 'bytes=-9'}
228        response = self.app.get(self.__class__.HTML_REL_URIPATH,
229                                headers=headers, 
230                                status=httplib.PARTIAL_CONTENT)
231        self.assert_('content-range' in response.header_dict)
232        print('Content-range: %s' % response.header_dict['content-range'])
233        print("output = %r" % response.body)
234        self.assert_(len(response.body) == 10)
235       
236    def test17RetrieveLast10Bytes(self):
237        # Specify a range of bytes for retrieval
238        headers = {'Range': 'bytes=170-'}
239        response = self.app.get(self.__class__.HTML_REL_URIPATH,
240                                headers=headers, 
241                                status=httplib.PARTIAL_CONTENT)
242        self.assert_('content-range' in response.header_dict)
243        print('Content-range: %s' % response.header_dict['content-range'])
244        print("output = %r" % response.body)
245        self.assert_(len(response.body) == 10)
246       
247    def test18CatchInvalidRangeSyntax(self):
248        # Specify invalid range syntax
249        headers = {'Range': 'bytes=a bad range'}
250        response = self.app.get(self.__class__.PDF_REL_URIPATH,
251                                headers=headers, 
252                                status=httplib.BAD_REQUEST)
253        print("output = %r" % response.body)
254       
255    def test19CatchInvalidRange(self):
256        # Specify a range of bytes for retrieval
257        headers = {'Range': '10000-0'}
258        response = self.app.get(self.__class__.PDF_REL_URIPATH,
259                                headers=headers, 
260                                status=httplib.REQUESTED_RANGE_NOT_SATISFIABLE)
261        self.assert_('content-range' in response.header_dict)   
262        print('Content-range: %s' % response.header_dict['content-range'])
263       
264             
265class FileServerAppPasterTestCase(FileServerAppTestCaseBase): 
266    """Base class for common Paste Deploy related set-up"""
267    INI_FILENAME = 'fileserver.ini'
268    INI_FILEPATH = path.join(FileServerAppTestCaseBase.THIS_DIR, INI_FILENAME)
269   
270    SERVICE_PORTNUM = 5080
271    MOUNT_POINT = '/fileserver'
272    URI_BASE = 'http://localhost:%d%s' % (SERVICE_PORTNUM, MOUNT_POINT)
273   
274    def __init__(self, *args, **kwargs):
275        wsgiapp = loadapp('config:' + self.__class__.INI_FILEPATH)
276        self.app = paste.fixture.TestApp(wsgiapp)
277       
278        self.services = []
279       
280        # Start the file server in a separate thread
281        self.addService(cfgFilePath=self.__class__.INI_FILEPATH, 
282                        port=self.__class__.SERVICE_PORTNUM)
283               
284        FileServerAppTestCaseBase.__init__(self, *args, **kwargs) 
285       
286    def test01GetPNG(self):
287        uri = self.__class__.URI_BASE + self.__class__.PNG_REL_URIPATH
288        response = urllib2.urlopen(uri)
289        self.assert_(response)
290        body = response.read()
291        self.assert_(len(body) > 0)
292        self.assert_(response.code == httplib.OK)
293       
294    def test02MissingMountPoint(self):
295        # Try request with incorrect path missing out mount point
296        uri = 'http://localhost:%d%s' % (self.__class__.SERVICE_PORTNUM,
297                                         self.__class__.PNG_REL_URIPATH)
298        self.assertRaises(urllib2.HTTPError, urllib2.urlopen, uri)
299       
300    def addService(self, *arg, **kw):
301        """Utility for setting up threads to run Paste HTTP based services with
302        unit tests
303       
304        @param arg: tuple contains ini file path setting for the service
305        @type arg: tuple
306        @param kw: keywords including "port" - port number to run the service
307        from
308        @type kw: dict
309        """
310        try:
311            self.services.append(PasteDeployAppServer(*arg, **kw))
312            self.services[-1].startThread()
313           
314        except socket.error:
315            pass
316
317    def __del__(self):
318        """Stop any services started with the addService method
319        """
320        if hasattr(self, 'services'):
321            for service in self.services:
322                service.terminateThread()
323               
324               
325if __name__ == "__main__":
326    unittest.main()
327
Note: See TracBrowser for help on using the repository browser.