mirror of
				https://github.com/zulip/zulip.git
				synced 2025-11-03 21:43:21 +00:00 
			
		
		
		
	email_mirror: Add tests for messages with attachments.
This commit is contained in:
		
				
					committed by
					
						
						Tim Abbott
					
				
			
			
				
	
			
			
			
						parent
						
							ede0991ef5
						
					
				
				
					commit
					f892ef1d3f
				
			@@ -17,6 +17,7 @@ from zerver.models import (
 | 
				
			|||||||
    get_display_recipient,
 | 
					    get_display_recipient,
 | 
				
			||||||
    get_realm,
 | 
					    get_realm,
 | 
				
			||||||
    get_stream,
 | 
					    get_stream,
 | 
				
			||||||
 | 
					    get_system_bot,
 | 
				
			||||||
    Recipient,
 | 
					    Recipient,
 | 
				
			||||||
)
 | 
					)
 | 
				
			||||||
 | 
					
 | 
				
			||||||
@@ -307,6 +308,65 @@ class TestStreamEmailMessagesSuccess(ZulipTestCase):
 | 
				
			|||||||
        self.assertEqual(get_display_recipient(message.recipient), stream.name)
 | 
					        self.assertEqual(get_display_recipient(message.recipient), stream.name)
 | 
				
			||||||
        self.assertEqual(message.topic_name(), incoming_valid_message['Subject'])
 | 
					        self.assertEqual(message.topic_name(), incoming_valid_message['Subject'])
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					class TestEmailMirrorMessagesWithAttachments(ZulipTestCase):
 | 
				
			||||||
 | 
					    def test_message_with_valid_attachment(self) -> None:
 | 
				
			||||||
 | 
					        user_profile = self.example_user('hamlet')
 | 
				
			||||||
 | 
					        self.login(user_profile.email)
 | 
				
			||||||
 | 
					        self.subscribe(user_profile, "Denmark")
 | 
				
			||||||
 | 
					        stream = get_stream("Denmark", user_profile.realm)
 | 
				
			||||||
 | 
					        stream_to_address = encode_email_address(stream)
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					        incoming_valid_message = MIMEMultipart()
 | 
				
			||||||
 | 
					        text_msg = MIMEText("Test body")
 | 
				
			||||||
 | 
					        incoming_valid_message.attach(text_msg)
 | 
				
			||||||
 | 
					        with open(os.path.join(settings.DEPLOY_ROOT, "static/images/default-avatar.png"), 'rb') as f:
 | 
				
			||||||
 | 
					            image_bytes = f.read()
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					        attachment_msg = MIMEImage(image_bytes)
 | 
				
			||||||
 | 
					        attachment_msg.add_header('Content-Disposition', 'attachment', filename="image.png")
 | 
				
			||||||
 | 
					        incoming_valid_message.attach(attachment_msg)
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					        incoming_valid_message['Subject'] = 'TestStreamEmailMessages Subject'
 | 
				
			||||||
 | 
					        incoming_valid_message['From'] = self.example_email('hamlet')
 | 
				
			||||||
 | 
					        incoming_valid_message['To'] = stream_to_address
 | 
				
			||||||
 | 
					        incoming_valid_message['Reply-to'] = self.example_email('othello')
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					        with mock.patch('zerver.lib.email_mirror.upload_message_file',
 | 
				
			||||||
 | 
					                        return_value='https://test_url') as upload_message_file:
 | 
				
			||||||
 | 
					            process_message(incoming_valid_message)
 | 
				
			||||||
 | 
					            upload_message_file.assert_called_with('image.png', len(image_bytes),
 | 
				
			||||||
 | 
					                                                   'image/png', image_bytes,
 | 
				
			||||||
 | 
					                                                   get_system_bot(settings.EMAIL_GATEWAY_BOT),
 | 
				
			||||||
 | 
					                                                   target_realm=user_profile.realm)
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					        message = most_recent_message(user_profile)
 | 
				
			||||||
 | 
					        self.assertEqual(message.content, "Test body[image.png](https://test_url)")
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					    def test_message_with_invalid_attachment(self) -> None:
 | 
				
			||||||
 | 
					        user_profile = self.example_user('hamlet')
 | 
				
			||||||
 | 
					        self.login(user_profile.email)
 | 
				
			||||||
 | 
					        self.subscribe(user_profile, "Denmark")
 | 
				
			||||||
 | 
					        stream = get_stream("Denmark", user_profile.realm)
 | 
				
			||||||
 | 
					        stream_to_address = encode_email_address(stream)
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					        incoming_valid_message = MIMEMultipart()
 | 
				
			||||||
 | 
					        text_msg = MIMEText("Test body")
 | 
				
			||||||
 | 
					        incoming_valid_message.attach(text_msg)
 | 
				
			||||||
 | 
					        # Create an invalid attachment:
 | 
				
			||||||
 | 
					        attachment_msg = MIMEMultipart()
 | 
				
			||||||
 | 
					        attachment_msg.add_header('Content-Disposition', 'attachment', filename="some_attachment")
 | 
				
			||||||
 | 
					        incoming_valid_message.attach(attachment_msg)
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					        incoming_valid_message['Subject'] = 'TestStreamEmailMessages Subject'
 | 
				
			||||||
 | 
					        incoming_valid_message['From'] = self.example_email('hamlet')
 | 
				
			||||||
 | 
					        incoming_valid_message['To'] = stream_to_address
 | 
				
			||||||
 | 
					        incoming_valid_message['Reply-to'] = self.example_email('othello')
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					        with mock.patch('zerver.lib.email_mirror.logger.warning') as mock_warn:
 | 
				
			||||||
 | 
					            process_message(incoming_valid_message)
 | 
				
			||||||
 | 
					            mock_warn.assert_called_with("Payload is not bytes (invalid attachment %s in message from %s)." %
 | 
				
			||||||
 | 
					                                         ('some_attachment', self.example_email('hamlet')))
 | 
				
			||||||
 | 
					
 | 
				
			||||||
class TestStreamEmailMessagesEmptyBody(ZulipTestCase):
 | 
					class TestStreamEmailMessagesEmptyBody(ZulipTestCase):
 | 
				
			||||||
    def test_receive_stream_email_messages_empty_body(self) -> None:
 | 
					    def test_receive_stream_email_messages_empty_body(self) -> None:
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 
 | 
				
			|||||||
		Reference in New Issue
	
	Block a user