KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use status_ready method in localstack
Best Python code snippet using localstack_python
test_mail_parser.py
Source:test_mail_parser.py 
1import unittest2 as unittest2from donomo.archive.models import *3from donomo.archive.service import mail_parser4from donomo.archive import operations5from django.test.utils import setup_test_environment6from donomo.archive.utils import sqs7import os8import time9# ----------------------------------------------------------------------------10class TestConversationNewsletter(unittest.TestCase):11 """12 """13 user = None14 processor = None15 16 # ------------------------------------------------------------------------17 @staticmethod18 def _init_infra():19 operations.initialize_infrastructure()20 sqs.clear_all_messages()21 # ------------------------------------------------------------------------22 @staticmethod23 def _init_user():24 try:25 return manager(User).get(username = 'testuser' )26 except User.DoesNotExist:27 # Make sure we have a test user28 return manager(User).create_user(29 username = 'testuser',30 email = 'testuser@donomo.com',31 password = manager(User).make_random_password())32 33 def _init_processor(self):34 MODULE_NAME = 'mail_parser'35 36 DEFAULT_INPUTS = [ AssetClass.UPLOAD ]37 DEFAULT_OUTPUTS = [ AssetClass.MESSAGE_PART ]38 DEFAULT_ACCEPTED_MIME_TYPES = [ MimeType.MAIL ]39 40 return operations.initialize_processor(41 MODULE_NAME,42 DEFAULT_INPUTS,43 DEFAULT_OUTPUTS,44 DEFAULT_ACCEPTED_MIME_TYPES ) [0]45 46 def _enqueue_message(self, message_file, accountname, label, flags):47 new_item = operations.create_asset_from_file(48 file_name = message_file,49 owner = self.user,50 producer = self.processor,51 asset_class = AssetClass.UPLOAD,52 child_number = 0,53 mime_type = MimeType.MAIL )54 new_item.orig_file_name += ',ACCOUNT=%s' % accountname55 new_item.orig_file_name += ',LABEL=%s' % label56 new_item.orig_file_name += ',FLAGS='57 for f in flags:58 new_item.orig_file_name += ',' + f59 new_item.save()60 operations.publish_work_item(new_item)61 def _process_queue(self):62 while True:63 work_item = operations.retrieve_work_item(max_wait_time = 5)64 if work_item:65 mail_parser.handle_work_item(self.processor, work_item)66 operations.close_work_item(work_item, True)67 else:68 break69 70 def setUp(self):71 self._init_infra();72 self.user = self._init_user();73 self.processor = self._init_processor();74 self.message_dir = os.path.join(os.path.dirname(__file__), 'data')75 76 def tearDown(self):77 sqs.clear_all_messages()78 79 # ------------------------------------------------------------------------80 def test_init(self):81 self.assert_( self.user is not None )82 self.assert_( self.processor is not None )83 def test_simple_message(self):84 """85 Adding a message should produce 1 message object and 1 conversation86 """87 MessageAggregate.objects.all().delete()88 Message.objects.all().delete()89 MessageRule.objects.all().delete()90 91 self._enqueue_message(os.path.join(self.message_dir, 'simple_message.eml'), 'testuser@donomo.com', 'label1', 'S,F')92 self._process_queue();93 94 self.assertEqual(Message.objects.all().count(), 1)95 self.assertEqual(MessageAggregate.objects.all().count(), 1)96 def test_conversation(self): 97 """98 3 messages in the same thread, added in reverse order. This means 'References' will be unresolved.99 Once all 2 are processed there should be 3 more messages and 1 more conversation.100 Also tests unicode of the subject line101 """102 original_message_count = Message.objects.all().count()103 original_aggregate_count = MessageAggregate.objects.all().count()104 105 self._enqueue_message(os.path.join(self.message_dir, 'conversation_message3.eml'), 'testuser@donomo.com', 'label2', 'S,F')106 self._enqueue_message(os.path.join(self.message_dir, 'conversation_message2.eml'), 'testuser@donomo.com', 'label2', 'S,F')107 self._enqueue_message(os.path.join(self.message_dir, 'conversation_message1.eml'), 'testuser@donomo.com', 'label2', 'S,F')108 self._process_queue();109 110 self.assertEqual(Message.objects.all().count() - original_message_count, 3 )111 self.assertEqual(MessageAggregate.objects.filter(status = MessageAggregate.STATUS_READY).count() - original_aggregate_count, 1)112 113 new_conversation = MessageAggregate.objects.get(status = MessageAggregate.STATUS_READY, owner = self.user, tags__label = 'label2')114 115 self.assertEqual(new_conversation.creator.type, MessageRule.CONVERSATION)116 self.assertEqual(new_conversation.messages.all().count(), 3)117 def test_newsletter(self):118 """119 - receive a message (new conversation is created)120 - flag it as newsletter (conversation is replaced with a newsletter)121 - receive a message of the same type (they are all put in the newsletter)122 - forward one newsletter (that message ends up in a new conversation and still sits in the newsletter aggregate)123 - receive another newsletter124 """125 MessageAggregate.objects.all().delete()126 Message.objects.all().delete()127 MessageRule.objects.all().delete()128 original_message_count = Message.objects.all().count()129 original_conversation_count = MessageAggregate.objects.filter(status = MessageAggregate.STATUS_READY, creator__type = MessageRule.CONVERSATION).count()130 original_newsletter_count = MessageAggregate.objects.filter(status = MessageAggregate.STATUS_READY, creator__type = MessageRule.NEWSLETTER).count()131 self._enqueue_message(os.path.join(self.message_dir, 'newsletter_message1.eml'), 'testuser@donomo.com', 'label3', 'S,F')132 self._process_queue();133 # new conversation is created134 new_conversation = MessageAggregate.objects.get(owner = self.user, tags__label = 'label3')135 self.assertEqual(new_conversation.creator.type, MessageRule.CONVERSATION)136 self.assertEqual(MessageAggregate.objects.filter(status = MessageAggregate.STATUS_READY, creator__type = MessageRule.CONVERSATION).count(), original_conversation_count + 1)137 138 # another message arrives, another conversation created139 self._enqueue_message(os.path.join(self.message_dir, 'newsletter_message2.eml'), 'testuser@donomo.com', 'label3', 'S,F')140 self._process_queue();141 # got two conversations142 self.assertEqual(MessageAggregate.objects.filter(status = MessageAggregate.STATUS_READY, creator__type = MessageRule.CONVERSATION).count(), original_conversation_count + 2)143 # add a newsletter rule for the sender144 rule = MessageRule(type = MessageRule.NEWSLETTER, owner = self.user, sender_address = new_conversation.messages.all()[0].sender_address)145 rule.save()146 operations.apply_message_rule(rule)147 148 # conversations got replaced by a newsletter149 self.assertEqual(MessageAggregate.objects.filter(status = MessageAggregate.STATUS_READY, creator__type = MessageRule.CONVERSATION).count(), original_conversation_count)150 self.assertEqual(MessageAggregate.objects.filter(status = MessageAggregate.STATUS_READY, creator__type = MessageRule.NEWSLETTER).count(), 1 + original_newsletter_count)151 152 # a newsletter was forwarded153 self._enqueue_message(os.path.join(self.message_dir, 'newsletter_message3.eml'), 'testuser@donomo.com', 'label3', 'S,F')154 self._process_queue();155 # a new conversation is created156 self.assertEqual(MessageAggregate.objects.filter(status = MessageAggregate.STATUS_READY, creator__type = MessageRule.CONVERSATION).count(), 1 + original_conversation_count)157 158 # a new newsletter was received159 self._enqueue_message(os.path.join(self.message_dir, 'newsletter_message4.eml'), 'testuser@donomo.com', 'label3', 'S,F')160 self._process_queue();161 162 # the new newsletter message goes into the existing aggregate163 # the conversation is still present, with 2 messages164 self.assertEqual(MessageAggregate.objects.filter(status = MessageAggregate.STATUS_READY, creator__type = MessageRule.NEWSLETTER).count(), 1 + original_newsletter_count)165 newsletter = MessageAggregate.objects.get(status = MessageAggregate.STATUS_READY, owner = self.user, creator__type = MessageRule.NEWSLETTER, tags__label = 'label3')166 self.assertEqual(newsletter.messages.all().count(), 3)167 conversation = MessageAggregate.objects.get(status = MessageAggregate.STATUS_READY, owner = self.user, creator__type = MessageRule.CONVERSATION, tags__label = 'label3')168 self.assertEqual(conversation.messages.all().count(), 2)169 def test_newsletter_delayed_rule_application(self):170 """171 - receive several messages from the same source (newsletter) and one message with a forward172 - apply a newsletter rule173 - check there's a single newsletter aggreate and a single conversation (for the forward)174 - check the names of conversation and newsletter175 """176 MessageAggregate.objects.all().delete()177 Message.objects.all().delete()178 MessageRule.objects.all().delete()179 180 original_message_count = Message.objects.all().count()181 original_conversation_count = MessageAggregate.objects.filter(status = MessageAggregate.STATUS_READY, creator__type = MessageRule.CONVERSATION).count()182 original_newsletter_count = MessageAggregate.objects.filter(status = MessageAggregate.STATUS_READY, creator__type = MessageRule.NEWSLETTER).count()183 self._enqueue_message(os.path.join(self.message_dir, 'newsletter_message1.eml'), 'testuser@donomo.com', 'label4', 'S,F')184 self._enqueue_message(os.path.join(self.message_dir, 'newsletter_message2.eml'), 'testuser@donomo.com', 'label4', 'S,F')185 self._enqueue_message(os.path.join(self.message_dir, 'newsletter_message3.eml'), 'testuser@donomo.com', 'label4', 'S,F')186 self._enqueue_message(os.path.join(self.message_dir, 'newsletter_message4.eml'), 'testuser@donomo.com', 'label4', 'S,F')187 self._process_queue();188 new_conversation = MessageAggregate.objects.filter(status = MessageAggregate.STATUS_READY, owner = self.user, tags__label = 'label4')[0]189 190 # add a newsletter rule for the sender191 rule = MessageRule(type = MessageRule.NEWSLETTER, owner = self.user, sender_address = new_conversation.messages.all()[0].sender_address)192 rule.save()193 time.sleep(2)194 operations.apply_message_rule(rule)195 time.sleep(2)196 # the new newsletter message goes into the existing aggregate197 # the conversation is still present, with 2 messages198 self.assertEqual(MessageAggregate.objects.filter(status = MessageAggregate.STATUS_READY, 199 owner = self.user, 200 creator__type = MessageRule.NEWSLETTER).count(), 201 1 + original_newsletter_count)202 newsletter = MessageAggregate.objects.get(status = MessageAggregate.STATUS_READY, 203 owner = self.user, 204 creator__type = MessageRule.NEWSLETTER, 205 tags__label = 'label4')206 self.assertEqual(newsletter.messages.all().count(), 3)207 conversation = MessageAggregate.objects.get(status = MessageAggregate.STATUS_READY, owner = self.user, creator__type = MessageRule.CONVERSATION, tags__label = 'label4')208 self.assertEqual(conversation.messages.all().count(), 2)209 210 # Aggregate's name is the subject name of the first message211 self.assertEqual(conversation.name, 'New on The Economist online - 22nd October 2010')212 self.assertEqual(newsletter.name, 'New on The Economist online - 18th October 2010')213# ----------------------------------------------------------------------------214if __name__ == '__main__':215 setup_test_environment()...
test_with.py
Source:test_with.py
1#!/usr/bin/env python2# test_ctxman.py - unit test for connection and cursor used as context manager3#4# Copyright (C) 2012 Daniele Varrazzo <daniele.varrazzo@gmail.com>5#6# psycopg2 is free software: you can redistribute it and/or modify it7# under the terms of the GNU Lesser General Public License as published8# by the Free Software Foundation, either version 3 of the License, or9# (at your option) any later version.10#11# In addition, as a special exception, the copyright holders give12# permission to link this program with the OpenSSL library (or with13# modified versions of OpenSSL that use the same license as OpenSSL),14# and distribute linked combinations including the two.15#16# You must obey the GNU Lesser General Public License in all respects for17# all of the code used other than OpenSSL.18#19# psycopg2 is distributed in the hope that it will be useful, but WITHOUT20# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or21# FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public22# License for more details.23from __future__ import with_statement24import psycopg225import psycopg2.extensions as ext26from testutils import unittest, ConnectingTestCase27class WithTestCase(ConnectingTestCase):28 def setUp(self):29 ConnectingTestCase.setUp(self)30 curs = self.conn.cursor()31 try:32 curs.execute("delete from test_with")33 self.conn.commit()34 except psycopg2.ProgrammingError:35 # assume table doesn't exist36 self.conn.rollback()37 curs.execute("create table test_with (id integer primary key)")38 self.conn.commit()39class WithConnectionTestCase(WithTestCase):40 def test_with_ok(self):41 with self.conn as conn:42 self.assert_(self.conn is conn)43 self.assertEqual(conn.status, ext.STATUS_READY)44 curs = conn.cursor()45 curs.execute("insert into test_with values (1)")46 self.assertEqual(conn.status, ext.STATUS_BEGIN)47 self.assertEqual(self.conn.status, ext.STATUS_READY)48 self.assert_(not self.conn.closed)49 curs = self.conn.cursor()50 curs.execute("select * from test_with")51 self.assertEqual(curs.fetchall(), [(1,)])52 def test_with_connect_idiom(self):53 with self.connect() as conn:54 self.assertEqual(conn.status, ext.STATUS_READY)55 curs = conn.cursor()56 curs.execute("insert into test_with values (2)")57 self.assertEqual(conn.status, ext.STATUS_BEGIN)58 self.assertEqual(self.conn.status, ext.STATUS_READY)59 self.assert_(not self.conn.closed)60 curs = self.conn.cursor()61 curs.execute("select * from test_with")62 self.assertEqual(curs.fetchall(), [(2,)])63 def test_with_error_db(self):64 def f():65 with self.conn as conn:66 curs = conn.cursor()67 curs.execute("insert into test_with values ('a')")68 self.assertRaises(psycopg2.DataError, f)69 self.assertEqual(self.conn.status, ext.STATUS_READY)70 self.assert_(not self.conn.closed)71 curs = self.conn.cursor()72 curs.execute("select * from test_with")73 self.assertEqual(curs.fetchall(), [])74 def test_with_error_python(self):75 def f():76 with self.conn as conn:77 curs = conn.cursor()78 curs.execute("insert into test_with values (3)")79 1/080 self.assertRaises(ZeroDivisionError, f)81 self.assertEqual(self.conn.status, ext.STATUS_READY)82 self.assert_(not self.conn.closed)83 curs = self.conn.cursor()84 curs.execute("select * from test_with")85 self.assertEqual(curs.fetchall(), [])86 def test_with_closed(self):87 def f():88 with self.conn:89 pass90 self.conn.close()91 self.assertRaises(psycopg2.InterfaceError, f)92 def test_subclass_commit(self):93 commits = []94 class MyConn(ext.connection):95 def commit(self):96 commits.append(None)97 super(MyConn, self).commit()98 with self.connect(connection_factory=MyConn) as conn:99 curs = conn.cursor()100 curs.execute("insert into test_with values (10)")101 self.assertEqual(conn.status, ext.STATUS_READY)102 self.assert_(commits)103 curs = self.conn.cursor()104 curs.execute("select * from test_with")105 self.assertEqual(curs.fetchall(), [(10,)])106 def test_subclass_rollback(self):107 rollbacks = []108 class MyConn(ext.connection):109 def rollback(self):110 rollbacks.append(None)111 super(MyConn, self).rollback()112 try:113 with self.connect(connection_factory=MyConn) as conn:114 curs = conn.cursor()115 curs.execute("insert into test_with values (11)")116 1/0117 except ZeroDivisionError:118 pass119 else:120 self.assert_("exception not raised")121 self.assertEqual(conn.status, ext.STATUS_READY)122 self.assert_(rollbacks)123 curs = conn.cursor()124 curs.execute("select * from test_with")125 self.assertEqual(curs.fetchall(), [])126class WithCursorTestCase(WithTestCase):127 def test_with_ok(self):128 with self.conn as conn:129 with conn.cursor() as curs:130 curs.execute("insert into test_with values (4)")131 self.assert_(not curs.closed)132 self.assertEqual(self.conn.status, ext.STATUS_BEGIN)133 self.assert_(curs.closed)134 self.assertEqual(self.conn.status, ext.STATUS_READY)135 self.assert_(not self.conn.closed)136 curs = self.conn.cursor()137 curs.execute("select * from test_with")138 self.assertEqual(curs.fetchall(), [(4,)])139 def test_with_error(self):140 try:141 with self.conn as conn:142 with conn.cursor() as curs:143 curs.execute("insert into test_with values (5)")144 1/0145 except ZeroDivisionError:146 pass147 self.assertEqual(self.conn.status, ext.STATUS_READY)148 self.assert_(not self.conn.closed)149 self.assert_(curs.closed)150 curs = self.conn.cursor()151 curs.execute("select * from test_with")152 self.assertEqual(curs.fetchall(), [])153 def test_subclass(self):154 closes = []155 class MyCurs(ext.cursor):156 def close(self):157 closes.append(None)158 super(MyCurs, self).close()159 with self.conn.cursor(cursor_factory=MyCurs) as curs:160 self.assert_(isinstance(curs, MyCurs))161 self.assert_(curs.closed)162 self.assert_(closes)163 def test_exception_swallow(self):164 # bug #262: __exit__ calls cur.close() that hides the exception165 # with another error.166 try:167 with self.conn as conn:168 with conn.cursor('named') as cur:169 cur.execute("select 1/0")170 cur.fetchone()171 except psycopg2.DataError, e:172 self.assertEqual(e.pgcode, '22012')173 else:174 self.fail("where is my exception?")175def test_suite():176 return unittest.TestLoader().loadTestsFromName(__name__)177if __name__ == "__main__":...
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 localstack 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?
