witten
/
luminotes
Archived
1
0
Fork 0
Code Releases Activity

Implemented several unit tests for controller.Files.

This commit is contained in:
Dan Helfman 2008-02-23 08:14:19 +00:00
parent e7c96cadf5
commit 3f5d5d2a89
4 changed files with 174 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
controller/Files.py
View File

@ -56,7 +56,7 @@ class Upload_file( object ):
File-like object for storing file uploads.
"""
def __init__( self, file_id, filename, content_length ):
self.__file = file( self.make_server_filename( file_id ), "w+" )
self.__file = self.open_file( file_id, "w+" )
self.__file_id = file_id
self.__filename = filename
self.__content_length = content_length
@ -94,7 +94,7 @@ class Upload_file( object ):
def delete( self ):
self.__file.close()
os.remove( self.make_server_filename( self.__file_id ) )
self.delete_file( self.__file_id )
def wait_for_complete( self ):
self.__complete.wait( timeout = cherrypy.server.socket_timeout )
@ -103,6 +103,16 @@ class Upload_file( object ):
def make_server_filename( file_id ):
return u"files/%s" % file_id
@staticmethod
def open_file( file_id, mode = None ):
if mode:
return file( Upload_file.make_server_filename( file_id ), mode )
return file( Upload_file.make_server_filename( file_id ) )
@staticmethod
def delete_file( file_id ):
return os.remove( Upload_file.make_server_filename( file_id ) )
filename = property( lambda self: self.__filename )
# expected byte count of the entire form upload, including the file and other form parameters
@ -251,7 +261,7 @@ class Files( object ):
def stream():
CHUNK_SIZE = 8192
local_file = file( Upload_file.make_server_filename( file_id ) )
local_file = Upload_file.open_file( file_id )
while True:
data = local_file.read( CHUNK_SIZE )
@ -493,7 +503,7 @@ class Files( object ):
user = self.__users.update_storage( user_id, commit = False )
self.__database.commit()
os.remove( Upload_file.make_server_filename( file_id ) )
Upload_file.delete_file( file_id )
return dict(
storage_bytes = user.storage_bytes,
@ -555,6 +565,6 @@ class Files( object ):
# filesystem
for ( file_id, db_file ) in files_to_delete.items():
self.__database.execute( db_file.sql_delete(), commit = False )
os.remove( Upload_file.make_server_filename( file_id ) )
Upload_file.delete_file( file_id )
self.__database.commit()

50
controller/test/Test_controller.py
View File

@ -7,17 +7,25 @@ from StringIO import StringIO
from copy import copy
class Truncated_StringIO( StringIO ):
class Wrapped_StringIO( StringIO ):
"""
A wrapper for StringIO that forcibly closes the file when only some of it has been read. Used
for simulating an upload that is canceled part of the way through.
A wrapper for StringIO that includes a bytes_read property, needed to work with
controller.Files.Upload_file.
"""
bytes_read = property( lambda self: self.tell() )
class Truncated_StringIO( Wrapped_StringIO ):
"""
A wrapper for Wrapped_StringIO that forcibly closes the file when only some of it has been read.
Used for simulating an upload that is canceled part of the way through.
"""
def readline( self, size = None ):
if self.tell() >= len( self.getvalue() ) * 0.25:
self.close()
return ""
return StringIO.readline( self, 256 )
return Wrapped_StringIO.readline( self, 256 )
class Test_controller( object ):
@ -27,6 +35,7 @@ class Test_controller( object ):
from model.Note import Note
from model.Invite import Invite
from model.User_revision import User_revision
from model.File import File
# Since Stub_database isn't a real database and doesn't know SQL, replace some of the
# SQL-returning methods in User, Note, and Notebook to return functions that manipulate data in
@ -313,6 +322,19 @@ class Test_controller( object ):
Invite.sql_revoke_invites = lambda self: \
lambda database: sql_revoke_invites( self, database )
def sql_load_note_files( note_id, database ):
files = []
for ( object_id, obj_list ) in database.objects.items():
obj = obj_list[ -1 ]
if isinstance( obj, File ) and obj.note_id == note_id:
files.append( obj )
return files
File.sql_load_note_files = staticmethod( lambda note_id:
lambda database: sql_load_note_files( note_id, database ) )
def setUp( self ):
from controller.Root import Root
@ -331,22 +353,26 @@ class Test_controller( object ):
u"luminotes.rate_plans": [
{
u"name": u"super",
u"storage_quota_bytes": 1337,
u"storage_quota_bytes": 1337 * 10,
u"notebook_collaboration": True,
u"fee": 1.99,
u"button": u"[subscribe here user %s!] button",
},
{
u"name": "extra super",
u"storage_quota_bytes": 31337,
u"storage_quota_bytes": 31337 * 10,
u"notebook_collaboration": True,
u"fee": 9.00,
u"button": u"[or here user %s!] button",
},
],
},
u"/files/upload": {
u"stream_response": True
u"/files/download": {
u"stream_response": True,
u"encoding_filter.on": False,
},
u"/files/progress": {
u"stream_response": True,
},
}
@ -440,7 +466,7 @@ class Test_controller( object ):
finally:
request.close()
def http_upload( self, http_path, form_args, filename, file_data, simulate_cancel = False, headers = None, session_id = None ):
def http_upload( self, http_path, form_args, filename, file_data, content_type, simulate_cancel = False, headers = None, session_id = None ):
"""
Perform an HTTP POST with the given path on the test server, sending the provided form_args
and file_data as a multipart form file upload. Return the result dict as returned by the
@ -457,8 +483,8 @@ class Test_controller( object ):
post_data.append( 'Content-Disposition: form-data; name="upload"; filename="%s"\n' % (
filename
) )
post_data.append( "Content-Type: image/png\n\n%s\n--%s--\n" % (
file_data, boundary
post_data.append( "Content-Type: %s\n\n%s\n--%s--\n" % (
content_type, file_data, boundary
) )
if headers is None:
@ -476,7 +502,7 @@ class Test_controller( object ):
if simulate_cancel:
file_wrapper = Truncated_StringIO( post_data )
else:
file_wrapper = StringIO( post_data )
file_wrapper = Wrapped_StringIO( post_data )
request = cherrypy.server.request( ( u"127.0.0.1", 1234 ), u"127.0.0.5" )
response = request.run( "POST %s HTTP/1.0" % str( http_path ), headers = headers, rfile = file_wrapper )

137
controller/test/Test_files.py
View File

@ -1,11 +1,14 @@
import types
import cherrypy
from StringIO import StringIO
from Test_controller import Test_controller
from model.Notebook import Notebook
from model.Note import Note
from model.User import User
from model.Invite import Invite
from model.File import File
from controller.Notebooks import Access_error
from controller.Files import Upload_file
class Test_files( Test_controller ):
@ -26,6 +29,37 @@ class Test_files( Test_controller ):
self.session_id = None
self.filename = "file.png"
self.file_data = "foobar\x07`-=[]\;',./~!@#$%^&*()_+{}|:\"<>?" * 100
self.content_type = "image/png"
# make Upload_file deal in fake files rather than actually using the filesystem
Upload_file.fake_files = {} # map of filename to fake file object
@staticmethod
def open_file( file_id, mode = None ):
fake_file = Upload_file.fake_files.get( Upload_file.make_server_filename( file_id ) )
if fake_file:
return fake_file
if mode not in ( "w", "w+" ):
raise IOError()
fake_file = StringIO()
Upload_file.fake_files[ file_id ] = fake_file
return fake_file
@staticmethod
def delete_file( file_id ):
fake_file = Upload_file.fake_files.get( Upload_file.make_server_filename( file_id ) )
if fake_file is None:
raise IOError()
del( fake_file[ file_id ] )
Upload_file.open_file = open_file
Upload_file.delete_file = delete_file
Upload_file.close = lambda self: None
self.make_users()
self.make_notebooks()
@ -60,51 +94,58 @@ class Test_files( Test_controller ):
self.anonymous = User.create( self.database.next_id( User ), u"anonymous" )
self.database.save( self.anonymous, commit = False )
def test_download( self ):
raise NotImplementedError()
def test_upload_page( self ):
self.login()
result = self.http_get(
"/files/upload_page?notebook_id=%s&note_id=%s" % ( self.notebook.object_id, self.note.object_id ),
session_id = self.session_id,
)
assert result.get( u"notebook_id" ) == self.notebook.object_id
assert result.get( u"note_id" ) == self.note.object_id
assert result.get( u"file_id" )
def test_upload_page_without_login( self ):
result = self.http_get(
"/files/upload_page?notebook_id=%s&note_id=%s" % ( self.notebook.object_id, self.note.object_id ),
)
assert u"access" in result.get( u"error" )
def test_upload( self ):
self.login()
file_id = "22"
result = self.http_upload(
"/files/upload",
"/files/upload?file_id=%s" % file_id,
dict(
notebook_id = self.notebook.object_id,
note_id = self.note.object_id,
),
filename = self.filename,
file_data = self.file_data,
content_type = self.content_type,
session_id = self.session_id,
)
gen = result[ u"body" ]
assert isinstance( gen, types.GeneratorType )
assert u"error" not in result
assert u"script" not in result
tick_count = 0
tick_done = False
# assert that the file metadata was actually stored in the database
db_file = self.database.load( File, file_id )
assert db_file
assert db_file.notebook_id == self.notebook.object_id
assert db_file.note_id == self.note.object_id
assert db_file.filename == self.filename
assert db_file.size_bytes == len( self.file_data )
assert db_file.content_type == self.content_type
try:
for piece in gen:
if u"tick(" in piece:
tick_count += 1
if u"tick(1.0)" in piece:
tick_done = True
# during this unit test, full session info isn't available, so swallow an expected
# exception about session_storage
except AttributeError, exc:
if u"session_storage" not in str( exc ):
raise exc
# assert that the progress bar is moving, and then completes
assert tick_count >= 2
assert tick_done
# TODO: assert that the uploaded file actually got stored somewhere
# assert that the file data was actually stored
assert Upload_file.open_file( file_id ).read() == self.file_data
def test_upload_without_login( self ):
result = self.http_upload(
@ -219,7 +260,57 @@ class Test_files( Test_controller ):
self.assert_streaming_error( result )
def test_upload_over_quota( self ):
raise NotImplementError()
raise NotImplementedError()
def test_progress( self ):
raise NotImplementedError()
self.login()
result = self.http_upload(
"/files/progress",
dict(
notebook_id = self.notebook.object_id,
note_id = self.note.object_id,
),
filename = self.filename,
file_data = self.file_data,
session_id = self.session_id,
)
gen = result[ u"body" ]
assert isinstance( gen, types.GeneratorType )
tick_count = 0
tick_done = False
try:
for piece in gen:
if u"tick(" in piece:
tick_count += 1
if u"tick(1.0)" in piece:
tick_done = True
# during this unit test, full session info isn't available, so swallow an expected
# exception about session_storage
except AttributeError, exc:
if u"session_storage" not in str( exc ):
raise exc
# assert that the progress bar is moving, and then completes
assert tick_count >= 2
assert tick_done
def test_stats( self ):
raise NotImplementedError()
def test_delete( self ):
raise NotImplementedError()
def test_rename( self ):
raise NotImplementedError()
def test_purge_unused( self ):
raise NotImplementedError()
def login( self ):
result = self.http_post( "/users/login", dict(

7
controller/test/Test_users.py
View File

@ -3419,6 +3419,13 @@ class Test_users( Test_controller ):
assert u"Thank you" in result[ u"notes" ][ 0 ].contents
assert u"confirmation" in result[ u"notes" ][ 0 ].contents
def test_rate_plan( self ):
plan_index = 1
rate_plan = cherrypy.root.users.rate_plan( plan_index )
assert rate_plan
assert rate_plan == self.settings[ u"global" ][ u"luminotes.rate_plans" ][ plan_index ]
def login( self ):
result = self.http_post( "/users/login", dict(
username = self.username,