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

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@7032
Revision 7032, 10.4 KB checked in by pjkersha, 10 years ago (diff)

Incomplete - task 9: Data Browser Replacement

  • adding support byte ranges in header requests
  • 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('sub.dir-1', '.sub-dir2')
40    HTDOCS_SUBDIR3 = path.join('sub.dir-1', '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        self.app = paste.fixture.TestApp(self.fileServerApp)
68         
69        FileServerAppTestCaseBase.__init__(self, *args, **kwargs)
70
71    def test01Assert(self):
72        response = self.app.get('/', status=200)
73        print(response.body)
74        self.assert_(response.body)
75        self.assert_(self.__class__.HTML_REL_FILEPATH in response.body)
76
77    def test02CheckBlkSizeValidation(self):
78        def _set(x):
79            self.fileServerApp.readBlkSize = x
80           
81        self.assertRaises(ValueError, _set, 'my block size') 
82        self.assertRaises(ValueError, _set, -10)
83       
84    def test03CheckHttpMethodValidation(self):
85        def _set():
86            self.fileServerApp.httpMethodMap = None
87           
88        self.assertRaises(TypeError, _set)
89       
90    def test04TrapHttpMethodNotAllowed(self):
91        self.fileServerApp.httpMethodMap = {}
92        response = self.app.get('/', status=httplib.METHOD_NOT_ALLOWED)
93
94    def test05GetJPEG(self):
95        # Test jpeg is returned with correct content type
96        response = self.app.get(self.__class__.JPEG_REL_URIPATH, 
97                                status=httplib.OK)
98        print(response.headers)
99        print(response.body)
100        self.assert_(response.header_dict['content-type'] == 'image/jpeg')
101
102    def test06GetPngWithUppercaseSuffix(self):
103        # Test PNG is returned with correct content type
104        response = self.app.get(self.__class__.PNG_REL_URIPATH, 
105                                status=httplib.OK)
106        print(response.headers)
107        print(response.body)
108        self.assert_(response.header_dict['content-type'] == 'image/png')
109
110    def test07WriteBlocks(self):
111        # Test writing output from server in blocks
112        self.fileServerApp.writeBlkSize = 8
113       
114        localFilePath = path.join(self.__class__.HTDOCS_DIR, 
115                                  self.__class__.PNG_REL_FILEPATH)
116        fileContent = open(localFilePath, 'rb').read()
117       
118        response = self.app.get(self.__class__.PNG_REL_URIPATH, 
119                                status=httplib.OK)
120       
121        print(response.headers)
122        print(response.body)
123        self.assert_(response.header_dict['content-type'] == 'image/png')
124        self.assert_(response.body == fileContent)
125
126    def test08FileNotFound(self):
127        response = self.app.get('/abc', status=httplib.NOT_FOUND)
128        print(response.headers)
129        print(response.body)
130
131    def test09ListSubDir(self):
132        # list content of a sub-directory
133        response = self.app.get(self.__class__.HTDOCS_SUBDIR3_URIPATH, 
134                                status=httplib.OK)
135        print(response.body)
136        self.assert_(self.__class__.PDF_FILENAME in response.body)
137
138    def test10PathFilter(self):
139        # Test filter which filters out path matching a given regular expression
140        self.fileServerApp.fileFilterRegEx = '.*\.(PNG|png)'
141       
142        # Test direct access to the JPEG URI
143        response1 = self.app.get(self.__class__.PNG_REL_URIPATH, 
144                                 status=httplib.NOT_FOUND)
145       
146        print(response1.headers)
147        print(response1.body)
148       
149        # Now test the list for the PNG file's directory
150        response2 = self.app.get(self.__class__.HTDOCS_SUBDIR1_URIPATH, 
151                                 status=httplib.OK)
152       
153        print(response2.headers)
154        print(response2.body)
155       
156        self.assert_(self.__class__.PNG_REL_FILEPATH not in response2.body)
157       
158    def test11AccessForbidden(self):
159       
160        jpegFilePath = path.join(self.__class__.HTDOCS_DIR,
161                                 self.__class__.JPEG_REL_FILEPATH)
162        try:
163            # Get file permissions for the file
164            mode = os.stat(jpegFilePath).st_mode
165           
166            # Make file inaccessible
167            os.chmod(jpegFilePath, 0000)
168
169            response = self.app.get(self.__class__.JPEG_REL_URIPATH, 
170                                    status=httplib.FORBIDDEN)
171            print(response.headers)
172            print(response.body)
173           
174        finally:
175            # Restore the original file permissions
176            os.chmod(jpegFilePath, mode)
177   
178    def test12WithAltMountPoint(self):
179        # Test with alternative mount point to the default '/'
180        self.fileServerApp.mountPoint = '/file-server'
181        requestPath = '/file-server' + self.__class__.HTDOCS_SUBDIR3_URIPATH
182       
183        response = self.app.get(requestPath, status=httplib.OK)
184        print(response.body)
185        self.assert_(self.__class__.PDF_FILENAME in response.body)
186       
187    def test13WithEnvironScriptName(self):
188        # Test with the application mounted at environ['SCRIPT_NAME']
189        extra_environ = {'SCRIPT_NAME': '/myfileserver'}
190        requestPath = '/myfileserver' + self.__class__.HTDOCS_SUBDIR3_URIPATH
191       
192        app = paste.fixture.TestApp(self.fileServerApp,
193                                    extra_environ=extra_environ)
194       
195        response = app.get(requestPath,
196           extra_environ = {'PATH_INFO': self.__class__.HTDOCS_SUBDIR3_URIPATH},
197           status=httplib.OK)
198        print(response.body)
199        self.assert_(self.__class__.PDF_FILENAME in response.body)
200       
201    def test14CheckResponseTypeForSubDir(self):
202        # Check the response type for sub-directory page listing retrieval
203        # It must be text/html or application/xhtml+xml for WGet recursive
204        # retrieval.
205        response = self.app.get(self.__class__.HTDOCS_SUBDIR3_URIPATH, 
206                                status=httplib.OK)
207        print(response.headers)
208        self.assert_('text/html' in response.header_dict['content-type'])
209       
210    def test15RetrieveRange(self):
211        # Specify a range of bytes for retrieval
212        headers = {'Range': '0-599'}
213        response = self.app.get(self.__class__.PDF_REL_URIPATH,
214                                headers=headers, 
215                                status=httplib.PARTIAL_CONTENT)
216        self.assert_('content-range' in response.header_dict)
217       
218class FileServerAppPasterTestCase(FileServerAppTestCaseBase): 
219    """Base class for common Paste Deploy related set-up"""
220    INI_FILENAME = 'fileserver.ini'
221    INI_FILEPATH = path.join(FileServerAppTestCaseBase.THIS_DIR, INI_FILENAME)
222   
223    SERVICE_PORTNUM = 5080
224    MOUNT_POINT = '/fileserver'
225    URI_BASE = 'http://localhost:%d%s' % (SERVICE_PORTNUM, MOUNT_POINT)
226   
227    def __init__(self, *args, **kwargs):
228        wsgiapp = loadapp('config:' + self.__class__.INI_FILEPATH)
229        self.app = paste.fixture.TestApp(wsgiapp)
230       
231        self.services = []
232       
233        # Start the file server in a separate thread
234        self.addService(cfgFilePath=self.__class__.INI_FILEPATH, 
235                        port=self.__class__.SERVICE_PORTNUM)
236               
237        FileServerAppTestCaseBase.__init__(self, *args, **kwargs) 
238       
239    def test01GetPNG(self):
240        uri = self.__class__.URI_BASE + self.__class__.PNG_REL_URIPATH
241        response = urllib2.urlopen(uri)
242        self.assert_(response)
243        body = response.read()
244        self.assert_(len(body) > 0)
245        self.assert_(response.code == httplib.OK)
246       
247    def test02MissingMountPoint(self):
248        # Try request with incorrect path missing out mount point
249        uri = 'http://localhost:%d%s' % (self.__class__.SERVICE_PORTNUM,
250                                         self.__class__.PNG_REL_URIPATH)
251        self.assertRaises(urllib2.HTTPError, urllib2.urlopen, uri)
252       
253    def addService(self, *arg, **kw):
254        """Utility for setting up threads to run Paste HTTP based services with
255        unit tests
256       
257        @param arg: tuple contains ini file path setting for the service
258        @type arg: tuple
259        @param kw: keywords including "port" - port number to run the service
260        from
261        @type kw: dict
262        """
263        try:
264            self.services.append(PasteDeployAppServer(*arg, **kw))
265            self.services[-1].startThread()
266           
267        except socket.error:
268            pass
269
270    def __del__(self):
271        """Stop any services started with the addService method
272        """
273        if hasattr(self, 'services'):
274            for service in self.services:
275                service.terminateThread()
276               
277               
278if __name__ == "__main__":
279    unittest.main()
280
Note: See TracBrowser for help on using the repository browser.