Implemented server-side portion of revoke_invite(), so now the UI for revoking invites functions properly.

NEWS

@@ -1,4 +1,4 @@
-1.1.0: December ??, 2007
+1.0.4: December ??, 2007
  * Ability to invite people to view your notebook.
  * When the web browser is resized, all notes are automatically resized as well.
  * Fixed note focusing in Safari.

controller/Users.py

@@ -677,7 +677,7 @@ class Users( object ):
     @param user_id: id of current logged-in user (if any), determined by @grab_user_id
     @rtype: json dict
     @return: { 'message': message, 'invites': invites }
-    @raise Password_reset_error: an error occured when sending the password reset email
+    @raise Invite_error: an error occured when sending the invite
     @raise Validation_error: one of the arguments is invalid
     @raise Access_error: user_id doesn't have owner-level notebook access to send an invite or
                          doesn't have a rate plan supporting notebook collaboration
@@ -690,7 +690,7 @@ class Users( object ):
     if not self.check_access( user_id, notebook_id, read_write = True, owner = True ):
       raise Access_error()
 
-    # this feature requires a rate plan above basic
+    # except for viewer-only invites, this feature requires a rate plan above basic
     user = self.__database.load( User, user_id )
     if user is None or user.username is None or ( user.rate_plan == 0 and access != u"viewer" ):
       raise Access_error()
@@ -771,3 +771,43 @@ class Users( object ):
         message = u"%s invitations have been sent." % email_count,
         invites = invites,
       )
+
+  @expose( view = Json )
+  @grab_user_id
+  @validate(
+    notebook_id = Valid_id(),
+    invite_id = Valid_id(),
+    user_id = Valid_id( none_okay = True ),
+  )
+  def revoke_invite( self, notebook_id, invite_id, user_id = None ):
+    """
+    Revoke the invite's access to the given notebook.
+
+    @type notebook_id: unicode
+    @param notebook_id: id of the notebook that the invitation is for
+    @type invite_id: unicode
+    @param invite_id: id of the invite to revoke
+    @type user_id: unicode
+    @param user_id: id of current logged-in user (if any), determined by @grab_user_id
+    @rtype: json dict
+    @return: { 'message': message, 'invites': invites }
+    @raise Validation_error: one of the arguments is invalid
+    @raise Access_error: user_id doesn't have owner-level notebook access to revoke an invite
+    """
+    if not self.check_access( user_id, notebook_id, read_write = True, owner = True ):
+      raise Access_error()
+
+    invite = self.__database.load( Invite, invite_id )
+    if not invite or not invite.email_address or invite.notebook_id != notebook_id:
+      raise Access_error()
+
+    self.__database.execute( invite.sql_revoke_user_access(), commit = False )
+    self.__database.execute( invite.sql_revoke_invites(), commit = False )
+    self.__database.commit()
+
+    invites = self.__database.select_many( Invite, Invite.sql_load_notebook_invites( notebook_id ) )
+
+    return dict(
+      message = u"Notebook access for %s has been revoked." % invite.email_address,
+      invites = invites,
+    )

controller/test/Test_controller.py

@@ -244,6 +244,34 @@ class Test_controller( object ):
     Invite.sql_load_notebook_invites = staticmethod( lambda notebook_id:
       lambda database: sql_load_notebook_invites( notebook_id, database ) )
 
+    def sql_revoke_user_access( self, database ):
+      invites = []
+
+      for ( user_id, notebook_infos ) in database.user_notebook.items():
+        for ( index, ( notebook_id, read_write, owner ) ) in enumerate( notebook_infos ):
+          if notebook_id != self.notebook_id: continue
+          for ( object_id, obj_list ) in database.objects.items():
+            obj = obj_list[ -1 ]
+            if isinstance( obj, Invite ) and obj.notebook_id == self.notebook_id and \
+               obj.email_address == self.email_address:
+              del( database.user_notebook[ user_id ][ index ] )
+
+    Invite.sql_revoke_user_access = lambda self: \
+      lambda database: sql_revoke_user_access( self, database )
+
+    def sql_revoke_invites( self, database ):
+      invites = []
+
+      for ( object_id, obj_list ) in database.objects.items():
+        obj = obj_list[ -1 ]
+        if isinstance( obj, Invite ) and obj.notebook_id == self.notebook_id and \
+           obj.email_address == self.email_address:
+          del( database.objects[ object_id ] )
+
+    Invite.sql_revoke_invites = lambda self: \
+      lambda database: sql_revoke_invites( self, database )
+
+
   def setUp( self ):
     from controller.Root import Root
     cherrypy.lowercase_api = True

controller/test/Test_users.py

@@ -1345,6 +1345,144 @@ class Test_users( Test_controller ):
     assert result[ u"error" ]
     assert "access" in result[ u"error" ]
 
+  def test_revoke_invite( self ):
+    # trick revoke_invite() into using a fake SMTP server
+    Stub_smtp.reset()
+    smtplib.SMTP = Stub_smtp
+    self.login()
+
+    self.user.rate_plan = 1
+    self.database.save( self.user )
+
+    email_addresses_list = [ u"foo@example.com" ]
+    email_addresses = email_addresses_list[ 0 ]
+
+    result = self.http_post( "/users/send_invites", dict(
+      notebook_id = self.notebooks[ 0 ].object_id,
+      email_addresses = email_addresses,
+      access = u"viewer",
+      invite_button = u"send invites",
+    ), session_id = self.session_id )
+    
+    assert len( result[ u"invites" ] ) == 1
+    matches = self.INVITE_LINK_PATTERN.search( smtplib.SMTP.message )
+    invite_id = matches.group( 2 )
+
+    result = self.http_post( "/users/revoke_invite", dict(
+      notebook_id = self.notebooks[ 0 ].object_id,
+      invite_id = invite_id,
+    ), session_id = self.session_id )
+
+    assert result[ u"message" ]
+    assert len( result[ u"invites" ] ) == 0
+
+  def test_revoke_invite_multiple( self ):
+    Stub_smtp.reset()
+    smtplib.SMTP = Stub_smtp
+    self.login()
+
+    self.user.rate_plan = 1
+    self.database.save( self.user )
+
+    email_addresses_list = [ u"foo@example.com", u"bar@example.com", u"foo@example.com" ]
+    email_addresses = u" ".join( email_addresses_list )
+
+    result = self.http_post( "/users/send_invites", dict(
+      notebook_id = self.notebooks[ 0 ].object_id,
+      email_addresses = email_addresses,
+      access = u"viewer",
+      invite_button = u"send invites",
+    ), session_id = self.session_id )
+    
+    assert len( result[ u"invites" ] ) == 2
+    ( from_address, to_addresses, message ) = smtplib.SMTP.emails[ 0 ]
+    matches = self.INVITE_LINK_PATTERN.search( message )
+    invite_id = matches.group( 2 )
+
+    result = self.http_post( "/users/revoke_invite", dict(
+      notebook_id = self.notebooks[ 0 ].object_id,
+      invite_id = invite_id,
+    ), session_id = self.session_id )
+
+    assert result[ u"message" ]
+    assert len( result[ u"invites" ] ) == 1
+    assert result[ u"invites" ][ 0 ].email_address == email_addresses_list[ 1 ]
+
+  def test_revoke_invite_without_login( self ):
+    Stub_smtp.reset()
+    smtplib.SMTP = Stub_smtp
+
+    # login to send the invites, but don't send the logged-in session id for revoke_invite() below
+    self.login()
+
+    self.user.rate_plan = 1
+    self.database.save( self.user )
+
+    email_addresses_list = [ u"foo@example.com" ]
+    email_addresses = email_addresses_list[ 0 ]
+
+    result = self.http_post( "/users/send_invites", dict(
+      notebook_id = self.notebooks[ 0 ].object_id,
+      email_addresses = email_addresses,
+      access = u"viewer",
+      invite_button = u"send invites",
+    ), session_id = self.session_id )
+    
+    assert len( result[ u"invites" ] ) == 1
+    matches = self.INVITE_LINK_PATTERN.search( smtplib.SMTP.message )
+    invite_id = matches.group( 2 )
+
+    result = self.http_post( "/users/revoke_invite", dict(
+      notebook_id = self.notebooks[ 0 ].object_id,
+      invite_id = invite_id,
+    ) )
+
+    assert result[ u"error" ]
+    assert "access" in result[ u"error" ]
+
+  def test_revoke_invite_unknown( self ):
+    self.login()
+
+    invite_id = u"unknowninviteid"
+
+    result = self.http_post( "/users/revoke_invite", dict(
+      notebook_id = self.notebooks[ 0 ].object_id,
+      invite_id = invite_id,
+    ), session_id = self.session_id )
+
+    assert result[ u"error" ]
+    assert "access" in result[ u"error" ]
+
+  def test_revoke_invite_for_incorrect_notebook( self ):
+    Stub_smtp.reset()
+    smtplib.SMTP = Stub_smtp
+    self.login()
+
+    self.user.rate_plan = 1
+    self.database.save( self.user )
+
+    email_addresses_list = [ u"foo@example.com" ]
+    email_addresses = email_addresses_list[ 0 ]
+
+    result = self.http_post( "/users/send_invites", dict(
+      notebook_id = self.notebooks[ 0 ].object_id,
+      email_addresses = email_addresses,
+      access = u"viewer",
+      invite_button = u"send invites",
+    ), session_id = self.session_id )
+    
+    assert len( result[ u"invites" ] ) == 1
+    matches = self.INVITE_LINK_PATTERN.search( smtplib.SMTP.message )
+    invite_id = matches.group( 2 )
+
+    result = self.http_post( "/users/revoke_invite", dict(
+      notebook_id = self.notebooks[ 1 ].object_id,
+      invite_id = invite_id,
+    ), session_id = self.session_id )
+
+    assert result[ u"error" ]
+    assert "access" in result[ u"error" ]
+
   def login( self ):
     result = self.http_post( "/users/login", dict(
       username = self.username,

model/Invite.py

@@ -103,6 +103,15 @@ class Invite( Persistent ):
     return "select id, revision, from_user_id, notebook_id, email_address, read_write, owner, redeemed_user_id from invite " + \
            "where id in ( select max( id ) from invite where notebook_id = %s group by email_address ) order by email_address;" % quote( notebook_id )
 
+  def sql_revoke_user_access( self ):
+    return "delete from user_notebook where notebook_id = %s and user_id in " % quote( self.__notebook_id ) + \
+           "( select redeemed_user_id from invite where notebook_id = %s and email_address = %s );" % \
+           ( quote( self.__notebook_id ), quote( self.__email_address ) )
+
+  def sql_revoke_invites( self ):
+    return "delete from invite where notebook_id = %s and email_address = %s;" % \
+           ( quote( self.__notebook_id ), quote( self.__email_address ) )
+
   def to_dict( self ):
     d = Persistent.to_dict( self )
     d.update( dict(