amorfo77/sogo
1
0
Fork 0
mirror of https://github.com/inverse-inc/sogo.git synced 2026-02-17 07:33:57 +00:00
Code Issues Projects Releases Wiki Activity

test: migration from Python to JavaScript

Browse Source
This commit is contained in:
Francis Lachapelle
2021-11-17 19:13:36 -05:00
parent 8cec92ea87
commit 1a7ba3d4ef
36 changed files with 14 additions and 6633 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
Tests/Integration/GNUmakefile
View File

@@ -1,8 +0,0 @@
include $(GNUSTEP_MAKEFILES)/common.make
TOOL_NAME = teststrings
teststrings_OBJC_FILES += \
teststrings.m
-include GNUmakefile.preamble
include $(GNUSTEP_MAKEFILES)/tool.make

19
Tests/Integration/GNUmakefile.preamble
View File

@@ -1,19 +0,0 @@
# compile settings
ADDITIONAL_CPPFLAGS += \
-DSOGO_MAJOR_VERSION=$(MAJOR_VERSION) \
-DSOGO_MINOR_VERSION=$(MINOR_VERSION) \
-DSOGO_SUBMINOR_VERSION=$(SUBMINOR_VERSION) \
-DSOGO_LIBDIR="@\"$(SOGO_LIBDIR)\""
ADDITIONAL_INCLUDE_DIRS += \
-D_GNU_SOURCE -I../../SOPE/ -I../../SoObjects/
ADDITIONAL_LIB_DIRS += \
-L../../SoObjects/SOGo/SOGo.framework/Versions/Current/sogo -lSOGo \
-L../../SOPE/GDLContentStore/$(GNUSTEP_OBJ_DIR)/ -lGDLContentStore \
-L../../SOPE/NGCards/$(GNUSTEP_OBJ_DIR)/ -lNGCards \
-L/usr/local/lib/sogo -L/usr/lib/sogo -L/usr/lib64/sogo -lEOControl -lNGStreams -lNGMime -lNGExtensions
ADDITIONAL_LDFLAGS += -Wl,--no-as-needed -Wl,--rpath,$(GNUSTEP_SYSTEM_LIBRARIES)/sogo

58
Tests/Integration/README
View File

@@ -1,58 +0,0 @@
setup
-----
(you need "python-simplejson", "python-xml", "python-vobject", "python-dateutil" and "python-m2crypto"
in order to run the scripts on Debian)
1) copy config.py.in to config.py (make sure to never EVER add it to git)
2) edit config.py to suit your environment
3) make sure that you use a fresh database, with no prior information in it
4) make sure that SOGoCalendarDefaultRoles and SOGoContactsDefaultRoles are empty or undefined
5) run the test scripts
runnable scripts
----------------
all.py - run all scripts below at once
test-webdavsync.py - explicit
test-davacl.py - dav acl tests for calendar and addressbook modules
other scripts
-------------
propfind.py - a sample implementation of a PROPFIND request using webdavlib
* developers
------------
- Test methods are always prefixed with "test". Sometimes, it's easier to
track down a problem by enabling only one test at a time. One possible method
is to replace "def test" with "def xtest" and replace it back when the
problems are solved.
- Test failures start with "FAIL:". Those are the ones that indicate possible
bugs in the application, if the test is itself known to work.
For example like this:
======================================================================
FAIL: 'modify' PUBLIC, 'view all' PRIVATE, 'view d&t' confidential
----------------------------------------------------------------------
Traceback (most recent call last):
File "./davacl.py", line 75, in testModifyPublicViewAllPrivateViewDConfidential
self._testRights({ "pu": "m", "pr": "v", "co": "d" })
File "./davacl.py", line 119, in _testRights
self._testCreate(rights)
File "./davacl.py", line 165, in _testCreate
exp_code)
File "./davacl.py", line 107, in _putEvent
% (exp_status, put.response["status"]))
AssertionError: event creation/modification: expected status code '403' (received '201')
- Test errors start with "ERRORS" and most likely indicate a bug in the test
code itself.
- Always set a doc string on the test methods, especially for complex test
cases.
- When writing tests, be aware that contrarily to unit tests, functional tests
often imply a logical order between the different steps.

64
Tests/Integration/all.py
View File

@@ -1,64 +0,0 @@
#!/usr/bin/python
import os, sys, unittest, getopt, traceback, time
import preferences
import sogotests
import unittest
if __name__ == "__main__":
unittest._TextTestResult.oldStartTest = unittest._TextTestResult.startTest
unittest._TextTestResult.startTest = sogotests.UnitTestTextTestResultNewStartTest
unittest._TextTestResult.stopTest = sogotests.UnitTestTextTestResultNewStopTest
loader = unittest.TestLoader()
modules = []
languages = preferences.SOGoSupportedLanguages
# We can disable testing all languages
testLanguages = False
opts, args = getopt.getopt (sys.argv[1:], [], ["enable-languages"])
for o, a in opts:
if o == "--enable-languages":
testLanguages = True
for mod in os.listdir("."):
if mod.startswith("test-") and mod.endswith(".py"):
modules.append(mod[:-3])
__import__(mod[:-3])
if len(modules) > 0:
suite = loader.loadTestsFromNames(modules)
print "%d tests in modules: '%s'" % (suite.countTestCases(),
"', '".join(modules))
runner = unittest.TextTestRunner(verbosity=2)
if testLanguages:
prefs = preferences.preferences()
# Get the current language
userLanguageString = prefs.get ("SOGoLanguage")
if userLanguageString:
userLanguage = languages.index (userLanguageString)
else:
userLanguage = languages.index ("English")
for i in range (0, len (languages)):
try:
prefs.set ("SOGoLanguage", i)
except Exception, inst:
print '-' * 60
traceback.print_exc ()
print '-' * 60
print "Running test in %s (%d/%d)" % \
(languages[i], i + 1, len (languages))
runner.verbosity = 2
runner.run(suite)
# Revert to the original language
prefs.set ("SOGoLanguage", userLanguage)
else:
runner.run(suite)
else:
print "No test available."

144
Tests/Integration/carddav.py
View File

@@ -1,144 +0,0 @@
from config import hostname, port, username, password
import webdavlib
import simplejson
import sogoLogin
DEBUG=True
DEBUG=False
class HTTPPreferencesPOST (webdavlib.HTTPPOST):
cookie = None
def prepare_headers (self):
headers = webdavlib.HTTPPOST.prepare_headers(self)
if self.cookie:
headers["Cookie"] = self.cookie
return headers
class HTTPPreferencesGET (webdavlib.HTTPGET):
cookie = None
def prepare_headers (self):
headers = webdavlib.HTTPGET.prepare_headers(self)
if self.cookie:
headers["Cookie"] = self.cookie
return headers
class Carddav:
login = username
passw = password
def __init__(self, otherLogin = None, otherPassword = None):
if otherLogin and otherPassword:
self.login = otherLogin
self.passw = otherPassword
self.client = webdavlib.WebDAVClient(hostname, port)
authCookie = sogoLogin.getAuthCookie(hostname, port, self.login, self.passw)
self.cookie = authCookie
self.cards = None
self.calendars = None
self.fields = None
self.events = None
#- If this is not set, we CAN'T save preferences
self.preferences = None
def _get(self, url):
get = HTTPPreferencesGET(url)
get.cookie = self.cookie
self.client.execute(get)
if DEBUG: print "(url):", url
if DEBUG: print "(status):", get.response["status"]
if DEBUG: print "(body):", get.response['body']
content = simplejson.loads(get.response['body'])
return content
def _post(self, url, data, errormsg="failure POST"):
if DEBUG: print "URL:", url
post = HTTPPreferencesPOST(url, simplejson.dumps(data))
post.content_type = "application/json"
post.cookie = self.cookie
self.client.execute(post)
# Raise an exception if the pref wasn't properly set
if post.response["status"] != 200:
raise Exception ("%s, (code = %d)" % (errormsg, post.response["status"]))
def load_cards(self):
if not self.cards:
url = "/SOGo/so/%s/Contacts/personal/view" % (self.login)
content = self._get(url)
#print "\nCONTENT:", content
if 'headers' in content:
self.cards = []
fields = content['headers'][0]
for h in content['headers'][1:]:
card = {}
for i, f in enumerate(fields):
card[f] = h[i]
self.cards.append(card)
else:
self.cards = []
return self.cards
def get_cards(self, pattern):
self.load_cards()
return [a for a in self.cards if pattern in a.values()]
def get_card(self, idstr):
url = "/SOGo/so/%s/Contacts/personal/%s/view" % (self.login, idstr)
content = self._get(url)
return content
def save_card(self, card):
url = "/SOGo/so/%s/Contacts/personal/%s/saveAsContact" % (self.login, card['id'])
self._post(url, card, "failure saving card")
def load_calendars(self):
if not self.calendars:
url = "/SOGo/so/%s/Calendar/calendarslist" % (self.login)
content = self._get(url)
self.calendars = content['calendars']
return self.calendars
def get_calendars(self, pattern):
self.load_calendars()
return [a for a in self.calendars if pattern in a.values()]
def get_calendar(self, idstr):
self.load_calendars()
callist = [a for a in self.calendars if a['id'] == idstr]
if len(callist):
return callist[0]
return None
def save_calendar(self, calendar):
url = "/SOGo/so/%s/Contacts/personal/%s/saveAsContact" % (self.login, calendar['id'])
self._post(url, card, "failure saving calendar")
def load_events(self):
if not self.events:
url = "/SOGo/so/%s/Calendar/eventslist" % (self.login)
content = self._get(url)
self.fields = content['fields']
self.events = []
months = content['events']
for month in months.keys():
days = months[month]['days']
for day in days.keys():
tmp_events = days[day]['events']
self.events.extend(dict(zip(self.fields, event)) for event in tmp_events)
return self.events
def newguid(self, folderpath):
url = "/SOGo/so/%s/%s/newguid" % (self.login, folderpath)
content = self._get(url)
return content['id']
def save_event(self, event, folder, gid):
#url = "/SOGo/so/%s/%s/%s/save" % (self.login, event['c_folder'], event['c_name'])
url = "/SOGo/so/%s/%s/%s/saveAsAppointment" % (self.login, folder, gid)
self._post(url, event, "failure saving event")

38
Tests/Integration/config.py.in
View File

@@ -1,38 +0,0 @@
# setup: 4 user are needed: username, superuser, attendee1, attendee1_delegate
# superuser must be a sogo superuser...
hostname = "localhost"
port = "80"
username = "myuser"
password = "mypass"
superuser = "super"
superuser_password="pass"
subscriber_username = "otheruser"
subscriber_password = "otherpass"
attendee1 = "user@domain.com"
attendee1_username = "user"
attendee1_password = "pass"
attendee1_delegate = "user2@domain.com"
attendee1_delegate_username = "sogo2"
attendee1_delegate_password = "sogo"
resource_no_overbook = "res"
resource_can_overbook = "res-nolimit"
white_listed_attendee = '{"sogo1":"John Doe <sogo1@example.com>"}'
mailserver = "imaphost"
testput_nbrdays = 30
sieve_server = "localhost"
sieve_port = 2000
sogo_user = "sogo"
sogo_tool_path = "/usr/local/sbin/sogo-tool"
webCalendarURL = "http://inverse.ca/sogo-integration-tests/CanadaHolidays.ics"

92
Tests/Integration/ev_generator.py
View File

@@ -1,92 +0,0 @@
import time
def hours(nbr):
return nbr * 3600
def days(nbr):
return nbr * hours(24)
class ev_generator:
ev_templ = """
BEGIN:VCALENDAR\r
VERSION:2.0\r
PRODID:-//Inverse//Event Generator//EN\r
CALSCALE:GREGORIAN\r
BEGIN:VTIMEZONE\r
TZID:America/Montreal\r
BEGIN:DAYLIGHT\r
TZOFFSETFROM:-0500\r
TZOFFSETTO:-0400\r
DTSTART:20070311T020000\r
RRULE:FREQ=YEARLY;BYMONTH=3;BYDAY=2SU\r
TZNAME:EDT\r
END:DAYLIGHT\r
BEGIN:STANDARD\r
TZOFFSETFROM:-0400\r
TZOFFSETTO:-0500\r
DTSTART:20071104T020000\r
RRULE:FREQ=YEARLY;BYMONTH=11;BYDAY=1SU\r
TZNAME:EST\r
END:STANDARD\r
END:VTIMEZONE\r
BEGIN:VEVENT\r
SEQUENCE:4\r
TRANSP:OPAQUE\r
UID:%(uid)s\r
SUMMARY:%(summary)s\r
DTSTART;TZID=America/Montreal:%(start)s\r
DTEND;TZID=America/Montreal:%(end)s\r
CREATED:20080711T231608Z\r
DTSTAMP:20080711T231640Z\r
END:VEVENT\r
END:VCALENDAR\r
"""
def __init__(self, maxDays):
self.reset(maxDays)
def reset(self, maxDays):
self.maxDays = maxDays
self.currentDay = 0
self.currentStart = 0
today = time.mktime(time.localtime())
self.firstDay = today - days(maxDays + 30)
def _calendarDate(self, eventTime):
timeStruct = time.localtime(eventTime)
return time.strftime("%Y%m%dT%H0000", timeStruct)
def _iterValues(self):
event = None
if (self.currentDay < self.maxDays):
eventStart = (self.firstDay
+ days(self.currentDay)
+ hours(self.currentStart + 8))
eventEnd = eventStart + hours(1)
thatDay = time.localtime(int(eventStart))
uid = "Event%d%d" % (eventStart, eventEnd)
summary = "%s - event %d" % (time.strftime("%Y-%m-%d", thatDay),
self.currentStart)
start = self._calendarDate(eventStart)
end = self._calendarDate(eventEnd)
event = {'uid': uid,
'summary': summary,
'start': start,
'end': end}
self.currentStart = self.currentStart + 1
if (self.currentStart > 7):
self.currentStart = 0
self.currentDay = self.currentDay + 1
return event
def iter(self):
hasMore = False
entryValues = self._iterValues()
if (entryValues is not None):
self.event = (self.ev_templ % entryValues).strip()
hasMore = True
return hasMore

631
Tests/Integration/managesieve.py
View File

@@ -1,631 +0,0 @@
"""Sieve management client.
A Protocol for Remotely Managing Sieve Scripts
Based on <draft-martin-managesieve-04.txt>
"""
__version__ = "0.4.2"
__author__ = """Hartmut Goebel <h.goebel@crazy-compilers.com>
Ulrich Eck <ueck@net-labs.de> April 2001
"""
import binascii, re, socket, time, random, sys
try:
import ssl
ssl_wrap_socket = ssl.wrap_socket
except ImportError:
ssl_wrap_socket = socket.ssl
__all__ = [ 'MANAGESIEVE', 'SIEVE_PORT', 'OK', 'NO', 'BYE', 'Debug']
Debug = 0
CRLF = '\r\n'
SIEVE_PORT = 2000
OK = 'OK'
NO = 'NO'
BYE = 'BYE'
AUTH_PLAIN = "PLAIN"
AUTH_LOGIN = "LOGIN"
# authentication mechanisms currently supported
# in order of preference
AUTHMECHS = [AUTH_PLAIN, AUTH_LOGIN]
# todo: return results or raise exceptions?
# todo: on result 'BYE' quit immediatly
# todo: raise exception on 'BYE'?
# Commands
commands = {
# name valid states
'STARTTLS': ('NONAUTH',),
'AUTHENTICATE': ('NONAUTH',),
'LOGOUT': ('NONAUTH', 'AUTH', 'LOGOUT'),
'CAPABILITY': ('NONAUTH', 'AUTH'),
'GETSCRIPT': ('AUTH', ),
'PUTSCRIPT': ('AUTH', ),
'SETACTIVE': ('AUTH', ),
'DELETESCRIPT': ('AUTH', ),
'LISTSCRIPTS': ('AUTH', ),
'HAVESPACE': ('AUTH', ),
# bogus command to receive a NO after STARTTLS (see starttls() )
'BOGUS': ('NONAUTH', 'AUTH', 'LOGOUT'),
}
### needed
Oknobye = re.compile(r'(?P<type>(OK|NO|BYE))'
r'( \((?P<code>.*)\))?'
r'( (?P<data>.*))?')
# draft-martin-managesieve-04.txt defines the size tag of literals to
# contain a '+' (plus sign) behind the digits, but timsieved does not
# send one. Thus we are less strikt here:
Literal = re.compile(r'.*{(?P<size>\d+)\+?}$')
re_dquote = re.compile(r'"(([^"\\]|\\.)*)"')
re_esc_quote = re.compile(r'\\([\\"])')
class SSLFakeSocket:
"""A fake socket object that really wraps a SSLObject.
It only supports what is needed in managesieve.
"""
def __init__(self, realsock, sslobj):
self.realsock = realsock
self.sslobj = sslobj
def send(self, str):
self.sslobj.write(str)
return len(str)
sendall = send
def close(self):
self.realsock.close()
class SSLFakeFile:
"""A fake file like object that really wraps a SSLObject.
It only supports what is needed in managesieve.
"""
def __init__(self, sslobj):
self.sslobj = sslobj
def readline(self):
str = ""
chr = None
while chr != "\n":
chr = self.sslobj.read(1)
str += chr
return str
def read(self, size=0):
if size == 0:
return ''
else:
return self.sslobj.read(size)
def close(self):
pass
def sieve_name(name):
# todo: correct quoting
return '"%s"' % name
def sieve_string(string):
return '{%d+}%s%s' % ( len(string), CRLF, string )
class MANAGESIEVE:
"""Sieve client class.
Instantiate with: MANAGESIEVE(host [, port])
host - host's name (default: localhost)
port - port number (default: standard Sieve port).
use_tls - switch to TLS automatically, if server supports
keyfile - keyfile to use for TLS (optional)
certfile - certfile to use for TLS (optional)
All Sieve commands are supported by methods of the same
name (in lower-case).
Each command returns a tuple: (type, [data, ...]) where 'type'
is usually 'OK' or 'NO', and 'data' is either the text from the
tagged response, or untagged results from command.
All arguments to commands are converted to strings, except for
AUTHENTICATE.
"""
"""
However, the 'password' argument to the LOGIN command is always
quoted. If you want to avoid having an argument string quoted (eg:
the 'flags' argument to STORE) then enclose the string in
parentheses (eg: "(\Deleted)").
Errors raise the exception class <instance>.error("<reason>").
IMAP4 server errors raise <instance>.abort("<reason>"),
which is a sub-class of 'error'. Mailbox status changes
from READ-WRITE to READ-ONLY raise the exception class
<instance>.readonly("<reason>"), which is a sub-class of 'abort'.
"error" exceptions imply a program error.
"abort" exceptions imply the connection should be reset, and
the command re-tried.
"readonly" exceptions imply the command should be re-tried.
Note: to use this module, you must read the RFCs pertaining
to the IMAP4 protocol, as the semantics of the arguments to
each IMAP4 command are left to the invoker, not to mention
the results.
"""
class error(Exception): """Logical errors - debug required"""
class abort(error): """Service errors - close and retry"""
def __clear_knowledge(self):
"""clear/init any knowledge obtained from the server"""
self.capabilities = []
self.loginmechs = []
self.implementation = ''
self.supports_tls = 0
def __init__(self, host='', port=SIEVE_PORT,
use_tls=False, keyfile=None, certfile=None):
self.host = host
self.port = port
self.debug = Debug
self.state = 'NONAUTH'
self.response_text = self.response_code = None
self.__clear_knowledge()
# Open socket to server.
self._open(host, port)
if __debug__:
self._cmd_log_len = 10
self._cmd_log_idx = 0
self._cmd_log = {} # Last `_cmd_log_len' interactions
if self.debug >= 1:
self._mesg('managesieve version %s' % __version__)
# Get server welcome message,
# request and store CAPABILITY response.
typ, data = self._get_response()
if typ == 'OK':
self._parse_capabilities(data)
if use_tls and self.supports_tls:
typ, data = self.starttls(keyfile=keyfile, certfile=certfile)
if typ == 'OK':
self._parse_capabilities(data)
def _parse_capabilities(self, lines):
for line in lines:
if len(line) == 2:
typ, data = line
else:
assert len(line) == 1, 'Bad Capabilities line: %r' % line
typ = line[0]
data = None
if __debug__:
if self.debug >= 3:
self._mesg('%s: %r' % (typ, data))
if typ == "IMPLEMENTATION":
self.implementation = data
elif typ == "SASL":
self.loginmechs = data.split()
elif typ == "SIEVE":
self.capabilities = data.split()
elif typ == "STARTTLS":
self.supports_tls = 1
else:
# A client implementation MUST ignore any other
# capabilities given that it does not understand.
pass
return
def __getattr__(self, attr):
# Allow UPPERCASE variants of MANAGESIEVE command methods.
if commands.has_key(attr):
return getattr(self, attr.lower())
raise AttributeError("Unknown MANAGESIEVE command: '%s'" % attr)
#### Private methods ###
def _open(self, host, port):
"""Setup 'self.sock' and 'self.file'."""
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.sock.connect((self.host, self.port))
self.file = self.sock.makefile('r')
def _close(self):
self.file.close()
self.sock.close()
def _read(self, size):
"""Read 'size' bytes from remote."""
data = ""
while len(data) < size:
data += self.file.read(size - len(data))
return data
def _readline(self):
"""Read line from remote."""
return self.file.readline()
def _send(self, data):
return self.sock.send(data)
def _get_line(self):
line = self._readline()
if not line:
raise self.abort('socket error: EOF')
# Protocol mandates all lines terminated by CRLF
line = line[:-2]
if __debug__:
if self.debug >= 4:
self._mesg('< %s' % line)
else:
self._log('< %s' % line)
return line
def _simple_command(self, *args):
"""Execute a command which does only return status.
Returns (typ) with
typ = response type
The responce code and text may be found in <instance>.response_code
and <instance>.response_text, respectivly.
"""
return self._command(*args)[0] # only return typ, ignore data
def _command(self, name, arg1=None, arg2=None, *options):
"""
Returns (typ, data) with
typ = response type
data = list of lists of strings read (only meaningfull if OK)
The responce code and text may be found in <instance>.response_code
and <instance>.response_text, respectivly.
"""
if self.state not in commands[name]:
raise self.error(
'Command %s illegal in state %s' % (name, self.state))
# concatinate command and arguments (if any)
data = " ".join(filter(None, (name, arg1, arg2)))
if __debug__:
if self.debug >= 4: self._mesg('> %r' % data)
else: self._log('> %s' % data)
try:
try:
self._send('%s%s' % (data, CRLF))
for o in options:
if __debug__:
if self.debug >= 4: self._mesg('> %r' % o)
else: self._log('> %r' % data)
self._send('%s%s' % (o, CRLF))
except (socket.error, OSError), val:
raise self.abort('socket error: %s' % val)
return self._get_response()
except self.abort, val:
if __debug__:
if self.debug >= 1:
self.print_log()
raise
def _readstring(self, data):
if data[0] == ' ': # space -> error
raise self.error('Unexpected space: %r' % data)
elif data[0] == '"': # handle double quote:
if not self._match(re_dquote, data):
raise self.error('Unmatched quote: %r' % data)
snippet = self.mo.group(1)
return re_esc_quote.sub(r'\1', snippet), data[self.mo.end():]
elif self._match(Literal, data):
# read a 'literal' string
size = int(self.mo.group('size'))
if __debug__:
if self.debug >= 4:
self._mesg('read literal size %s' % size)
return self._read(size), self._get_line()
else:
data = data.split(' ', 1)
if len(data) == 1:
data.append('')
return data
def _get_response(self):
"""
Returns (typ, data) with
typ = response type
data = list of lists of strings read (only meaningfull if OK)
The responce code and text may be found in <instance>.response_code
and <instance>.response_text, respectivly.
"""
"""
response-deletescript = response-oknobye
response-authenticate = *(string CRLF) (response-oknobye)
response-capability = *(string [SP string] CRLF) response-oknobye
response-listscripts = *(string [SP "ACTIVE"] CRLF) response-oknobye
response-oknobye = ("OK" / "NO" / "BYE") [SP "(" resp-code ")"] [SP string] CRLF
string = quoted / literal
quoted = <"> *QUOTED-CHAR <">
literal = "{" number "+}" CRLF *OCTET
;; The number represents the number of octets
;; MUST be literal-utf8 except for values
--> a response either starts with a quote-charakter, a left-bracket or
OK, NO, BYE
"quoted" CRLF
"quoted" SP "quoted" CRLF
{size} CRLF *OCTETS CRLF
{size} CRLF *OCTETS CRLF
[A-Z-]+ CRLF
"""
data = [] ; dat = None
resp = self._get_line()
while 1:
if self._match(Oknobye, resp):
typ, code, dat = self.mo.group('type','code','data')
if __debug__:
if self.debug >= 1:
self._mesg('%s response: %s %s' % (typ, code, dat))
self.response_code = code
self.response_text = None
if dat:
self.response_text = self._readstring(dat)[0]
# if server quits here, send code instead of empty data
if typ == "BYE":
return typ, code
return typ, data
## elif 0:
## dat2 = None
## dat, resp = self._readstring(resp)
## if resp.startswith(' '):
## dat2, resp = self._readstring(resp[1:])
## data.append( (dat, dat2))
## resp = self._get_line()
else:
dat = []
while 1:
dat1, resp = self._readstring(resp)
if __debug__:
if self.debug >= 4:
self._mesg('read: %r' % (dat1,))
if self.debug >= 5:
self._mesg('rest: %r' % (resp,))
dat.append(dat1)
if not resp.startswith(' '):
break
resp = resp[1:]
if len(dat) == 1:
dat.append(None)
data.append(dat)
resp = self._get_line()
return self.error('Should not come here')
def _match(self, cre, s):
# Run compiled regular expression match method on 's'.
# Save result, return success.
self.mo = cre.match(s)
if __debug__:
if self.mo is not None and self.debug >= 5:
self._mesg("\tmatched r'%s' => %s" % (cre.pattern, `self.mo.groups()`))
return self.mo is not None
if __debug__:
def _mesg(self, s, secs=None):
if secs is None:
secs = time.time()
tm = time.strftime('%M:%S', time.localtime(secs))
sys.stderr.write(' %s.%02d %s\n' % (tm, (secs*100)%100, s))
sys.stderr.flush()
def _log(self, line):
# Keep log of last `_cmd_log_len' interactions for debugging.
self._cmd_log[self._cmd_log_idx] = (line, time.time())
self._cmd_log_idx += 1
if self._cmd_log_idx >= self._cmd_log_len:
self._cmd_log_idx = 0
def print_log(self):
self.self._mesg('last %d SIEVE interactions:' % len(self._cmd_log))
i, n = self._cmd_log_idx, self._cmd_log_len
while n:
try:
self.self._mesg(*self._cmd_log[i])
except:
pass
i += 1
if i >= self._cmd_log_len:
i = 0
n -= 1
### Public methods ###
def authenticate(self, mechanism, *authobjects):
"""Authenticate command - requires response processing."""
# command-authenticate = "AUTHENTICATE" SP auth-type [SP string] *(CRLF string)
# response-authenticate = *(string CRLF) (response-oknobye)
mech = mechanism.upper()
if not mech in self.loginmechs:
raise self.error("Server doesn't allow %s authentication." % mech)
if mech == AUTH_LOGIN:
authobjects = [ sieve_name(binascii.b2a_base64(ao)[:-1])
for ao in authobjects
]
elif mech == AUTH_PLAIN:
if len(authobjects) < 3:
# assume authorization identity (authzid) is missing
# and these two authobjects are username and password
authobjects.insert(0, '')
ao = '\0'.join(authobjects)
ao = binascii.b2a_base64(ao)[:-1]
authobjects = [ sieve_string(ao) ]
else:
raise self.error("managesieve doesn't support %s authentication." % mech)
typ, data = self._command('AUTHENTICATE',
sieve_name(mech), *authobjects)
if typ == 'OK':
self.state = 'AUTH'
return typ
def login(self, auth, user, password):
"""
Authenticate to the Sieve server using the best mechanism available.
"""
for authmech in AUTHMECHS:
if authmech in self.loginmechs:
authobjs = [auth, user, password]
if authmech == AUTH_LOGIN:
authobjs = [user, password]
return self.authenticate(authmech, *authobjs)
else:
raise self.abort('No matching authentication mechanism found.')
def logout(self):
"""Terminate connection to server."""
# command-logout = "LOGOUT" CRLF
# response-logout = response-oknobye
typ = self._simple_command('LOGOUT')
self.state = 'LOGOUT'
self._close()
return typ
def listscripts(self):
"""Get a list of scripts on the server.
(typ, [data]) = <instance>.listscripts()
if 'typ' is 'OK', 'data' is list of (scriptname, active) tuples.
"""
# command-listscripts = "LISTSCRIPTS" CRLF
# response-listscripts = *(sieve-name [SP "ACTIVE"] CRLF) response-oknobye
typ, data = self._command('LISTSCRIPTS')
if typ != 'OK': return typ, data
scripts = []
for dat in data:
if __debug__:
if not len(dat) in (1, 2):
self.error("Unexpected result from LISTSCRIPTS: %r" (dat,))
scripts.append( (dat[0], dat[1] is not None ))
return typ, scripts
def getscript(self, scriptname):
"""Get a script from the server.
(typ, scriptdata) = <instance>.getscript(scriptname)
'scriptdata' is the script data.
"""
# command-getscript = "GETSCRIPT" SP sieve-name CRLF
# response-getscript = [string CRLF] response-oknobye
typ, data = self._command('GETSCRIPT', sieve_name(scriptname))
if typ != 'OK': return typ, data
if len(data) != 1:
self.error('GETSCRIPT returned more than one string/script')
# todo: decode data?
return typ, data[0][0]
def putscript(self, scriptname, scriptdata):
"""Put a script onto the server."""
# command-putscript = "PUTSCRIPT" SP sieve-name SP string CRLF
# response-putscript = response-oknobye
return self._simple_command('PUTSCRIPT',
sieve_name(scriptname),
sieve_string(scriptdata)
)
def deletescript(self, scriptname):
"""Delete a scripts at the server."""
# command-deletescript = "DELETESCRIPT" SP sieve-name CRLF
# response-deletescript = response-oknobye
return self._simple_command('DELETESCRIPT', sieve_name(scriptname))
def setactive(self, scriptname):
"""Mark a script as the 'active' one."""
# command-setactive = "SETACTIVE" SP sieve-name CRLF
# response-setactive = response-oknobye
return self._simple_command('SETACTIVE', sieve_name(scriptname))
def havespace(self, scriptname, size):
# command-havespace = "HAVESPACE" SP sieve-name SP number CRLF
# response-havespace = response-oknobye
return self._simple_command('HAVESPACE',
sieve_name(scriptname),
str(size))
def capability(self):
"""
Isse a CAPABILITY command and return the result.
As a side-effect, on succes these attributes are (re)set:
self.implementation
self.loginmechs
self.capabilities
self.supports_tls
"""
# command-capability = "CAPABILITY" CRLF
# response-capability = *(string [SP string] CRLF) response-oknobye
typ, data = self._command('CAPABILITY')
if typ == 'OK':
self._parse_capabilities(data)
return typ, data
def starttls(self, keyfile=None, certfile=None):
"""Puts the connection to the SIEVE server into TLS mode.
If the server supports TLS, this will encrypt the rest of the SIEVE
session. If you provide the keyfile and certfile parameters,
the identity of the SIEVE server and client can be checked. This,
however, depends on whether the socket module really checks the
certificates.
"""
# command-starttls = "STARTTLS" CRLF
# response-starttls = response-oknobye
typ, data = self._command('STARTTLS')
if typ == 'OK':
sslobj = ssl_wrap_socket(self.sock, keyfile, certfile)
self.sock = SSLFakeSocket(self.sock, sslobj)
self.file = SSLFakeFile(sslobj)
# MUST discard knowledge obtained from the server
self.__clear_knowledge()
# Some servers send capabilities after TLS handshake, some
# do not. We send a bogus command, and expect a NO. If you
# get something else instead, read the extra NO to clear
# the buffer.
typ, data = self._command('BOGUS')
if typ != 'NO':
typ, data = self._get_response()
# server may not advertise capabilities, thus we need to ask
self.capability()
if self.debug >= 3: self._mesg('started Transport Layer Security (TLS)')
return typ, data

164
Tests/Integration/preferences.py
View File

@@ -1,164 +0,0 @@
from config import hostname, port, username, password
import webdavlib
import urllib
import base64
import simplejson
import sogoLogin
DEBUG=False
#DEBUG=True
# must be kept in sync with SoObjects/SOGo/SOGoDefaults.plist
# this should probably be fetched magically...
SOGoSupportedLanguages = [ "Arabic", "Basque", "Bulgarian", "Catalan", "ChineseChina", "ChineseTaiwan", "Croatian",
"Czech", "Dutch", "Danish", "Welsh", "English", "Finnish",
"SpanishSpain", "SpanishArgentina", "French", "German", "Hebrew",
"Hungarian", "Indonesian", "Icelandic", "Italian", "Japanese",
"Latvian", "Lithuanian", "Macedonian", "Montenegrin", "Portuguese", "BrazilianPortuguese",
"NorwegianBokmal", "NorwegianNynorsk", "Polish", "Romanian", "Russian",
"Serbian", "SerbianLatin", "Slovak", "Slovenian", "Swedish", "TurkishTurkey", "Ukrainian" ];
daysBetweenResponseList=[1,2,3,5,7,14,21,30]
class HTTPPreferencesPOST (webdavlib.HTTPPOST):
cookie = None
def prepare_headers (self):
headers = webdavlib.HTTPPOST.prepare_headers(self)
if self.cookie:
headers["Cookie"] = self.cookie
return headers
class HTTPPreferencesGET (webdavlib.HTTPGET):
cookie = None
def prepare_headers (self):
headers = webdavlib.HTTPGET.prepare_headers(self)
if self.cookie:
headers["Cookie"] = self.cookie
return headers
class preferences:
login = username
passw = password
def __init__(self, otherLogin = None, otherPassword = None):
if otherLogin and otherPassword:
self.login = otherLogin
self.passw = otherPassword
self.client = webdavlib.WebDAVClient(hostname, port)
authCookie = sogoLogin.getAuthCookie(hostname, port, self.login, self.passw)
self.cookie = authCookie
#- If this is not set, we CAN'T save preferences
self.preferences = None
def find_key(self, d, key):
if key in d:
return d
subdicts = [a[1] for a in d.iteritems() if type(a[1]) == dict]
for subd in subdicts:
ret = self.find_key(subd, key)
if ret:
return ret
return None
def load_preferences(self):
defaults = self.get_defaults()
settings = self.get_settings()
self.preferences = {'defaults': defaults, 'settings': settings}
#print "LOAD PREFS:", self.preferences
def get(self, preference=None):
if not self.preferences:
self.load_preferences()
#- Want the whole thing
if not preference:
return self.preferences
else:
tmpdict = self.find_key(self.preferences, preference)
if tmpdict:
return tmpdict[preference]
else:
return None
def _get(self, subtype='jsonDefault', preference=None):
url = "/SOGo/so/%s/%s" % (self.login, subtype)
get = HTTPPreferencesGET(url)
get.cookie = self.cookie
self.client.execute(get)
if DEBUG: print "DEBUG (url):", url
if DEBUG: print "DEBUG (status):", get.response["status"]
if DEBUG: print "DEBUG (body):", get.response['body']
content = simplejson.loads(get.response['body'])
result = None
try:
if preference:
result = content[preference]
else:
result = content
except:
pass
return result
def get_defaults(self, preference=None):
return self._get('jsonDefaults', preference)
def get_settings(self, preference=None):
return self._get('jsonSettings', preference)
def set_nosave(self, preference, value=None):
# First check if we did a get, if not, must get first
if not self.preferences:
self.load_preferences()
# Get the right sub-dict and change the key/value
subdict = self.find_key(self.preferences, preference)
if not subdict:
raise AttributeError("ERROR(nosubdict): looking for %s in: %s" %(preference, str(self.preferences)))
subdict[preference] = value
def set(self, preference, value=None):
self.set_nosave(preference, value)
self.save()
def set_multiple(self, preferences={}):
for key, value in preferences.iteritems():
self.set_nosave(key, value)
self.save()
def set_or_create(self, preference, value, paths=['defaults']):
if not self.preferences:
self.load_preferences()
subdict = self.find_key(self.preferences, preference)
#- Pref is not set
if not subdict:
subdict = self.preferences
for path in paths:
subdict = subdict.setdefault(path, {})
subdict[preference] = value
def save(self):
url = "/SOGo/so/%s/Preferences/save" % self.login
post = HTTPPreferencesPOST(url, simplejson.dumps(self.preferences))
post.content_type = "application/json"
post.cookie = self.cookie
self.client.execute(post)
# Raise an exception if the pref wasn't properly set
if post.response["status"] != 200:
raise Exception ("failure setting prefs, (code = %d)" \
% post.response["status"])
# Simple main to test this class
if __name__ == "__main__":
p = preferences ()
print p.get ("SOGoLanguage")
p.set ("SOGoLanguage", SOGoSupportedLanguages.index("French"))
print p.get ("SOGoLanguage")

49
Tests/Integration/propfind.py
View File

@@ -1,49 +0,0 @@
#!/usr/bin/python
from config import hostname, port, username, password
import webdavlib
import sys
import getopt
import xml.dom.minidom
def parseArguments():
arguments = {}
depth = "0"
quiet = False
(opts, args) = getopt.getopt(sys.argv[1:], "d:q", ["depth=", "quiet"])
for pair in opts:
if (pair[0] == "-d" or pair[0] == "--depth"):
depth = pair[1]
elif (pair[0] == "-q" or pair[0] == "--quiet"):
quiet = True
# print "depth: " + depth
nargs = len(args)
if (nargs > 0):
resource = args[0]
if (nargs > 1):
properties = args[1:]
else:
properties = [ "allprop" ]
else:
print "resource required"
sys.exit(-1)
client = webdavlib.WebDAVClient(hostname, port, username, password)
propfind = webdavlib.WebDAVPROPFIND(resource, properties, depth)
client.execute(propfind)
sys.stderr.write("response:\n\n")
print propfind.response["body"]
if propfind.response.has_key("document"):
sys.stderr.write("document tree:\n")
elem = propfind.response["document"]
dom = xml.dom.minidom.parseString(xml.etree.ElementTree.tostring(elem))
print dom.toprettyxml()

57
Tests/Integration/sogoLogin.py
View File

@@ -1,57 +0,0 @@
#!/usr/bin/python
from config import hostname, port, username, password
import sys, getopt
import webdavlib
import urllib
import urllib2
import base64
import simplejson
import cookielib
def getAuthCookie(hostname, port, username, password) :
cjar = cookielib.CookieJar();
opener = urllib2.build_opener(urllib2.HTTPCookieProcessor(cjar))
urllib2.install_opener(opener)
creds2 = simplejson.dumps({"userName":username, "password": password})
req = urllib2.Request("http://%s:%s/SOGo/connect" % (hostname, port), creds2,
{'Content-Type': 'application/json'})
fd = urllib2.urlopen(req)
for cookie in cjar :
if cookie.name == "0xHIGHFLYxSOGo":
authinfo = cookie.value
break
return "0xHIGHFLYxSOGo="+authinfo
def usage() :
msg ="""Usage:
%s [-h] [-H | --host=hostname] [-p|--passwd=password] \
[-P|--port=port] [-u|--user=username]\n""" % sys.argv[0]
sys.stderr.write(msg);
if __name__ == "__main__" :
try:
opts, args = getopt.getopt (sys.argv[1:], "hH:p:P:u:", \
("host=", "passwd=", "port=", "user="));
except getopt.GetoptError:
usage()
exit(1)
for o, v in opts :
if o == "-h" :
usage()
exit(1)
elif o == "-H" or o == "--host" :
hostname = v
elif o == "-p" or o == "--passwd" :
password = v
elif o == "-P" or o == "--port" :
port = v
elif o == "-u" or o == "--user" :
username = v
print getAuthCookie(hostname, port, username, password)

23
Tests/Integration/sogotests.py
View File

@@ -1,23 +0,0 @@
import sys
import unittest
import time
def UnitTestTextTestResultNewStartTest(self, test):
self.xstartTime = time.time()
self.oldStartTest(test)
def UnitTestTextTestResultNewStopTest(self, test):
unittest.TestResult.stopTest(self, test)
endTime = time.time()
delta = endTime - self.xstartTime
print " %f ms" % delta
def runTests():
unittest._TextTestResult.oldStartTest = unittest._TextTestResult.startTest
unittest._TextTestResult.startTest = UnitTestTextTestResultNewStartTest
unittest._TextTestResult.stopTest = UnitTestTextTestResultNewStopTest
argv = []
argv.extend(sys.argv)
argv.append("-v")
unittest.main(argv=argv)

1081
Tests/Integration/test-caldav-scheduling.py
View File

File diff suppressed because it is too large Load Diff

180
Tests/Integration/test-carddav.py
View File

@@ -1,180 +0,0 @@
#!/usr/bin/python
from config import hostname, port, username, password
import carddav
import sogotests
import unittest
import webdavlib
import time
class JsonDavEventTests(unittest.TestCase):
def setUp(self):
self._connect_as_user()
def _connect_as_user(self, newuser=username, newpassword=password):
self.dv = carddav.Carddav(newuser, newpassword)
def _create_new_event(self, path):
gid = self.dv.newguid(path)
event = {'startDate': "2015-12-25",
'startTime': "10:00",
'endDate': "2015-12-25",
'endTime': "23:00",
'isTransparent': 0,
'sendAppointmentNotifications': 0,
'summary': "Big party",
'alarm': {'action': 'display',
'quantity': 10,
'unit': "MINUTES",
'reference': "BEFORE",
'relation': "START",
'email': "sogo1@example.com"},
'organizer': {'name': u"Balthazar C\xe9sar",
'email': "sogo2@example.com"},
'c_name': gid,
'c_folder': path
}
return (event, path, gid)
def _get_dav_data(self, filename, user=username, passwd=password):
w = webdavlib.WebDAVClient(hostname, port, user, passwd)
query = webdavlib.HTTPGET("http://localhost/SOGo/dav/%s/Calendar/personal/%s" % (username, filename))
w.execute(query)
self.assertEquals(query.response['status'], 200)
return query.response['body'].split("\r\n")
def _get_dav_field(self, davdata, fieldname):
try:
data = [a.split(':')[1] for a in davdata if fieldname in a][0]
except IndexError:
data = ''
return data
def test_create_new_event(self):
path = 'Calendar/personal'
(event, folder, gid) = self._create_new_event(path)
#print "Saving Event to:", folder, gid
self.dv.save_event(event, folder, gid)
#- Get the event back with JSON
self._connect_as_user()
self.dv.load_events()
elist = [e for e in self.dv.events if e['c_name'] == gid]
#- MUST have this event -- only once
self.assertEquals(len(elist), 1)
strdate = "%d-%.02d-%.02d" % time.gmtime(elist[0]['c_startdate'])[0:3]
self.assertEquals(strdate, event['startDate'])
#- Get the event back with DAV
dav = self._get_dav_data(gid, username, password)
self.assertEquals(self._get_dav_field(dav, 'SUMMARY:'), event['summary'])
class JsonDavPhoneTests(unittest.TestCase):
def setUp(self):
self._connect_as_user()
self.newphone = [{'type': 'home', 'value': '123.456.7890'}]
self.newphones_difftype = [{'type': 'home', 'value': '123.456.7890'},
{'type': 'work', 'value': '987.654.3210'},
{'type': 'fax', 'value': '555.666.7777'}]
self.newphones_sametype = [{'type': 'work', 'value': '123.456.7890'},
{'type': 'work', 'value': '987.654.3210'}]
# Easier to erase them all in tearDown
self.allphones = list(self.newphone)
self.allphones.extend(self.newphones_difftype)
self.allphones.extend(self.newphones_sametype)
#- In case there are no cards for this user
try:
self._get_card()
except IndexError:
path = 'Contacts/personal'
(card, path, gid) = self._create_new_card(path)
self._save_card(card)
def tearDown(self):
self._connect_as_user()
self._get_card()
#- Remove the phones we just added
for phone in self.allphones:
try:
self.card['phones'].pop(self.card['phones'].index(phone))
except ValueError:
#print "Can't find", phone
pass
self._save_card()
def _connect_as_user(self, newuser=username, newpassword=password):
self.dv = carddav.Carddav(newuser, newpassword)
def _create_new_card(self, path):
gid = self.dv.newguid(path)
card = {'c_categories': None,
'c_cn': 'John Doe',
'c_component': 'vcard',
'c_givenname': 'John Doe',
'c_mail': 'johndoe@nothere.com',
'c_name': gid,
'c_o': '',
'c_screenname': '',
'c_sn': '',
'c_telephonenumber': '123.456.7890',
'emails': [{'type': 'pref', 'value': 'johndoe@nothere.com'}],
'phones': [{'type': 'home', 'value': '111.222.3333'}],
'id': gid}
return (card, path, gid)
def _get_card(self, name="John Doe"):
tmp_card = self.dv.get_cards(name)[0]
self.card = self.dv.get_card(tmp_card['c_name'])
def _save_card(self, card=None):
if card:
self.dv.save_card(card)
else:
self.dv.save_card(self.card)
def _get_dav_data(self, filename, user=username, passwd=password):
w = webdavlib.WebDAVClient(hostname, port, user, passwd)
query = webdavlib.HTTPGET("http://localhost/SOGo/dav/%s/Contacts/personal/%s" % (username, filename))
w.execute(query)
self.assertEquals(query.response['status'], 200)
return query.response['body'].split("\r\n")
def _phone_to_dav_str(self, phonedict):
return "TEL;TYPE=%s:%s" % (phonedict['type'], phonedict['value'])
def _testMultiplePhones(self, phones):
""" Add Multiple Phones to Contact JSON and verify with DAV """
#- Use JSON to get CARD and add a phone and save it back
self._get_card()
oldphones = self.card['phones']
oldphones.extend(phones)
self._save_card()
#- Make sure that the phone is there when using JSON
self._connect_as_user()
self._get_card()
#print "C:::", self.card
testphones = self.card['phones']
#print "P1:", oldphones
#print "P2:", testphones
self.assertEquals(sorted(oldphones), sorted(testphones))
#- Verify that DAV has the same values
dav = self._get_dav_data(self.card['id'], username, password)
for phone in phones:
found = dav.index(self._phone_to_dav_str(phone))
self.assertTrue(found > 0)
def testSinglePhone(self):
self._testMultiplePhones(self.newphone)
def testMultipleDifferentPhones(self):
self._testMultiplePhones(self.newphones_difftype)
def testMultipleSameTypePhones(self):
self._testMultiplePhones(self.newphones_sametype)
if __name__ == "__main__":
sogotests.runTests()

56
Tests/Integration/test-config.py
View File

@@ -1,56 +0,0 @@
#!/usr/bin/python
from config import hostname, port, username, password, mailserver, subscriber_username, attendee1, attendee1_delegate
import sogotests
import unittest
import time
class CalDAVITIPDelegationTest(unittest.TestCase):
def testConfigPY(self):
""" config.py validation """
try:
test = hostname
except:
self.fail("'hostname' is not defined")
try:
test = username
except:
self.fail("'username' is not defined")
try:
test = subscriber_username
except:
self.fail("'subscriber_username' is not defined")
try:
test = attendee1
except:
self.fail("'attendee1' is not defined")
try:
test = attendee1_delegate
except:
self.fail("'attendee1_delegate' is not defined")
self.assertEquals(subscriber_username, attendee1,
"'subscriber_username' and 'attendee1'"
+ " must be the same user")
try:
test = mailserver
except:
self.fail("'mailserver' is not defined")
userHash = {}
userList = [ username, subscriber_username, attendee1_delegate ]
for user in userList:
self.assertFalse(userHash.has_key(user),
"username, attendee1, attendee1_delegate must"
+ " all be different users ('%s')"
% user)
userHash[user] = True
if __name__ == "__main__":
sogotests.runTests()

63
Tests/Integration/test-contact-categories.py
View File

@@ -1,63 +0,0 @@
#!/usr/bin/python
import sogotests
import unittest
import webdavlib
from config import *
class HTTPContactCategoriesTest(unittest.TestCase):
def _setCategories(self, user, categories = None):
resource = '/SOGo/dav/%s/Contacts/' % user
if categories is None:
categories = []
elements = [ { "{urn:inverse:params:xml:ns:inverse-dav}category": x }
for x in categories ]
props = { "{urn:inverse:params:xml:ns:inverse-dav}contacts-categories": elements }
proppatch = webdavlib.WebDAVPROPPATCH(resource, props)
client = webdavlib.WebDAVClient(hostname, port, username, password)
client.execute(proppatch)
self.assertEquals(proppatch.response["status"], 207,
"failure (%s) setting '%s' categories on %s's contacts"
% (proppatch.response["status"],
"', '".join(categories), user))
def _getCategories(self, user):
resource = '/SOGo/dav/%s/Contacts/' % user
props = [ "{urn:inverse:params:xml:ns:inverse-dav}contacts-categories" ]
propfind = webdavlib.WebDAVPROPFIND(resource, props, "0")
client = webdavlib.WebDAVClient(hostname, port, username, password)
client.execute(propfind)
self.assertEquals(propfind.response["status"], 207,
"failure (%s) getting categories on %s's contacts"
% (propfind.response["status"], user))
categories = []
prop_nodes = propfind.response["document"].findall("{DAV:}response/{DAV:}propstat/{DAV:}prop/{urn:inverse:params:xml:ns:inverse-dav}contacts-categories")
for prop_node in prop_nodes:
cat_nodes = prop_node.findall("{urn:inverse:params:xml:ns:inverse-dav}category")
if cat_nodes is not None:
for cat_node in cat_nodes:
categories.append(cat_node.text)
return categories
def test(self):
self._setCategories(username, [])
cats = self._getCategories(username)
self.assertTrue(cats is not None and len(cats) == 0)
self._setCategories(username, [ "Coucou" ])
cats = self._getCategories(username)
self.assertTrue(cats is not None and len(cats) == 1)
self.assertEquals(cats[0], "Coucou")
self._setCategories(username, [ "Toto", "Cuicui" ])
cats = self._getCategories(username)
self.assertTrue(cats is not None and len(cats) == 2)
self.assertEquals(cats[0], "Toto")
self.assertEquals(cats[1], "Cuicui")
if __name__ == "__main__":
sogotests.runTests()

1130
Tests/Integration/test-davacl.py
View File

File diff suppressed because it is too large Load Diff

52
Tests/Integration/test-default-classification.py
View File

@@ -1,52 +0,0 @@
#!/usr/bin/python
import sogotests
import unittest
import webdavlib
from config import *
class HTTPDefaultClassificationTest(unittest.TestCase):
def _setClassification(self, user, component, classification = ""):
resource = '/SOGo/dav/%s/Calendar/' % user
props = { "{urn:inverse:params:xml:ns:inverse-dav}%s-default-classification" % component: classification }
proppatch = webdavlib.WebDAVPROPPATCH(resource, props)
client = webdavlib.WebDAVClient(hostname, port, username, password)
client.execute(proppatch)
return (proppatch.response["status"] == 207);
def _getClassification(self, user, component):
resource = '/SOGo/dav/%s/Calendar/' % user
property_name = "{urn:inverse:params:xml:ns:inverse-dav}%s-default-classification" % component
propfind = webdavlib.WebDAVPROPFIND(resource, [ property_name ], "0")
client = webdavlib.WebDAVClient(hostname, port, username, password)
client.execute(propfind)
classification = None
propstat_nodes = propfind.response["document"].findall("{DAV:}response/{DAV:}propstat")
for propstat_node in propstat_nodes:
status_nodes = propstat_node.findall("{DAV:}status")
if status_nodes[0].text.lower() == "http/1.1 200 ok":
property_nodes = propstat_node.findall("{DAV:}prop/%s" % property_name)
if len(property_nodes) > 0:
classification = property_nodes[0].text
return classification
def test(self):
self.assertFalse(self._setClassification(username, "123456", "PUBLIC"),
"expected failure when setting a classification with an invalid property")
self.assertFalse(self._setClassification(username, "events", ""),
"expected failure when setting an empty classification")
self.assertFalse(self._setClassification(username, "events", "pouet"),
"expected failure when setting an invalid classification")
for component in [ "events", "tasks" ]:
for classification in [ "PUBLIC", "PRIVATE", "CONFIDENTIAL" ]:
self.assertTrue(self._setClassification(username, component, classification),
"error when setting classification to '%s'" % classification)
fetched_class = self._getClassification(username, component)
self.assertTrue(classification == fetched_class,
"set and fetched classifications do not match (%s != %s)" % (classification, fetched_class))
if __name__ == "__main__":
sogotests.runTests()

208
Tests/Integration/test-ical.py
View File

@@ -1,208 +0,0 @@
#!/usr/bin/python
# FIXME: we should avoid using superuser if possible
from config import hostname, port, username, password, subscriber_username, \
superuser, superuser_password
import unittest
import sogotests
import utilities
import webdavlib
class iCalTest(unittest.TestCase):
def testPrincipalCollectionSet(self):
"""principal-collection-set: 'DAV' header must be returned with iCal 4"""
client = webdavlib.WebDAVClient(hostname, port, username, password)
resource = '/SOGo/dav/%s/' % username
# NOT iCal4
propfind = webdavlib.WebDAVPROPFIND(resource,
["{DAV:}principal-collection-set"],
0)
client.execute(propfind)
self.assertEquals(propfind.response["status"], 207)
headers = propfind.response["headers"]
self.assertFalse(headers.has_key("dav"),
"DAV header must not be returned when user-agent is NOT iCal 4")
# iCal4
propfind = webdavlib.WebDAVPROPFIND(resource,
["{DAV:}principal-collection-set"],
0)
client.user_agent = "DAVKit/4.0.1 (730); CalendarStore/4.0.1 (973); iCal/4.0.1 (1374); Mac OS X/10.6.2 (10C540)"
client.execute(propfind)
self.assertEquals(propfind.response["status"], 207)
headers = propfind.response["headers"]
self.assertTrue(headers.has_key("dav"),
"DAV header must be returned when user-agent is iCal 4")
expectedDAVClasses = ["1", "2", "access-control", "calendar-access",
"calendar-schedule", "calendar-auto-schedule",
"calendar-proxy"]
davClasses = [x.strip() for x in headers["dav"].split(",")]
for davClass in expectedDAVClasses:
self.assertTrue(davClass in davClasses,
"DAV class '%s' not found" % davClass)
def _setMemberSet(self, owner, members, perm):
resource = '/SOGo/dav/%s/calendar-proxy-%s/' % (owner, perm)
membersHref = [ { "{DAV:}href": '/SOGo/dav/%s/' % x }
for x in members ]
props = { "{DAV:}group-member-set": membersHref }
proppatch = webdavlib.WebDAVPROPPATCH(resource, props)
client = webdavlib.WebDAVClient(hostname, port, superuser, superuser_password)
client.user_agent = "DAVKit/4.0.1 (730); CalendarStore/4.0.1 (973); iCal/4.0.1 (1374); Mac OS X/10.6.2 (10C540)"
client.execute(proppatch)
self.assertEquals(proppatch.response["status"], 207,
"failure (%s) setting '%s' permission for '%s' on %s's calendars"
% (proppatch.response["status"], perm,
"', '".join(members), owner))
def _getMembership(self, user):
resource = '/SOGo/dav/%s/' % user
propfind = webdavlib.WebDAVPROPFIND(resource,
["{DAV:}group-membership"], 0)
client = webdavlib.WebDAVClient(hostname, port, superuser, superuser_password)
client.user_agent = "DAVKit/4.0.1 (730); CalendarStore/4.0.1 (973); iCal/4.0.1 (1374); Mac OS X/10.6.2 (10C540)"
client.execute(propfind)
hrefs = propfind.response["document"].findall("{DAV:}response/{DAV:}propstat/{DAV:}prop/{DAV:}group-membership/{DAV:}href")
members = [x.text for x in hrefs]
return members
def _getProxyFor(self, user, perm):
resource = '/SOGo/dav/%s/' % user
prop = "{http://calendarserver.org/ns/}calendar-proxy-%s-for" % perm
propfind = webdavlib.WebDAVPROPFIND(resource, [prop], 0)
client = webdavlib.WebDAVClient(hostname, port, superuser, superuser_password)
client.user_agent = "DAVKit/4.0.1 (730); CalendarStore/4.0.1 (973); iCal/4.0.1 (1374); Mac OS X/10.6.2 (10C540)"
client.execute(propfind)
hrefs = propfind.response["document"].findall("{DAV:}response/{DAV:}propstat/{DAV:}prop/{http://calendarserver.org/ns/}calendar-proxy-%s-for/{DAV:}href"
% perm)
members = [x.text[len("/SOGo/dav/"):-1] for x in hrefs]
return members
def testCalendarProxy(self):
"""calendar-proxy as used from iCal"""
self._setMemberSet(username, [], "read")
self._setMemberSet(username, [], "write")
self._setMemberSet(subscriber_username, [], "read")
self._setMemberSet(subscriber_username, [], "write")
self.assertEquals([], self._getMembership(username),
"'%s' must have no membership"
% username)
self.assertEquals([], self._getMembership(subscriber_username),
"'%s' must have no membership"
% subscriber_username)
self.assertEquals([], self._getProxyFor(username, "read"),
"'%s' must not be a proxy for anyone" % username)
self.assertEquals([], self._getProxyFor(username, "write"),
"'%s' must not be a proxy for anyone" % username)
self.assertEquals([], self._getProxyFor(subscriber_username, "read"),
"'%s' must not be a proxy for anyone" % subscriber_username)
self.assertEquals([], self._getProxyFor(subscriber_username, "write"),
"'%s' must not be a proxy for anyone" % subscriber_username)
for perm in ("read", "write"):
for users in ((username, subscriber_username),
(subscriber_username, username)):
self._setMemberSet(users[0], [users[1]], perm)
membership = self._getMembership(users[1])
self.assertEquals(['/SOGo/dav/%s/calendar-proxy-%s/'
% (users[0], perm)],
membership,
"'%s' must have %s access to %s's calendars"
% (users[1], perm, users[0]))
proxyFor = self._getProxyFor(users[1], perm)
self.assertEquals([users[0]], proxyFor,
"'%s' expected to be %s proxy for %s: %s"
% (users[1], perm, users[0], proxyFor))
def _testMapping(self, client, perm, resource, rights):
dav_utility = utilities.TestCalendarACLUtility(self, client, resource)
dav_utility.setupRights(subscriber_username, rights)
membership = self._getMembership(subscriber_username)
self.assertEquals(['/SOGo/dav/%s/calendar-proxy-%s/'
% (username, perm)],
membership,
"'%s' must have %s access to %s's calendars:\n%s"
% (subscriber_username, perm, username, membership))
proxyFor = self._getProxyFor(subscriber_username, perm)
self.assertEquals([username], proxyFor,
"'%s' expected to be %s proxy for %s: %s"
% (subscriber_username, perm, username, proxyFor))
def testCalendarProxy2(self):
"""calendar-proxy as used from SOGo"""
client = webdavlib.WebDAVClient(hostname, port, superuser, superuser_password)
client.user_agent = "DAVKit/4.0.1 (730); CalendarStore/4.0.1 (973); iCal/4.0.1 (1374); Mac OS X/10.6.2 (10C540)"
personal_resource = "/SOGo/dav/%s/Calendar/personal/" % username
dav_utility = utilities.TestCalendarACLUtility(self,
client,
personal_resource)
dav_utility.setupRights(subscriber_username, {})
dav_utility.subscribe([subscriber_username])
other_resource = ("/SOGo/dav/%s/Calendar/test-calendar-proxy2/"
% username)
delete = webdavlib.WebDAVDELETE(other_resource)
client.execute(delete)
mkcol = webdavlib.WebDAVMKCOL(other_resource)
client.execute(mkcol)
dav_utility = utilities.TestCalendarACLUtility(self,
client,
other_resource)
dav_utility.setupRights(subscriber_username, {})
dav_utility.subscribe([subscriber_username])
## we test the rights mapping
# write: write on 'personal', none on 'test-calendar-proxy2'
self._testMapping(client, "write", personal_resource,
{ "c": True, "d": False, "pu": "v" })
self._testMapping(client, "write", personal_resource,
{ "c": False, "d": True, "pu": "v" })
self._testMapping(client, "write", personal_resource,
{ "c": False, "d": False, "pu": "m" })
self._testMapping(client, "write", personal_resource,
{ "c": False, "d": False, "pu": "r" })
# read: read on 'personal', none on 'test-calendar-proxy2'
self._testMapping(client, "read", personal_resource,
{ "c": False, "d": False, "pu": "d" })
self._testMapping(client, "read", personal_resource,
{ "c": False, "d": False, "pu": "v" })
# write: read on 'personal', write on 'test-calendar-proxy2'
self._testMapping(client, "write", other_resource,
{ "c": False, "d": False, "pu": "r" })
## we test the unsubscription
# unsubscribed from personal, subscribed to 'test-calendar-proxy2'
dav_utility = utilities.TestCalendarACLUtility(self, client,
personal_resource)
dav_utility.unsubscribe([subscriber_username])
membership = self._getMembership(subscriber_username)
self.assertEquals(['/SOGo/dav/%s/calendar-proxy-write/' % username],
membership,
"'%s' must have write access to %s's calendars"
% (subscriber_username, username))
# unsubscribed from personal, unsubscribed from 'test-calendar-proxy2'
dav_utility = utilities.TestCalendarACLUtility(self, client,
other_resource)
dav_utility.unsubscribe([subscriber_username])
membership = self._getMembership(subscriber_username)
self.assertEquals([],
membership,
"'%s' must have no access to %s's calendars"
% (subscriber_username, username))
delete = webdavlib.WebDAVDELETE(other_resource)
client.execute(delete)
if __name__ == "__main__":
sogotests.runTests()

677
Tests/Integration/test-maildav.py
View File

@@ -1,677 +0,0 @@
#!/usr/bin/python
from config import hostname, port, username, password, mailserver, subscriber_username, subscriber_password
import sys
import sogotests
import unittest
import webdavlib
import time
# TODO
# add test with multiple sort criterias
def fetchUserEmail(login):
client = webdavlib.WebDAVClient(hostname, port,
username, password)
resource = '/SOGo/dav/%s/' % login
propfind = webdavlib.WebDAVPROPFIND(resource,
["{urn:ietf:params:xml:ns:caldav}calendar-user-address-set"],
0)
client.execute(propfind)
nodes = propfind.xpath_evaluate('{DAV:}response/{DAV:}propstat/{DAV:}prop/C:calendar-user-address-set/{DAV:}href',
None)
return nodes[0].childNodes[0].nodeValue
message1 = """Return-Path: <cyril@cyril.dev>
Received: from cyril.dev (localhost [127.0.0.1])
by cyril.dev (Cyrus v2.3.8-Debian-2.3.8-1) with LMTPA;
Tue, 17 Dec 2009 07:42:16 -0400
Received: from aloha.dev (localhost [127.0.0.1])
by aloha.dev (Cyrus v2.3.8-Debian-2.3.8-1) with LMTPA;
Tue, 29 Sep 2009 07:42:16 -0400
Message-ID: <4AC1F29sept6.5060801@cyril.dev>
Date: Mon, 28 Sep 2009 07:42:14 -0400
From: Cyril <message1from@cyril.dev>
User-Agent: Thunderbird 2.0.0.22 (Macintosh/20090605)
References: <4AC3BF1B.3010806@inverse.ca>
MIME-Version: 1.0
To: message1to@cyril.dev
CC: 2message1cc@cyril.dev, user10@cyril.dev
Subject: message1subject
Content-Type: text/plain; charset=us-ascii; format=flowed
Content-Transfer-Encoding: 7bit
Reply-To: support@inverse.ca
Hello Jacques,
Can you read me?
--
Cyril <cyril@cyril.dev>
"""
message2 = """Return-Path: <cyril@cyril.dev>
Received: from cyril.dev (localhost [127.0.0.1])
by cyril.dev (Cyrus v2.3.8-Debian-2.3.8-1) with LMTPA;
Tue, 09 Dec 2009 07:42:16 -0400
Message-ID: <410sepAC1F296.5060801a@cyril.dev>
Date: Tue, 10 Sep 2009 07:42:14 -0400
User-Agent: Thunderbird 2.0.0.22 (Macintosh/20090605)
MIME-Version: 1.0
From: Cyril <message2from@cyril.dev>
To: message2to@cyril.dev
CC: 3message2cc@cyril.dev
Subject: message2subject
Content-Type: text/plain; charset=us-ascii; format=flowed
Content-Transfer-Encoding: 7bit
Reply-To: support@inverse.ca
Hello Jacques,
Can you read me?
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
Stuff StuffStuffStuffStuff StuffStuffStuff StuffStuff
--
Cyril <cyril@cyril.dev>
"""
message3 = """Return-Path: <cyril@cyril.dev>
Received: from cyril.dev (localhost [127.0.0.1])
by cyril.dev (Cyrus v2.3.8-Debian-2.3.8-1) with LMTPA;
Tue, 15 Dec 2009 07:42:16 -0400
Message-ID: <4AC1aF2dec96.5060801a@cyril.dev>
Date: Tue, 10 Dec 2009 07:42:14 -0400
User-Agent: Thunderbird 2.0.0.22 (Macintosh/20090605)
MIME-Version: 1.0
From: Cyril <message3from@cyril.dev>
To: message3to@cyril.dev
CC: 1message3cc@cyril.dev
Subject: Hallo
Content-Type: text/plain; charset=us-ascii; format=flowed
Content-Transfer-Encoding: 7bit
Reply-To: support@inverse.ca
Hello Jacques,
Can you read me?
This message is just a bit larger than message1 but smaller than message2
--
Cyril <cyril@cyril.dev>
"""
message1_received = """Received: from cyril.dev (localhost [127.0.0.1])
by cyril.dev (Cyrus v2.3.8-Debian-2.3.8-1) with LMTPA;
Tue, 17 Dec 2009 07:42:16 -0400"""
class DAVMailCollectionTest():
resource = '/SOGo/dav/%s/Mail/' % username
user_email = None
def setUp(self):
self.client = webdavlib.WebDAVClient(hostname, port,
username, password)
if self.user_email is None:
self.user_email = fetchUserEmail(username)
if self.user_email.startswith("mailto:"):
self.user_email = self.user_email[7:]
self.resource = '/SOGo/dav/%s/Mail/%s_A_%s/' \
% (username,
username.replace("@", "_A_").replace(".", "_D_"),
mailserver.replace(".", "_D_"))
## helper methods
def _makeCollection(self, name, status = 201):
url = "%s%s" % (self.resource, name)
mkcol = webdavlib.WebDAVMKCOL(url)
self.client.execute(mkcol)
self.assertEquals(mkcol.response["status"], status,
"failure creating collection"
"(code = %d)" % mkcol.response["status"])
def _deleteCollection(self, name, status = 204):
url = "%sfolder%s" % (self.resource, name)
delete = webdavlib.WebDAVDELETE(url)
self.client.execute(delete)
self.assertEquals(delete.response["status"], status,
"failure deleting collection"
"(code = %d)" % delete.response["status"])
def _putMessage(self, client, folder, message,
exp_status = 201):
url = "%sfolder%s" % (self.resource, folder)
put = webdavlib.HTTPPUT(url, message)
put.content_type = "message/rfc822"
client.execute(put)
if (exp_status is not None):
self.assertEquals(put.response["status"], exp_status,
"message creation/modification:"
" expected status code '%d' (received '%d')"
% (exp_status, put.response["status"]))
return put.response["headers"]["location"]
def _testProperty (self, url, property, expected, isDate = 0):
propfind = webdavlib.WebDAVPROPFIND(url, (property, ), 0)
self.client.execute(propfind)
key = property.replace("{urn:schemas:httpmail:}", "a:")
key = key.replace("{urn:schemas:mailheader:}", "a:")
tmp = propfind.xpath_evaluate("{DAV:}response/{DAV:}propstat/{DAV:}prop")
prop = tmp[0].firstChild;
result = None
if prop:
result = prop._get_firstChild()._get_nodeValue()
#print key, result
if isDate:
tstruct = time.strptime (result, "%a, %d %b %Y %H:%M:%S %Z")
result = int (time.mktime (tstruct))
self.assertEquals(result, expected,
"failure in propfind"
"(%s != %s)" % (result, expected))
def testMKCOL(self):
"""Folder creation"""
self._deleteCollection("test-dav-mail-%40-abc")
self._deleteCollection("test-dav-mail-@-def")
self._deleteCollection("test-dav-mail-%20-ghi")
self._makeCollection("test-dav-mail-%40-abc")
self._makeCollection("test-dav-mail-@-def")
self._makeCollection("test-dav-mail-%20-ghi")
self._makeCollection("test-dav-mail-%25-jkl", 500)
# Test MOVE
# self._makeCollection ("test-dav-mail-movable")
# url = "%sfolder%s" % (self.resource, "test-dav-mail-movable")
# move = webdavlib.WebDAVMOVE (url)
# move.destination = "http://cyril.dev%s%s2" % (self.resource, "test-dav-mail-movable")
# move.host = "cyril.dev"
# self.client.execute (move)
# self.assertEquals(move.response["status"], 204,
# "failure creating collection"
# "(code = %d)" % move.response["status"])
def testPUT(self):
"""Message creation"""
self._deleteCollection("test-dav-mail")
self._makeCollection("test-dav-mail")
# message creation on collection url
url = "%s%s" % (self.resource, "foldertest-dav-mail/")
put = webdavlib.HTTPPUT(url, message1)
put.content_type = "message/rfc822"
self.client.execute(put)
self.assertEquals(put.response["status"], 201,
"failure putting message"
"(code = %d)" % put.response["status"])
itemLocation = put.response["headers"]["location"]
get = webdavlib.WebDAVGET(itemLocation)
self.client.execute(get)
self.assertEquals(get.response["status"], 200,
"failure getting item"
"(code = %d)" % get.response["status"])
# message creation with explicit filename
url = "%s%s" %(self.resource, "foldertest-dav-mail/blabla.eml")
put = webdavlib.HTTPPUT(url, message1)
put.content_type = "message/rfc822"
self.client.execute(put)
self.assertEquals(put.response["status"], 201,
"failure putting message"
"(code = %d)" % put.response["status"])
itemLocation = put.response["headers"]["location"]
get = webdavlib.WebDAVGET(itemLocation)
self.client.execute(get)
self.assertEquals(get.response["status"], 200,
"failure getting item"
"(code = %d)" % get.response["status"])
self._deleteCollection("test-dav-mail")
def _testFilters(self, filters):
for filter in filters:
self._testFilter(filter)
def _testFilter(self, filter):
expected_hrefs = {}
expected_count = len(filter[1])
for href in filter[1]:
expected_hrefs[href] = True
received_count = 0
url = "%sfolder%s" % (self.resource, "test-dav-mail")
query = webdavlib.MailDAVMailQuery(url, ["displayname"], filter[0])
self.client.execute(query)
self.assertEquals(query.response["status"], 207,
"filter %s:\n\tunexpected status: %d"
% (filter[0], query.response["status"]))
response_nodes = query.xpath_evaluate("{DAV:}response")
for response_node in response_nodes:
href_node = query.xpath_evaluate("{DAV:}href", response_node)[0]
href = href_node.childNodes[0].nodeValue
received_count = received_count + 1
self.assertTrue(expected_hrefs.has_key(href),
"filter %s:\n\tunexpected href: %s" % (filter[0], href))
self.assertEquals(len(filter[1]), received_count,
"filter %s:\n\tunexpected amount of refs: %d"
% (filter[0], received_count))
def _testSort(self, sortOrder, ascending = True):
expected_hrefs = sortOrder[1]
expected_count = len(expected_hrefs)
received_count = 0
url = "%sfolder%s" % (self.resource, "test-dav-mail")
query = webdavlib.MailDAVMailQuery(url, ["displayname"],
None, sortOrder[0], ascending)
self.client.execute(query)
self.assertEquals(query.response["status"], 207,
"sortOrder %s:\n\tunexpected status: %d"
% (sortOrder[0], query.response["status"]))
response_nodes = query.response["document"].findall("{DAV:}response")
for response_node in response_nodes:
href_node = response_node.find("{DAV:}href")
href = href_node.text
self.assertEquals(expected_hrefs[received_count], href,
"sortOrder %s:\n\tunexpected href: %s (expecting: %s)"
% (sortOrder[0], href,
expected_hrefs[received_count]))
received_count = received_count + 1
self.assertEquals(expected_count, received_count,
"sortOrder %s:\n\tunexpected amount of refs: %d"
% (sortOrder[0], received_count))
def testREPORTMailQueryFilters(self):
"""mail-query filters"""
self._deleteCollection("test-dav-mail")
self._makeCollection("test-dav-mail")
msg1Loc = self._putMessage(self.client, "test-dav-mail", message1)
parsed = webdavlib.HTTPUnparsedURL(msg1Loc)
msg1Path = parsed.path
msg2Loc = self._putMessage(self.client, "test-dav-mail", message2)
parsed = webdavlib.HTTPUnparsedURL(msg2Loc)
msg2Path = parsed.path
msg3Loc = self._putMessage(self.client, "test-dav-mail", message3)
parsed = webdavlib.HTTPUnparsedURL(msg3Loc)
msg3Path = parsed.path
properties = ["{DAV:}displayname"]
## 1. test filter: receive-date
# SINCE, BEFORE, ON
# q = MailDAVMailQuery(self.resource, properties, filters = None)
filters = (({ "receive-date": { "from": "20091201T000000Z",
"to": "20091208T000000Z" } },
[]),
({ "receive-date": { "from": "20091208T000000Z",
"to": "20091213T134300Z" } },
[ msg2Loc ]),
({ "receive-date": { "from": "20091208T000000Z",
"to": "20091216T134300Z" } },
[ msg2Loc, msg3Loc ]),
({ "receive-date": { "from": "20091216T000000Z",
"to": "20091220T134300Z" } },
[ msg1Loc ]),
({ "receive-date": { "from": "20091220T000000Z",
"to": "20091229T134300Z" } },
[]))
# receive-date seems to be considered the same as date by imapd
print "Warning, receive-date test disabled"
#self._testFilters(filters)
## 1. test filter: date
# SENTSINCE, SENTBEFORE, SENTON
filters = (({ "date": { "from": "20090101T000000Z",
"to": "20090201T000000Z" } },
[]),
({ "date": { "from": "20090912T000000Z",
"to": "20090929T134300Z" } },
[ msg1Loc ]),
({ "date": { "from": "20090929T134300Z",
"to": "20091209T000000Z" } },
[]),
({ "date": { "from": "20090901T134300Z",
"to": "20091209T000000Z" } },
[ msg1Loc, msg2Loc ]),
({ "date": { "from": "20091201T000000Z",
"to": "20091211T000000Z" } },
[ msg3Loc ]),
({ "date": { "from": "20091211T000000Z",
"to": "20101211T000000Z" } },
[]),
({ "date": { "from": "20090101T000000Z",
"to": "20100101T000000Z" } },
[ msg1Loc, msg2Loc, msg3Loc ]))
self._testFilters(filters)
## 1. test filter: sequence
# x:y
filters = (({ "sequence": { "from": "1" }},
[ msg1Loc, msg2Loc, msg3Loc ]),
({ "sequence": { "from": "5" }},
[]),
({ "sequence": { "to": "5" }},
[ msg1Loc, msg2Loc, msg3Loc ]),
({ "sequence": { "from": "1",
"to": "2" }},
[ msg1Loc, msg2Loc ]))
# Sequence not yet implemented
print "Warning, sequence test disabled"
#self._testFilters(filters)
## 1. test filter: uid
# UID
filters = (({ "uid": { "from": "1" }},
[ msg1Loc, msg2Loc, msg3Loc ]),
# disabled because we get 3
#({ "uid": { "from": "5" }},
# []),
({ "uid": { "to": "5" }},
[ msg1Loc, msg2Loc, msg3Loc ]),
({ "uid": { "from": "1",
"to": "2" }},
[ msg1Loc, msg2Loc ]))
print "Warning, one of the uid tests is disabled"
self._testFilters(filters)
## 1. test filter: from
# FROM
filters = (({ "from": { "match": "message" }},
[ msg1Loc, msg2Loc, msg3Loc ]),
({ "from": { "match": "Cyril" }},
[ msg1Loc, msg2Loc, msg3Loc ]),
({ "from": { "match": "cyril.dev" }},
[ msg1Loc, msg2Loc, msg3Loc ]),
({ "from": { "match": "message1from" }},
[ msg1Loc ]),
({ "from": { "match": "message2from" }},
[ msg2Loc ]),
({ "from": { "match": "message3from" }},
[ msg3Loc ]))
self._testFilters(filters)
## 1. test filter: to
# TO
filters = (({ "to": { "match": "message" }},
[ msg1Loc, msg2Loc, msg3Loc ]),
({ "to": { "match": "Cyril" }},
[ msg1Loc, msg2Loc, msg3Loc ]),
({ "to": { "match": "cyril.dev" }},
[ msg1Loc, msg2Loc, msg3Loc ]),
({ "to": { "match": "message1to" }},
[ msg1Loc ]),
({ "to": { "match": "message2to" }},
[ msg2Loc ]),
({ "to": { "match": "message3to" }},
[ msg3Loc ]))
self._testFilters(filters)
## 1. test filter: cc
# CC
filters = (({ "cc": { "match": "message" }},
[ msg1Loc, msg2Loc, msg3Loc ]),
({ "cc": { "match": "Cyril" }},
[ msg1Loc, msg2Loc, msg3Loc ]),
({ "cc": { "match": "cyril.dev" }},
[ msg1Loc, msg2Loc, msg3Loc ]),
({ "cc": { "match": "message1cc" }},
[ msg1Loc ]),
({ "cc": { "match": "message2cc" }},
[ msg2Loc ]),
({ "cc": { "match": "message3cc" }},
[ msg3Loc ]))
self._testFilters(filters)
## 1. test filter: bcc
# BCC
## TODO
## 1. test filter: body
# BODY
filters = (({ "body": { "match": "Hello" }},
[ msg1Loc, msg2Loc, msg3Loc ]),
({ "body": { "match": "Stuff" }},
[ msg2Loc ]),
({ "body": { "match": "DOESNOT MATCH" }},
[]))
self._testFilters(filters)
## 1. test filter: size
# LARGER, SMALLER
#1 848
#2 4308
#3 699
filters = (({ "size": { "min": "300",
"max": "300" }},
[]),
({ "size": { "min": "800",
"max": "800" }},
[]),
({ "size": { "min": "5000",
"max": "5000" }},
[]),
({ "size": { "min": "838",
"max": "838" }},
[ msg1Loc ]),
({ "size": { "min": "699",
"max": "4308" }},
[ msg1Loc, msg2Loc, msg3Loc ]),
({ "size": { "min": "700",
"max": "4308" }},
[ msg1Loc, msg2Loc ]),
({ "size": { "min": "698",
"max": "848" }},
[ msg1Loc, msg3Loc ]),
({ "size": { "min": "300",
"max": "5000" },
"size": { "min": "840",
"max": "850",
"not": "true" }},
[ msg2Loc, msg3Loc ]))
print "message flags are not handled yet"
## 1. test filter: answered
# ANSWERED, UNANSWERED
## 1. test filter: draft
# DRAFT
## 1. test filter: flagged
# FLAGGED
## 1. test filter: recent
# RECENT
## 1. test filter: seen
# SEEN
## 1. test filter: deleted
# DELETED
## 1. test filter: keywords
# KEYWORD x
## 1. test filter: multiple combinations
filters = (({ "body": { "match": "Hello" },
"cc": { "match": "message1cc" }},
[ msg1Loc ]),
({ "to": { "match": "message" },
"uid": { "from": "1",
"to": "2" }},
[ msg1Loc, msg2Loc ]),
({ "to": { "match": "message" },
"uid": { "from": "1",
"to": "2" },
"cc": { "match": "message3cc" }},
[]))
self._testFilters(filters)
self._deleteCollection("test-dav-mail")
def testREPORTMailQuerySort(self):
"""mail-query sort"""
self._deleteCollection("test-dav-mail")
self._makeCollection("test-dav-mail")
msg1Loc = self._putMessage(self.client, "test-dav-mail", message1)
parsed = webdavlib.HTTPUnparsedURL(msg1Loc)
msg1Path = parsed.path
msg2Loc = self._putMessage(self.client, "test-dav-mail", message2)
parsed = webdavlib.HTTPUnparsedURL(msg2Loc)
msg2Path = parsed.path
msg3Loc = self._putMessage(self.client, "test-dav-mail", message3)
parsed = webdavlib.HTTPUnparsedURL(msg3Loc)
msg3Path = parsed.path
# 1. test sort: (receive-date) ARRIVAL
self._testSort(([ "{urn:schemas:mailheader:}received" ],
[ msg1Loc, msg2Loc, msg3Loc ]))
# 1. test sort: (date) DATE
self._testSort(([ "{urn:schemas:mailheader:}date" ],
[ msg2Loc, msg1Loc, msg3Loc ]))
# 1. test sort: FROM
self._testSort(([ "{urn:schemas:mailheader:}from" ],
[ msg1Loc, msg2Loc, msg3Loc ]))
# 1. test sort: TO
self._testSort(([ "{urn:schemas:mailheader:}to" ],
[ msg1Loc, msg2Loc, msg3Loc ]))
# 1. test sort: CC
self._testSort(([ "{urn:schemas:mailheader:}cc" ],
[ msg3Loc, msg1Loc, msg2Loc ]))
# 1. test sort: SUBJECT
self._testSort(([ "{DAV:}displayname" ],
[ msg3Loc, msg1Loc, msg2Loc ]))
self._testSort(([ "{urn:schemas:mailheader:}subject" ],
[ msg3Loc, msg1Loc, msg2Loc ]))
# 1. test sort: SIZE
self._testSort(([ "{DAV:}getcontentlength" ],
[ msg3Loc, msg1Loc, msg2Loc ]))
# 1. test sort: REVERSE CC
self._testSort(([ "{urn:schemas:mailheader:}cc" ],
[ msg2Loc, msg1Loc, msg3Loc ]),
False)
self._deleteCollection("test-dav-mail")
def testPROPFIND(self):
"""Message properties"""
self._deleteCollection ("test-dav-mail")
self._makeCollection ("test-dav-mail")
url = "%s%s" % (self.resource, "foldertest-dav-mail/")
put = webdavlib.HTTPPUT (url, message1)
put.content_type = "message/rfc822"
self.client.execute (put)
self.assertEquals(put.response["status"], 201,
"failure putting message"
"(code = %d)" % put.response["status"])
itemLocation = put.response["headers"]["location"]
tests = (("{urn:schemas:httpmail:}date", 1254156134, 1),
("{urn:schemas:httpmail:}hasattachment", "0", 0),
("{urn:schemas:httpmail:}read", "0", 0),
("{urn:schemas:httpmail:}textdescription",
"<![CDATA[%s]]>" % message1, 0),
("{urn:schemas:httpmail:}unreadcount", None, 0),
("{urn:schemas:mailheader:}cc",
"2message1cc@cyril.dev, user10@cyril.dev", 0),
("{urn:schemas:mailheader:}date",
"Mon, 28 Sep 2009 11:42:14 GMT", 0),
("{urn:schemas:mailheader:}from",
"Cyril <message1from@cyril.dev>", 0),
("{urn:schemas:mailheader:}in-reply-to", None, 0),
("{urn:schemas:mailheader:}message-id",
"<4AC1F29sept6.5060801@cyril.dev>", 0),
("{urn:schemas:mailheader:}received", message1_received, 0),
("{urn:schemas:mailheader:}references",
"<4AC3BF1B.3010806@inverse.ca>", 0),
("{urn:schemas:mailheader:}subject", "message1subject", 0),
("{urn:schemas:mailheader:}to", "message1to@cyril.dev", 0))
for test in tests:
property, expected, isDate = test
self._testProperty(itemLocation, property, expected, isDate)
self._deleteCollection ("test-dav-mail")
if __name__ == "__main__":
sogotests.runTests()

60
Tests/Integration/test-preferences.py
View File

@@ -1,60 +0,0 @@
#!/usr/bin/python
from config import hostname, port, username, password, white_listed_attendee
import preferences
import simplejson
import sogotests
import unittest
import utilities
class preferencesTest(unittest.TestCase):
def setUp(self):
self.prefs = preferences.preferences()
# because if not set in vacation will not be found later
# we must make sure they are there at the start
self.prefs.set_or_create("autoReplyText", '', ["defaults", "Vacation"])
self.prefs.set_or_create("PreventInvitations", 0, ["settings", "Calendar"])
self.prefs.set_or_create("PreventInvitationsWhitelist", [], ["settings", "Calendar"])
def tearDown(self):
self.prefs.set("autoReplyText", "")
def _setTextPref(self, prefText = None ):
""" set a text preference to a known value """
self.prefs.set("autoReplyText", prefText)
# make sure it was set correctly
prefData = self.prefs.get("Vacation")
self.assertEqual(prefData["autoReplyText"], prefText,
"%s != %s" % (prefData["autoReplyText"], prefText))
def testSetTextPreferences(self):
""" Set/get a text preference - normal characters"""
self._setTextPref("defaultText")
def testSetTextPreferencesWeirdChars(self):
""" Set/get a text preference - weird characters - used to crash on 1.3.12"""
prefText = "weird data \ ' \"; ^"
self._setTextPref(prefText)
def testSetPreventInvitation(self):
""" Set/get the PreventInvitation pref"""
self.prefs.set('PreventInvitations', 0)
notset = self.prefs.get('Calendar')['PreventInvitations']
self.assertEqual(notset, 0)
self.prefs.set('PreventInvitations', 1)
isset = self.prefs.get('Calendar')['PreventInvitations']
self.assertEqual(isset, 1)
def testPreventInvitationsWhiteList(self):
"""Add to the PreventInvitations Whitelist"""
self.prefs.set("PreventInvitationsWhitelist", white_listed_attendee)
whitelist = self.prefs.get('Calendar')['PreventInvitationsWhitelist']
self.assertEqual(whitelist, white_listed_attendee)
if __name__ == "__main__":
sogotests.runTests()

219
Tests/Integration/test-prevent-invitations.py
View File

@@ -1,219 +0,0 @@
#!/usr/bin/python
from config import hostname, port, username, password, \
superuser, superuser_password, \
attendee1, attendee1_username, \
attendee1_password, \
attendee1_delegate, attendee1_delegate_username, \
attendee1_delegate_password, \
resource_no_overbook, resource_can_overbook, \
white_listed_attendee
import preferences
import simplejson
import sogotests
import unittest
import utilities
import datetime
import dateutil.tz
import vobject
import vobject.base
import vobject.icalendar
import webdavlib
import StringIO
class preventInvitationsTest(unittest.TestCase):
def setUp(self):
self.prefs = preferences.preferences(attendee1, attendee1_password)
self.caldav = CalDAVSchedulingTest(self)
cal = self.prefs.get("Calendar")
if "PreventInvitationsWhitelist" not in cal:
cal["PreventInvitationsWhitelist"] = None
self.prefs.set("PreventInvitationsWhitelist", None)
if "PreventInvitations" not in cal:
cal["PreventInvitations"] = 0
self.prefs.set("PreventInvitations", 0)
def tearDown(self):
#self.prefs.set("autoReplyText", "")
self.prefs.set('PreventInvitations', 0)
self.prefs.set("PreventInvitationsWhitelist", None)
#- Manual Cleanup, not called because classs is not derived from unittest
self.caldav.tearDown()
def not_test_empty_string_instead_of_null(self):
self.prefs.set('PreventInvitationsWhitelist', "")
def testDontPreventInvitation(self):
""" Set/get the PreventInvitation pref"""
#- First accept the invitation
self.prefs.set('PreventInvitations', 0)
notset = self.prefs.get_settings('')['Calendar']['PreventInvitations']
self.assertEqual(notset, 0)
self.caldav.AddAttendee()
self.caldav.VerifyEvent()
def testPreventInvitation(self):
""" Set PreventInvitation and don't accept the Invitation"""
#- Second, enable PreventInviation and refuse it
self.prefs.set('PreventInvitations', 1)
isset = self.prefs.get_settings('')['Calendar']['PreventInvitations']
self.assertEqual(isset, 1)
self.caldav.AddAttendee(409)
self.caldav.VerifyEvent(404)
def testPreventInvitationWhiteList(self):
""" Set PreventInvitation add to WhiteList and accept the Invitation"""
#- First, add the Organiser to the Attendee's whitelist
self.prefs.set('PreventInvitations', 1)
self.prefs.set("PreventInvitationsWhitelist", simplejson.dumps(white_listed_attendee))
whitelist = self.prefs.get_settings('Calendar')['PreventInvitationsWhitelist']
self.assertEqual(whitelist, white_listed_attendee)
#- Second, try again to invite, it should work
self.prefs.set('PreventInvitations', 1)
isset = self.prefs.get_settings('')['Calendar']['PreventInvitations']
self.assertEqual(isset, 1)
self.caldav.AddAttendee()
self.caldav.VerifyEvent()
class CalDAVSchedulingTest(object):
def __init__(self, parent_self):
self.test = parent_self # used for utilities
self.setUp()
def setUp(self):
self.superuser_client = webdavlib.WebDAVClient(hostname, port,
superuser, superuser_password)
self.client = webdavlib.WebDAVClient(hostname, port,
username, password)
self.attendee1_client = webdavlib.WebDAVClient(hostname, port,
attendee1_username, attendee1_password)
self.attendee1_delegate_client = webdavlib.WebDAVClient(hostname, port,
attendee1_delegate_username, attendee1_delegate_password)
utility = utilities.TestUtility(self.test, self.client)
(self.user_name, self.user_email) = utility.fetchUserInfo(username)
(self.attendee1_name, self.attendee1_email) = utility.fetchUserInfo(attendee1)
(self.attendee1_delegate_name, self.attendee1_delegate_email) = utility.fetchUserInfo(attendee1_delegate)
self.user_calendar = "/SOGo/dav/%s/Calendar/personal/" % username
self.attendee1_calendar = "/SOGo/dav/%s/Calendar/personal/" % attendee1
self.attendee1_delegate_calendar = "/SOGo/dav/%s/Calendar/personal/" % attendee1_delegate
# fetch non existing event to let sogo create the calendars in the db
self._getEvent(self.client, "%snonexistent" % self.user_calendar, exp_status=404)
self._getEvent(self.attendee1_client, "%snonexistent" % self.attendee1_calendar, exp_status=404)
self._getEvent(self.attendee1_delegate_client, "%snonexistent" %
self.attendee1_delegate_calendar, exp_status=404)
# list of ics used by the test.
# tearDown will loop over this and wipe them in all users' calendar
self.ics_list = []
def tearDown(self):
# delete all created events from all users' calendar
for ics in self.ics_list:
self._deleteEvent(self.superuser_client,
"%s%s" % (self.user_calendar, ics), None)
self._deleteEvent(self.superuser_client,
"%s%s" % (self.attendee1_calendar, ics), None)
self._deleteEvent(self.superuser_client,
"%s%s" % (self.attendee1_delegate_calendar, ics), None)
def _newEvent(self, summary="test event", uid="test", transp=0):
transparency = ("OPAQUE", "TRANSPARENT")
newCal = vobject.iCalendar()
vevent = newCal.add('vevent')
vevent.add('summary').value = summary
vevent.add('transp').value = transparency[transp]
now = datetime.datetime.now(dateutil.tz.gettz("America/Montreal"))
startdate = vevent.add('dtstart')
startdate.value = now
enddate = vevent.add('dtend')
enddate.value = now + datetime.timedelta(0, 3600)
vevent.add('uid').value = uid
vevent.add('dtstamp').value = now
vevent.add('last-modified').value = now
vevent.add('created').value = now
vevent.add('class').value = "PUBLIC"
vevent.add('sequence').value = "0"
return newCal
def _putEvent(self, client, filename, event, exp_status = 201):
put = webdavlib.HTTPPUT(filename, event.serialize())
put.content_type = "text/calendar; charset=utf-8"
client.execute(put)
if exp_status is not None:
self.test.assertEquals(put.response["status"], exp_status)
def _getEvent(self, client, filename, exp_status = 200):
get = webdavlib.HTTPGET(filename)
client.execute(get)
if exp_status is not None:
self.test.assertEquals(get.response["status"], exp_status)
if get.response["headers"]["content-type"].startswith("text/calendar"):
stream = StringIO.StringIO(get.response["body"])
event = vobject.base.readComponents(stream).next()
else:
event = None
return event
def _deleteEvent(self, client, filename, exp_status = 204):
delete = webdavlib.WebDAVDELETE(filename)
client.execute(delete)
if exp_status is not None:
self.test.assertEquals(delete.response["status"], exp_status)
def AddAttendee(self, exp_status=204):
""" add attendee after event creation """
# make sure the event doesn't exist
ics_name = "test-add-attendee.ics"
self.ics_list += [ics_name]
self._deleteEvent(self.client,
"%s%s" % (self.user_calendar,ics_name), None)
self._deleteEvent(self.attendee1_client,
"%s%s" % (self.attendee1_calendar,ics_name), None)
# 1. create an event in the organiser's calendar
event = self._newEvent(summary="Test add attendee", uid="Test add attendee")
organizer = event.vevent.add('organizer')
organizer.cn_param = self.user_name
organizer.value = self.user_email
self._putEvent(self.client, "%s%s" % (self.user_calendar, ics_name), event)
# 2. add an attendee
event.add("method").value = "REQUEST"
attendee = event.vevent.add('attendee')
attendee.cn_param = self.attendee1_name
attendee.rsvp_param = "TRUE"
attendee.partstat_param = "NEEDS-ACTION"
attendee.value = self.attendee1_email
self._putEvent(self.client, "%s%s" % (self.user_calendar, ics_name), event,
exp_status=exp_status)
#- save event for VerifyEvent
self.event = event
self.ics_name = ics_name
def VerifyEvent(self, exp_status=200):
# 1. verify that the attendee has the event
attendee_event = self._getEvent(self.attendee1_client, "%s%s" % (self.attendee1_calendar, self.ics_name), exp_status)
# 2. make sure the received event match the original one
if attendee_event:
self.test.assertEquals(self.event.vevent.uid, attendee_event.vevent.uid)
if __name__ == "__main__":
sogotests.runTests()

164
Tests/Integration/test-sieve.py
View File

@@ -1,164 +0,0 @@
#!/usr/bin/python
from config import hostname, port, username, password, sieve_port, sieve_server
import managesieve
import preferences
import sogotests
import unittest
import utilities
import webdavlib
sieve_simple_vacation="""require ["vacation"];\r\nvacation :days %(days)d :addresses ["%(mailaddr)s"] text:\r\n%(vacation_msg)s\r\n.\r\n;\r\n"""
sieve_vacation_ignoreLists="""require ["vacation"];\r\nif allof ( not exists ["list-help", "list-unsubscribe", "list-subscribe", "list-owner", "list-post", "list-archive", "list-id", "Mailing-List"], not header :comparator "i;ascii-casemap" :is "Precedence" ["list", "bulk", "junk"], not header :comparator "i;ascii-casemap" :matches "To" "Multiple recipients of*" ) { vacation :days %(days)d :addresses ["%(mailaddr)s"] text:\r\n%(vacation_msg)s\r\n.\r\n;\r\n}\r\n"""
sieve_simple_forward="""redirect "%(redirect_mailaddr)s";\r\n"""
sieve_forward_keep="""redirect "%(redirect_mailaddr)s";\r\nkeep;\r\n"""
sieve_simple_filter="""require ["fileinto"];\r\nif anyof (header :contains "subject" "%(subject)s") {\r\n fileinto "%(folderName)s";\r\n}\r\n"""
class sieveTest(unittest.TestCase):
def _killFilters(self):
self.prefs=preferences.preferences()
# kill existing filters
self.prefs.set_or_create("SOGoSieveFilters", [], ["defaults"])
# vacation filters
self.prefs.set_or_create("autoReplyText", "", ["defaults", "Vacation"])
self.prefs.set_or_create("autoReplyEmailAddresses", [], ["defaults", "Vacation"])
self.prefs.set_or_create("daysBetweenResponse", 7, ["defaults", "Vacation"])
self.prefs.set_or_create("ignoreLists", 0, ["defaults", "Vacation"])
self.prefs.set_or_create("enabled", 0, ["defaults", "Vacation"])
# forwarding filters
self.prefs.set_or_create("forwardAddress", [], ["defaults", "Forward"])
self.prefs.set_or_create("keepCopy", 0, ["defaults", "Forward"])
def setUp(self):
ret = ""
self.client = webdavlib.WebDAVClient(hostname, port,
username, password)
utility = utilities.TestUtility(self, self.client)
(self.user_name, self.user_email) = utility.fetchUserInfo(username)
self.user_email = self.user_email.replace("mailto:", "")
self.ms = managesieve.MANAGESIEVE(sieve_server, sieve_port)
self.assertEqual(self.ms.login("", username, password), "OK",
"Couldn't login")
self._killFilters()
def tearDown(self):
self._killFilters()
def _getSogoSieveScript(self):
sieveFoundsogo=0
createdSieveScript=""
(ret, sieveScriptList) = self.ms.listscripts()
self.assertEqual(ret, "OK", "Couldn't get sieve script list")
for (script, isActive) in sieveScriptList:
if (script == "sogo"):
sieveFoundsogo=1
self.assertEqual(isActive, True, "sogo sieve script is not active!")
(ret, createdSieveScript) = self.ms.getscript(script)
self.assertEqual(ret, "OK", "Couldn't get sogo sieve script")
self.assertEqual(sieveFoundsogo, 1, "sogo sieve script not found!")
return createdSieveScript
def testSieveSimpleVacation(self):
""" enable simple vacation script """
vacation_msg="vacation test"
daysSelect=3
sieveScript = sieve_simple_vacation % { "mailaddr": self.user_email,
"vacation_msg": vacation_msg,
"days": preferences.daysBetweenResponseList[daysSelect],
}
# Enabling Vacation now is an 'enabled' setting in the subdict Vacation
# We need to get that subdict first -- next save/set will also save this
vacation = self.prefs.get("Vacation")
vacation['enabled'] = 1
self.prefs.set_nosave("autoReplyText", vacation_msg)
self.prefs.set_nosave("daysBetweenResponse", "%d" % preferences.daysBetweenResponseList[daysSelect])
self.prefs.set_nosave("autoReplyEmailAddresses", [self.user_email])
self.prefs.save()
createdSieveScript=self._getSogoSieveScript()
self.assertEqual(sieveScript, createdSieveScript)
def testSieveVacationIgnoreLists(self):
""" enable vacation script - ignore lists"""
vacation_msg="vacation test - ignore list"
daysSelect=2
sieveScript = sieve_vacation_ignoreLists % { "mailaddr": self.user_email,
"vacation_msg": vacation_msg,
"days": preferences.daysBetweenResponseList[daysSelect],
}
# Enabling Vacation now is an 'enabled' setting in the subdict Vacation
# We need to get that subdict first -- next save/set will also save this
vacation = self.prefs.get("Vacation")
vacation['enabled'] = 1
self.prefs.set_nosave("autoReplyText", vacation_msg)
self.prefs.set_nosave("daysBetweenResponse", "%d" % preferences.daysBetweenResponseList[daysSelect])
self.prefs.set_nosave("autoReplyEmailAddresses", [self.user_email])
self.prefs.set_nosave("ignoreLists", 1)
self.prefs.save()
createdSieveScript=self._getSogoSieveScript()
self.assertEqual(sieveScript, createdSieveScript)
def testSieveSimpleForward(self):
""" enable simple forwarding """
redirect_mailaddr="nonexistent@inverse.com"
sieveScript = sieve_simple_forward % { "redirect_mailaddr": redirect_mailaddr }
# Enabling Forward now is an 'enabled' setting in the subdict Forward
# We need to get that subdict first -- next save/set will also save this
forward = self.prefs.get("Forward")
forward['enabled'] = 1
self.prefs.set("forwardAddress", [redirect_mailaddr])
createdSieveScript = self._getSogoSieveScript()
self.assertEqual(sieveScript, createdSieveScript)
def testSieveForwardKeepCopy(self):
""" enable email forwarding - keep a copy """
redirect_mailaddr="nonexistent@inverse.com"
sieveScript = sieve_forward_keep % { "redirect_mailaddr": redirect_mailaddr }
# Enabling Forward now is an 'enabled' setting in the subdict Forward
# We need to get that subdict first -- next save/set will also save this
forward = self.prefs.get("Forward")
forward['enabled'] = 1
self.prefs.set_nosave("forwardAddress", [redirect_mailaddr])
self.prefs.set_nosave("keepCopy", 1)
self.prefs.save()
createdSieveScript = self._getSogoSieveScript()
self.assertEqual(sieveScript, createdSieveScript)
def testSieveSimpleFilter(self):
""" add simple sieve filter """
folderName="Sent"
subject=__name__
sieveScript=sieve_simple_filter % { "subject": subject, "folderName": folderName }
self.prefs.set("SOGoSieveFilters", [{"active": True, "actions": [{"method": "fileinto", "argument": "Sent"}], "rules": [{"operator": "contains", "field": "subject", "value": subject}], "match": "any", "name": folderName}])
createdSieveScript = self._getSogoSieveScript()
self.assertEqual(sieveScript, createdSieveScript)
if __name__ == "__main__":
sogotests.runTests()

44
Tests/Integration/test-sogo-tool.py
View File

@@ -1,44 +0,0 @@
#!/usr/bin/python
# XXX this script has to be run as root because it su to sogo_user
# in order to use its .GNUstepDefaults prefs
# Would be much better to have another way to specify which Defaults to use
from config import sogo_user, sogo_tool_path
import os
import pwd
import shutil
import sogotests
import tempfile
import unittest
class sogoToolTest(unittest.TestCase):
def setUp(self):
self.backupdir = tempfile.mkdtemp()
def tearDown(self):
os.chdir("/")
shutil.rmtree(self.backupdir)
def testBackupAll(self):
""" sogo-tool backup ALL """
(uid, gid) = pwd.getpwnam(sogo_user)[2:4]
# We need to run as root since there's no way
# of using another user's GNUstepDefaults
self.assertEqual(os.getuid(), 0, "this test must run as root...")
os.chown(self.backupdir, uid, gid)
cmd = "sudo -u %s bash -c \"(cd %s && %s backup . ALL >/dev/null 2>&1)\"" % (sogo_user, self.backupdir, sogo_tool_path)
#print "sogo-tool cmd to execute %s" % cmd
status = os.system(cmd)
#print "Exit status of os.system(): %d" % status
rc = os.WEXITSTATUS(status)
#self.assertEqual(rc, 0, "sogo-tool failed RC=%d" % rc)
if __name__ == "__main__":
sogotests.runTests()

73
Tests/Integration/test-ui-posts.py
View File

@@ -1,73 +0,0 @@
#!/usr/bin/python
from config import hostname, port, username, password, \
webCalendarURL
import simplejson
import sogoLogin
import sogotests
import unittest
import utilities
import webdavlib
import httplib
class UIPostsTests(unittest.TestCase):
def setUp(self):
self.client = webdavlib.WebDAVClient(hostname, port)
self.gcClient = webdavlib.WebDAVClient(hostname, port)
self.cookie = sogoLogin.getAuthCookie(hostname, port, username, password)
def _urlPostData(self, client, url, data, exp_status=200):
post = webdavlib.HTTPPOST(url, simplejson.dumps(data))
post.content_type = "application/json"
post.cookie = self.cookie
client.execute(post)
if (exp_status is not None):
self.assertEquals(post.response["status"], exp_status)
return post.response
def _urlGet(self, client, url, exp_status=200):
get = webdavlib.HTTPGET(url)
get.cookie = self.cookie
client.execute(get)
if (exp_status is not None):
self.assertEquals(get.response["status"], exp_status)
return get.response
def testAddWebCalendar(self):
""" Add Web Calendar """
ret=True
data = {"url":"%s" % webCalendarURL}
calendarBaseURL="/SOGo/so/%s/Calendar" % username
addWebCalendarURL = "%s/addWebCalendar" % calendarBaseURL
response = self._urlPostData(self.client, addWebCalendarURL, data)
respJSON = simplejson.loads(response['body'])
calID = respJSON['id']
self.assertNotEqual(calID, None)
# reload the cal
calURL = "%s/%s" % (calendarBaseURL, calID)
try:
response = self._urlGet(self.client, "%s/reload" % calURL, exp_status=None)
except httplib.BadStatusLine:
# that's bad, the server probably reset the connection. fake a 502
response['status'] = 502
# cleanup our trash
self._urlPostData(self.gcClient, "%s/delete" % calURL, "", exp_status=None)
# delayed assert to allow cal deletion on failure
self.assertEqual(response['status'], 200)
if __name__ == "__main__":
sogotests.runTests()

155
Tests/Integration/test-webdav.py
View File

@@ -1,155 +0,0 @@
#!/usr/bin/python
from config import hostname, port, username, password, subscriber_username
import sogotests
import unittest
import utilities
import webdavlib
class WebDAVTest(unittest.TestCase):
def __init__(self, arg):
unittest.TestCase.__init__(self, arg)
self.client = webdavlib.WebDAVClient(hostname, port,
username, password)
self.dav_utility = utilities.TestUtility(self, self.client)
def testPrincipalCollectionSet(self):
"""property: 'principal-collection-set' on collection object"""
resource = '/SOGo/dav/%s/' % username
propfind = webdavlib.WebDAVPROPFIND(resource,
["{DAV:}principal-collection-set"],
0)
self.client.execute(propfind)
self.assertEquals(propfind.response["status"], 207)
nodes = propfind.response["document"] \
.findall('{DAV:}response/{DAV:}propstat/{DAV:}prop/{DAV:}principal-collection-set/{DAV:}href')
responseHref = nodes[0].text
if responseHref[0:4] == "http":
self.assertEquals("http://%s/SOGo/dav/" % hostname, responseHref,
"{DAV:}principal-collection-set returned %s instead of 'http../SOGo/dav/'"
% ( responseHref, resource ))
else:
self.assertEquals("/SOGo/dav/", responseHref,
"{DAV:}principal-collection-set returned %s instead of '/SOGo/dav/'"
% responseHref)
def testPrincipalCollectionSet2(self):
"""property: 'principal-collection-set' on non-collection object"""
resource = '/SOGo/dav/%s/freebusy.ifb' % username
propfind = webdavlib.WebDAVPROPFIND(resource,
["{DAV:}principal-collection-set"],
0)
self.client.execute(propfind)
self.assertEquals(propfind.response["status"], 207)
node = propfind.response["document"] \
.find('{DAV:}response/{DAV:}propstat/{DAV:}prop/{DAV:}principal-collection-set/{DAV:}href')
responseHref = node.text
expectedHref = '/SOGo/dav/'
if responseHref[0:4] == "http":
self.assertEquals("http://%s%s" % (hostname, expectedHref), responseHref,
"{DAV:}principal-collection-set returned %s instead of '%s'"
% ( responseHref, expectedHref ))
else:
self.assertEquals(expectedHref, responseHref,
"{DAV:}principal-collection-set returned %s instead of '%s'"
% ( responseHref, expectedHref ))
def _testPropfindURL(self, resource):
resourceWithSlash = resource[-1] == '/'
propfind = webdavlib.WebDAVPROPFIND(resource,
["{DAV:}displayname", "{DAV:}resourcetype"],
1)
self.client.execute(propfind)
self.assertEquals(propfind.response["status"], 207)
nodes = propfind.response["document"].findall('{DAV:}response')
for node in nodes:
responseHref = node.find('{DAV:}href').text
hasSlash = responseHref[-1] == '/'
resourcetype = node.find('{DAV:}propstat/{DAV:}prop/{DAV:}resourcetype')
isCollection = len(resourcetype.getchildren()) > 0
if isCollection:
self.assertEquals(hasSlash, resourceWithSlash,
"failure with href '%s' while querying '%s'"
% (responseHref, resource))
else:
self.assertEquals(hasSlash, False,
"failure with href '%s' while querying '%s'"
% (responseHref, resource))
def testPropfindURL(self):
"""propfind: ensure various NSURL work-arounds"""
# a collection without /
self._testPropfindURL('/SOGo/dav/%s' % username)
# a collection with /
self._testPropfindURL('/SOGo/dav/%s/' % username)
# a non-collection
self._testPropfindURL('/SOGo/dav/%s/freebusy.ifb' % username)
## REPORT
def testPrincipalPropertySearch(self):
"""principal-property-search"""
resource = '/SOGo/dav'
userInfo = self.dav_utility.fetchUserInfo(username)
# subscriber_userInfo = self.dav_utility.fetchUserInfo(subscriber_username)
matches = [["{urn:ietf:params:xml:ns:caldav}calendar-home-set",
"/SOGo/dav/%s/Calendar" % username]]
## the SOGo implementation does not support more than one
## property-search at a time:
# ["{urn:ietf:params:xml:ns:caldav}calendar-home-set",
# "/SOGo/dav/%s/Calendar" % subscriber_username]]
query = webdavlib.WebDAVPrincipalPropertySearch(resource,
["displayname"], matches)
self.client.execute(query)
self.assertEquals(query.response["status"], 207)
response = query.response["document"].findall('{DAV:}response')[0]
href = response.find('{DAV:}href').text
self.assertEquals("/SOGo/dav/%s/" % username, href)
displayname = response.find('{DAV:}propstat/{DAV:}prop/{DAV:}displayname')
value = displayname.text
if value is None:
value = ""
self.assertEquals(userInfo[0], value)
# http://tools.ietf.org/html/rfc3253.html#section-3.8
def testExpandProperty(self):
"""expand-property"""
resource = '/SOGo/dav/%s/' % username
userInfo = self.dav_utility.fetchUserInfo(username)
query_props = {"{DAV:}owner": { "{DAV:}href": resource,
"{DAV:}displayname": userInfo[0]},
"{DAV:}principal-collection-set": { "{DAV:}href": "/SOGo/dav/",
"{DAV:}displayname": "SOGo"}}
query = webdavlib.WebDAVExpandProperty(resource, query_props.keys(),
["displayname"])
self.client.execute(query)
self.assertEquals(query.response["status"], 207)
topResponse = query.response["document"].find('{DAV:}response')
topHref = topResponse.find('{DAV:}href')
self.assertEquals(resource, topHref.text)
for query_prop in query_props.keys():
propResponse = topResponse.find('{DAV:}propstat/{DAV:}prop/%s'
% query_prop)
propHref = propResponse.find('{DAV:}response/{DAV:}href')
self.assertEquals(query_props[query_prop]["{DAV:}href"],
propHref.text,
"'%s', href mismatch: exp. '%s', got '%s'"
% (query_prop,
query_props[query_prop]["{DAV:}href"],
propHref.text))
propDisplayname = propResponse.find('{DAV:}response/{DAV:}propstat/{DAV:}prop/{DAV:}displayname')
displayName = propDisplayname.text
if displayName is None:
displayName = ""
self.assertEquals(query_props[query_prop]["{DAV:}displayname"],
displayName,
"'%s', displayname mismatch: exp. '%s', got '%s'"
% (query_prop,
query_props[query_prop]["{DAV:}displayname"],
propDisplayname))
if __name__ == "__main__":
sogotests.runTests()

51
Tests/Integration/test-webdavlib.py
View File

@@ -1,51 +0,0 @@
#!/usr/bin/python
import sogotests
import unittest
from webdavlib import *
class HTTPUnparsedURLTest(unittest.TestCase):
def testURLParse(self):
fullURL = "http://username:password@hostname:123/folder/folder/object?param1=value1&param2=value2"
testURL = HTTPUnparsedURL(fullURL)
self.assertEquals(testURL.protocol, "http")
self.assertEquals(testURL.username, "username")
self.assertEquals(testURL.password, "password")
self.assertEquals(testURL.hostname, "hostname")
self.assertEquals(testURL.port, "123")
self.assertEquals(testURL.path, "/folder/folder/object")
exp_params = { "param1": "value1",
"param2": "value2" }
self.assertEquals(exp_params, testURL.parameters)
pathURL = "/folder/folder/simplereference"
testURL = HTTPUnparsedURL(pathURL)
self.assertEquals(testURL.protocol, None)
self.assertEquals(testURL.username, None)
self.assertEquals(testURL.password, None)
self.assertEquals(testURL.hostname, None)
self.assertEquals(testURL.port, None)
self.assertEquals(testURL.path, "/folder/folder/simplereference")
pathURL = "http://user:secret@bla.com/hooray"
testURL = HTTPUnparsedURL(pathURL)
self.assertEquals(testURL.protocol, "http")
self.assertEquals(testURL.username, "user")
self.assertEquals(testURL.password, "secret")
self.assertEquals(testURL.hostname, "bla.com")
self.assertEquals(testURL.port, None)
self.assertEquals(testURL.path, "/hooray")
pathURL = "http://user@bla.com:80/hooray"
testURL = HTTPUnparsedURL(pathURL)
self.assertEquals(testURL.protocol, "http")
self.assertEquals(testURL.username, "user")
self.assertEquals(testURL.password, None)
self.assertEquals(testURL.hostname, "bla.com")
self.assertEquals(testURL.port, "80")
self.assertEquals(testURL.path, "/hooray")
if __name__ == "__main__":
sogotests.runTests()

71
Tests/Integration/test-webdavsync.py
View File

@@ -1,71 +0,0 @@
#!/usr/bin/python
from config import hostname, port, username, password
import math
import sys
import sogotests
import time
import unittest
import webdavlib
resource = '/SOGo/dav/%s/Calendar/test-webdavsync/' % username
class WebdavSyncTest(unittest.TestCase):
def setUp(self):
self.client = webdavlib.WebDAVClient(hostname, port,
username, password)
def tearDown(self):
delete = webdavlib.WebDAVDELETE(resource)
self.client.execute(delete)
def test(self):
"""webdav sync"""
# missing tests:
# invalid tokens: negative, non-numeric, > current timestamp
# non-empty collections: token validity, status codes for added,
# modified and removed elements
# preparation
mkcol = webdavlib.WebDAVMKCOL(resource)
self.client.execute(mkcol)
self.assertEquals(mkcol.response["status"], 201,
"preparation: failure creating collection (code != 201)")
# test queries:
# empty collection:
# without a token (query1)
# with a token (query2)
# (when done, non-empty collection:
# without a token (query3)
# with a token (query4))
query1 = webdavlib.WebDAVSyncQuery(resource, None, [ "getetag" ])
self.client.execute(query1)
self.assertEquals(query1.response["status"], 207,
("query1: invalid status code: %d (!= 207)"
% query1.response["status"]))
token_node = query1.response["document"].find("{DAV:}sync-token")
# Implicit "assertion": we expect SOGo to return a token node, with a
# non-empty numerical value. Anything else will trigger an exception
token = int(token_node.text)
self.assertTrue(token >= 0)
query1EndTime = int(math.ceil(query1.start + query1.duration))
self.assertTrue(token <= query1EndTime,
"token = %d > query1EndTime = %d" % (token, query1EndTime))
# we make sure that any token is accepted when the collection is
# empty, but that the returned token differs
query2 = webdavlib.WebDAVSyncQuery(resource, "1234", [ "getetag" ])
self.client.execute(query2)
self.assertEquals(query2.response["status"], 207)
token_node = query2.response["document"].find("{DAV:}sync-token")
self.assertTrue(token_node is not None,
"expected 'sync-token' tag")
token = int(token_node.text)
self.assertTrue(token > 0)
if __name__ == "__main__":
sogotests.runTests()

82
Tests/Integration/teststrings.m
View File

@@ -1,82 +0,0 @@
/* teststrings.m - this file is part of SOGO
*
* Copyright (C) 2010 Inverse inc.
*
* Author: Wolfgang Sourdeau <wsourdeau@inverse.ca>
*
* This file 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 2, or (at your option)
* any later version.
*
* This file 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; see the file COPYING. If not, write to
* the Free Software Foundation, Inc., 59 Temple Place - Suite 330,
* Boston, MA 02111-1307, USA.
*/
#include <Foundation/NSAutoreleasePool.h>
#include <Foundation/NSDictionary.h>
#include <Foundation/NSException.h>
#include <Foundation/NSString.h>
#include <SOGo/NSDictionary+Utilities.h>
static int
performTest (char *filename)
{
NSDictionary *testDict;
NSString *nsFilename;
int rc;
nsFilename = [NSString stringWithFormat: @"%s", filename];
NS_DURING
{
testDict = [NSDictionary dictionaryFromStringsFile: nsFilename];
if ([testDict count] == 0)
{
NSLog (@"Bad or empty strings file");
rc = 2;
testDict = nil;
}
else
rc = 0;
}
NS_HANDLER
{
NSLog (@"An exception was caught: %@", localException);
rc = 1;
testDict = nil;
}
NS_ENDHANDLER;
return rc;
}
int
main (int argc, char *argv[])
{
NSAutoreleasePool *pool;
int rc;
pool = [NSAutoreleasePool new];
if (argc == 2)
{
rc = performTest (argv[1]);
}
else
{
NSLog (@"Usage: %s file.strings", argv[0]);
rc = 1;
}
[pool release];
return rc;
}

24
Tests/Integration/teststrings.sh
View File

@@ -1,24 +0,0 @@
#!/bin/bash
TOPDIR=../..
RC=0
if [ ! -f teststrings ]
then
make
fi
for stringfile in ${TOPDIR}/*/*/*.lproj/Localizable.strings
do
./obj/teststrings "$stringfile" > /dev/null
code=$?
if test $code -eq 0;
then
echo "$stringfile: passed";
else
echo "$stringfile: FAILED (code: $code)";
RC=$(($RC+$code))
fi
done
exit $RC

199
Tests/Integration/utilities.py
View File

@@ -1,199 +0,0 @@
#!/usr/bin/python
import StringIO
import sys
import unittest
import vobject
import vobject.ics_diff
import webdavlib
import xml.sax.saxutils
class ics_compare():
def __init__(self, event1, event2):
self.event1 = event1
self.event2 = event2
self.diffs = None
def _vcalendarComponent(self, event):
event_component = None
for item in vobject.readComponents(event):
if item.name == "VCALENDAR":
event_component = item
return event_component
def areEqual(self):
s_event1 = StringIO.StringIO(self.event1)
s_event2 = StringIO.StringIO(self.event2)
event1_vcalendar = self._vcalendarComponent(s_event1)
if event1_vcalendar is None:
raise Exception("No VCALENDAR component in event1")
event2_vcalendar = self._vcalendarComponent(s_event2)
if event2_vcalendar is None:
raise Exception("No VCALENDAR component in event2")
self.diffs = vobject.ics_diff.diff(event1_vcalendar, event2_vcalendar)
if not self.diffs:
return True
else:
return False
def textDiff(self):
saved_stdout = sys.stdout
out = StringIO.StringIO()
sys.stdout = out
try :
if self.diffs is not None:
for (left, right) in self.diffs:
left.prettyPrint()
right.prettyPrint()
finally:
sys.stdout = saved_stdout
return out.getvalue().strip()
class TestUtility():
def __init__(self, test, client, resource = None):
self.test = test
self.client = client
self.userInfo = {}
def fetchUserInfo(self, login):
if not self.userInfo.has_key(login):
resource = "/SOGo/dav/%s/" % login
propfind = webdavlib.WebDAVPROPFIND(resource,
["displayname",
"{urn:ietf:params:xml:ns:caldav}calendar-user-address-set"],
0)
self.client.execute(propfind)
self.test.assertEquals(propfind.response["status"], 207)
common_tree = "{DAV:}response/{DAV:}propstat/{DAV:}prop"
name_nodes = propfind.response["document"] \
.findall('%s/{DAV:}displayname' % common_tree)
email_nodes = propfind.response["document"] \
.findall('%s/{urn:ietf:params:xml:ns:caldav}calendar-user-address-set/{DAV:}href'
% common_tree)
if len(name_nodes[0].text) > 0:
displayName = name_nodes[0].text
else:
displayName = ""
self.userInfo[login] = (displayName, email_nodes[0].text)
return self.userInfo[login]
class TestACLUtility(TestUtility):
def __init__(self, test, client, resource):
TestUtility.__init__(self, test, client, resource)
self.resource = resource
def _subscriptionOperation(self, subscribers, operation):
subscribeQuery = ("<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n"
+ "<%s" % operation
+ " xmlns=\"urn:inverse:params:xml:ns:inverse-dav\"")
if (subscribers is not None):
subscribeQuery = (subscribeQuery
+ " users=\"%s\"" % ",".join(subscribers))
subscribeQuery = subscribeQuery + "/>"
post = webdavlib.HTTPPOST(self.resource, subscribeQuery)
post.content_type = "application/xml; charset=\"utf-8\""
self.client.execute(post)
self.test.assertEquals(post.response["status"], 200,
"subscribtion failure to '%s' for '%s' (status: %d)"
% (self.resource, "', '".join(subscribers),
post.response["status"]))
def subscribe(self, subscribers=None):
self._subscriptionOperation(subscribers, "subscribe")
def unsubscribe(self, subscribers=None):
self._subscriptionOperation(subscribers, "unsubscribe")
def rightsToSOGoRights(self, rights):
self.fail("subclass must implement this method")
def setupRights(self, username, rights = None):
if rights is not None:
rights_str = "".join(["<%s/>"
% x for x in self.rightsToSOGoRights(rights) ])
aclQuery = ("<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n"
+ "<acl-query"
+ " xmlns=\"urn:inverse:params:xml:ns:inverse-dav\">"
+ "<set-roles user=\"%s\">%s</set-roles>" % (xml.sax.saxutils.escape(username),
rights_str)
+ "</acl-query>")
else:
aclQuery = ("<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n"
+ "<acl-query"
+ " xmlns=\"urn:inverse:params:xml:ns:inverse-dav\">"
+ "<remove-user user=\"%s\"/>" % xml.sax.saxutils.escape(username)
+ "</acl-query>")
post = webdavlib.HTTPPOST(self.resource, aclQuery)
post.content_type = "application/xml; charset=\"utf-8\""
self.client.execute(post)
if rights is None:
err_msg = ("rights modification: failure to remove entry (status: %d)"
% post.response["status"])
else:
err_msg = ("rights modification: failure to set '%s' (status: %d)"
% (rights_str, post.response["status"]))
self.test.assertEquals(post.response["status"], 204, err_msg)
# Calendar:
# rights:
# v: view all
# d: view date and time
# m: modify
# r: respond
# short rights notation: { "c": create,
# "d": delete,
# "pu": public,
# "pr": private,
# "co": confidential }
class TestCalendarACLUtility(TestACLUtility):
def rightsToSOGoRights(self, rights):
sogoRights = []
if rights.has_key("c") and rights["c"]:
sogoRights.append("ObjectCreator")
if rights.has_key("d") and rights["d"]:
sogoRights.append("ObjectEraser")
classes = { "pu": "Public",
"pr": "Private",
"co": "Confidential" }
rights_table = { "v": "Viewer",
"d": "DAndTViewer",
"m": "Modifier",
"r": "Responder" }
for k in classes.keys():
if rights.has_key(k):
right = rights[k]
sogo_right = "%s%s" % (classes[k], rights_table[right])
sogoRights.append(sogo_right)
return sogoRights
# Addressbook:
# short rights notation: { "c": create,
# "d": delete,
# "e": edit,
# "v": view }
class TestAddressBookACLUtility(TestACLUtility):
def rightsToSOGoRights(self, rights):
sogoRightsTable = { "c": "ObjectCreator",
"d": "ObjectEraser",
"v": "ObjectViewer",
"e": "ObjectEditor" }
sogoRights = []
for k in rights.keys():
sogoRights.append(sogoRightsTable[k])
return sogoRights

580
Tests/Integration/webdavlib.py
View File

@@ -1,580 +0,0 @@
# webdavlib.py - A versatile WebDAV Python Library
#
# Copyright (C) 2009, 2010 Inverse inc.
#
# Author: Wolfgang Sourdeau <wsourdeau@inverse.ca>
#
# webdavlib is free software; you can redistribute it and/or modify it under
# the terms of the GNU Lesser General Public License as published by the Free
# Software Foundation; either version 2, or (at your option) any later
# version.
#
# webdavlib 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 Lesser General Public License for more
# details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with webdavlib; see the file COPYING. If not, write to the Free
# Software Foundation, 59 Temple Place - Suite 330, Boston, MA 02111-1307,
# USA.
import httplib
import re
import time
import xml.dom.expatbuilder
import xml.etree.cElementTree
import xml.sax.saxutils
import sys
xmlns_dav = "DAV:"
xmlns_caldav = "urn:ietf:params:xml:ns:caldav"
xmlns_carddav = "urn:ietf:params:xml:ns:carddav"
xmlns_inversedav = "urn:inverse:params:xml:ns:inverse-dav"
url_re = None
class HTTPUnparsedURL:
def __init__(self, url):
self._parse(url)
def _parse(self, url):
# ((proto)://((username(:(password)?)@)?hostname(:(port))))(path)?
# if url_re is None:
url_parts = url.split("?")
alpha_match = "[a-zA-Z0-9%\._-]+"
num_match = "[0-9]+"
pattern = ("((%s)://(((%s)(:(%s))?@)?(%s)(:(%s))?))?(/.*)"
% (alpha_match, alpha_match, alpha_match,
alpha_match, num_match))
url_re = re.compile(pattern)
re_match = url_re.match(url_parts[0])
if re_match is None:
raise Exception, "URL expression could not be parsed: %s" % url
(trash, self.protocol, trash, trash, self.username, trash,
self.password, self.hostname, trash, self.port, self.path) = re_match.groups()
self.parameters = {}
if len(url_parts) > 1:
param_elms = url_parts[1].split("&")
for param_pair in param_elms:
parameter = param_pair.split("=")
self.parameters[parameter[0]] = parameter[1]
class WebDAVClient:
user_agent = "Mozilla/5.0"
def __init__(self, hostname, port, username = None, password = "",
forcessl = False):
if int(port) == 443 or forcessl:
import M2Crypto.httpslib
self.conn = M2Crypto.httpslib.HTTPSConnection(hostname, int(port),
True)
else:
self.conn = httplib.HTTPConnection(hostname, port, True)
if username is None:
self.simpleauth_hash = None
else:
self.simpleauth_hash = (("%s:%s" % (username, password))
.encode('base64')[:-1])
def prepare_headers(self, query, body):
headers = { "User-Agent": self.user_agent }
if self.simpleauth_hash is not None:
headers["authorization"] = "Basic %s" % self.simpleauth_hash
if body is not None:
headers["content-length"] = len(body)
if query.__dict__.has_key("depth") and query.depth is not None:
headers["depth"] = query.depth
if query.__dict__.has_key("content_type"):
headers["content-type"] = query.content_type
if not query.__dict__.has_key("accept-language"):
headers["accept-language"] = 'en-us,en;q=0.5'
query_headers = query.prepare_headers()
if query_headers is not None:
for key in query_headers.keys():
headers[key] = query_headers[key]
return headers
def execute(self, query):
body = query.render()
query.start = time.time()
self.conn.request(query.method, query.url,
body, self.prepare_headers(query, body))
try:
query.set_response(self.conn.getresponse())
except httplib.BadStatusLine, e:
print e
query.set_response(self.conn.getresponse())
query.duration = time.time() - query.start
class HTTPSimpleQuery:
method = None
def __init__(self, url):
self.url = url
self.response = None
self.start = -1
self.duration = -1
def prepare_headers(self):
return {}
def render(self):
return None
def set_response(self, http_response):
headers = {}
for rk, rv in http_response.getheaders():
k = rk.lower()
headers[k] = rv
self.response = { "headers": headers,
"status": http_response.status,
"version": http_response.version,
"body": http_response.read() }
class HTTPGET(HTTPSimpleQuery):
method = "GET"
cookie = None
def prepare_headers (self):
headers = HTTPSimpleQuery.prepare_headers(self)
if self.cookie:
headers["Cookie"] = self.cookie
return headers
class HTTPOPTIONS(HTTPSimpleQuery):
method = "OPTIONS"
class HTTPQuery(HTTPSimpleQuery):
def __init__(self, url):
HTTPSimpleQuery.__init__(self, url)
self.content_type = "application/octet-stream"
class HTTPPUT(HTTPQuery):
method = "PUT"
def __init__(self, url, content,
content_type="application/octet-stream",
exclusive=False):
HTTPQuery.__init__(self, url)
self.content = content
self.content_type = content_type
self.exclusive = exclusive
def render(self):
return self.content
def prepare_headers(self):
headers = HTTPQuery.prepare_headers(self)
if self.exclusive:
headers["if-none-match"] = "*"
return headers
class HTTPPOST(HTTPPUT):
method = "POST"
cookie = None
def prepare_headers (self):
headers = HTTPPUT.prepare_headers(self)
if self.cookie:
headers["Cookie"] = self.cookie
return headers
class WebDAVQuery(HTTPQuery):
method = None
def __init__(self, url, depth = None):
HTTPQuery.__init__(self, url)
self.content_type = "application/xml; charset=\"utf-8\""
self.depth = depth
self.ns_mgr = _WD_XMLNS_MGR()
self.top_node = None
# helper for PROPFIND and REPORT (only)
def _initProperties(self, properties):
props = _WD_XMLTreeElement("prop")
self.top_node.append(props)
for prop in properties:
prop_tag = self.render_tag(prop)
props.append(_WD_XMLTreeElement(prop_tag))
def render(self):
if self.top_node is not None:
text = ("<?xml version=\"1.0\" encoding=\"utf-8\"?>\n%s"
% self.top_node.render(self.ns_mgr.render()))
else:
text = ""
return text
def render_tag(self, tag):
cb = tag.find("}")
if cb > -1:
ns = tag[1:cb]
real_tag = tag[cb+1:]
new_tag = self.ns_mgr.register(real_tag, ns)
else:
new_tag = tag
return new_tag
def set_response(self, http_response):
HTTPQuery.set_response(self, http_response)
headers = self.response["headers"]
if (headers.has_key("content-type")
and headers.has_key("content-length")
and (headers["content-type"].startswith("application/xml")
or headers["content-type"].startswith("text/xml"))
and int(headers["content-length"]) > 0):
document = xml.etree.cElementTree.fromstring(self.response["body"])
self.response["document"] = document
class WebDAVMKCOL(WebDAVQuery):
method = "MKCOL"
class WebDAVDELETE(WebDAVQuery):
method = "DELETE"
class WebDAVREPORT(WebDAVQuery):
method = "REPORT"
class WebDAVGET(WebDAVQuery):
method = "GET"
class WebDAVPROPFIND(WebDAVQuery):
method = "PROPFIND"
def __init__(self, url, properties, depth = None):
WebDAVQuery.__init__(self, url, depth)
self.top_node = _WD_XMLTreeElement("propfind")
if properties is not None and len(properties) > 0:
self._initProperties(properties)
class WebDAVPROPPATCH(WebDAVQuery):
method = "PROPPATCH"
# <x0:propertyupdate xmlns:x1="urn:ietf:params:xml:ns:caldav" xmlns:x0="DAV:"><x0:set><x0:prop>
def __init__(self, url, properties):
WebDAVQuery.__init__(self, url, None)
self.top_node = _WD_XMLTreeElement("propertyupdate")
set_node = _WD_XMLTreeElement("set")
self.top_node.append(set_node)
prop_node = _WD_XMLTreeElement("prop")
set_node.append(prop_node)
prop_node.appendSubtree(self, properties)
class WebDAVMOVE(WebDAVQuery):
method = "MOVE"
destination = None
host = None
def prepare_headers(self):
headers = WebDAVQuery.prepare_headers(self)
print "DESTINATION", self.destination
if self.destination is not None:
headers["Destination"] = self.destination
if self.host is not None:
headers["Host"] = self.host
return headers
class WebDAVPrincipalPropertySearch(WebDAVREPORT):
def __init__(self, url, properties, matches):
WebDAVQuery.__init__(self, url)
ppsearch_tag = self.ns_mgr.register("principal-property-search",
xmlns_dav)
self.top_node = _WD_XMLTreeElement(ppsearch_tag)
self._initMatches(matches)
if properties is not None and len(properties) > 0:
self._initProperties(properties)
def _initMatches(self, matches):
for match in matches:
psearch = _WD_XMLTreeElement("property-search")
self.top_node.append(psearch)
prop = _WD_XMLTreeElement("prop")
psearch.append(prop)
match_tag = self.render_tag(match[0])
prop.append(_WD_XMLTreeElement(match_tag))
match_tag = _WD_XMLTreeElement("match")
psearch.append(match_tag)
match_tag.appendSubtree(self, match[1])
class WebDAVSyncQuery(WebDAVREPORT):
def __init__(self, url, token, properties):
WebDAVQuery.__init__(self, url)
self.top_node = _WD_XMLTreeElement("sync-collection")
sync_token = _WD_XMLTreeElement("sync-token")
self.top_node.append(sync_token)
if token is not None:
sync_token.append(_WD_XMLTreeTextNode(token))
if properties is not None and len(properties) > 0:
self._initProperties(properties)
class WebDAVExpandProperty(WebDAVREPORT):
def _parseTag(self, tag):
result = []
cb = tag.find("}")
if cb > -1:
result.append(tag[cb+1:])
result.append(tag[1:cb])
else:
result.append(tag)
result.append("DAV:")
return result;
def _propElement(self, tag):
parsedTag = self._parseTag(tag)
parameters = { "name": parsedTag[0] }
if len(parsedTag) > 1:
parameters["namespace"] = parsedTag[1]
return _WD_XMLTreeElement("property", parameters)
def __init__(self, url, query_properties, properties):
WebDAVQuery.__init__(self, url)
self.top_node = _WD_XMLTreeElement("expand-property")
for query_tag in query_properties:
property_query = self._propElement(query_tag)
self.top_node.append(property_query)
for tag in properties:
property = self._propElement(tag)
property_query.append(property)
class CalDAVPOST(WebDAVQuery):
method = "POST"
def __init__(self, url, content,
originator = None, recipients = None):
WebDAVQuery.__init__(self, url)
self.content_type = "text/calendar; charset=utf-8"
self.originator = originator
self.recipients = recipients
self.content = content
def prepare_headers(self):
headers = WebDAVQuery.prepare_headers(self)
if self.originator is not None:
headers["originator"] = self.originator
if self.recipients is not None:
headers["recipient"] = ",".join(self.recipients)
return headers
def render(self):
return self.content
class CalDAVCalendarMultiget(WebDAVREPORT):
def __init__(self, url, properties, hrefs, depth = None):
WebDAVQuery.__init__(self, url, depth)
multiget_tag = self.ns_mgr.register("calendar-multiget", xmlns_caldav)
self.top_node = _WD_XMLTreeElement(multiget_tag)
if properties is not None and len(properties) > 0:
self._initProperties(properties)
for href in hrefs:
href_node = _WD_XMLTreeElement("href")
self.top_node.append(href_node)
href_node.append(_WD_XMLTreeTextNode(href))
class CalDAVCalendarQuery(WebDAVREPORT):
def __init__(self, url, properties, component = None, timerange = None):
WebDAVQuery.__init__(self, url)
multiget_tag = self.ns_mgr.register("calendar-query", xmlns_caldav)
self.top_node = _WD_XMLTreeElement(multiget_tag)
if properties is not None and len(properties) > 0:
self._initProperties(properties)
if component is not None:
filter_tag = self.ns_mgr.register("filter",
xmlns_caldav)
compfilter_tag = self.ns_mgr.register("comp-filter",
xmlns_caldav)
filter_node = _WD_XMLTreeElement(filter_tag)
cal_filter_node = _WD_XMLTreeElement(compfilter_tag,
{ "name": "VCALENDAR" })
comp_node = _WD_XMLTreeElement(compfilter_tag,
{ "name": component })
## TODO
# if timerange is not None:
cal_filter_node.append(comp_node)
filter_node.append(cal_filter_node)
self.top_node.append(filter_node)
class CardDAVAddressBookQuery(WebDAVREPORT):
def __init__(self, url, properties, searchProperty = None, searchValue = None):
WebDAVQuery.__init__(self, url)
query_tag = self.ns_mgr.register("addressbook-query", xmlns_carddav)
ns_key = self.ns_mgr.xmlns[xmlns_carddav]
self.top_node = _WD_XMLTreeElement(query_tag)
if properties is not None and len(properties) > 0:
self._initProperties(properties)
if searchProperty is not None:
filter_node = _WD_XMLTreeElement("%s:filter" % ns_key)
self.top_node.append(filter_node)
propfilter_node = _WD_XMLTreeElement("%s:prop-filter" % ns_key, { "name": searchProperty })
filter_node.append(propfilter_node)
match_node = _WD_XMLTreeElement("%s:text-match" % ns_key,
{ "collation": "i;unicasemap", "match-type": "starts-with" })
propfilter_node.append(match_node)
match_node.appendSubtree(None, searchValue)
class MailDAVMailQuery(WebDAVREPORT):
def __init__(self, url, properties, filters = None,
sort = None, ascending = True):
WebDAVQuery.__init__(self, url)
mailquery_tag = self.ns_mgr.register("mail-query",
xmlns_inversedav)
self.top_node = _WD_XMLTreeElement(mailquery_tag)
if properties is not None and len(properties) > 0:
self._initProperties(properties)
if filters is not None and len(filters) > 0:
self._initFilters(filters)
if sort is not None and len(sort) > 0:
self._initSort(sort, ascending)
def _initFilters(self, filters):
mailfilter_tag = self.ns_mgr.register("mail-filters",
xmlns_inversedav)
mailfilter_node = _WD_XMLTreeElement(mailfilter_tag)
self.top_node.append(mailfilter_node)
for filterk in filters.keys():
filter_tag = self.ns_mgr.register(filterk,
xmlns_inversedav)
filter_node = _WD_XMLTreeElement(filter_tag,
filters[filterk])
mailfilter_node.append(filter_node)
def _initSort(self, sort, ascending):
sort_tag = self.ns_mgr.register("sort", xmlns_inversedav)
if ascending:
sort_attrs = { "order": "ascending" }
else:
sort_attrs = { "order": "descending" }
sort_node = _WD_XMLTreeElement(sort_tag, sort_attrs)
self.top_node.append(sort_node)
for item in sort:
sort_subnode = _WD_XMLTreeElement(self.render_tag(item))
sort_node.append(sort_subnode)
# private classes to handle XML stuff
class _WD_XMLNS_MGR:
def __init__(self):
self.xmlns = {}
self.counter = 0
def render(self):
text = " xmlns=\"DAV:\""
for k in self.xmlns:
text = text + " xmlns:%s=\"%s\"" % (self.xmlns[k], k)
return text
def create_key(self, namespace):
new_nssym = "n%d" % self.counter
self.counter = self.counter + 1
self.xmlns[namespace] = new_nssym
return new_nssym
def register(self, tag, namespace):
if namespace != xmlns_dav:
if self.xmlns.has_key(namespace):
key = self.xmlns[namespace]
else:
key = self.create_key(namespace)
else:
key = None
if key is not None:
newTag = "%s:%s" % (key, tag)
else:
newTag = tag
return newTag
class _WD_XMLTreeElement:
typeNum = type(0)
typeStr = type("")
typeUnicode = type(u"")
typeList = type([])
typeDict = type({})
def __init__(self, tag, attributes = {}):
self.tag = tag
self.children = []
self.attributes = attributes
def append(self, child):
self.children.append(child)
def appendSubtree(self, query, subtree):
if type(subtree) == self.typeNum:
strValue = "%d" % subtree
textNode = _WD_XMLTreeTextNode(strValue)
self.append(textNode)
elif type(subtree) == self.typeUnicode:
textNode = _WD_XMLTreeTextNode(subtree.encode("utf-8"))
self.append(textNode)
elif type(subtree) == self.typeStr:
textNode = _WD_XMLTreeTextNode(subtree)
self.append(textNode)
elif type(subtree) == self.typeList:
for x in subtree:
self.appendSubtree(query, x)
elif type(subtree) == self.typeDict:
for x in subtree.keys():
tag = query.render_tag(x)
node = _WD_XMLTreeElement(tag)
node.appendSubtree(query, subtree[x])
self.append(node)
else:
None
def render(self, ns_text = None):
text = "<" + self.tag
if ns_text is not None:
text = text + ns_text
for k in self.attributes:
text = text + " %s=\"%s\"" % (k, self.attributes[k])
if len(self.children) > 0:
text = text + ">"
for child in self.children:
text = text + child.render()
text = text + "</" + self.tag + ">"
else:
text = text + "/>"
return text
class _WD_XMLTreeTextNode:
def __init__(self, text):
self.text = xml.sax.saxutils.escape(text)
def render(self):
return self.text

106
Tests/Integration/webdavsync-tool.py
View File

@@ -1,106 +0,0 @@
#!/usr/bin/python
import getopt
import sys
import urlparse
import webdavlib
import xml.dom.minidom
def usage() :
msg ="""Usage:
%s [-h] [-s sync-token] -u uri\n""" % sys.argv[0]
sys.stderr.write(msg);
def getAllCollections(client, uri):
collections = []
depth = 1
propfind = webdavlib.WebDAVPROPFIND(uri, ["allprop"], depth)
client.execute(propfind)
client.conn.close()
doc = propfind.response["document"]
for response in doc.iter("{DAV:}response"):
propstat = response.find("{DAV:}propstat")
if propstat is not None:
prop = propstat.find("{DAV:}prop")
if prop is not None:
resourcetype = prop.find("{DAV:}resourcetype")
if resourcetype.find("{DAV:}collection") is not None:
href = prop.find("{DAV:}href")
if href is not None and href.text != uri:
collections.append(href.text)
return collections
def changedItemsFromCollection(client, collection, synctoken=None):
# get all changed hrefs since synctoken
hrefs = []
syncquery = webdavlib.WebDAVSyncQuery(collection, synctoken, [ "getcontenttype", "getetag" ])
client.execute(syncquery)
client.conn.close()
if (syncquery.response["status"] != 207):
raise Exception("Bad http response code: %d" % syncquery.response["status"])
doc = syncquery.response["document"]
# extract all hrefs
for syncResponse in doc.iter("{DAV:}response"):
href = syncResponse.find("{DAV:}href")
if href is not None:
hrefs.append(href.text)
return hrefs
def main():
depth = 1
synctoken = "1"
url = None
try:
opts, args = getopt.getopt (sys.argv[1:], "hs:u:", \
("sync-token=", "url="));
except getopt.GetoptError:
usage()
exit(1)
for o, v in opts :
if o == "-h" :
usage()
exit(1)
elif o == "-s" or o == "--sync-token" :
synctoken = v
elif o == "-u" or o == "--url" :
url = v
if url is None:
usage()
exit(1)
o = urlparse.urlparse(url)
hostname = o.hostname
port = o.port
username = o.username
password = o.password
uri = o.path
client = webdavlib.WebDAVClient(hostname, port, username, password)
collections = getAllCollections(client, uri)
if len(collections) == 0:
print "No collections found!"
sys.exit(1)
for collection in collections:
changedItems = changedItemsFromCollection(client, collection)
# fetch the href data
if len(changedItems) > 0:
multiget = webdavlib.CalDAVCalendarMultiget(collection,
["getetag", "{%s}calendar-data" % webdavlib.xmlns_caldav],
changedItems, depth)
client.execute(multiget)
client.conn.close()
if (multiget.response["status"] != 207):
raise Exception("Bad http response code: %d" % multiget.response["status"])
if __name__ == "__main__":
main()

8
Tests/README
View File

@@ -1,8 +0,0 @@
This directory holds automated tests for SOGo.
We currrently have:
- Intregation holds all interated tests that are used to
validate overall DAV functionality right now
- Unit holds all unit tests

14
Tests/README.md Normal file
View File

@@ -0,0 +1,14 @@
# Tests
This directory holds automated tests for SOGo.
- `spec` and `lib`: hold JavaScript driven interated tests that are used to validate overall DAV functionality
- `Unit`: holds all unit tests
## Tools
* [Jasmin](https://jasmine.github.io/) - testing framework
* [tsdav](https://tsdav.vercel.app/) - webdav request helper
* [ical.js](https://github.com/mozilla-comm/ical.js) - ics and vcard parser
* [cross-fetch](https://github.com/lquixada/cross-fetch) - fetch API
* [xml-js](https://github.com/nashwaan/xml-js) - convert JS object to XML

3
packaging/debian/rules
View File

@@ -74,9 +74,6 @@ clean:
dh_testdir
dh_testroot
rm -f build-arch-stamp
( cd Tests/Integration; make clean )
rm -f Tests/Integration/config.py
-find Tests -name "*.pyc" -exec rm -f {} \;
if [ -f config.make ]; \
then \
$(MAKE) clean; \