diff --git a/integration/P/all.py b/integration/P/all.py index 8fcba71..9bb1ce2 100644 --- a/integration/P/all.py +++ b/integration/P/all.py @@ -1,47 +1,51 @@ import stick import unittest from test_kolabd import TestKolabDaemon from test_wap_client_connect import TestWAPClientConnect from test_wap_client_users import TestWAPClientUserAdd from test_wap_client_resources import TestWAPClientResourceAdd from test_wap_client_form_value_select import TestWAPClientFormValueListOptions from test_wallace_forward import TestWallaceEmailForward from test_wallace_footer import TestWallaceFooter from test_wallace_nonascii import TestWallaceNonAscii from test_wallace_module_resources import TestWallaceModuleResources +from test_wallace_resource_invitation import TestWallaceResourcesInvitation +from test_wallace_invitationpolicy import TestWallaceInvitationpolicy from test_T249 import TestKolabCliT249 class TestPykolab(unittest.TestSuite): """ TestSuite that can be run with `nosetests all.py` or unittest.main() """ def __init__(self, *args, **kw): super(TestPykolab, self).__init__(*args, **kw) opts = stick.conf.cli_keywords loader = stick.KolabTestLoader(opts) # add tests in order of their dependencies self.addTests(loader.loadTestsFromTestCase(TestWAPClientConnect)) self.addTests(loader.loadTestsFromTestCase(TestWAPClientFormValueListOptions)) self.addTests(loader.loadTestsFromTestCase(TestWAPClientUserAdd)) self.addTests(loader.loadTestsFromTestCase(TestWAPClientResourceAdd)) # kolabd self.addTests(loader.loadTestsFromTestCase(TestKolabDaemon)) # wallace self.addTests(loader.loadTestsFromTestCase(TestWallaceEmailForward)) self.addTests(loader.loadTestsFromTestCase(TestWallaceFooter)) self.addTests(loader.loadTestsFromTestCase(TestWallaceNonAscii)) self.addTests(loader.loadTestsFromTestCase(TestWallaceModuleResources)) + self.addTests(loader.loadTestsFromTestCase(TestWallaceResourcesInvitation)) + self.addTests(loader.loadTestsFromTestCase(TestWallaceInvitationpolicy)) # kolab cli self.addTests(loader.loadTestsFromTestCase(TestKolabCliT249)) \ No newline at end of file diff --git a/integration/P/test_wallace_invitationpolicy.py b/integration/P/test_wallace_invitationpolicy.py new file mode 100644 index 0000000..e29daf1 --- /dev/null +++ b/integration/P/test_wallace_invitationpolicy.py @@ -0,0 +1,1185 @@ +import stick +import pykolab +import time +import smtplib +import email +import datetime +import pytz +import uuid +import kolabformat + +from pykolab.imap import IMAP +from wallace import module_resources + +from pykolab.translate import _ +from pykolab.xml import event_from_message +from pykolab.xml import todo_from_message +from pykolab.xml import participant_status_label +from pykolab.itip import events_from_message + +from test_wallace_invitationpolicy_data import * + +conf = pykolab.getConf() + + +class TestWallaceInvitationpolicy(stick.KolabIntegrationTest): + john = None + itip_reply_subject = None + + @classmethod + def setUpClass(self, *args, **kw): + # check if resource module is activated in kolab.conf + if not 'invitationpolicy' in conf.get_list('wallace', 'modules'): + raise Exception("'invitationpolicy' module required in wallace.modules from kolab.conf") + + # check if autoupdate is enabled in kolab.conf + if not conf.get('wallace', 'invitationpolicy_autoupdate_other_attendees_on_reply'): + raise Exception("wallace.invitationpolicy_autoupdate_other_attendees_on_reply required to be enabled in kolab.conf") + + # set language to default + pykolab.translate.setUserLanguage(conf.get('kolab','default_locale')) + + self.itip_reply_subject = _('"%(summary)s" has been %(status)s') + + # create users with their individual invitation policies + self.john = self.get_user_prop(self.require_user("John", "Doe", + kolabinvitationpolicy=['ACT_UPDATE_AND_NOTIFY','ACT_MANUAL'], + preferredlanguage='en_US' + )) + self.mark = self.get_user_prop(self.require_user("Mark", "German", + kolabinvitationpolicy=['ACT_ACCEPT','ACT_UPDATE_AND_NOTIFY'], + preferredlanguage='de_DE' + )) + self.jane = self.get_user_prop(self.require_user("Jane", "Manager", + kolabinvitationpolicy=['ACT_ACCEPT_IF_NO_CONFLICT','ACT_REJECT_IF_CONFLICT','TASK_ACCEPT','TASK_UPDATE_AND_NOTIFY','ACT_UPDATE'], + kolabdelegate=[self.mark['dn']], + preferredlanguage='en_US' + )) + self.jack = self.get_user_prop(self.require_user("Jack", "Tentative", + kolabinvitationpolicy=['ACT_TENTATIVE_IF_NO_CONFLICT','ALL_SAVE_TO_FOLDER','ACT_UPDATE'], + preferredlanguage='en_US' + )) + self.lucy = self.get_user_prop(self.require_user("Lucy", "Meyer", + kolabinvitationpolicy=['ALL_SAVE_AND_FORWARD','ACT_CANCEL_DELETE_AND_NOTIFY','ACT_UPDATE_AND_NOTIFY'], + preferredlanguage='en_US' + )) + self.bill = self.get_user_prop(self.require_user("Bill", "Mayor", + kolabinvitationpolicy=['ALL_SAVE_TO_FOLDER:lucy.meyer@example.org','ALL_REJECT'], + preferredlanguage='en_US' + )) + + self.external = { + 'displayname': 'Bob External', + 'mail': 'bob.external@gmail.com' + } + + # do synchronize + super(TestWallaceInvitationpolicy, self).setUpClass(*args, **kw) + + # create confidential calendar folder for jane + imap = IMAP() + imap.connect(domain=conf.get('kolab', 'primary_domain')) # sets self.domain + imap.user_mailbox_create_additional_folders(self.jane['mail'], { + 'Calendar/Confidential': { + 'annotations': { + '/shared/vendor/kolab/folder-type': "event", + '/private/vendor/kolab/folder-type': "event.confidential" + } + } + }) + # grant full access for Mark to Jane's calendar + imap.set_acl(imap.folder_quote(self.jane['kolabcalendarfolder']), self.mark['mail'], "lrswipkxtecda") + imap.disconnect() + + @classmethod + def get_user_prop(self, data): + (local, domain) = data['mail'].split('@') + data.update({ + 'displayname': "%(cn)s" % data, + 'mailbox': "user/%(mail)s" % data, + 'kolabcalendarfolder': "user/%s/Calendar@%s" % (local, domain), + 'kolabtasksfolder': "user/%s/Tasks@%s" % (local, domain), + 'kolabconfidentialcalendar': "user/%s/Calendar/Confidential@%s" % (local, domain), + }) + return dict([(k, v.encode('utf-8') if isinstance(v, unicode) else v) for (k,v) in data.iteritems()]) + + def send_message(self, itip_payload, to_addr, from_addr=None, method="REQUEST"): + if from_addr is None: + from_addr = self.john['mail'] + + smtp = smtplib.SMTP('localhost', 10026) + smtp.sendmail(from_addr, to_addr, mime_message % (to_addr, method, itip_payload)) + + def send_itip_invitation(self, attendee_email, start=None, allday=False, template=None, summary="test", sequence=0, partstat='NEEDS-ACTION', from_addr=None, uid=None, instance=None): + if start is None: + start = datetime.datetime.now() + + if uid is None: + uid = str(uuid.uuid4()) + + recurrence_id = '' + + if allday: + default_template = itip_allday + end = start + datetime.timedelta(days=1) + date_format = '%Y%m%d' + else: + end = start + datetime.timedelta(hours=4) + default_template = itip_invitation + date_format = '%Y%m%dT%H%M%S' + + if from_addr is not None: + if template: + template = template.replace("john.doe@example.org", from_addr) + else: + default_template = default_template.replace("john.doe@example.org", from_addr) + + if instance is not None: + recurrence_id = "\nRECURRENCE-ID;TZID=Europe/Berlin:" + instance.strftime(date_format) + + self.send_message((template if template is not None else default_template) % { + 'uid': uid, + 'recurrenceid': recurrence_id, + 'start': start.strftime(date_format), + 'end': end.strftime(date_format), + 'mailto': attendee_email, + 'summary': summary, + 'sequence': sequence, + 'partstat': partstat + }, + attendee_email, from_addr=from_addr) + + return uid + + def send_itip_update(self, attendee_email, uid, start=None, template=None, summary="test", sequence=1, partstat='ACCEPTED', instance=None): + if start is None: + start = datetime.datetime.now() + + end = start + datetime.timedelta(hours=4) + + date_format = '%Y%m%dT%H%M%S' + recurrence_id = '' + + if instance is not None: + recurrence_id = "\nRECURRENCE-ID;TZID=Europe/Berlin:" + instance.strftime(date_format) + + self.send_message((template if template is not None else itip_invitation) % { + 'uid': uid, + 'recurrenceid': recurrence_id, + 'start': start.strftime(date_format), + 'end': end.strftime(date_format), + 'mailto': attendee_email, + 'summary': summary, + 'sequence': sequence, + 'partstat': partstat + }, + attendee_email) + + return uid + + def send_itip_reply(self, uid, attendee_email, mailto, start=None, template=None, summary="test", sequence=0, partstat='ACCEPTED', instance=None): + if start is None: + start = datetime.datetime.now() + + end = start + datetime.timedelta(hours=4) + + date_format = '%Y%m%dT%H%M%S' + recurrence_id = '' + + if instance is not None: + recurrence_id = "\nRECURRENCE-ID;TZID=Europe/Berlin:" + instance.strftime(date_format) + + self.send_message((template if template is not None else itip_reply) % { + 'uid': uid, + 'recurrenceid': recurrence_id, + 'start': start.strftime(date_format), + 'end': end.strftime(date_format), + 'mailto': attendee_email, + 'organizer': mailto, + 'summary': summary, + 'sequence': sequence, + 'partstat': partstat + }, + mailto, + attendee_email, + method='REPLY') + + return uid + + def send_itip_cancel(self, attendee_email, uid, template=None, summary="test", sequence=1, instance=None, thisandfuture=False): + recurrence_id = '' + + if instance is not None: + recurrence_id = "\nRECURRENCE-ID;TZID=Europe/Berlin%s:%s" % ( + ';RANGE=THISANDFUTURE' if thisandfuture else '', + instance.strftime('%Y%m%dT%H%M%S') + ) + + self.send_message((template if template is not None else itip_cancellation) % { + 'uid': uid, + 'recurrenceid': recurrence_id, + 'mailto': attendee_email, + 'summary': summary, + 'sequence': sequence, + }, + attendee_email, + method='CANCEL') + + return uid + + def create_calendar_event(self, start=None, summary="test", sequence=0, user=None, attendees=None, folder=None, recurring=False, uid=None): + if start is None: + start = datetime.datetime.now(pytz.timezone("Europe/Berlin")) + if user is None: + user = self.john + if attendees is None: + attendees = [self.jane] + if folder is None: + folder = user['kolabcalendarfolder'] + + end = start + datetime.timedelta(hours=4) + + event = pykolab.xml.Event() + event.set_start(start) + event.set_end(end) + event.set_organizer(user['mail'], user['displayname']) + + if uid: + event.set_uid(uid) + + for attendee in attendees: + event.add_attendee(attendee['mail'], attendee['displayname'], role="REQ-PARTICIPANT", participant_status="NEEDS-ACTION", rsvp=True) + + event.set_summary(summary) + event.set_sequence(sequence) + + if recurring and isinstance(recurring, kolabformat.RecurrenceRule): + event.set_recurrence(rrule) + else: + rrule = kolabformat.RecurrenceRule() + rrule.setFrequency(kolabformat.RecurrenceRule.Daily) + rrule.setCount(10) + event.set_recurrence(rrule) + + # create event with attachment + vattach = event.get_attachments() + attachment = kolabformat.Attachment() + attachment.setLabel('attach.txt') + attachment.setData('This is a text attachment', 'text/plain') + vattach.append(attachment) + event.event.setAttachments(vattach) + + imap = IMAP() + imap.connect() + + mailbox = imap.folder_quote(folder) + imap.set_acl(mailbox, "cyrus-admin", "lrswipkxtecda") + imap.imap.m.select(mailbox) + + result = imap.imap.m.append( + mailbox, + None, + None, + event.to_message().as_string() + ) + + return event.get_uid() + + def create_task_assignment(self, due=None, summary="test", sequence=0, user=None, attendees=None): + if due is None: + due = datetime.datetime.now(pytz.timezone("Europe/Berlin")) + datetime.timedelta(days=2) + if user is None: + user = self.john + if attendees is None: + attendees = [self.jane] + + todo = pykolab.xml.Todo() + todo.set_due(due) + todo.set_organizer(user['mail'], user['displayname']) + + for attendee in attendees: + todo.add_attendee(attendee['mail'], attendee['displayname'], role="REQ-PARTICIPANT", participant_status="NEEDS-ACTION", rsvp=True) + + todo.set_summary(summary) + todo.set_sequence(sequence) + + imap = IMAP() + imap.connect() + + mailbox = imap.folder_quote(user['kolabtasksfolder']) + imap.set_acl(mailbox, "cyrus-admin", "lrswipkxtecda") + imap.imap.m.select(mailbox) + + result = imap.imap.m.append( + mailbox, + None, + None, + todo.to_message().as_string() + ) + + return todo.get_uid() + + def update_calendar_event(self, uid, start=None, summary=None, sequence=0, user=None): + if user is None: + user = self.john + + event = self.check_user_calendar_event(user['kolabcalendarfolder'], uid) + if event: + if start is not None: + event.set_start(start) + if summary is not None: + event.set_summary(summary) + if sequence is not None: + event.set_sequence(sequence) + + imap = IMAP() + imap.connect() + + mailbox = imap.folder_quote(user['kolabcalendarfolder']) + imap.set_acl(mailbox, "cyrus-admin", "lrswipkxtecda") + imap.imap.m.select(mailbox) + + return imap.imap.m.append( + mailbox, + None, + None, + event.to_message().as_string() + ) + + return False + + def check_message_received(self, subject, from_addr=None, mailbox=None): + if mailbox is None: + mailbox = self.john['mailbox'] + + imap = IMAP() + imap.connect() + + mailbox = imap.folder_quote(mailbox) + imap.set_acl(mailbox, "cyrus-admin", "lrs") + imap.imap.m.select(mailbox) + + found = None + retries = 15 + + while not found and retries > 0: + retries -= 1 + + typ, data = imap.imap.m.search(None, '(UNDELETED HEADER FROM "%s")' % (from_addr) if from_addr else 'UNDELETED') + for num in data[0].split(): + typ, msg = imap.imap.m.fetch(num, '(RFC822)') + message = email.message_from_string(msg[0][1]) + if message['Subject'] == subject: + found = message + break + + time.sleep(1) + + imap.disconnect() + + return found + + def check_user_calendar_event(self, mailbox, uid=None): + return self.check_user_imap_object(mailbox, uid) + + def check_user_imap_object(self, mailbox, uid=None, type='event'): + imap = IMAP() + imap.connect() + + mailbox = imap.folder_quote(mailbox) + imap.set_acl(mailbox, "cyrus-admin", "lrs") + imap.imap.m.select(mailbox) + + found = None + retries = 15 + + while not found and retries > 0: + retries -= 1 + + typ, data = imap.imap.m.search(None, '(UNDELETED HEADER SUBJECT "%s")' % (uid) if uid else '(UNDELETED HEADER X-Kolab-Type "application/x-vnd.kolab.' + type + '")') + for num in data[0].split(): + typ, data = imap.imap.m.fetch(num, '(RFC822)') + object_message = email.message_from_string(data[0][1]) + + # return matching UID or first event found + if uid and object_message['subject'] != uid: + continue + + if type == 'task': + found = todo_from_message(object_message) + else: + found = event_from_message(object_message) + + if found: + break + + time.sleep(1) + + return found + + def purge_mailbox(self, mailbox): + imap = IMAP() + imap.connect() + mailbox = imap.folder_quote(mailbox) + imap.set_acl(mailbox, "cyrus-admin", "lrwcdest") + imap.imap.m.select(mailbox) + + typ, data = imap.imap.m.search(None, 'ALL') + for num in data[0].split(): + imap.imap.m.store(num, '+FLAGS', '\\Deleted') + + imap.imap.m.expunge() + imap.disconnect() + + + def test_001_invite_accept_udate(self): + start = datetime.datetime(2014,8,13, 10,0,0) + uid = self.send_itip_invitation(self.jane['mail'], start) + + response = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('ACCEPTED') }, self.jane['mail']) + self.assertIsInstance(response, email.message.Message) + + event = self.check_user_calendar_event(self.jane['kolabcalendarfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + self.assertEqual(event.get_summary(), "test") + + # send update with the same sequence: no re-scheduling + self.send_itip_update(self.jane['mail'], uid, start, summary="test updated", sequence=0, partstat='ACCEPTED') + + time.sleep(10) + event = self.check_user_calendar_event(self.jane['kolabcalendarfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + self.assertEqual(event.get_summary(), "test updated") + self.assertEqual(event.get_attendee(self.jane['mail']).get_participant_status(), kolabformat.PartAccepted) + + + # @depends on test_001_invite_user + def test_002_invite_conflict_reject(self): + uid = self.send_itip_invitation(self.jane['mail'], datetime.datetime(2014,8,13, 11,0,0), summary="test2") + + response = self.check_message_received(self.itip_reply_subject % { 'summary':'test2', 'status':participant_status_label('DECLINED') }, self.jane['mail']) + self.assertIsInstance(response, email.message.Message) + + event = self.check_user_calendar_event(self.jane['kolabcalendarfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + self.assertEqual(event.get_summary(), "test2") + + + def test_003_invite_accept_tentative(self): + self.purge_mailbox(self.john['mailbox']) + + uid = self.send_itip_invitation(self.jack['mail'], datetime.datetime(2014,7,24, 8,0,0)) + + response = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('TENTATIVE') }, self.jack['mail']) + self.assertIsInstance(response, email.message.Message) + + + def test_004_copy_to_calendar(self): + self.purge_mailbox(self.john['mailbox']) + + self.send_itip_invitation(self.jack['mail'], datetime.datetime(2014,7,29, 8,0,0)) + response = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('TENTATIVE') }, self.jack['mail']) + self.assertIsInstance(response, email.message.Message) + + # send conflicting request to jack + uid = self.send_itip_invitation(self.jack['mail'], datetime.datetime(2014,7,29, 10,0,0), summary="test2") + response = self.check_message_received(self.itip_reply_subject % { 'summary':'test2', 'status':participant_status_label('DECLINED') }, self.jack['mail']) + self.assertEqual(response, None, "No reply expected") + + event = self.check_user_calendar_event(self.jack['kolabcalendarfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + self.assertEqual(event.get_summary(), "test2") + self.assertEqual(event.get_attendee(self.jack['mail']).get_participant_status(), kolabformat.PartNeedsAction) + + + def test_004_copy_to_calendar_and_forward(self): + uid = self.send_itip_invitation(self.lucy['mail'], datetime.datetime(2015,2,11, 14,0,0), summary="test forward") + response = self.check_message_received(self.itip_reply_subject % { 'summary':'test forward', 'status':participant_status_label('ACCEPTED') }, self.lucy['mail'], self.lucy['mailbox']) + self.assertEqual(response, None, "No reply expected") + + event = self.check_user_calendar_event(self.lucy['kolabcalendarfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + self.assertEqual(event.get_summary(), "test forward") + self.assertEqual(event.get_attendee(self.lucy['mail']).get_participant_status(), kolabformat.PartNeedsAction) + + # find original itip invitation in user's inbox + message = self.check_message_received('"test"', 'john.doe@example.org', self.lucy['mailbox']) + self.assertIsInstance(message, email.message.Message) + + itips = events_from_message(message) + self.assertEqual(len(itips), 1); + self.assertEqual(itips[0]['method'], 'REQUEST'); + self.assertEqual(itips[0]['uid'], uid); + + + def test_005_invite_rescheduling_accept(self): + self.purge_mailbox(self.john['mailbox']) + + start = datetime.datetime(2014,8,14, 9,0,0, tzinfo=pytz.timezone("Europe/Berlin")) + uid = self.send_itip_invitation(self.jane['mail'], start) + + response = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('ACCEPTED') }, self.jane['mail']) + self.assertIsInstance(response, email.message.Message) + + event = self.check_user_calendar_event(self.jane['kolabcalendarfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + self.assertEqual(event.get_summary(), "test") + + self.purge_mailbox(self.john['mailbox']) + + # send update with new date and incremented sequence + new_start = pytz.timezone("Europe/Berlin").localize(datetime.datetime(2014,8,15, 15,0,0)) + self.send_itip_update(self.jane['mail'], uid, new_start, summary="test", sequence=1) + + response = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('ACCEPTED') }, self.jane['mail']) + self.assertIsInstance(response, email.message.Message) + + event = self.check_user_calendar_event(self.jane['kolabcalendarfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + self.assertEqual(event.get_start(), new_start) + self.assertEqual(event.get_sequence(), 1) + + + def test_005_invite_rescheduling_reject(self): + self.purge_mailbox(self.john['mailbox']) + self.purge_mailbox(self.jack['kolabcalendarfolder']) + + start = datetime.datetime(2014,8,9, 17,0,0, tzinfo=pytz.timezone("Europe/Berlin")) + uid = self.send_itip_invitation(self.jack['mail'], start) + + response = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('TENTATIVE') }, self.jack['mail']) + self.assertIsInstance(response, email.message.Message) + + # send update with new but conflicting date and incremented sequence + self.create_calendar_event(datetime.datetime(2014,8,10, 10,30,0, tzinfo=pytz.timezone("Europe/Berlin")), user=self.jack) + new_start = pytz.timezone("Europe/Berlin").localize(datetime.datetime(2014,8,10, 9,30,0)) + self.send_itip_update(self.jack['mail'], uid, new_start, summary="test (updated)", sequence=1) + + response = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('DECLINED') }, self.jack['mail']) + self.assertEqual(response, None) + + # verify re-scheduled copy in jack's calendar with NEEDS-ACTION + event = self.check_user_calendar_event(self.jack['kolabcalendarfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + self.assertEqual(event.get_start(), new_start) + self.assertEqual(event.get_sequence(), 1) + + attendee = event.get_attendee(self.jack['mail']) + self.assertTrue(attendee.get_rsvp()) + self.assertEqual(attendee.get_participant_status(), kolabformat.PartNeedsAction) + + + def test_006_invitation_reply(self): + self.purge_mailbox(self.john['mailbox']) + + start = datetime.datetime(2014,8,18, 14,30,0, tzinfo=pytz.timezone("Europe/Berlin")) + uid = self.create_calendar_event(start, user=self.john) + + event = self.check_user_calendar_event(self.john['kolabcalendarfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + + # send a reply from jane to john + self.send_itip_reply(uid, self.jane['mail'], self.john['mail'], start=start) + + # check for the updated event in john's calendar + time.sleep(10) + event = self.check_user_calendar_event(self.john['kolabcalendarfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + + attendee = event.get_attendee(self.jane['mail']) + self.assertIsInstance(attendee, pykolab.xml.Attendee) + self.assertEqual(attendee.get_participant_status(), kolabformat.PartAccepted) + + # check attachments in update event + attachments = event.get_attachments() + self.assertEqual(len(attachments), 1) + self.assertEqual(event.get_attachment_data(0), 'This is a text attachment') + + + def test_006_invitation_reply_delegated(self): + self.purge_mailbox(self.john['mailbox']) + + start = datetime.datetime(2014,8,28, 14,30,0, tzinfo=pytz.timezone("Europe/Berlin")) + uid = self.create_calendar_event(start, user=self.john) + + event = self.check_user_calendar_event(self.john['kolabcalendarfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + + # send a reply from jane to john + self.send_itip_reply(uid, self.jane['mail'], self.john['mail'], start=start, template=itip_delegated, partstat='NEEDS-ACTION') + + # check for the updated event in john's calendar + time.sleep(10) + event = self.check_user_calendar_event(self.john['kolabcalendarfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + + attendee = event.get_attendee(self.jane['mail']) + self.assertIsInstance(attendee, pykolab.xml.Attendee) + self.assertEqual(attendee.get_participant_status(), kolabformat.PartDelegated) + self.assertEqual(len(attendee.get_delegated_to()), 1) + self.assertEqual(attendee.get_delegated_to(True)[0], 'jack@ripper.com') + + delegatee = event.get_attendee('jack@ripper.com') + self.assertIsInstance(delegatee, pykolab.xml.Attendee) + self.assertEqual(delegatee.get_participant_status(), kolabformat.PartNeedsAction) + self.assertEqual(len(delegatee.get_delegated_from()), 1) + self.assertEqual(delegatee.get_delegated_from(True)[0], self.jane['mail']) + + + def test_007_invitation_cancel(self): + self.purge_mailbox(self.john['mailbox']) + + uid = self.send_itip_invitation(self.jane['mail'], summary="cancelled") + + response = self.check_message_received(self.itip_reply_subject % { 'summary':'cancelled', 'status':participant_status_label('ACCEPTED') }, self.jane['mail']) + self.assertIsInstance(response, email.message.Message) + + self.send_itip_cancel(self.jane['mail'], uid, summary="cancelled") + + time.sleep(10) + event = self.check_user_calendar_event(self.jane['kolabcalendarfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + self.assertEqual(event.get_summary(), "cancelled") + self.assertEqual(event.get_status(True), 'CANCELLED') + self.assertTrue(event.get_transparency()) + + + def test_007_invitation_cancel_and_delete(self): + self.purge_mailbox(self.john['mailbox']) + + uid = self.send_itip_invitation(self.lucy['mail'], summary="cancel-delete") + event = self.check_user_calendar_event(self.lucy['kolabcalendarfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + + self.send_itip_cancel(self.lucy['mail'], uid, summary="cancel-delete") + + response = self.check_message_received(_('"%s" has been cancelled') % ('cancel-delete'), self.john['mail'], mailbox=self.lucy['mailbox']) + self.assertIsInstance(response, email.message.Message) + + # verify event was removed from the user's calendar + self.assertEqual(self.check_user_calendar_event(self.lucy['kolabcalendarfolder'], uid), None) + + + def test_008_inivtation_reply_notify(self): + self.purge_mailbox(self.john['mailbox']) + + start = datetime.datetime(2014,8,12, 16,0,0, tzinfo=pytz.timezone("Europe/Berlin")) + uid = self.create_calendar_event(start, user=self.john, attendees=[self.jane, self.mark, self.jack]) + + # send a reply from jane to john + self.send_itip_reply(uid, self.jane['mail'], self.john['mail'], start=start) + + # check for notification message + # this notification should be suppressed until mark has replied, too + notification = self.check_message_received(_('"%s" has been updated') % ('test'), self.john['mail']) + self.assertEqual(notification, None) + + # send a reply from mark to john + self.send_itip_reply(uid, self.mark['mail'], self.john['mail'], start=start, partstat='ACCEPTED') + + notification = self.check_message_received(_('"%s" has been updated') % ('test'), self.john['mail']) + self.assertIsInstance(notification, email.message.Message) + + notification_text = str(notification.get_payload()); + self.assertIn(self.jane['mail'], notification_text) + self.assertIn(_("PENDING"), notification_text) + + self.purge_mailbox(self.john['mailbox']) + + # send a reply from mark to john + self.send_itip_reply(uid, self.jack['mail'], self.john['mail'], start=start, partstat='ACCEPTED') + + # this triggers an additional notification + notification = self.check_message_received(_('"%s" has been updated') % ('test'), self.john['mail']) + self.assertIsInstance(notification, email.message.Message) + + notification_text = str(notification.get_payload()); + self.assertNotIn(_("PENDING"), notification_text) + + + def test_008_notify_translated(self): + self.purge_mailbox(self.mark['mailbox']) + + start = datetime.datetime(2014,8,12, 16,0,0, tzinfo=pytz.timezone("Europe/Berlin")) + uid = self.create_calendar_event(start, user=self.mark, attendees=[self.jane]) + + # send a reply from jane to mark + self.send_itip_reply(uid, self.jane['mail'], self.mark['mail'], start=start) + + # change translations to de_DE + pykolab.translate.setUserLanguage(self.mark['preferredlanguage']) + notification = self.check_message_received(_('"%s" has been updated') % ('test'), self.mark['mail'], self.mark['mailbox']) + self.assertIsInstance(notification, email.message.Message) + + notification_text = str(notification.get_payload()); + self.assertIn(self.jane['mail'], notification_text) + self.assertIn(participant_status_label("ACCEPTED")+":", notification_text) + + # reset localization + pykolab.translate.setUserLanguage(conf.get('kolab','default_locale')) + + + def test_009_outdated_reply(self): + self.purge_mailbox(self.john['mailbox']) + + start = datetime.datetime(2014,9,2, 11,0,0, tzinfo=pytz.timezone("Europe/Berlin")) + uid = self.create_calendar_event(start, user=self.john, sequence=2) + + # send a reply from jane to john + self.send_itip_reply(uid, self.jane['mail'], self.john['mail'], start=start, sequence=1) + + # verify jane's attendee status was not updated + time.sleep(10) + event = self.check_user_calendar_event(self.john['kolabcalendarfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + self.assertEqual(event.get_sequence(), 2) + self.assertEqual(event.get_attendee(self.jane['mail']).get_participant_status(), kolabformat.PartNeedsAction) + + + def test_010_partstat_update_propagation(self): + # ATTENTION: this test requires wallace.invitationpolicy_autoupdate_other_attendees_on_reply to be enabled in config + + start = datetime.datetime(2014,8,21, 13,0,0, tzinfo=pytz.timezone("Europe/Berlin")) + uid = self.create_calendar_event(start, user=self.john, attendees=[self.jane, self.jack, self.external]) + + event = self.check_user_calendar_event(self.john['kolabcalendarfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + + # send invitations to jack and jane + event_itip = event.as_string_itip() + self.send_itip_invitation(self.jane['mail'], start, template=event_itip) + self.send_itip_invitation(self.jack['mail'], start, template=event_itip) + + # wait for replies from jack and jane to be processed and propagated + time.sleep(10) + event = self.check_user_calendar_event(self.john['kolabcalendarfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + + # check updated event in organizer's calendar + self.assertEqual(event.get_attendee(self.jane['mail']).get_participant_status(), kolabformat.PartAccepted) + self.assertEqual(event.get_attendee(self.jack['mail']).get_participant_status(), kolabformat.PartTentative) + + # check updated partstats in jane's calendar + janes = self.check_user_calendar_event(self.jane['kolabcalendarfolder'], uid) + self.assertEqual(janes.get_attendee(self.jane['mail']).get_participant_status(), kolabformat.PartAccepted) + self.assertEqual(janes.get_attendee(self.jack['mail']).get_participant_status(), kolabformat.PartTentative) + + # check updated partstats in jack's calendar + jacks = self.check_user_calendar_event(self.jack['kolabcalendarfolder'], uid) + self.assertEqual(jacks.get_attendee(self.jane['mail']).get_participant_status(), kolabformat.PartAccepted) + self.assertEqual(jacks.get_attendee(self.jack['mail']).get_participant_status(), kolabformat.PartTentative) + + # PART 2: create conflicting event in jack's calendar + new_start = datetime.datetime(2014,8,21, 6,0,0, tzinfo=pytz.timezone("Europe/Berlin")) + self.create_calendar_event(new_start, user=self.jack, attendees=[], summary="blocker") + + # re-schedule initial event to new date + self.update_calendar_event(uid, start=new_start, sequence=1, user=self.john) + self.send_itip_update(self.jane['mail'], uid, new_start, summary="test (updated)", sequence=1) + self.send_itip_update(self.jack['mail'], uid, new_start, summary="test (updated)", sequence=1) + + # wait for replies to be processed and propagated + time.sleep(10) + event = self.check_user_calendar_event(self.john['kolabcalendarfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + + # check updated event in organizer's calendar (jack didn't reply yet) + self.assertEqual(event.get_attendee(self.jane['mail']).get_participant_status(), kolabformat.PartAccepted) + self.assertEqual(event.get_attendee(self.jack['mail']).get_participant_status(), kolabformat.PartTentative) + + # check partstats in jack's calendar: jack's status should remain needs-action + jacks = self.check_user_calendar_event(self.jack['kolabcalendarfolder'], uid) + self.assertEqual(jacks.get_attendee(self.jane['mail']).get_participant_status(), kolabformat.PartAccepted) + self.assertEqual(jacks.get_attendee(self.jack['mail']).get_participant_status(), kolabformat.PartNeedsAction) + + + def test_011_manual_schedule_auto_update(self): + self.purge_mailbox(self.john['mailbox']) + + # create an event in john's calendar as it was manually accepted + start = datetime.datetime(2014,9,2, 11,0,0, tzinfo=pytz.timezone("Europe/Berlin")) + uid = self.create_calendar_event(start, user=self.jane, sequence=1, folder=self.john['kolabcalendarfolder']) + + # send update with the same sequence: no re-scheduling + templ = itip_invitation.replace("RSVP=TRUE", "RSVP=FALSE").replace("Doe, John", self.jane['displayname']).replace("john.doe@example.org", self.jane['mail']) + self.send_itip_update(self.john['mail'], uid, start, summary="test updated", sequence=1, partstat='ACCEPTED', template=templ) + + time.sleep(10) + event = self.check_user_calendar_event(self.john['kolabcalendarfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + self.assertEqual(event.get_summary(), "test updated") + self.assertEqual(event.get_attendee(self.john['mail']).get_participant_status(), kolabformat.PartAccepted) + + # this should also trigger an update notification + notification = self.check_message_received(_('"%s" has been updated') % ('test updated'), self.jane['mail'], mailbox=self.john['mailbox']) + self.assertIsInstance(notification, email.message.Message) + + # send outdated update: should not be saved + self.send_itip_update(self.john['mail'], uid, start, summary="old test", sequence=0, partstat='NEEDS-ACTION', template=templ) + notification = self.check_message_received(_('"%s" has been updated') % ('old test'), self.jane['mail'], mailbox=self.john['mailbox']) + self.assertEqual(notification, None) + + + def test_012_confidential_invitation(self): + start = datetime.datetime(2014,9,21, 9,30,0) + uid = self.send_itip_invitation(self.jane['mail'], start, summary='confidential', template=itip_invitation.replace("DESCRIPTION:test", "CLASS:CONFIDENTIAL")) + + # check event being stored in the folder annotared with event.confidential + event = self.check_user_calendar_event(self.jane['kolabconfidentialcalendar'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + self.assertEqual(event.get_summary(), "confidential") + + + def test_013_update_shared_folder(self): + # create an event organized by Mark (a delegate of Jane) into Jane's calendar + start = datetime.datetime(2015,3,10, 9,30,0, tzinfo=pytz.timezone("Europe/Berlin")) + uid = self.create_calendar_event(start, user=self.mark, attendees=[self.jane, self.john], folder=self.jane['kolabcalendarfolder']) + + event = self.check_user_calendar_event(self.jane['kolabcalendarfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + + # send a reply from john to mark + self.send_itip_reply(uid, self.john['mail'], self.mark['mail'], start=start) + + # check for the updated event in jane's calendar + time.sleep(10) + event = self.check_user_calendar_event(self.jane['kolabcalendarfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + self.assertEqual(event.get_attendee(self.john['mail']).get_participant_status(), kolabformat.PartAccepted) + + def test_014_per_sender_policy(self): + # send invitation from john => REJECT + start = datetime.datetime(2015,2,28, 14,0,0) + uid = self.send_itip_invitation(self.bill['mail'], start) + + response = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('DECLINED') }, self.bill['mail']) + self.assertIsInstance(response, email.message.Message) + + # send invitation from lucy => SAVE + start = datetime.datetime(2015,3,11, 10,0,0) + templ = itip_invitation.replace("Doe, John", self.lucy['displayname']) + uid = self.send_itip_invitation(self.bill['mail'], start, template=templ, from_addr=self.lucy['mail']) + + event = self.check_user_calendar_event(self.bill['kolabcalendarfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + + + def test_015_update_single_occurrence(self): + self.purge_mailbox(self.john['mailbox']) + + start = datetime.datetime(2015,4,2, 14,0,0) + uid = self.send_itip_invitation(self.jane['mail'], start, template=itip_recurring) + + response = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('ACCEPTED') }, self.jane['mail']) + self.assertIsInstance(response, email.message.Message) + + event = self.check_user_calendar_event(self.jane['kolabcalendarfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + self.assertTrue(event.is_recurring()) + + # send update to a single instance with the same sequence: no re-scheduling + exdate = start + datetime.timedelta(days=14) + self.send_itip_update(self.jane['mail'], uid, exdate, summary="test exception", sequence=0, partstat='ACCEPTED', instance=exdate) + + time.sleep(10) + event = self.check_user_calendar_event(self.jane['kolabcalendarfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + self.assertEqual(len(event.get_exceptions()), 1) + + exception = event.get_instance(exdate) + self.assertEqual(exception.get_summary(), "test exception") + self.assertEqual(exception.get_attendee(self.jane['mail']).get_participant_status(), kolabformat.PartAccepted) + + + def test_015_reschedule_single_occurrence(self): + self.purge_mailbox(self.john['mailbox']) + + start = datetime.datetime(2015,4,10, 9,0,0) + uid = self.send_itip_invitation(self.jane['mail'], start, template=itip_recurring) + + response = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('ACCEPTED') }, self.jane['mail']) + self.assertIsInstance(response, email.message.Message) + + # send update to a single instance with the same sequence: no re-scheduling + exdate = start + datetime.timedelta(days=14) + exstart = exdate + datetime.timedelta(hours=5) + self.send_itip_update(self.jane['mail'], uid, exstart, summary="test resceduled", sequence=1, partstat='NEEDS-ACTION', instance=exdate) + + response = self.check_message_received(self.itip_reply_subject % { 'summary':'test resceduled', 'status':participant_status_label('ACCEPTED') }, self.jane['mail']) + self.assertIsInstance(response, email.message.Message) + + time.sleep(10) + event = self.check_user_calendar_event(self.jane['kolabcalendarfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + self.assertEqual(len(event.get_exceptions()), 1) + + # re-schedule again, conflicts with itself + exstart = exdate + datetime.timedelta(hours=6) + self.send_itip_update(self.jane['mail'], uid, exstart, summary="test new", sequence=2, partstat='NEEDS-ACTION', instance=exdate) + + response = self.check_message_received(self.itip_reply_subject % { 'summary':'test new', 'status':participant_status_label('ACCEPTED') }, self.jane['mail']) + self.assertIsInstance(response, email.message.Message) + + # check for updated excaption + time.sleep(10) + event = self.check_user_calendar_event(self.jane['kolabcalendarfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + self.assertEqual(len(event.get_exceptions()), 1) + + exception = event.get_instance(exdate) + self.assertIsInstance(exception, pykolab.xml.Event) + self.assertEqual(exception.get_start().strftime('%Y%m%dT%H%M%S'), exstart.strftime('%Y%m%dT%H%M%S')) + + + def test_016_reply_single_occurrence(self): + self.purge_mailbox(self.john['mailbox']) + + start = datetime.datetime(2015,3,7, 10,0,0, tzinfo=pytz.timezone("Europe/Zurich")) + uid = self.create_calendar_event(start, attendees=[self.jane, self.mark], recurring=True) + + event = self.check_user_calendar_event(self.john['kolabcalendarfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + + # store a copy in mark's calendar, too + self.create_calendar_event(start, attendees=[self.jane, self.mark], recurring=True, folder=self.mark['kolabcalendarfolder'], uid=uid) + + # send a reply for a single occurrence from jane + exdate = start + datetime.timedelta(days=7) + self.send_itip_reply(uid, self.jane['mail'], self.john['mail'], start=exdate, instance=exdate) + + # check for the updated event in john's calendar + time.sleep(10) + event = self.check_user_calendar_event(self.john['kolabcalendarfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + self.assertEqual(len(event.get_exceptions()), 1) + + exception = event.get_instance(exdate) + self.assertEqual(exception.get_attendee(self.jane['mail']).get_participant_status(), kolabformat.PartAccepted) + + # check mark's copy for partstat update being stored in an exception, too + marks = self.check_user_calendar_event(self.mark['kolabcalendarfolder'], uid) + self.assertIsInstance(marks, pykolab.xml.Event) + self.assertEqual(len(marks.get_exceptions()), 1) + + exception = marks.get_instance(exdate) + self.assertEqual(exception.get_attendee(self.jane['mail']).get_participant_status(), kolabformat.PartAccepted) + + # send a reply for a the entire series from mark + self.send_itip_reply(uid, self.mark['mail'], self.john['mail'], start=start) + + # check for the updated event in john's calendar + time.sleep(10) + event = self.check_user_calendar_event(self.john['kolabcalendarfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + self.assertEqual(len(event.get_exceptions()), 1) + + exception = event.get_instance(exdate) + self.assertEqual(exception.get_attendee(self.mark['mail']).get_participant_status(), kolabformat.PartAccepted) + + def test_017_cancel_single_occurrence(self): + self.purge_mailbox(self.john['mailbox']) + + start = datetime.datetime(2015,3,20, 19,0,0, tzinfo=pytz.timezone("Europe/Zurich")) + uid = self.send_itip_invitation(self.jane['mail'], summary="recurring", start=start, template=itip_recurring) + + response = self.check_message_received(self.itip_reply_subject % { 'summary':'recurring', 'status':participant_status_label('ACCEPTED') }, self.jane['mail']) + self.assertIsInstance(response, email.message.Message) + + event = self.check_user_calendar_event(self.jane['kolabcalendarfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + + exdate = start + datetime.timedelta(days=14) + self.send_itip_cancel(self.jane['mail'], uid, summary="recurring cancelled", instance=exdate) + + time.sleep(10) + event = self.check_user_calendar_event(self.jane['kolabcalendarfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + self.assertEqual(len(event.get_exceptions()), 1) + + exception = event.get_instance(exdate) + self.assertEqual(exception.get_status(True), 'CANCELLED') + self.assertTrue(exception.get_transparency()) + + # send a new invitation for the cancelled slot + uid = self.send_itip_invitation(self.jane['mail'], summary="new booking", start=exdate) + + response = self.check_message_received(self.itip_reply_subject % { 'summary':'new booking', 'status':participant_status_label('ACCEPTED') }, self.jane['mail']) + self.assertIsInstance(response, email.message.Message) + + def test_017_cancel_delete_single_occurrence(self): + self.purge_mailbox(self.john['mailbox']) + + start = datetime.datetime(2015,3,24, 13,0,0, tzinfo=pytz.timezone("Europe/Zurich")) + uid = self.send_itip_invitation(self.lucy['mail'], summary="recurring cancel-delete", start=start, template=itip_recurring) + + event = self.check_user_calendar_event(self.lucy['kolabcalendarfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + + # send update to a single instance with the same sequence: no re-scheduling + exdate = start + datetime.timedelta(days=14) + exstart = exdate + datetime.timedelta(hours=5) + self.send_itip_update(self.lucy['mail'], uid, exstart, summary="recurring rescheduled", sequence=1, partstat='NEEDS-ACTION', instance=exdate) + + time.sleep(10) + event = self.check_user_calendar_event(self.lucy['kolabcalendarfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + self.assertEqual(len(event.get_exceptions()), 1) + + # send cancellation for exception + self.send_itip_cancel(self.lucy['mail'], uid, summary="recurring rescheduled", instance=exdate) + + response = self.check_message_received(_('"%s" has been cancelled') % ('recurring rescheduled'), self.john['mail'], mailbox=self.lucy['mailbox']) + self.assertIsInstance(response, email.message.Message) + + event = self.check_user_calendar_event(self.lucy['kolabcalendarfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + self.assertEqual(len(event.get_exception_dates()), 1) + self.assertEqual(event.get_exception_dates()[0].strftime("%Y-%m-%d"), exdate.strftime("%Y-%m-%d")) + self.assertEqual(len(event.get_exceptions()), 0) + + def test_017_cancel_thisandfuture(self): + self.purge_mailbox(self.john['mailbox']) + + start = datetime.datetime(2015,5,4, 6,30,0) + uid = self.send_itip_invitation(self.mark['mail'], summary="recurring", start=start, template=itip_recurring) + + pykolab.translate.setUserLanguage(self.mark['preferredlanguage']) + response = self.check_message_received(_('"%(summary)s" has been %(status)s') % { 'summary':'recurring', 'status':participant_status_label('ACCEPTED') }, self.mark['mail']) + self.assertIsInstance(response, email.message.Message) + + pykolab.translate.setUserLanguage(conf.get('kolab','default_locale')) + + event = self.check_user_calendar_event(self.mark['kolabcalendarfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + + exdate = start + datetime.timedelta(days=14) + self.send_itip_cancel(self.mark['mail'], uid, summary="recurring ended", instance=exdate, thisandfuture=True) + + time.sleep(10) + event = self.check_user_calendar_event(self.mark['kolabcalendarfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + + rrule = event.get_recurrence().to_dict() + self.assertIsInstance(rrule['until'], datetime.datetime) + self.assertEqual(rrule['until'].strftime('%Y%m%d'), (exdate - datetime.timedelta(days=1)).strftime('%Y%m%d')) + + + def test_018_invite_individual_occurrences(self): + self.purge_mailbox(self.john['mailbox']) + + start = datetime.datetime(2015,1,30, 17,0,0, tzinfo=pytz.timezone("Europe/Zurich")) + uid = self.send_itip_invitation(self.jane['mail'], summary="single", start=start, instance=start) + + response = self.check_message_received(self.itip_reply_subject % { 'summary':'single', 'status':participant_status_label('ACCEPTED') }, self.jane['mail']) + self.assertIsInstance(response, email.message.Message) + self.assertIn("RECURRENCE-ID", str(response)) + + event = self.check_user_calendar_event(self.jane['kolabcalendarfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + self.assertIsInstance(event.get_recurrence_id(), datetime.datetime) + + # send another invitation with the same UID for different RECURRENCE-ID + newstart = datetime.datetime(2015,2,6, 17,0,0, tzinfo=pytz.timezone("Europe/Zurich")) + self.send_itip_invitation(self.jane['mail'], summary="single #2", uid=uid, start=newstart, instance=newstart) + + response = self.check_message_received(self.itip_reply_subject % { 'summary':'single #2', 'status':participant_status_label('ACCEPTED') }, self.jane['mail']) + self.assertIsInstance(response, email.message.Message) + self.assertIn("RECURRENCE-ID", str(response)) + + event = self.check_user_calendar_event(self.jane['kolabcalendarfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + self.assertEqual(len(event.get_exceptions()), 1) + + exception = event.get_instance(newstart) + self.assertEqual(exception.get_summary(), "single #2") + self.assertEqual(exception.get_recurrence_id(), newstart) + + # send an update occurrence #1 + self.send_itip_invitation(self.jane['mail'], summary="updated #1", uid=uid, start=start, instance=start) + time.sleep(5) + + # send an update occurrence #2 + self.send_itip_invitation(self.jane['mail'], summary="updated #2", uid=uid, start=newstart, instance=newstart) + + time.sleep(10) + event = self.check_user_calendar_event(self.jane['kolabcalendarfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + self.assertEqual(event.get_summary(), "updated #1") + + exception = event.get_instance(newstart) + self.assertEqual(exception.get_summary(), "updated #2") + + + def test_020_task_assignment_accept(self): + start = datetime.datetime(2014,9,10, 19,0,0) + uid = self.send_itip_invitation(self.jane['mail'], start, summary='work', template=itip_todo) + + response = self.check_message_received(self.itip_reply_subject % { 'summary':'work', 'status':participant_status_label('ACCEPTED') }, self.jane['mail']) + self.assertIsInstance(response, email.message.Message) + + todo = self.check_user_imap_object(self.jane['kolabtasksfolder'], uid, 'task') + self.assertIsInstance(todo, pykolab.xml.Todo) + self.assertEqual(todo.get_summary(), "work") + + # send update with the same sequence: no re-scheduling + self.send_itip_update(self.jane['mail'], uid, start, summary='work updated', template=itip_todo, sequence=0, partstat='ACCEPTED') + + response = self.check_message_received(self.itip_reply_subject % { 'summary':'work updated', 'status':participant_status_label('ACCEPTED') }, self.jane['mail']) + self.assertEqual(response, None) + + time.sleep(10) + todo = self.check_user_imap_object(self.jane['kolabtasksfolder'], uid, 'task') + self.assertIsInstance(todo, pykolab.xml.Todo) + self.assertEqual(todo.get_summary(), "work updated") + self.assertEqual(todo.get_attendee(self.jane['mail']).get_participant_status(), kolabformat.PartAccepted) + + + def test_021_task_assignment_reply(self): + self.purge_mailbox(self.john['mailbox']) + + due = datetime.datetime(2014,9,12, 14,0,0, tzinfo=pytz.timezone("Europe/Berlin")) + uid = self.create_task_assignment(due, user=self.john) + + todo = self.check_user_imap_object(self.john['kolabtasksfolder'], uid, 'task') + self.assertIsInstance(todo, pykolab.xml.Todo) + + # send a reply from jane to john + partstat = 'COMPLETED' + self.send_itip_reply(uid, self.jane['mail'], self.john['mail'], start=due, template=itip_todo_reply, partstat=partstat) + + # check for the updated task in john's tasklist + time.sleep(10) + todo = self.check_user_imap_object(self.john['kolabtasksfolder'], uid, 'task') + self.assertIsInstance(todo, pykolab.xml.Todo) + + attendee = todo.get_attendee(self.jane['mail']) + self.assertIsInstance(attendee, pykolab.xml.Attendee) + self.assertEqual(attendee.get_participant_status(True), partstat) + + # this should trigger an update notification + notification = self.check_message_received(_('"%s" has been updated') % ('test'), self.john['mail']) + self.assertIsInstance(notification, email.message.Message) + + notification_text = str(notification.get_payload()); + self.assertIn(participant_status_label(partstat), notification_text) + + + def test_022_task_cancellation(self): + uid = self.send_itip_invitation(self.jane['mail'], summary='more work', template=itip_todo) + + time.sleep(10) + self.send_itip_cancel(self.jane['mail'], uid, template=itip_todo_cancel, summary="cancelled") + + time.sleep(10) + todo = self.check_user_imap_object(self.jane['kolabtasksfolder'], uid, 'task') + self.assertIsInstance(todo, pykolab.xml.Todo) + self.assertEqual(todo.get_summary(), "more work") + self.assertEqual(todo.get_status(True), 'CANCELLED') + + # this should trigger a notification message + notification = self.check_message_received(_('"%s" has been cancelled') % ('more work'), self.john['mail'], mailbox=self.jane['mailbox']) + self.assertIsInstance(notification, email.message.Message) + + +if __name__ == '__main__': + stick.main() diff --git a/integration/P/test_wallace_invitationpolicy_data.py b/integration/P/test_wallace_invitationpolicy_data.py new file mode 100644 index 0000000..db5c295 --- /dev/null +++ b/integration/P/test_wallace_invitationpolicy_data.py @@ -0,0 +1,199 @@ + +itip_invitation = """ +BEGIN:VCALENDAR +VERSION:2.0 +PRODID:-//Roundcube Webmail 0.9-0.3.el6.kolab_3.0//NONSGML Calendar//EN +CALSCALE:GREGORIAN +METHOD:REQUEST +BEGIN:VEVENT +UID:%(uid)s%(recurrenceid)s +DTSTAMP:20140213T125414Z +DTSTART;TZID=Europe/Berlin:%(start)s +DTEND;TZID=Europe/Berlin:%(end)s +SUMMARY:%(summary)s +DESCRIPTION:test +ORGANIZER;CN="Doe, John":mailto:john.doe@example.org +ATTENDEE;ROLE=REQ-PARTICIPANT;PARTSTAT=%(partstat)s;RSVP=TRUE:mailto:%(mailto)s +ATTENDEE;ROLE=OPT-PARTICIPANT;PARTSTAT=TENTATIVE;RSVP=FALSE:mailto:somebody@else.com +TRANSP:OPAQUE +SEQUENCE:%(sequence)d +END:VEVENT +END:VCALENDAR +""" + +itip_cancellation = """ +BEGIN:VCALENDAR +VERSION:2.0 +PRODID:-//Roundcube Webmail 0.9-0.3.el6.kolab_3.0//NONSGML Calendar//EN +CALSCALE:GREGORIAN +METHOD:CANCEL +BEGIN:VEVENT +UID:%(uid)s%(recurrenceid)s +DTSTAMP:20140218T125414Z +DTSTART;TZID=Europe/Berlin:20120713T100000 +DTEND;TZID=Europe/Berlin:20120713T110000 +SUMMARY:%(summary)s +DESCRIPTION:test +ORGANIZER;CN="Doe, John":mailto:john.doe@example.org +ATTENDEE;ROLE=REQ-PARTICIPANT;PARTSTAT=ACCEPTED;RSVP=TRUE:mailto:%(mailto)s +TRANSP:OPAQUE +SEQUENCE:%(sequence)d +END:VEVENT +END:VCALENDAR +""" + +itip_recurring = """ +BEGIN:VCALENDAR +VERSION:2.0 +PRODID:-//Apple Inc.//Mac OS X 10.9.2//EN +CALSCALE:GREGORIAN +METHOD:REQUEST +BEGIN:VEVENT +UID:%(uid)s +DTSTAMP:20140213T125414Z +DTSTART;TZID=Europe/Berlin:%(start)s +DTEND;TZID=Europe/Berlin:%(end)s +RRULE:FREQ=WEEKLY;INTERVAL=1;COUNT=10 +SUMMARY:%(summary)s +DESCRIPTION:test +ORGANIZER;CN="Doe, John":mailto:john.doe@example.org +ATTENDEE;ROLE=REQ-PARTICIPANT;PARTSTAT=%(partstat)s;RSVP=TRUE:mailto:%(mailto)s +TRANSP:OPAQUE +SEQUENCE:%(sequence)d +END:VEVENT +END:VCALENDAR +""" + +itip_reply = """ +BEGIN:VCALENDAR +VERSION:2.0 +PRODID:-//pykolab-0.6.9-1//kolab.org// +CALSCALE:GREGORIAN +METHOD:REPLY +BEGIN:VEVENT +SUMMARY:%(summary)s +UID:%(uid)s%(recurrenceid)s +DTSTART;TZID=Europe/Berlin;VALUE=DATE-TIME:%(start)s +DTEND;TZID=Europe/Berlin;VALUE=DATE-TIME:%(end)s +DTSTAMP;VALUE=DATE-TIME:20140706T171038Z +ORGANIZER;CN="Doe, John":MAILTO:%(organizer)s +ATTENDEE;CUTYPE=INDIVIDUAL;PARTSTAT=%(partstat)s;ROLE=REQ-PARTICIPANT:mailto:%(mailto)s +PRIORITY:0 +SEQUENCE:%(sequence)d +END:VEVENT +END:VCALENDAR +""" + +itip_delegated = """ +BEGIN:VCALENDAR +VERSION:2.0 +PRODID:-//pykolab-0.6.9-1//kolab.org// +CALSCALE:GREGORIAN +METHOD:REPLY +BEGIN:VEVENT +SUMMARY:%(summary)s +UID:%(uid)s%(recurrenceid)s +DTSTART;TZID=Europe/Berlin;VALUE=DATE-TIME:%(start)s +DTEND;TZID=Europe/Berlin;VALUE=DATE-TIME:%(end)s +DTSTAMP;VALUE=DATE-TIME:20140706T171038Z +ORGANIZER;CN="Doe, John":MAILTO:%(organizer)s +ATTENDEE;PARTSTAT=DELEGATED;DELEGATED-TO=jack@ripper.com;ROLE=NON-PARTICIPANT:mailto:%(mailto)s +ATTENDEE;PARTSTAT=%(partstat)s;DELEGATED-FROM=%(mailto)s;ROLE=REQ-PARTICIPANT:mailto:jack@ripper.com +PRIORITY:0 +SEQUENCE:%(sequence)d +END:VEVENT +END:VCALENDAR +""" + +itip_todo = """ +BEGIN:VCALENDAR +VERSION:2.0 +PRODID:-//Roundcube//Roundcube libcalendaring 1.1-git//Sabre//Sabre VObject + 2.1.3//EN +CALSCALE:GREGORIAN +METHOD:REQUEST +BEGIN:VTODO +UID:%(uid)s +CREATED;VALUE=DATE-TIME:20140731T100704Z +DTSTAMP;VALUE=DATE-TIME:20140820T101333Z +DTSTART;VALUE=DATE-TIME;TZID=Europe/Berlin:%(start)s +DUE;VALUE=DATE-TIME;TZID=Europe/Berlin:%(end)s +SUMMARY:%(summary)s +SEQUENCE:%(sequence)d +PRIORITY:1 +STATUS:NEEDS-ACTION +PERCENT-COMPLETE:0 +ORGANIZER;CN="Doe, John":mailto:john.doe@example.org +ATTENDEE;PARTSTAT=%(partstat)s;ROLE=REQ-PARTICIPANT:mailto:%(mailto)s +END:VTODO +END:VCALENDAR +""" + +itip_todo_reply = """ +BEGIN:VCALENDAR +VERSION:2.0 +PRODID:-//Roundcube//Roundcube libcalendaring 1.1-git//Sabre//Sabre VObject + 2.1.3//EN +CALSCALE:GREGORIAN +METHOD:REPLY +BEGIN:VTODO +UID:%(uid)s +CREATED;VALUE=DATE-TIME:20140731T100704Z +DTSTAMP;VALUE=DATE-TIME:20140821T085424Z +DTSTART;VALUE=DATE-TIME;TZID=Europe/Berlin:%(start)s +DUE;VALUE=DATE-TIME;TZID=Europe/Berlin:%(end)s +SUMMARY:%(summary)s +SEQUENCE:%(sequence)d +PRIORITY:1 +STATUS:NEEDS-ACTION +PERCENT-COMPLETE:40 +ATTENDEE;PARTSTAT=%(partstat)s;ROLE=REQ-PARTICIPANT;CUTYPE=INDIVIDUAL:mailto:%(mailto)s +ORGANIZER;CN="Doe, John":mailto:%(organizer)s +END:VTODO +END:VCALENDAR +""" + +itip_todo_cancel = """ +BEGIN:VCALENDAR +VERSION:2.0 +PRODID:-//Roundcube//Roundcube libcalendaring 1.1-git//Sabre//Sabre VObject + 2.1.3//EN +CALSCALE:GREGORIAN +METHOD:CANCEL +BEGIN:VTODO +UID:%(uid)s +CREATED;VALUE=DATE-TIME:20140731T100704Z +DTSTAMP;VALUE=DATE-TIME:20140820T101333Z +SUMMARY:%(summary)s +SEQUENCE:%(sequence)d +PRIORITY:1 +STATUS:CANCELLED +ORGANIZER;CN="Doe, John":mailto:john.doe@example.org +ATTENDEE;PARTSTAT=ACCEPTED;ROLE=REQ-PARTICIPANT:mailto:%(mailto)s +END:VTODO +END:VCALENDAR +""" + +mime_message = """MIME-Version: 1.0 +Content-Type: multipart/mixed; + boundary="=_c8894dbdb8baeedacae836230e3436fd" +From: "Doe, John" +Date: Tue, 25 Feb 2014 13:54:14 +0100 +Message-ID: <240fe7ae7e139129e9eb95213c1016d7@example.org> +To: %s +Subject: "test" + +--=_c8894dbdb8baeedacae836230e3436fd +Content-Type: text/plain; charset=UTF-8; format=flowed +Content-Transfer-Encoding: quoted-printable + +*test* + +--=_c8894dbdb8baeedacae836230e3436fd +Content-Type: text/calendar; charset=UTF-8; method=%s; name=event.ics +Content-Disposition: attachment; filename=event.ics +Content-Transfer-Encoding: 8bit + +%s +--=_c8894dbdb8baeedacae836230e3436fd-- +""" diff --git a/integration/P/test_wallace_resource_invitation.py b/integration/P/test_wallace_resource_invitation.py new file mode 100644 index 0000000..d8f1fe5 --- /dev/null +++ b/integration/P/test_wallace_resource_invitation.py @@ -0,0 +1,849 @@ +import stick +import pykolab +import time +import datetime +import smtplib +import email +import uuid +import re + +from pykolab.auth import Auth +from pykolab.imap import IMAP +from wallace import module_resources + +from pykolab.translate import _ +from pykolab.xml import utils as xmlutils +from pykolab.xml import event_from_message +from pykolab.xml import participant_status_label +from pykolab.itip import events_from_message + +from test_wallace_resource_invitation_data import * + +conf = pykolab.getConf() + + +class TestWallaceResourcesInvitation(stick.KolabIntegrationTest): + john = None + jane = None + + @classmethod + def setUpClass(self, *args, **kw): + # check if resource module is activated in kolab.conf + if not 'resources' in conf.get_list('wallace', 'modules'): + raise Exception("'resources' module required in wallace.modules from kolab.conf") + + # set language to default + pykolab.translate.setUserLanguage(conf.get('kolab','default_locale')) + + self.itip_reply_subject = _("Reservation Request for %(summary)s was %(status)s") + + self.john = self.require_user("John", "Doe", kolabinvitationpolicy='ALL_MANUAL') + self.john['sender'] = "%(givenname)s %(sn)s <%(mail)s>" % self.john + self.john['mailbox'] = "user/%(mail)s" % self.john + self.john['displayname'] = "%(givenname)s %(sn)s" % self.john + + self.jane = self.require_user("Jane", "Manager", kolabinvitationpolicy='ALL_MANUAL') + self.jane['sender'] = "%(givenname)s %(sn)s <%(mail)s>" % self.jane + self.jane['mailbox'] = "user/%(mail)s" % self.jane + self.jane['displayname'] = "%(givenname)s %(sn)s" % self.jane + + self.audi = self.require_resource("car", "Audi A4") + self.passat = self.require_resource("car", "VW Passat") + self.boxter = self.require_resource("car", "Porsche Boxter S") + self.cars = self.require_resource("collection", "Company Cars", members=[ + self.audi['dn'], self.passat['dn'], self.boxter['dn'] + ]) + + self.room1 = self.require_resource("confroom", "Room 101", owner=self.jane['dn'], kolabinvitationpolicy='ACT_ACCEPT_AND_NOTIFY') + self.room2 = self.require_resource("confroom", "Conference Room B-222") + self.rooms = self.require_resource("collection", "Rooms", members=[ self.room1['dn'], self.room2['dn'] ], owner=self.jane['dn'], kolabinvitationpolicy='ACT_ACCEPT_AND_NOTIFY') + + self.room3 = self.require_resource("confroom", "CEOs Office 303") + self.viprooms = self.require_resource("collection", "VIP Rooms", members=[ self.room3['dn'] ], owner=self.jane['dn'], kolabinvitationpolicy='ACT_MANUAL') + + super(TestWallaceResourcesInvitation, self).setUpClass(*args, **kw) + + + def send_message(self, itip_payload, to_addr, from_addr=None): + if from_addr is None: + from_addr = self.john['mail'] + + smtp = smtplib.SMTP('localhost', 10026) + smtp.sendmail(from_addr, to_addr, mime_message % (to_addr, itip_payload)) + smtp.quit() + + def send_itip_invitation(self, resource_email, start=None, allday=False, template=None, uid=None, instance=None): + if start is None: + start = datetime.datetime.now() + + if uid is None: + uid = str(uuid.uuid4()) + + if allday: + default_template = itip_allday + end = start + datetime.timedelta(days=1) + date_format = '%Y%m%d' + else: + end = start + datetime.timedelta(hours=4) + default_template = itip_invitation + date_format = '%Y%m%dT%H%M%S' + + recurrence_id = '' + if instance is not None: + recurrence_id = "\nRECURRENCE-ID;TZID=Europe/London:" + instance.strftime(date_format) + + self.send_message((template if template is not None else default_template) % ( + uid, + recurrence_id, + start.strftime(date_format), + end.strftime(date_format), + resource_email + ), + resource_email) + + return uid + + def send_itip_update(self, resource_email, uid, start=None, template=None, sequence=None, instance=None): + if start is None: + start = datetime.datetime.now() + + end = start + datetime.timedelta(hours=4) + + recurrence_id = '' + if instance is not None: + recurrence_id = "\nRECURRENCE-ID;TZID=Europe/London:" + instance.strftime('%Y%m%dT%H%M%S') + + if sequence is not None: + if not template: + template = itip_update + template = re.sub(r'SEQUENCE:\d+', 'SEQUENCE:' + str(sequence), template) + + self.send_message((template if template is not None else itip_update) % ( + uid, + recurrence_id, + start.strftime('%Y%m%dT%H%M%S'), + end.strftime('%Y%m%dT%H%M%S'), + resource_email + ), + resource_email) + + return uid + + def send_itip_cancel(self, resource_email, uid, instance=None): + recurrence_id = '' + if instance is not None: + recurrence_id = "\nRECURRENCE-ID;TZID=Europe/London:" + instance.strftime('%Y%m%dT%H%M%S') + + self.send_message(itip_cancellation % ( + uid, + recurrence_id, + resource_email + ), + resource_email) + + return uid + + + def send_owner_response(self, event, partstat, from_addr=None): + if from_addr is None: + from_addr = self.jane['mail'] + + itip_reply = event.to_message_itip(from_addr, + method="REPLY", + participant_status=partstat, + message_text="Request " + partstat, + subject="Booking has been %s" % (partstat) + ) + + smtp = smtplib.SMTP('localhost', 10026) + smtp.sendmail(from_addr, str(event.get_organizer().email()), str(itip_reply)) + smtp.quit() + + def check_message_received(self, subject, from_addr=None, mailbox=None): + if mailbox is None: + mailbox = self.john['mailbox'] + + imap = IMAP() + imap.connect() + imap.set_acl(mailbox, "cyrus-admin", "lrs") + imap.imap.m.select(mailbox) + + found = None + retries = 15 + + while not found and retries > 0: + retries -= 1 + + typ, data = imap.imap.m.search(None, '(UNDELETED HEADER FROM "%s")' % (from_addr) if from_addr else 'UNDELETED') + for num in data[0].split(): + typ, msg = imap.imap.m.fetch(num, '(RFC822)') + message = email.message_from_string(msg[0][1]) + if message['Subject'] == subject: + found = message + break + + time.sleep(1) + + imap.disconnect() + + return found + + def check_resource_calendar_event(self, mailbox, uid=None): + imap = IMAP() + imap.connect() + + imap.set_acl(mailbox, "cyrus-admin", "lrs") + imap.imap.m.select(imap.folder_quote(mailbox)) + + found = None + retries = 10 + + while not found and retries > 0: + retries -= 1 + + typ, data = imap.imap.m.search(None, '(UNDELETED HEADER SUBJECT "%s")' % (uid) if uid else '(UNDELETED HEADER X-Kolab-Type "application/x-vnd.kolab.event")') + for num in data[0].split(): + typ, data = imap.imap.m.fetch(num, '(RFC822)') + event_message = email.message_from_string(data[0][1]) + + # return matching UID or first event found + if uid and event_message['subject'] != uid: + continue + + found = event_from_message(event_message) + if found: + break + + time.sleep(1) + + imap.disconnect() + + return found + + def purge_mailbox(self, mailbox): + imap = IMAP() + imap.connect() + imap.set_acl(mailbox, "cyrus-admin", "lrwcdest") + imap.imap.m.select(u'"'+mailbox+'"') + + typ, data = imap.imap.m.search(None, 'ALL') + for num in data[0].split(): + imap.imap.m.store(num, '+FLAGS', '\\Deleted') + + imap.imap.m.expunge() + imap.disconnect() + + + def find_resource_by_email(self, email): + resource = None + for r in [self.audi, self.passat, self.boxter, self.room1, self.room2]: + if (email.find(r['mail']) >= 0): + resource = r + break + return resource + + + def test_001_resource_from_email_address(self): + resource = module_resources.resource_record_from_email_address(self.audi['mail']) + self.assertEqual(len(resource), 1) + self.assertEqual(resource[0], self.audi['dn']) + + collection = module_resources.resource_record_from_email_address(self.cars['mail']) + self.assertEqual(len(collection), 1) + self.assertEqual(collection[0], self.cars['dn']) + + + def test_002_invite_resource(self): + uid = self.send_itip_invitation(self.audi['mail'], datetime.datetime(2014,7,13, 10,0,0)) + + response = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('ACCEPTED') }, self.audi['mail']) + self.assertIsInstance(response, email.message.Message) + + event = self.check_resource_calendar_event(self.audi['kolabtargetfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + self.assertEqual(event.get_summary(), "test") + + # @depends test_002_invite_resource + def test_003_invite_resource_conflict(self): + uid = self.send_itip_invitation(self.audi['mail'], datetime.datetime(2014,7,13, 12,0,0)) + + response = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('DECLINED') }, self.audi['mail']) + self.assertIsInstance(response, email.message.Message) + + self.assertEqual(self.check_resource_calendar_event(self.audi['kolabtargetfolder'], uid), None) + + + def test_004_invite_resource_collection(self): + self.purge_mailbox(self.john['mailbox']) + + uid = self.send_itip_invitation(self.cars['mail'], datetime.datetime(2014,7,13, 12,0,0)) + + # one of the collection members accepted the reservation + accept = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('ACCEPTED') }) + self.assertIsInstance(accept, email.message.Message) + + delegatee = self.find_resource_by_email(accept['from']) + self.assertIn(delegatee['mail'], accept['from']) + + # check booking in the delegatee's resource calendar + self.assertIsInstance(self.check_resource_calendar_event(delegatee['kolabtargetfolder'], uid), pykolab.xml.Event) + + # resource collection responds with a DELEGATED message + response = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('DELEGATED') }, self.cars['mail']) + self.assertIsInstance(response, email.message.Message) + self.assertIn("ROLE=NON-PARTICIPANT;RSVP=FALSE", str(response)) + + + def test_005_rescheduling_reservation(self): + self.purge_mailbox(self.john['mailbox']) + + uid = self.send_itip_invitation(self.audi['mail'], datetime.datetime(2014,4,1, 10,0,0)) + + response = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('ACCEPTED') }, self.audi['mail']) + self.assertIsInstance(response, email.message.Message) + + self.purge_mailbox(self.john['mailbox']) + self.send_itip_update(self.audi['mail'], uid, datetime.datetime(2014,4,1, 12,0,0)) # conflict with myself + + response = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('ACCEPTED') }, self.audi['mail']) + self.assertIsInstance(response, email.message.Message) + + event = self.check_resource_calendar_event(self.audi['kolabtargetfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + self.assertEqual(event.get_start().hour, 12) + self.assertEqual(event.get_sequence(), 2) + + + def test_005_rescheduling_collection(self): + self.purge_mailbox(self.john['mailbox']) + + uid = self.send_itip_invitation(self.cars['mail'], datetime.datetime(2014,4,24, 12,0,0)) + + # one of the collection members accepted the reservation + accept = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('ACCEPTED') }) + self.assertIsInstance(accept, email.message.Message) + delegatee = self.find_resource_by_email(accept['from']) + + # book that resource for the next day + self.send_itip_invitation(delegatee['mail'], datetime.datetime(2014,4,25, 14,0,0)) + accept2 = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('ACCEPTED') }) + + # re-schedule first booking to a conflicting date + self.purge_mailbox(self.john['mailbox']) + update_template = itip_delegated.replace("resource-car-audia4@example.org", delegatee['mail']) + self.send_itip_update(delegatee['mail'], uid, datetime.datetime(2014,4,25, 12,0,0), template=update_template) + + # expect response from another member of the initially delegated collection + new_accept = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('ACCEPTED') }) + self.assertIsInstance(new_accept, email.message.Message) + + new_delegatee = self.find_resource_by_email(new_accept['from']) + self.assertNotEqual(delegatee['mail'], new_delegatee['mail']) + + # event now booked into new delegate's calendar + event = self.check_resource_calendar_event(new_delegatee['kolabtargetfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + + # old resource responds with a DELEGATED message + response = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('DELEGATED') }, delegatee['mail']) + self.assertIsInstance(response, email.message.Message) + + # old reservation was removed from old delegate's calendar + self.assertEqual(self.check_resource_calendar_event(delegatee['kolabtargetfolder'], uid), None) + + + def test_006_cancelling_revervation(self): + self.purge_mailbox(self.john['mailbox']) + + uid = self.send_itip_invitation(self.boxter['mail'], datetime.datetime(2014,5,1, 10,0,0)) + self.assertIsInstance(self.check_resource_calendar_event(self.boxter['kolabtargetfolder'], uid), pykolab.xml.Event) + + self.send_itip_cancel(self.boxter['mail'], uid) + + time.sleep(2) # wait for IMAP to update + self.assertEqual(self.check_resource_calendar_event(self.boxter['kolabtargetfolder'], uid), None) + + # make new reservation to the now free'd slot + self.send_itip_invitation(self.boxter['mail'], datetime.datetime(2014,5,1, 9,0,0)) + + response = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('ACCEPTED') }, self.boxter['mail']) + self.assertIsInstance(response, email.message.Message) + + + def test_007_update_delegated(self): + self.purge_mailbox(self.john['mailbox']) + + dt = datetime.datetime(2014,8,1, 12,0,0) + uid = self.send_itip_invitation(self.cars['mail'], dt) + + # wait for accept notification + accept = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('ACCEPTED') }) + self.assertIsInstance(accept, email.message.Message) + delegatee = self.find_resource_by_email(accept['from']) + + # send update message to all attendees (collection and delegatee) + self.purge_mailbox(self.john['mailbox']) + update_template = itip_delegated.replace("resource-car-audia4@example.org", delegatee['mail']) + self.send_itip_update(self.cars['mail'], uid, dt, template=update_template) + self.send_itip_update(delegatee['mail'], uid, dt, template=update_template) + + # get response from delegatee + accept = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('ACCEPTED') }) + self.assertIsInstance(accept, email.message.Message) + self.assertIn(delegatee['mail'], accept['from']) + + # no delegation response on updates + self.assertEqual(self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('DELEGATED') }, self.cars['mail']), None) + + + def test_008_allday_reservation(self): + self.purge_mailbox(self.john['mailbox']) + + uid = self.send_itip_invitation(self.audi['mail'], datetime.datetime(2014,6,2), True) + + accept = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('ACCEPTED') }) + self.assertIsInstance(accept, email.message.Message) + + event = self.check_resource_calendar_event(self.audi['kolabtargetfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + self.assertIsInstance(event.get_start(), datetime.date) + + uid2 = self.send_itip_invitation(self.audi['mail'], datetime.datetime(2014,6,2, 16,0,0)) + response = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('DECLINED') }, self.audi['mail']) + self.assertIsInstance(response, email.message.Message) + + + def test_009_recurring_events(self): + self.purge_mailbox(self.john['mailbox']) + + # register an infinitely recurring resource invitation + uid = self.send_itip_invitation(self.audi['mail'], datetime.datetime(2014,2,20, 12,0,0), + template=itip_recurring.replace(";COUNT=10", "")) + + accept = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('ACCEPTED') }) + self.assertIsInstance(accept, email.message.Message) + + # check non-recurring against recurring + uid2 = self.send_itip_invitation(self.audi['mail'], datetime.datetime(2014,3,13, 10,0,0)) + response = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('DECLINED') }, self.audi['mail']) + self.assertIsInstance(response, email.message.Message) + + self.purge_mailbox(self.john['mailbox']) + + # check recurring against recurring + uid3 = self.send_itip_invitation(self.audi['mail'], datetime.datetime(2014,2,22, 8,0,0), template=itip_recurring) + accept = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('ACCEPTED') }) + self.assertIsInstance(accept, email.message.Message) + + + def test_010_invalid_bookings(self): + self.purge_mailbox(self.john['mailbox']) + + itip_other = itip_invitation.replace("mailto:%s", "mailto:some-other-resource@example.org\nCOMMENT: Sent to %s") + self.send_itip_invitation(self.audi['mail'], datetime.datetime(2014,3,22, 8,0,0), template=itip_other) + + time.sleep(1) + + itip_invalid = itip_invitation.replace("DTSTART;", "X-DTSTART;") + self.send_itip_invitation(self.audi['mail'], datetime.datetime(2014,3,24, 19,30,0), template=itip_invalid) + + self.assertEqual(self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('ACCEPTED') }, self.audi['mail']), None) + + + def test_011_owner_info(self): + self.purge_mailbox(self.john['mailbox']) + + self.send_itip_invitation(self.room1['mail'], datetime.datetime(2014,6,19, 16,0,0)) + + accept = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('ACCEPTED') }, self.room1['mail']) + self.assertIsInstance(accept, email.message.Message) + respose_text = str(accept.get_payload(0)) + self.assertIn(self.jane['mail'], respose_text) + self.assertIn(self.jane['displayname'], respose_text) + + + def test_011_owner_info_from_collection(self): + self.purge_mailbox(self.john['mailbox']) + + self.send_itip_invitation(self.room2['mail'], datetime.datetime(2014,6,19, 16,0,0)) + + accept = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('ACCEPTED') }, self.room2['mail']) + self.assertIsInstance(accept, email.message.Message) + respose_text = str(accept.get_payload(0)) + print respose_text, self.jane['mail'] + self.assertIn(self.jane['mail'], respose_text) + self.assertIn(self.jane['displayname'], respose_text) + + + def test_012_owner_notification(self): + self.purge_mailbox(self.john['mailbox']) + self.purge_mailbox(self.jane['mailbox']) + + self.send_itip_invitation(self.room1['mail'], datetime.datetime(2014,8,4, 13,0,0)) + + # check notification message sent to resource owner (jane) + notify = self.check_message_received(_('Booking for %s has been %s') % (self.room1['cn'], participant_status_label('ACCEPTED')), self.room1['mail'], self.jane['mailbox']) + self.assertIsInstance(notify, email.message.Message) + + notification_text = str(notify.get_payload()) + self.assertIn(self.john['mail'], notification_text) + self.assertIn(participant_status_label('ACCEPTED'), notification_text) + + self.purge_mailbox(self.john['mailbox']) + + # check notification sent to collection owner (jane) + self.send_itip_invitation(self.rooms['mail'], datetime.datetime(2014,8,4, 12,30,0)) + + # one of the collection members accepted the reservation + accepted = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('ACCEPTED') }) + delegatee = self.find_resource_by_email(accepted['from']) + + notify = self.check_message_received(_('Booking for %s has been %s') % (delegatee['cn'], participant_status_label('ACCEPTED')), delegatee['mail'], self.jane['mailbox']) + self.assertIsInstance(notify, email.message.Message) + self.assertIn(self.john['mail'], notification_text) + + + def test_013_owner_confirmation_accept(self): + self.purge_mailbox(self.john['mailbox']) + self.purge_mailbox(self.jane['mailbox']) + + uid = self.send_itip_invitation(self.room3['mail'], datetime.datetime(2014,9,12, 14,0,0)) + + # requester (john) gets a TENTATIVE confirmation + response = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('TENTATIVE') }, self.room3['mail']) + self.assertIsInstance(response, email.message.Message) + + event = self.check_resource_calendar_event(self.room3['kolabtargetfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + self.assertEqual(event.get_summary(), "test") + self.assertEqual(event.get_attendee_by_email(self.room3['mail']).get_participant_status(True), 'TENTATIVE') + + # check confirmation message sent to resource owner (jane) + notify = self.check_message_received(_('Booking request for %s requires confirmation') % (self.room3['cn']), mailbox=self.jane['mailbox']) + self.assertIsInstance(notify, email.message.Message) + + # resource owner confirms reservation request + itip_event = events_from_message(notify)[0] + self.send_owner_response(itip_event['xml'], 'ACCEPTED', from_addr=self.jane['mail']) + + # requester (john) now gets the ACCEPTED response + response = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('ACCEPTED') }, self.room3['mail']) + self.assertIsInstance(response, email.message.Message) + + event = self.check_resource_calendar_event(self.room3['kolabtargetfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + self.assertEqual(event.get_status(True), 'CONFIRMED') + self.assertEqual(event.get_attendee_by_email(self.room3['mail']).get_participant_status(True), 'ACCEPTED') + + + def test_014_owner_confirmation_decline(self): + self.purge_mailbox(self.john['mailbox']) + self.purge_mailbox(self.jane['mailbox']) + + uid = self.send_itip_invitation(self.room3['mail'], datetime.datetime(2014,9,14, 9,0,0)) + + # requester (john) gets a TENTATIVE confirmation + response = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('TENTATIVE') }, self.room3['mail']) + self.assertIsInstance(response, email.message.Message) + + # check confirmation message sent to resource owner (jane) + notify = self.check_message_received(_('Booking request for %s requires confirmation') % (self.room3['cn']), mailbox=self.jane['mailbox']) + self.assertIsInstance(notify, email.message.Message) + + itip_event = events_from_message(notify)[0] + + # resource owner declines reservation request + itip_reply = itip_event['xml'].to_message_itip(self.jane['mail'], + method="REPLY", + participant_status='DECLINED', + message_text="Request declined", + subject=_('Booking for %s has been %s') % (self.room3['cn'], participant_status_label('DECLINED')) + ) + + smtp = smtplib.SMTP('localhost', 10026) + smtp.sendmail(self.jane['mail'], str(itip_event['organizer']), str(itip_reply)) + smtp.quit() + + # requester (john) now gets the DECLINED response + response = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('DECLINED') }, self.room3['mail']) + self.assertIsInstance(response, email.message.Message) + + # tentative reservation was set to cancelled + event = self.check_resource_calendar_event(self.room3['kolabtargetfolder'], uid) + self.assertEqual(event, None) + #self.assertEqual(event.get_status(True), 'CANCELLED') + #self.assertEqual(event.get_attendee_by_email(self.room3['mail']).get_participant_status(True), 'DECLINED') + + + def test_015_owner_confirmation_update(self): + self.purge_mailbox(self.john['mailbox']) + + uid = self.send_itip_invitation(self.room3['mail'], datetime.datetime(2014,8,19, 9,0,0), uid="http://a-totally.stupid/?uid") + + # requester (john) gets a TENTATIVE confirmation + response = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('TENTATIVE') }, self.room3['mail']) + self.assertIsInstance(response, email.message.Message) + + # check first confirmation message sent to resource owner (jane) + notify1 = self.check_message_received(_('Booking request for %s requires confirmation') % (self.room3['cn']), mailbox=self.jane['mailbox']) + self.assertIsInstance(notify1, email.message.Message) + + itip_event1 = events_from_message(notify1)[0] + self.assertEqual(itip_event1['start'].hour, 9) + + self.purge_mailbox(self.jane['mailbox']) + self.purge_mailbox(self.john['mailbox']) + + # send update with new date (and sequence) + self.send_itip_update(self.room3['mail'], uid, datetime.datetime(2014,8,19, 16,0,0)) + + event = self.check_resource_calendar_event(self.room3['kolabtargetfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + self.assertEqual(event.get_attendee_by_email(self.room3['mail']).get_participant_status(True), 'TENTATIVE') + + # check second confirmation message sent to resource owner (jane) + notify2 = self.check_message_received(_('Booking request for %s requires confirmation') % (self.room3['cn']), mailbox=self.jane['mailbox']) + self.assertIsInstance(notify2, email.message.Message) + + itip_event2 = events_from_message(notify2)[0] + self.assertEqual(itip_event2['start'].hour, 16) + + # resource owner declines the first reservation request + itip_reply = itip_event1['xml'].to_message_itip(self.jane['mail'], + method="REPLY", + participant_status='DECLINED', + message_text="Request declined", + subject=_('Booking for %s has been %s') % (self.room3['cn'], participant_status_label('DECLINED')) + ) + smtp = smtplib.SMTP('localhost', 10026) + smtp.sendmail(self.jane['mail'], str(itip_event1['organizer']), str(itip_reply)) + smtp.quit() + + time.sleep(5) + + # resource owner accpets the second reservation request + itip_reply = itip_event2['xml'].to_message_itip(self.jane['mail'], + method="REPLY", + participant_status='ACCEPTED', + message_text="Request accepred", + subject=_('Booking for %s has been %s') % (self.room3['cn'], participant_status_label('ACCEPTED')) + ) + smtp = smtplib.SMTP('localhost', 10026) + smtp.sendmail(self.jane['mail'], str(itip_event2['organizer']), str(itip_reply)) + smtp.quit() + + # requester (john) now gets the ACCEPTED response + response = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('ACCEPTED') }, self.room3['mail']) + self.assertIsInstance(response, email.message.Message) + + event = self.check_resource_calendar_event(self.room3['kolabtargetfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + self.assertEqual(event.get_attendee_by_email(self.room3['mail']).get_participant_status(True), 'ACCEPTED') + + + def test_016_collection_owner_confirmation(self): + self.purge_mailbox(self.john['mailbox']) + + uid = self.send_itip_invitation(self.viprooms['mail'], datetime.datetime(2014,8,15, 17,0,0)) + + # resource collection responds with a DELEGATED message + response = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('DELEGATED') }, self.viprooms['mail']) + self.assertIsInstance(response, email.message.Message) + + # the collection member tentatively accepted the reservation + accept = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('TENTATIVE') }) + self.assertIsInstance(accept, email.message.Message) + self.assertIn(self.room3['mail'], accept['from']) + + # check confirmation message sent to resource owner (jane) + notify = self.check_message_received(_('Booking request for %s requires confirmation') % (self.room3['cn']), mailbox=self.jane['mailbox']) + self.assertIsInstance(notify, email.message.Message) + + + def test_017_reschedule_single_occurrence(self): + self.purge_mailbox(self.john['mailbox']) + + # register a recurring resource invitation + start = datetime.datetime(2015,2,10, 12,0,0) + uid = self.send_itip_invitation(self.audi['mail'], start, template=itip_recurring) + + accept = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('ACCEPTED') }) + self.assertIsInstance(accept, email.message.Message) + + self.purge_mailbox(self.john['mailbox']) + + # send rescheduling request to a single instance + exdate = start + datetime.timedelta(days=14) + exstart = exdate + datetime.timedelta(hours=5) + self.send_itip_update(self.audi['mail'], uid, exstart, instance=exdate) + + accept = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('ACCEPTED') }) + self.assertIsInstance(accept, email.message.Message) + self.assertIn("RECURRENCE-ID;TZID=Europe/London:" + exdate.strftime('%Y%m%dT%H%M%S'), str(accept)) + + event = self.check_resource_calendar_event(self.audi['kolabtargetfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + self.assertEqual(len(event.get_exceptions()), 1) + + # send new invitation for now free slot + uid = self.send_itip_invitation(self.audi['mail'], exdate, template=itip_invitation.replace('SUMMARY:test', 'SUMMARY:new')) + + accept = self.check_message_received(self.itip_reply_subject % { 'summary':'new', 'status':participant_status_label('ACCEPTED') }) + self.assertIsInstance(accept, email.message.Message) + + # send rescheduling request to that single instance again: now conflicting + exdate = start + datetime.timedelta(days=14) + exstart = exdate + datetime.timedelta(hours=2) + self.send_itip_update(self.audi['mail'], uid, exstart, instance=exdate, sequence=3) + + response = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('DECLINED') }) + self.assertIsInstance(response, email.message.Message) + self.assertIn("RECURRENCE-ID;TZID=Europe/London:", str(response)) + + + def test_018_invite_single_occurrence(self): + self.purge_mailbox(self.john['mailbox']) + self.purge_mailbox(self.boxter['kolabtargetfolder']) + + start = datetime.datetime(2015,3,2, 18,30,0) + uid = self.send_itip_invitation(self.boxter['mail'], start, instance=start) + + accept = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('ACCEPTED') }) + self.assertIsInstance(accept, email.message.Message) + self.assertIn("RECURRENCE-ID;TZID=Europe/London:" + start.strftime('%Y%m%dT%H%M%S'), str(accept)) + + self.purge_mailbox(self.john['mailbox']) + + # send a second invitation for another instance with the same UID + nextstart = datetime.datetime(2015,3,9, 18,30,0) + self.send_itip_invitation(self.boxter['mail'], nextstart, uid=uid, instance=nextstart) + + accept = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('ACCEPTED') }) + self.assertIsInstance(accept, email.message.Message) + self.assertIn("RECURRENCE-ID;TZID=Europe/London:" + nextstart.strftime('%Y%m%dT%H%M%S'), str(accept)) + + self.purge_mailbox(self.john['mailbox']) + + # send rescheduling request to the first instance + exstart = start + datetime.timedelta(hours=2) + self.send_itip_update(self.boxter['mail'], uid, exstart, instance=start) + + accept = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('ACCEPTED') }) + self.assertIsInstance(accept, email.message.Message) + self.assertIn("RECURRENCE-ID;TZID=Europe/London:" + start.strftime('%Y%m%dT%H%M%S'), str(accept)) + + # the resource calendar now has two reservations stored in one object + one = self.check_resource_calendar_event(self.boxter['kolabtargetfolder'], uid) + self.assertIsInstance(one, pykolab.xml.Event) + self.assertIsInstance(one.get_recurrence_id(), datetime.datetime) + self.assertEqual(one.get_start().hour, exstart.hour) + + two = one.get_instance(nextstart) + self.assertIsInstance(two, pykolab.xml.Event) + self.assertIsInstance(two.get_recurrence_id(), datetime.datetime) + + self.purge_mailbox(self.john['mailbox']) + + # send rescheduling request to the 2nd instance + self.send_itip_update(self.boxter['mail'], uid, nextstart + datetime.timedelta(hours=2), sequence=2, instance=nextstart) + + accept = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('ACCEPTED') }) + self.assertIsInstance(accept, email.message.Message) + self.assertIn("RECURRENCE-ID;TZID=Europe/London:" + nextstart.strftime('%Y%m%dT%H%M%S'), str(accept)) + + event = self.check_resource_calendar_event(self.boxter['kolabtargetfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + self.assertEqual(len(event.get_exceptions()), 1) + + two = event.get_instance(nextstart) + self.assertIsInstance(two, pykolab.xml.Event) + self.assertEqual(two.get_sequence(), 2) + self.assertEqual(two.get_start().hour, 20) + + + def test_019_cancel_single_occurrence(self): + self.purge_mailbox(self.john['mailbox']) + + # register a recurring resource invitation + start = datetime.datetime(2015,2,12, 14,0,0) + uid = self.send_itip_invitation(self.passat['mail'], start, template=itip_recurring) + + accept = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('ACCEPTED') }) + self.assertIsInstance(accept, email.message.Message) + + exdate = start + datetime.timedelta(days=7) + self.send_itip_cancel(self.passat['mail'], uid, instance=exdate) + + time.sleep(5) # wait for IMAP to update + event = self.check_resource_calendar_event(self.passat['kolabtargetfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + self.assertEqual(len(event.get_exceptions()), 1) + + exception = event.get_instance(exdate) + self.assertEqual(exception.get_status(True), 'CANCELLED') + self.assertTrue(exception.get_transparency()) + + self.purge_mailbox(self.john['mailbox']) + + # store a single occurrence with recurrence-id + start = datetime.datetime(2015,3,2, 18,30,0) + uid = self.send_itip_invitation(self.passat['mail'], start, instance=start) + + accept = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('ACCEPTED') }) + self.assertIsInstance(accept, email.message.Message) + + self.send_itip_cancel(self.passat['mail'], uid, instance=start) + + time.sleep(5) # wait for IMAP to update + self.assertEqual(self.check_resource_calendar_event(self.passat['kolabtargetfolder'], uid), None) + + + def test_020_owner_confirmation_single_occurrence(self): + self.purge_mailbox(self.john['mailbox']) + self.purge_mailbox(self.jane['mailbox']) + + start = datetime.datetime(2015,4,18, 14,0,0) + uid = self.send_itip_invitation(self.room3['mail'], start, template=itip_recurring) + + notify = self.check_message_received(_('Booking request for %s requires confirmation') % (self.room3['cn']), mailbox=self.jane['mailbox']) + self.assertIsInstance(notify, email.message.Message) + + # resource owner confirms reservation request (entire series) + itip_event = events_from_message(notify)[0] + self.send_owner_response(itip_event['xml'], 'ACCEPTED', from_addr=self.jane['mail']) + + self.purge_mailbox(self.john['mailbox']) + self.purge_mailbox(self.jane['mailbox']) + + # send rescheduling request to a single instance + exdate = start + datetime.timedelta(days=14) + exstart = exdate + datetime.timedelta(hours=4) + self.send_itip_update(self.room3['mail'], uid, exstart, instance=exdate) + + # check confirmation message sent to resource owner (jane) + notify = self.check_message_received(_('Booking request for %s requires confirmation') % (self.room3['cn']), mailbox=self.jane['mailbox']) + self.assertIsInstance(notify, email.message.Message) + self.assertIn("RECURRENCE-ID;TZID=Europe/London:" + exdate.strftime('%Y%m%dT%H%M%S'), str(notify)) + + itip_event = events_from_message(notify)[0] + self.assertIsInstance(itip_event['xml'].get_recurrence_id(), datetime.datetime) + + # resource owner declines reservation request + self.send_owner_response(itip_event['xml'], 'DECLINED', from_addr=self.jane['mail']) + + # requester (john) now gets the DECLINED response + response = self.check_message_received(self.itip_reply_subject % { 'summary':'test', 'status':participant_status_label('DECLINED') }, self.room3['mail']) + self.assertIsInstance(response, email.message.Message) + self.assertIn("RECURRENCE-ID;TZID=Europe/London:" + exdate.strftime('%Y%m%dT%H%M%S'), str(response)) + + event = self.check_resource_calendar_event(self.room3['kolabtargetfolder'], uid) + self.assertIsInstance(event, pykolab.xml.Event) + self.assertEqual(len(event.get_exceptions()), 1) + + exception = event.get_instance(exdate) + self.assertEqual(exception.get_attendee_by_email(self.room3['mail']).get_participant_status(True), 'DECLINED') + + +if __name__ == '__main__': + stick.main() + \ No newline at end of file diff --git a/integration/P/test_wallace_resource_invitation_data.py b/integration/P/test_wallace_resource_invitation_data.py new file mode 100644 index 0000000..e58d132 --- /dev/null +++ b/integration/P/test_wallace_resource_invitation_data.py @@ -0,0 +1,157 @@ + +itip_invitation = """ +BEGIN:VCALENDAR +VERSION:2.0 +PRODID:-//Roundcube Webmail 0.9-0.3.el6.kolab_3.0//NONSGML Calendar//EN +CALSCALE:GREGORIAN +METHOD:REQUEST +BEGIN:VEVENT +UID:%s%s +DTSTAMP:20140213T125414Z +DTSTART;TZID=Europe/London:%s +DTEND;TZID=Europe/London:%s +SUMMARY:test +DESCRIPTION:test +ORGANIZER;CN="Doe, John":mailto:john.doe@example.org +ATTENDEE;ROLE=REQ-PARTICIPANT;CUTYPE=RESOURCE;PARTSTAT=NEEDS-ACTION;RSVP=TRUE:mailto:%s +ATTENDEE;ROLE=REQ-PARTICIPANT;PARTSTAT=TENTATIVE;CN=Somebody Else:mailto:somebody@else.com +TRANSP:OPAQUE +END:VEVENT +END:VCALENDAR +""" + +itip_update = """ +BEGIN:VCALENDAR +VERSION:2.0 +PRODID:-//Roundcube Webmail 0.9-0.3.el6.kolab_3.0//NONSGML Calendar//EN +CALSCALE:GREGORIAN +METHOD:REQUEST +BEGIN:VEVENT +UID:%s%s +DTSTAMP:20140215T125414Z +DTSTART;TZID=Europe/London:%s +DTEND;TZID=Europe/London:%s +SEQUENCE:2 +SUMMARY:test +DESCRIPTION:test +ORGANIZER;CN="Doe, John":mailto:john.doe@example.org +ATTENDEE;ROLE=REQ-PARTICIPANT;CUTYPE=RESOURCE;PARTSTAT=NEEDS-ACTION;RSVP=TRUE:mailto:%s +TRANSP:OPAQUE +END:VEVENT +END:VCALENDAR +""" + +itip_delegated = """ +BEGIN:VCALENDAR +VERSION:2.0 +PRODID:-//Roundcube//Roundcube libcalendaring 1.0-git//Sabre//Sabre VObject + 2.1.3//EN +CALSCALE:GREGORIAN +METHOD:REQUEST +BEGIN:VEVENT +UID:%s%s +DTSTAMP;VALUE=DATE-TIME:20140227T141939Z +DTSTART;VALUE=DATE-TIME;TZID=Europe/London:%s +DTEND;VALUE=DATE-TIME;TZID=Europe/London:%s +SUMMARY:test +SEQUENCE:4 +ATTENDEE;CN=Company Cars;PARTSTAT=DELEGATED;ROLE=NON-PARTICIPANT;CUTYPE=IND + IVIDUAL;RSVP=TRUE;DELEGATED-TO=resource-car-audia4@example.org:mailto:reso + urce-collection-companycars@example.org +ATTENDEE;CN=The Delegate;PARTSTAT=ACCEPTED;ROLE=REQ-PARTICIPANT;CUTYPE=INDI + VIDUAL;RSVP=TRUE;DELEGATED-FROM=resource-collection-companycars@example.or + g:mailto:resource-car-audia4@example.org +ORGANIZER;CN=:mailto:john.doe@example.org +DESCRIPTION:Sent to %s +END:VEVENT +END:VCALENDAR +""" + +itip_cancellation = """ +BEGIN:VCALENDAR +VERSION:2.0 +PRODID:-//Roundcube Webmail 0.9-0.3.el6.kolab_3.0//NONSGML Calendar//EN +CALSCALE:GREGORIAN +METHOD:CANCEL +BEGIN:VEVENT +UID:%s%s +DTSTAMP:20140218T125414Z +DTSTART;TZID=Europe/London:20120713T100000 +DTEND;TZID=Europe/London:20120713T110000 +SUMMARY:test +DESCRIPTION:test +ORGANIZER;CN="Doe, John":mailto:john.doe@example.org +ATTENDEE;ROLE=REQ-PARTICIPANT;PARTSTAT=ACCEPTED;RSVP=TRUE:mailt= + o:%s +TRANSP:OPAQUE +SEQUENCE:3 +END:VEVENT +END:VCALENDAR +""" + +itip_allday = """ +BEGIN:VCALENDAR +VERSION:2.0 +PRODID:-//Roundcube Webmail 0.9-0.3.el6.kolab_3.0//NONSGML Calendar//EN +CALSCALE:GREGORIAN +METHOD:REQUEST +BEGIN:VEVENT +UID:%s%s +DTSTAMP:20140213T125414Z +DTSTART;VALUE=DATE:%s +DTEND;VALUE=DATE:%s +SUMMARY:test +DESCRIPTION:test +ORGANIZER;CN="Doe, John":mailto:john.doe@example.org +ATTENDEE;ROLE=REQ-PARTICIPANT;CUTYPE=RESOURCE;PARTSTAT=NEEDS-ACTION;RSVP=TRUE:mailto:%s +TRANSP:OPAQUE +END:VEVENT +END:VCALENDAR +""" + + +itip_recurring = """ +BEGIN:VCALENDAR +VERSION:2.0 +PRODID:-//Apple Inc.//Mac OS X 10.9.2//EN +CALSCALE:GREGORIAN +METHOD:REQUEST +BEGIN:VEVENT +UID:%s%s +DTSTAMP:20140213T125414Z +DTSTART;TZID=Europe/London:%s +DTEND;TZID=Europe/London:%s +RRULE:FREQ=WEEKLY;INTERVAL=1;COUNT=10 +SUMMARY:test +DESCRIPTION:test +ORGANIZER;CN="Doe, John":mailto:john.doe@example.org +ATTENDEE;ROLE=REQ-PARTICIPANT;CUTYPE=RESOURCE;PARTSTAT=NEEDS-ACTION;RSVP=TRUE:mailto:%s +TRANSP:OPAQUE +END:VEVENT +END:VCALENDAR +""" + +mime_message = """MIME-Version: 1.0 +Content-Type: multipart/mixed; + boundary="=_c8894dbdb8baeedacae836230e3436fd" +From: "Doe, John" +Date: Tue, 25 Feb 2014 13:54:14 +0100 +Message-ID: <240fe7ae7e139129e9eb95213c1016d7@example.org> +User-Agent: Roundcube Webmail/0.9-0.3.el6.kolab_3.0 +To: %s +Subject: "test" + +--=_c8894dbdb8baeedacae836230e3436fd +Content-Type: text/plain; charset=UTF-8; format=flowed +Content-Transfer-Encoding: quoted-printable + +*test* + +--=_c8894dbdb8baeedacae836230e3436fd +Content-Type: text/calendar; charset=UTF-8; method=REQUEST; name=event.ics +Content-Disposition: attachment; filename=event.ics +Content-Transfer-Encoding: 8bit + +%s +--=_c8894dbdb8baeedacae836230e3436fd-- +"""