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

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

Incomplete - task 9: Data Browser Replacement

  • Major fix for handling environSCRIPT_NAME?
  • removed reconstruct_url function - not needed.
  • 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     
60class FileServerAppTestCase(FileServerAppTestCaseBase):
61    """Test ceda.server.wsgi.fileserver.app.FileServerApp with Paste Fixture
62    """
63   
64    def __init__(self, *args, **kwargs):
65        self.fileServerApp = FileServerApp(self.__class__.HTDOCS_DIR)
66        self.app = paste.fixture.TestApp(self.fileServerApp)
67         
68        FileServerAppTestCaseBase.__init__(self, *args, **kwargs)
69
70    def test01Assert(self):
71        response = self.app.get('/', status=200)
72        print(response.body)
73        self.assert_(response.body)
74        self.assert_(self.__class__.HTML_REL_FILEPATH in response.body)
75
76    def test02CheckBlkSizeValidation(self):
77        def _set(x):
78            self.fileServerApp.readBlkSize = x
79           
80        self.assertRaises(ValueError, _set, 'my block size') 
81        self.assertRaises(ValueError, _set, -10)
82       
83    def test03CheckHttpMethodValidation(self):
84        def _set():
85            self.fileServerApp.httpMethodMap = None
86           
87        self.assertRaises(TypeError, _set)
88       
89    def test04TrapHttpMethodNotAllowed(self):
90        self.fileServerApp.httpMethodMap = {}
91        response = self.app.get('/', status=httplib.METHOD_NOT_ALLOWED)
92
93    def test05GetJPEG(self):
94        # Test jpeg is returned with correct content type
95        response = self.app.get(self.__class__.JPEG_REL_URIPATH, 
96                                status=httplib.OK)
97        print(response.headers)
98        print(response.body)
99        self.assert_(response.header_dict['content-type'] == 'image/jpeg')
100
101    def test06GetPngWithUppercaseSuffix(self):
102        # Test PNG is returned with correct content type
103        response = self.app.get(self.__class__.PNG_REL_URIPATH, 
104                                status=httplib.OK)
105        print(response.headers)
106        print(response.body)
107        self.assert_(response.header_dict['content-type'] == 'image/png')
108
109    def test07WriteBlocks(self):
110        # Test writing output from server in blocks
111        self.fileServerApp.writeBlkSize = 8
112       
113        localFilePath = path.join(self.__class__.HTDOCS_DIR, 
114                                  self.__class__.PNG_REL_FILEPATH)
115        fileContent = open(localFilePath, 'rb').read()
116       
117        response = self.app.get(self.__class__.PNG_REL_URIPATH, 
118                                status=httplib.OK)
119       
120        print(response.headers)
121        print(response.body)
122        self.assert_(response.header_dict['content-type'] == 'image/png')
123        self.assert_(response.body == fileContent)
124
125    def test08FileNotFound(self):
126        response = self.app.get('/abc', status=httplib.NOT_FOUND)
127        print(response.headers)
128        print(response.body)
129
130    def test09ListSubDir(self):
131        # list content of a sub-directory
132        response = self.app.get(self.__class__.HTDOCS_SUBDIR3_URIPATH, 
133                                status=httplib.OK)
134        print(response.body)
135        self.assert_(self.__class__.PDF_FILENAME in response.body)
136
137    def test10PathFilter(self):
138        # Test filter which filters out path matching a given regular expression
139        self.fileServerApp.fileFilterRegEx = '.*\.(PNG|png)'
140       
141        # Test direct access to the JPEG URI
142        response1 = self.app.get(self.__class__.PNG_REL_URIPATH, 
143                                 status=httplib.NOT_FOUND)
144       
145        print(response1.headers)
146        print(response1.body)
147       
148        # Now test the list for the PNG file's directory
149        response2 = self.app.get(self.__class__.HTDOCS_SUBDIR1_URIPATH, 
150                                 status=httplib.OK)
151       
152        print(response2.headers)
153        print(response2.body)
154       
155        self.assert_(self.__class__.PNG_REL_FILEPATH not in response2.body)
156       
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               
202       
203class FileServerAppPasterTestCase(FileServerAppTestCaseBase): 
204    """Base class for common Paste Deploy related set-up"""
205    INI_FILENAME = 'fileserver.ini'
206    INI_FILEPATH = path.join(FileServerAppTestCaseBase.THIS_DIR, INI_FILENAME)
207   
208    SERVICE_PORTNUM = 5080
209    MOUNT_POINT = '/fileserver'
210    URI_BASE = 'http://localhost:%d%s' % (SERVICE_PORTNUM, MOUNT_POINT)
211   
212    def __init__(self, *args, **kwargs):
213        wsgiapp = loadapp('config:' + self.__class__.INI_FILEPATH)
214        self.app = paste.fixture.TestApp(wsgiapp)
215       
216        self.services = []
217       
218        # Start the file server in a separate thread
219        self.addService(cfgFilePath=self.__class__.INI_FILEPATH, 
220                        port=self.__class__.SERVICE_PORTNUM)
221               
222        FileServerAppTestCaseBase.__init__(self, *args, **kwargs) 
223       
224    def test01GetPNG(self):
225        uri = self.__class__.URI_BASE + self.__class__.PNG_REL_URIPATH
226        response = urllib2.urlopen(uri)
227        self.assert_(response)
228        body = response.read()
229        self.assert_(len(body) > 0)
230        self.assert_(response.code == httplib.OK)
231       
232    def test02MissingMountPoint(self):
233        # Try request with incorrect path missing out mount point
234        uri = 'http://localhost:%d%s' % (self.__class__.SERVICE_PORTNUM,
235                                         self.__class__.PNG_REL_URIPATH)
236        self.assertRaises(urllib2.HTTPError, urllib2.urlopen, uri)
237       
238    def addService(self, *arg, **kw):
239        """Utility for setting up threads to run Paste HTTP based services with
240        unit tests
241       
242        @param arg: tuple contains ini file path setting for the service
243        @type arg: tuple
244        @param kw: keywords including "port" - port number to run the service
245        from
246        @type kw: dict
247        """
248        try:
249            self.services.append(PasteDeployAppServer(*arg, **kw))
250            self.services[-1].startThread()
251           
252        except socket.error:
253            pass
254
255    def __del__(self):
256        """Stop any services started with the addService method
257        """
258        if hasattr(self, 'services'):
259            for service in self.services:
260                service.terminateThread()
261               
262               
263if __name__ == "__main__":
264    unittest.main()
265
Note: See TracBrowser for help on using the repository browser.