KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use mail method in avocado
Best Python code snippet using avocado_python
test_mail_features.py
Source:test_mail_features.py 
...165 self.assertEqual(composer.composition_mode, 'comment')166 self.assertEqual(composer.model, 'mail.channel')167 self.assertEqual(composer.subject, 'Re: %s' % self.group_pigs.name)168 self.assertEqual(composer.record_name, self.group_pigs.name)169 composer.send_mail()170 message = self.group_pigs.message_ids[0]171 composer = self.env['mail.compose.message'].with_context({172 'default_composition_mode': 'comment',173 'default_res_id': self.group_pigs.id,174 'default_parent_id': message.id175 }).sudo(self.user_employee).create({})176 self.assertEqual(composer.model, 'mail.channel')177 self.assertEqual(composer.res_id, self.group_pigs.id)178 self.assertEqual(composer.parent_id, message)179 self.assertEqual(composer.subject, 'Re: %s' % self.group_pigs.name)180 # TODO: test attachments ?181 @mute_logger('openerp.addons.mail.models.mail_mail')182 def test_message_compose_mass_mail(self):183 composer = self.env['mail.compose.message'].with_context({184 'default_composition_mode': 'mass_mail',185 'default_model': 'mail.channel',186 'default_res_id': False,187 'active_ids': [self.group_pigs.id, self.group_public.id]188 }).sudo(self.user_employee).create({189 'subject': 'Testing ${object.name}',190 'body': '<p>${object.description}</p>',191 'partner_ids': [(4, self.partner_1.id), (4, self.partner_2.id)]192 })193 composer.with_context({194 'default_res_id': -1,195 'active_ids': [self.group_pigs.id, self.group_public.id]196 }).send_mail()197 # check mail_mail198 mails = self.env['mail.mail'].search([('subject', 'ilike', 'Testing')])199 for mail in mails:200 self.assertEqual(mail.recipient_ids, self.partner_1 | self.partner_2,201 'compose wizard: mail_mail mass mailing: mail.mail in mass mail incorrect recipients')202 # check message on group_pigs203 message1 = self.group_pigs.message_ids[0]204 self.assertEqual(message1.subject, 'Testing %s' % self.group_pigs.name)205 self.assertEqual(message1.body, '<p>%s</p>' % self.group_pigs.description)206 # check message on group_public207 message1 = self.group_public.message_ids[0]208 self.assertEqual(message1.subject, 'Testing %s' % self.group_public.name)209 self.assertEqual(message1.body, '<p>%s</p>' % self.group_public.description)210 # # Test: Pigs and Bird did receive their message211 # # check logged messages212 # message1 = group_pigs.message_ids[0]213 # message2 = group_bird.message_ids[0]214 # # Test: Pigs and Bird did receive their message215 # messages = self.MailMessage.search([], limit=2)216 # mail = self.MailMail.search([('mail_message_id', '=', message2.id)], limit=1)217 # self.assertTrue(mail, 'message_send: mail.mail message should have in processing mail queue')218 # #check mass mail state...219 # mails = self.MailMail.search([('state', '=', 'exception')])220 # self.assertNotIn(mail.id, mails.ids, 'compose wizard: Mail sending Failed!!')221 # self.assertIn(message1.id, messages.ids, 'compose wizard: Pigs did not receive its mass mailing message')222 # self.assertIn(message2.id, messages.ids, 'compose wizard: Bird did not receive its mass mailing message')223 # check followers ?224 # Test: mail.channel followers: author not added as follower in mass mail mode225 # self.assertEqual(set(group_pigs.message_follower_ids.ids), set([self.partner_admin_id, p_b.id, p_c.id, p_d.id]),226 # 'compose wizard: mail_post_autofollow and mail_create_nosubscribe context keys not correctly taken into account')227 # self.assertEqual(set(group_bird.message_follower_ids.ids), set([self.partner_admin_id]),228 # 'compose wizard: mail_post_autofollow and mail_create_nosubscribe context keys not correctly taken into account')229 @mute_logger('openerp.addons.mail.models.mail_mail')230 def test_message_compose_mass_mail_active_domain(self):231 self.env['mail.compose.message'].with_context({232 'default_composition_mode': 'mass_mail',233 'default_model': 'mail.channel',234 'active_ids': [self.group_pigs.id],235 'active_domain': [('name', 'in', ['%s' % self.group_pigs.name, '%s' % self.group_public.name])],236 }).sudo(self.user_employee).create({237 'subject': 'From Composer Test',238 'body': '${object.description}',239 }).send_mail()240 self.assertEqual(self.group_pigs.message_ids[0].subject, 'From Composer Test')...
test_mail_message.py
Source:test_mail_message.py
1# -*- coding: utf-8 -*-2from .common import TestMail3from openerp.exceptions import AccessError4from openerp.exceptions import except_orm5from openerp.tools import mute_logger6class TestMailMessage(TestMail):7 def setUp(self):8 super(TestMailMessage, self).setUp()9 self.group_private = self.env['mail.channel'].with_context({10 'mail_create_nolog': True,11 'mail_create_nosubscribe': True12 }).create({13 'name': 'Private',14 'public': 'private'}15 ).with_context({'mail_create_nosubscribe': False})16 self.message = self.env['mail.message'].create({17 'body': 'My Body',18 'model': 'mail.channel',19 'res_id': self.group_private.id,20 })21 def test_mail_message_values_basic(self):22 self.env['ir.config_parameter'].search([('key', '=', 'mail.catchall.domain')]).unlink()23 msg = self.env['mail.message'].sudo(self.user_employee).create({24 'reply_to': 'test.reply@example.com',25 'email_from': 'test.from@example.com',26 })27 self.assertIn('-private', msg.message_id, 'mail_message: message_id for a void message should be a "private" one')28 self.assertEqual(msg.reply_to, 'test.reply@example.com')29 self.assertEqual(msg.email_from, 'test.from@example.com')30 def test_mail_message_values_default(self):31 self.env['ir.config_parameter'].search([('key', '=', 'mail.catchall.domain')]).unlink()32 msg = self.env['mail.message'].sudo(self.user_employee).create({})33 self.assertIn('-private', msg.message_id, 'mail_message: message_id for a void message should be a "private" one')34 self.assertEqual(msg.reply_to, '%s <%s>' % (self.user_employee.name, self.user_employee.email))35 self.assertEqual(msg.email_from, '%s <%s>' % (self.user_employee.name, self.user_employee.email))36 def test_mail_message_values_alias(self):37 alias_domain = 'example.com'38 self.env['ir.config_parameter'].set_param('mail.catchall.domain', alias_domain)39 self.env['ir.config_parameter'].search([('key', '=', 'mail.catchall.alias')]).unlink()40 msg = self.env['mail.message'].sudo(self.user_employee).create({})41 self.assertIn('-private', msg.message_id, 'mail_message: message_id for a void message should be a "private" one')42 self.assertEqual(msg.reply_to, '%s <%s@%s>' % (self.user_employee.name, self.user_employee.alias_name, alias_domain))43 self.assertEqual(msg.email_from, '%s <%s@%s>' % (self.user_employee.name, self.user_employee.alias_name, alias_domain))44 def test_mail_message_values_alias_catchall(self):45 alias_domain = 'example.com'46 alias_catchall = 'pokemon'47 self.env['ir.config_parameter'].set_param('mail.catchall.domain', alias_domain)48 self.env['ir.config_parameter'].set_param('mail.catchall.alias', alias_catchall)49 msg = self.env['mail.message'].sudo(self.user_employee).create({})50 self.assertIn('-private', msg.message_id, 'mail_message: message_id for a void message should be a "private" one')51 self.assertEqual(msg.reply_to, '%s <%s@%s>' % (self.env.user.company_id.name, alias_catchall, alias_domain))52 self.assertEqual(msg.email_from, '%s <%s@%s>' % (self.user_employee.name, self.user_employee.alias_name, alias_domain))53 def test_mail_message_values_document_no_alias(self):54 self.env['ir.config_parameter'].search([('key', '=', 'mail.catchall.domain')]).unlink()55 msg = self.env['mail.message'].sudo(self.user_employee).create({56 'model': 'mail.channel',57 'res_id': self.group_pigs.id58 })59 self.assertIn('-openerp-%d-mail.channel' % self.group_pigs.id, msg.message_id, 'mail_message: message_id for a void message should be a "private" one')60 self.assertEqual(msg.reply_to, '%s <%s>' % (self.user_employee.name, self.user_employee.email))61 self.assertEqual(msg.email_from, '%s <%s>' % (self.user_employee.name, self.user_employee.email))62 def test_mail_message_values_document_alias(self):63 alias_domain = 'example.com'64 self.env['ir.config_parameter'].set_param('mail.catchall.domain', alias_domain)65 self.env['ir.config_parameter'].search([('key', '=', 'mail.catchall.alias')]).unlink()66 msg = self.env['mail.message'].sudo(self.user_employee).create({67 'model': 'mail.channel',68 'res_id': self.group_pigs.id69 })70 self.assertIn('-openerp-%d-mail.channel' % self.group_pigs.id, msg.message_id, 'mail_message: message_id for a void message should be a "private" one')71 self.assertEqual(msg.reply_to, '%s %s <%s@%s>' % (self.env.user.company_id.name, self.group_pigs.name, self.group_pigs.alias_name, alias_domain))72 self.assertEqual(msg.email_from, '%s <%s@%s>' % (self.user_employee.name, self.user_employee.alias_name, alias_domain))73 def test_mail_message_values_document_alias_catchall(self):74 alias_domain = 'example.com'75 alias_catchall = 'pokemon'76 self.env['ir.config_parameter'].set_param('mail.catchall.domain', alias_domain)77 self.env['ir.config_parameter'].set_param('mail.catchall.alias', alias_catchall)78 msg = self.env['mail.message'].sudo(self.user_employee).create({79 'model': 'mail.channel',80 'res_id': self.group_pigs.id81 })82 self.assertIn('-openerp-%d-mail.channel' % self.group_pigs.id, msg.message_id, 'mail_message: message_id for a void message should be a "private" one')83 self.assertEqual(msg.reply_to, '%s %s <%s@%s>' % (self.env.user.company_id.name, self.group_pigs.name, self.group_pigs.alias_name, alias_domain))84 self.assertEqual(msg.email_from, '%s <%s@%s>' % (self.user_employee.name, self.user_employee.alias_name, alias_domain))85 def test_mail_message_values_no_auto_thread(self):86 msg = self.env['mail.message'].sudo(self.user_employee).create({87 'model': 'mail.channel',88 'res_id': self.group_pigs.id,89 'no_auto_thread': True,90 })91 self.assertIn('reply_to', msg.message_id)92 self.assertNotIn('mail.channel', msg.message_id)93 self.assertNotIn('-%d-' % self.group_pigs.id, msg.message_id)94 @mute_logger('openerp.addons.mail.models.mail_mail')95 def test_mail_message_access_search(self):96 # Data: various author_ids, partner_ids, documents97 msg1 = self.env['mail.message'].create({98 'subject': '_Test', 'body': 'A', 'subtype_id': self.ref('mail.mt_comment')})99 msg2 = self.env['mail.message'].create({100 'subject': '_Test', 'body': 'A+B', 'subtype_id': self.ref('mail.mt_comment'),101 'partner_ids': [(6, 0, [self.user_public.partner_id.id])]})102 msg3 = self.env['mail.message'].create({103 'subject': '_Test', 'body': 'A Pigs', 'subtype_id': False,104 'model': 'mail.channel', 'res_id': self.group_pigs.id})105 msg4 = self.env['mail.message'].create({106 'subject': '_Test', 'body': 'A+P Pigs', 'subtype_id': self.ref('mail.mt_comment'),107 'model': 'mail.channel', 'res_id': self.group_pigs.id,108 'partner_ids': [(6, 0, [self.user_public.partner_id.id])]})109 msg5 = self.env['mail.message'].create({110 'subject': '_Test', 'body': 'A+E Pigs', 'subtype_id': self.ref('mail.mt_comment'),111 'model': 'mail.channel', 'res_id': self.group_pigs.id,112 'partner_ids': [(6, 0, [self.user_employee.partner_id.id])]})113 msg6 = self.env['mail.message'].create({114 'subject': '_Test', 'body': 'A Birds', 'subtype_id': self.ref('mail.mt_comment'),115 'model': 'mail.channel', 'res_id': self.group_private.id})116 msg7 = self.env['mail.message'].sudo(self.user_employee).create({117 'subject': '_Test', 'body': 'B', 'subtype_id': self.ref('mail.mt_comment')})118 msg8 = self.env['mail.message'].sudo(self.user_employee).create({119 'subject': '_Test', 'body': 'B+E', 'subtype_id': self.ref('mail.mt_comment'),120 'partner_ids': [(6, 0, [self.user_employee.partner_id.id])]})121 # Test: Public: 2 messages (recipient)122 messages = self.env['mail.message'].sudo(self.user_public).search([('subject', 'like', '_Test')])123 self.assertEqual(messages, msg2 | msg4)124 # Test: Employee: 3 messages on Pigs Raoul can read (employee can read group with default values)125 messages = self.env['mail.message'].sudo(self.user_employee).search([('subject', 'like', '_Test'), ('body', 'ilike', 'A')])126 self.assertEqual(messages, msg3 | msg4 | msg5)127 # Test: Raoul: 3 messages on Pigs Raoul can read (employee can read group with default values), 0 on Birds (private group) + 2 messages as author128 messages = self.env['mail.message'].sudo(self.user_employee).search([('subject', 'like', '_Test')])129 self.assertEqual(messages, msg3 | msg4 | msg5 | msg7 | msg8)130 # Test: Admin: all messages131 messages = self.env['mail.message'].search([('subject', 'like', '_Test')])132 self.assertEqual(messages, msg1 | msg2 | msg3 | msg4 | msg5 | msg6 | msg7 | msg8)133 # Test: Portal: 0 (no access to groups, not recipient)134 messages = self.env['mail.message'].sudo(self.user_portal).search([('subject', 'like', '_Test')])135 self.assertFalse(messages)136 # Test: Portal: 2 messages (public group with a subtype)137 self.group_pigs.write({'public': 'public'})138 messages = self.env['mail.message'].sudo(self.user_portal).search([('subject', 'like', '_Test')])139 self.assertEqual(messages, msg4 | msg5)140 @mute_logger('openerp.addons.base.ir.ir_model', 'openerp.models')141 def test_mail_message_access_read_crash(self):142 # TODO: Change the except_orm to Warning ( Because here it's call check_access_rule143 # which still generate exception in except_orm.So we need to change all144 # except_orm to warning in mail module.)145 with self.assertRaises(except_orm):146 self.message.sudo(self.user_employee).read()147 @mute_logger('openerp.models')148 def test_mail_message_access_read_crash_portal(self):149 with self.assertRaises(except_orm):150 self.message.sudo(self.user_portal).read(['body', 'message_type', 'subtype_id'])151 def test_mail_message_access_read_ok_portal(self):152 self.message.write({'subtype_id': self.ref('mail.mt_comment'), 'res_id': self.group_public.id})153 self.message.sudo(self.user_portal).read(['body', 'message_type', 'subtype_id'])154 def test_mail_message_access_read_notification(self):155 attachment = self.env['ir.attachment'].create({156 'datas': 'My attachment'.encode('base64'),157 'name': 'doc.txt',158 'datas_fname': 'doc.txt'})159 # attach the attachment to the message160 self.message.write({'attachment_ids': [(4, attachment.id)]})161 self.message.write({'partner_ids': [(4, self.user_employee.partner_id.id)]})162 self.message.sudo(self.user_employee).read()163 # Test: Bert has access to attachment, ok because he can read message164 attachment.sudo(self.user_employee).read(['name', 'datas'])165 def test_mail_message_access_read_author(self):166 self.message.write({'author_id': self.user_employee.partner_id.id})167 self.message.sudo(self.user_employee).read()168 def test_mail_message_access_read_doc(self):169 self.message.write({'model': 'mail.channel', 'res_id': self.group_public.id})170 # Test: Bert reads the message, ok because linked to a doc he is allowed to read171 self.message.sudo(self.user_employee).read()172 @mute_logger('openerp.addons.base.ir.ir_model')173 def test_mail_message_access_create_crash_public(self):174 # Do: Bert creates a message on Pigs -> ko, no creation rights175 with self.assertRaises(AccessError):176 self.env['mail.message'].sudo(self.user_public).create({'model': 'mail.channel', 'res_id': self.group_pigs.id, 'body': 'Test'})177 # Do: Bert create a message on Jobs -> ko, no creation rights178 with self.assertRaises(AccessError):179 self.env['mail.message'].sudo(self.user_public).create({'model': 'mail.channel', 'res_id': self.group_public.id, 'body': 'Test'})180 @mute_logger('openerp.models')181 def test_mail_message_access_create_crash(self):182 # Do: Bert create a private message -> ko, no creation rights183 with self.assertRaises(except_orm):184 self.env['mail.message'].sudo(self.user_employee).create({'model': 'mail.channel', 'res_id': self.group_private.id, 'body': 'Test'})185 @mute_logger('openerp.models')186 def test_mail_message_access_create_doc(self):187 # TODO Change the except_orm to Warning188 Message = self.env['mail.message'].sudo(self.user_employee)189 # Do: Raoul creates a message on Jobs -> ok, write access to the related document190 Message.create({'model': 'mail.channel', 'res_id': self.group_public.id, 'body': 'Test'})191 # Do: Raoul creates a message on Priv -> ko, no write access to the related document192 with self.assertRaises(except_orm):193 Message.create({'model': 'mail.channel', 'res_id': self.group_private.id, 'body': 'Test'})194 def test_mail_message_access_create_private(self):195 self.env['mail.message'].sudo(self.user_employee).create({'body': 'Test'})196 def test_mail_message_access_create_reply(self):197 self.message.write({'partner_ids': [(4, self.user_employee.partner_id.id)]})198 self.env['mail.message'].sudo(self.user_employee).create({'model': 'mail.channel', 'res_id': self.group_private.id, 'body': 'Test', 'parent_id': self.message.id})199 def test_message_set_star(self):200 msg = self.group_pigs.message_post(body='My Body', subject='1')201 msg_emp = self.env['mail.message'].sudo(self.user_employee).browse(msg.id)202 # Admin set as starred203 msg.set_message_starred(True)204 self.assertTrue(msg.starred)205 # Employee set as starred206 msg_emp.set_message_starred(True)207 self.assertTrue(msg_emp.starred)208 # Do: Admin unstars msg209 msg.set_message_starred(False)210 self.assertFalse(msg.starred)...
service.py
Source:service.py
1#!/usr/bin/python2.42#3# Copyright 2008 Google Inc. All Rights Reserved.4#5# Licensed under the Apache License, Version 2.0 (the "License");6# you may not use this file except in compliance with the License.7# You may obtain a copy of the License at8#9# http://www.apache.org/licenses/LICENSE-2.010#11# Unless required by applicable law or agreed to in writing, software12# distributed under the License is distributed on an "AS IS" BASIS,13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.14# See the License for the specific language governing permissions and15# limitations under the License.16"""Contains the methods to import mail via Google Apps Email Migration API.17 MigrationService: Provides methods to import mail.18"""19__author__ = ('google-apps-apis@googlegroups.com',20 'pti@google.com (Prashant Tiwari)')21import base6422import threading23import time24from atom.service import deprecation25from gdata.apps import migration26from gdata.apps.migration import MailEntryProperties27import gdata.apps.service28import gdata.service29API_VER = '2.0'30class MigrationService(gdata.apps.service.AppsService):31 """Client for the EMAPI migration service. Use either ImportMail to import32 one message at a time, or AddMailEntry and ImportMultipleMails to import a33 bunch of messages at a time.34 """35 def __init__(self, email=None, password=None, domain=None, source=None,36 server='apps-apis.google.com', additional_headers=None):37 gdata.apps.service.AppsService.__init__(38 self, email=email, password=password, domain=domain, source=source,39 server=server, additional_headers=additional_headers)40 self.mail_batch = migration.BatchMailEventFeed()41 self.mail_entries = []42 self.exceptions = 043 def _BaseURL(self):44 return '/a/feeds/migration/%s/%s' % (API_VER, self.domain)45 def ImportMail(self, user_name, mail_message, mail_item_properties,46 mail_labels):47 """Imports a single mail message.48 Args:49 user_name: The username to import messages to.50 mail_message: An RFC822 format email message.51 mail_item_properties: A list of Gmail properties to apply to the message.52 mail_labels: A list of labels to apply to the message.53 Returns:54 A MailEntry representing the successfully imported message.55 Raises:56 AppsForYourDomainException: An error occurred importing the message.57 """58 uri = '%s/%s/mail' % (self._BaseURL(), user_name)59 mail_entry = migration.MailEntry()60 mail_entry.rfc822_msg = migration.Rfc822Msg(text=(base64.b64encode(61 mail_message)))62 mail_entry.rfc822_msg.encoding = 'base64'63 mail_entry.mail_item_property = map(64 lambda x: migration.MailItemProperty(value=x), mail_item_properties)65 mail_entry.label = map(lambda x: migration.Label(label_name=x),66 mail_labels)67 try:68 return migration.MailEntryFromString(str(self.Post(mail_entry, uri)))69 except gdata.service.RequestError, e:70 # Store the number of failed imports when importing several at a time 71 self.exceptions += 172 raise gdata.apps.service.AppsForYourDomainException(e.args[0])73 def AddBatchEntry(self, mail_message, mail_item_properties,74 mail_labels):75 """Adds a message to the current batch that you later will submit.76 77 Deprecated, use AddMailEntry instead78 Args:79 mail_message: An RFC822 format email message.80 mail_item_properties: A list of Gmail properties to apply to the message.81 mail_labels: A list of labels to apply to the message.82 Returns:83 The length of the MailEntry representing the message.84 """85 deprecation("calling deprecated method AddBatchEntry")86 mail_entry = migration.BatchMailEntry()87 mail_entry.rfc822_msg = migration.Rfc822Msg(text=(base64.b64encode(88 mail_message)))89 mail_entry.rfc822_msg.encoding = 'base64'90 mail_entry.mail_item_property = map(91 lambda x: migration.MailItemProperty(value=x), mail_item_properties)92 mail_entry.label = map(lambda x: migration.Label(label_name=x),93 mail_labels)94 self.mail_batch.AddBatchEntry(mail_entry)95 return len(str(mail_entry))96 def SubmitBatch(self, user_name):97 """Sends all the mail items you have added to the batch to the server.98 99 Deprecated, use ImportMultipleMails instead100 Args:101 user_name: The username to import messages to.102 Returns:103 An HTTPResponse from the web service call.104 Raises:105 AppsForYourDomainException: An error occurred importing the batch.106 """107 deprecation("calling deprecated method SubmitBatch")108 uri = '%s/%s/mail/batch' % (self._BaseURL(), user_name)109 try:110 self.result = self.Post(self.mail_batch, uri,111 converter=migration.BatchMailEventFeedFromString)112 except gdata.service.RequestError, e:113 raise gdata.apps.service.AppsForYourDomainException(e.args[0])114 self.mail_batch = migration.BatchMailEventFeed()115 return self.result116 def AddMailEntry(self, mail_message, mail_item_properties=None,117 mail_labels=None, identifier=None):118 """Prepares a list of mail messages to import using ImportMultipleMails.119 120 Args:121 mail_message: An RFC822 format email message as a string.122 mail_item_properties: List of Gmail properties to apply to the123 message.124 mail_labels: List of Gmail labels to apply to the message.125 identifier: The optional file identifier string126 127 Returns:128 The number of email messages to be imported.129 """130 mail_entry_properties = MailEntryProperties(131 mail_message=mail_message,132 mail_item_properties=mail_item_properties,133 mail_labels=mail_labels,134 identifier=identifier)135 self.mail_entries.append(mail_entry_properties)136 return len(self.mail_entries)137 def ImportMultipleMails(self, user_name, threads_per_batch=20):138 """Launches separate threads to import every message added by AddMailEntry.139 140 Args:141 user_name: The user account name to import messages to.142 threads_per_batch: Number of messages to import at a time.143 144 Returns:145 The number of email messages that were successfully migrated.146 147 Raises:148 Exception: An error occurred while importing mails.149 """150 num_entries = len(self.mail_entries)151 if not num_entries:152 return 0153 threads = []154 for mail_entry_properties in self.mail_entries:155 t = threading.Thread(name=mail_entry_properties.identifier,156 target=self.ImportMail,157 args=(user_name, mail_entry_properties.mail_message,158 mail_entry_properties.mail_item_properties,159 mail_entry_properties.mail_labels))160 threads.append(t)161 try:162 # Determine the number of batches needed with threads_per_batch in each163 batches = num_entries / threads_per_batch + (164 0 if num_entries % threads_per_batch == 0 else 1)165 batch_min = 0166 # Start the threads, one batch at a time167 for batch in range(batches):168 batch_max = ((batch + 1) * threads_per_batch169 if (batch + 1) * threads_per_batch < num_entries170 else num_entries)171 for i in range(batch_min, batch_max):172 threads[i].start()173 time.sleep(1)174 for i in range(batch_min, batch_max):175 threads[i].join()176 batch_min = batch_max177 self.mail_entries = []178 except Exception, e:179 raise Exception(e.args[0])180 else:...
mail_mail.py
Source:mail_mail.py
1# -*- coding: utf-8 -*-2# Part of Odoo. See LICENSE file for full copyright and licensing details.3import re4import urlparse5import werkzeug.urls6from openerp import tools7from openerp import SUPERUSER_ID8from openerp.osv import osv, fields9URL_REGEX = r'(\bhref=[\'"]([^\'"]+)[\'"])'10class MailMail(osv.Model):11 """Add the mass mailing campaign data to mail"""12 _name = 'mail.mail'13 _inherit = ['mail.mail']14 _columns = {15 'mailing_id': fields.many2one('mail.mass_mailing', 'Mass Mailing'),16 'statistics_ids': fields.one2many(17 'mail.mail.statistics', 'mail_mail_id',18 string='Statistics',19 ),20 }21 def create(self, cr, uid, values, context=None):22 """ Override mail_mail creation to create an entry in mail.mail.statistics """23 # TDE note: should be after 'all values computed', to have values (FIXME after merging other branch holding create refactoring)24 mail_id = super(MailMail, self).create(cr, uid, values, context=context)25 if values.get('statistics_ids'):26 mail = self.browse(cr, SUPERUSER_ID, mail_id, context=context)27 for stat in mail.statistics_ids:28 self.pool['mail.mail.statistics'].write(cr, uid, [stat.id], {'message_id': mail.message_id, 'state': 'outgoing'}, context=context)29 return mail_id30 def _get_tracking_url(self, cr, uid, mail, partner=None, context=None):31 base_url = self.pool.get('ir.config_parameter').get_param(cr, uid, 'web.base.url')32 track_url = urlparse.urljoin(33 base_url, 'mail/track/%(mail_id)s/blank.gif?%(params)s' % {34 'mail_id': mail.id,35 'params': werkzeug.url_encode({'db': cr.dbname})36 }37 )38 return '<img src="%s" alt=""/>' % track_url39 def _get_unsubscribe_url(self, cr, uid, mail, email_to, context=None):40 base_url = self.pool.get('ir.config_parameter').get_param(cr, uid, 'web.base.url')41 url = urlparse.urljoin(42 base_url, 'mail/mailing/%(mailing_id)s/unsubscribe?%(params)s' % {43 'mailing_id': mail.mailing_id.id,44 'params': werkzeug.url_encode({'db': cr.dbname, 'res_id': mail.res_id, 'email': email_to})45 }46 )47 return url48 def send_get_mail_body(self, cr, uid, ids, partner=None, context=None):49 """ Override to add the tracking URL to the body and to add50 Statistic_id in shorted urls """51 # TDE: temporary addition (mail was parameter) due to semi-new-API52 body = super(MailMail, self).send_get_mail_body(cr, uid, ids, partner=partner, context=context)53 mail = self.browse(cr, uid, ids[0], context=context)54 links_blacklist = ['/unsubscribe_from_list']55 if mail.mailing_id and body and mail.statistics_ids:56 for match in re.findall(URL_REGEX, mail.body_html):57 href = match[0]58 url = match[1]59 if not [s for s in links_blacklist if s in href]:60 new_href = href.replace(url, url + '/m/' + str(mail.statistics_ids[0].id))61 body = body.replace(href, new_href)62 # prepend <base> tag for images using absolute urls63 domain = self.pool.get("ir.config_parameter").get_param(cr, uid, "web.base.url", context=context)64 base = "<base href='%s'>" % domain65 body = tools.append_content_to_html(base, body, plaintext=False, container_tag='div')66 # resolve relative image url to absolute for outlook.com67 def _sub_relative2absolute(match):68 return match.group(1) + urlparse.urljoin(domain, match.group(2))69 body = re.sub('(<img(?=\s)[^>]*\ssrc=")(/[^/][^"]+)', _sub_relative2absolute, body)70 body = re.sub(r'(<[^>]+\bstyle="[^"]+\burl\(\'?)(/[^/\'][^\'")]+)', _sub_relative2absolute, body)71 # generate tracking URL72 if mail.statistics_ids:73 tracking_url = self._get_tracking_url(cr, uid, mail, partner, context=context)74 if tracking_url:75 body = tools.append_content_to_html(body, tracking_url, plaintext=False, container_tag='div')76 return body77 def send_get_email_dict(self, cr, uid, ids, partner=None, context=None):78 # TDE: temporary addition (mail was parameter) due to semi-new-API79 res = super(MailMail, self).send_get_email_dict(cr, uid, ids, partner, context=context)80 mail = self.browse(cr, uid, ids[0], context=context)81 base_url = self.pool.get('ir.config_parameter').get_param(cr, uid, 'web.base.url')82 if mail.mailing_id and res.get('body') and res.get('email_to'):83 emails = tools.email_split(res.get('email_to')[0])84 email_to = emails and emails[0] or False85 unsubscribe_url= self._get_unsubscribe_url(cr, uid, mail, email_to, context=context)86 link_to_replace = base_url+'/unsubscribe_from_list'87 if link_to_replace in res['body']:88 res['body'] = res['body'].replace(link_to_replace, unsubscribe_url if unsubscribe_url else '#')89 return res90 def _postprocess_sent_message(self, cr, uid, mail, context=None, mail_sent=True):91 if mail_sent is True and mail.statistics_ids:92 self.pool['mail.mail.statistics'].write(cr, uid, [s.id for s in mail.statistics_ids], {'sent': fields.datetime.now(), 'exception': False}, context=context)93 elif mail_sent is False and mail.statistics_ids:94 self.pool['mail.mail.statistics'].write(cr, uid, [s.id for s in mail.statistics_ids], {'exception': fields.datetime.now()}, context=context)...
Automation Testing Tutorials
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run avocado automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
