1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
|
#!/usr/bin/env python
"""
A handler to help with testing.
Copyright (C) 2014, 2015, 2016 Paul Boddie <paul@boddie.org.uk>
This program is free software; you can redistribute it and/or modify it under
the terms of the GNU General Public License as published by the Free Software
Foundation; either version 3 of the License, or (at your option) any later
version.
This program is distributed in the hope that it will be useful, but WITHOUT
ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
details.
You should have received a copy of the GNU General Public License along with
this program. If not, see <http://www.gnu.org/licenses/>.
"""
from imiptools.client import ClientForObject
from imiptools.data import Object, get_address, parse_object
from imiptools.dates import get_datetime, to_timezone
from imiptools.mail import Messenger
from imiptools.period import RecurringPeriod
from imiptools.stores import get_store, get_journal
from os.path import split
import sys
class TestClient(ClientForObject):
"""
A content handler for use in testing, as opposed to operating within the
mail processing pipeline.
"""
# Action methods.
def handle_request(self, action, start=None, end=None, recurrenceid=None):
"""
Process the current request for the current user. Return whether the
given 'action' was taken.
If 'start' and 'end' are specified, they will be used in any
counter-proposal.
Where 'recurrenceid' is specified and refers to a new recurrence, the
action will apply only to this new recurrence.
"""
have_new_recurrence = self.obj.get_recurrenceid() != recurrenceid
if have_new_recurrence:
self.obj["RECURRENCE-ID"] = [(recurrenceid, {})]
self.obj.remove_all(["RDATE", "RRULE"])
# Reply only on behalf of this user.
if action in ("accept", "decline"):
attendee_attr = self.update_participation(action == "accept" and "ACCEPTED" or "DECLINED")
method = "REPLY"
elif action == "counter":
attendee_attr = self.obj.get_value_map("ATTENDEE").get(self.user)
method = "COUNTER"
# Nothing else is supported.
else:
return None
# For counter-proposals or new recurrences, set a new main period for
# the event.
if action == "counter" or have_new_recurrence:
period = self.obj.get_main_period(self.get_tzid())
# Use the existing or configured time zone for the specified
# datetimes.
start = to_timezone(get_datetime(start), period.tzid)
end = to_timezone(get_datetime(end), period.tzid)
period = RecurringPeriod(start, end, period.tzid, period.origin, period.get_start_attr(), period.get_end_attr())
self.obj.set_period(period)
# Where no attendees remain, no message is generated.
if not attendee_attr:
return None
# NOTE: This is a simpler form of the code in imipweb.client.
organiser = get_address(self.obj.get_value("ORGANIZER"))
self.update_sender(attendee_attr)
self.obj["ATTENDEE"] = [(self.user, attendee_attr)]
self.update_dtstamp()
self.update_sequence(False)
message = self.messenger.make_outgoing_message(
[self.obj.to_part(method)],
[organiser],
outgoing_bcc=get_address(self.user)
)
return message.as_string()
# A simple main program that attempts to handle a stored request, writing the
# response message to standard output.
if __name__ == "__main__":
progname = split(sys.argv[0])[-1]
try:
action, store_type, store_dir, journal_dir, preferences_dir, user = sys.argv[1:7]
if len(sys.argv) >= 10:
start, end = sys.argv[7:9]
i = 9
else:
start, end = None, None
i = 7
uid, recurrenceid = (sys.argv[i:i+2] + [None] * 2)[:2]
except ValueError:
print >>sys.stderr, """\
Usage: %s <action> <store type> <store directory> <journal directory>
<preferences directory> <user URI> [ <start> <end> ]
<uid> <recurrence-id>
Need 'accept', 'counter' or 'decline', a store type, a store directory, a
journal directory, a preferences directory, user URI, any counter-proposal or
new recurrence datetimes (see below), plus the appropriate event UID and
RECURRENCE-ID (if a recurrence is involved).
The RECURRENCE-ID must be in exactly the form employed by the store, not a
different but equivalent representation, if the identifier is to refer to an
existing recurrence.
Alternatively, omit the UID and RECURRENCE-ID and provide event-only details on
standard input to force the script to handle an event not already present in the
store.
If 'counter' has been indicated, alternative start and end datetimes are also
required. If a specific recurrence is being separated from an event, such
datetimes are also required in order to set the main period of the recurrence.
"""
sys.exit(1)
store = get_store(store_type, store_dir)
journal = get_journal(store_type, journal_dir)
if uid is not None:
fragment = store.get_event(user, uid, recurrenceid)
# Permit new recurrences by getting the parent object.
if not fragment:
fragment = store.get_event(user, uid)
if not fragment:
print >>sys.stderr, "No such event:", uid, recurrenceid
sys.exit(1)
else:
fragment = parse_object(sys.stdin, "utf-8")
obj = Object(fragment)
handler = TestClient(obj, user, Messenger(), store, None, journal, preferences_dir)
response = handler.handle_request(action, start, end, recurrenceid)
if response:
if uid is not None:
store.dequeue_request(user, uid, recurrenceid)
print response
else:
sys.exit(1)
# vim: tabstop=4 expandtab shiftwidth=4
|
