See ChangeLog

Monotone-Parent: f1c001048ccc36181a09a459327f563dbc26e12f
Monotone-Revision: aab1e060a1d26a9624f7f84c30d1414ca1969560

Monotone-Author: ludovic@Sophos.ca
Monotone-Date: 2010-01-24T20:40:12
Monotone-Branch: ca.inverse.sogo

Tests/Integration/README

@@ -0,0 +1,56 @@
+setup
+-----
+
+(you need "python-simplejson", "python-xml", "python-vobject" 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 monotone)
+2) edit config.py to suit your environment
+3) 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 contrary to unit tests, functional tests
+  often imply a logical order between the different steps.

Tests/Integration/all.py

@@ -0,0 +1,59 @@
+#!/usr/bin/python
+
+import os, sys, unittest, getopt, traceback
+import preferences
+
+if __name__ == "__main__":
+    loader = unittest.TestLoader()
+    modules = []
+    # Duplicated from UIxPreferences.m
+    languages = ["Czech", "Dutch", "English", "French", 
+                 "German", "Hungarian", "Italian", "BrazilianPortuguese", 
+                 "Russian", "Spanish", "Swedish", "Welsh"]
+
+    # We can disable testing all languages
+    testLanguages = True
+    opts, args = getopt.getopt (sys.argv[1:], [], ["disable-languages"])
+    for o, a in opts:
+        if o == "--disable-languages":
+            testLanguages = False
+
+
+    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()
+
+        if testLanguages:
+            prefs = preferences.preferences()
+            # Get the current language
+            userLanguageString = prefs.get ("Language")
+            if userLanguageString:
+                userLanguage = languages.index (userLanguageString)
+            else:
+                userLanguage = 2
+
+            for i in range (0, len (languages)):
+                try:
+                    prefs.set ("language", 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.run(suite)
+            # Revert to the original language
+            prefs.set ("language", userLanguage)
+        else:
+            runner.run(suite)
+
+    else:
+        print "No test available."

Tests/Integration/config.py.in

@@ -0,0 +1,15 @@
+# setup: username must be super-user or have read-access to PUBLIC events in
+#        both attendee and delegate's personal calendar
+
+hostname = "localhost"
+port = "80"
+username = "myuser"
+password = "mypass"
+
+subscriber_username = "otheruser"
+subscriber_password = "otherpass"
+
+attendee1 = "user@domain.com"
+attendee1_delegate = "otheruser@domain.com"
+
+mailserver = "imaphost"

Tests/Integration/preferences.py

@@ -0,0 +1,77 @@
+from config import hostname, port, username, password
+import webdavlib
+import urllib
+import base64
+import simplejson
+
+
+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,
+                                         self.login, self.passw)
+
+    authString = \
+        base64.urlsafe_b64encode (username+":"+password).replace("=", "%3D")
+    self.cookie = \
+        "0xHIGHFLYxSOGo=basic%%20%s;%%20SOGo=6C0F6C0F014AC60ED5" % authString
+
+    self.preferencesMap = {"language": "2.1.0.3.0.1.3.1.1.3.1.2"}
+
+  def set(self, preference, value):
+    formKey = self.preferencesMap[preference]
+    content = "%s=%s" % (formKey, value)
+    url = "/SOGo/so/%s/preferences" % self.login
+
+    post = HTTPPreferencesPOST (url, content)
+    post.content_type = "application/x-www-form-urlencoded"
+    post.cookie = self.cookie
+
+    self.client.execute (post)
+
+    # Raise an exception if the language wasn't properly set
+    if post.response["status"] != 200:
+      raise Exception ("failure setting language, (code = %d)" \
+                       % post.response["status"])
+
+  def get(self, preference):
+    url = "/SOGo/so/%s/preferences/jsonDefaults" % self.login
+    get = HTTPPreferencesGET (url)
+    get.cookie = self.cookie
+    self.client.execute (get)
+    content = simplejson.loads(get.response['body'])
+    result = None
+    try:
+      result = content[preference]
+    except:
+      pass
+    return result
+
+# Simple main to test this class
+if __name__ == "__main__":
+  p = preferences ()
+  p.set ("language", 6)

Tests/Integration/propfind.py

@@ -0,0 +1,45 @@
+#!/usr/bin/python
+
+from config import hostname, port, username, password
+
+import webdavlib
+
+import sys
+import getopt
+import xml.dom.ext
+
+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)
+
+print "response:\n\n%s" % propfind.response["body"]
+
+if propfind.response.has_key("document"):
+    print "document tree:"
+    xml.dom.ext.PrettyPrint(propfind.response["document"])

Tests/Integration/test-caldav-scheduling.py

@@ -0,0 +1,416 @@
+#!/usr/bin/python
+
+# setup: username must be super-user or have read-access to PUBLIC events in
+#        both attendee and delegate's personal calendar
+
+from config import hostname, port, username, password, attendee1, attendee1_delegate
+
+import datetime
+import sys
+import time
+import unittest
+import vobject
+import vobject.base
+import vobject.icalendar
+import webdavlib
+import StringIO
+
+def fetchUserInfo(login):
+    client = webdavlib.WebDAVClient(hostname, port, username, password)
+    resource = "/SOGo/dav/%s/" % login
+    propfind = webdavlib.WebDAVPROPFIND(resource,
+                                        ["displayname",
+                                         "{urn:ietf:params:xml:ns:caldav}calendar-user-address-set"],
+                                        0)
+    propfind.xpath_namespace = { "D": "DAV:",
+                                 "C": "urn:ietf:params:xml:ns:caldav" }
+    client.execute(propfind)
+    assert(propfind.response["status"] == 207)
+    name_nodes = propfind.xpath_evaluate('/D:multistatus/D:response/D:propstat/D:prop/D:displayname',
+                                          None)
+    email_nodes = propfind.xpath_evaluate('/D:multistatus/D:response/D:propstat/D:prop/C:calendar-user-address-set/D:href',
+                                          None)
+
+    return (name_nodes[0].childNodes[0].nodeValue, email_nodes[0].childNodes[0].nodeValue)
+
+class CalDAVITIPDelegationTest(unittest.TestCase):
+    def setUp(self):
+        self.client = webdavlib.WebDAVClient(hostname, port,
+                                             username, password)
+        (self.user_name, self.user_email) = fetchUserInfo(username)
+        (self.attendee1_name, self.attendee1_email) = fetchUserInfo(attendee1)
+        (self.attendee1_delegate_name, self.attendee1_delegate_email) = 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
+
+    def _newEvent(self):
+        newCal = vobject.iCalendar()
+        vevent = newCal.add('vevent')
+        vevent.add('summary').value = "test event"
+        vevent.add('transp').value = "OPAQUE"
+
+        now = datetime.datetime.now()
+        startdate = vevent.add('dtstart')
+        startdate.value = now
+        enddate = vevent.add('dtend')
+        enddate.value = now + datetime.timedelta(0, 3600)
+        vevent.add('uid').value = "test-delegation"
+        vevent.add('dtstamp').value = now
+        vevent.add('last-modified').value = now
+        vevent.add('created').value = now
+        
+        vevent.add('sequence').value = "0"
+
+        return newCal
+
+    def tearDown(self):
+        self._deleteEvent(self.client,
+                          "%stest-delegation.ics" % self.user_calendar, None)
+        self._deleteEvent(self.client,
+                          "%stest-delegation.ics" % self.attendee1_calendar, None)
+        self._deleteEvent(self.client,
+                          "%stest-delegation.ics" % self.attendee1_delegate_calendar,
+                          None)
+
+    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.assertEquals(put.response["status"], exp_status)
+
+    def _postEvent(self, client, outbox, event, originator, recipients,
+                   exp_status = 200):
+        post = webdavlib.CalDAVPOST(outbox, event.serialize(),
+                                    originator, recipients)
+        client.execute(post)
+        if exp_status is not None:
+            self.assertEquals(post.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.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.assertEquals(delete.response["status"], exp_status)
+
+    def _eventAttendees(self, event):
+        attendees = {}
+
+        event_component = event.vevent
+        for child in event_component.getChildren():
+            if child.name == "ATTENDEE":
+                try:
+                    delegated_to = child.delegated_to_param
+                except:
+                    delegated_to = "(none)"
+                try:
+                    delegated_from = child.delegated_from_param
+                except:
+                    delegated_from = "(none)"
+                attendees[child.value] = ("%s/%s/%s"
+                                          % (child.partstat_param,
+                                             delegated_to,
+                                             delegated_from))
+
+        return attendees
+
+    def _compareAttendees(self, compared_event, event):
+        compared_attendees = self._eventAttendees(compared_event)
+        compared_emails = compared_attendees.keys()
+        self.assertTrue(len(compared_emails) > 0,
+                        "no attendee found")
+        compared_emails.sort()
+
+        attendees = self._eventAttendees(event)
+        emails = attendees.keys()
+        emails.sort()
+
+        self.assertEquals(len(compared_emails), len(emails),
+                          "number of attendees is not equal"
+                          + " (actual: %d, exp: %d)"
+                          % (len(compared_emails), len(emails)))
+
+        for email in emails:
+            self.assertEquals(compared_attendees[email],
+                              attendees[email],
+                              "partstat for attendee '%s' does not match"
+                              " (actual: '%s', expected: '%s')"
+                              % (email,
+                                 compared_attendees[email], attendees[email]))
+
+    def testInvitationDelegation(self):
+        """ invitation delegation """
+
+        # the invitation must not exist
+        self._deleteEvent(self.client,
+                          "%stest-delegation.ics" % self.user_calendar, None)
+        self._deleteEvent(self.client,
+                          "%stest-delegation.ics" % self.attendee1_calendar, None)
+        self._deleteEvent(self.client,
+                          "%stest-delegation.ics" % self.attendee1_delegate_calendar,
+                          None)
+
+        # 1. org -> attendee => org: 1, attendee: 1 (pst=N-A), delegate: 0
+
+        invitation = self._newEvent()
+        invitation.add("method").value = "REQUEST"
+        organizer = invitation.vevent.add('organizer')
+        organizer.cn_param = self.user_name
+        organizer.value = self.user_email
+        attendee = invitation.vevent.add('attendee')
+        attendee.cn_param = self.attendee1_name
+        attendee.rsvp_param = "TRUE"
+        attendee.partstat_param = "NEEDS-ACTION"
+        attendee.value = self.attendee1_email
+
+        self._postEvent(self.client, self.user_calendar, invitation,
+                        self.user_email, [self.attendee1_email])
+        del invitation.method
+        self._putEvent(self.client,
+                       "%stest-delegation.ics" % self.user_calendar,
+                       invitation)
+
+        att_inv = self._getEvent(self.client,
+                                 "%stest-delegation.ics"
+                                 % self.attendee1_calendar)
+        self._compareAttendees(att_inv, invitation)
+
+        # 2. attendee delegates to delegate
+        #    => org: 1 (updated), attendee: 1 (updated,pst=D),
+        #       delegate: 1 (new,pst=N-A)
+
+        invitation.add("method").value = "REQUEST"
+        attendee1 = invitation.vevent.attendee
+        attendee1.partstat_param = "DELEGATED"
+        attendee1.delegated_to_param = self.attendee1_delegate_email
+        delegate = invitation.vevent.add('attendee')
+        delegate.delegated_from_param = self.attendee1_email
+        delegate.cn_param = self.attendee1_delegate_name
+        delegate.rsvp_param = "TRUE"
+        delegate.partstat_param = "NEEDS-ACTION"
+        delegate.value = self.attendee1_delegate_email
+
+        self._postEvent(self.client,
+                        self.attendee1_calendar, invitation,
+                        self.attendee1_email, [self.attendee1_delegate_email])
+        invitation.method.value = "REPLY"
+        self._postEvent(self.client,
+                        self.attendee1_calendar, invitation,
+                        self.attendee1_email, [self.user_email])
+        del invitation.method
+        self._putEvent(self.client,
+                       "%stest-delegation.ics" % self.attendee1_calendar,
+                       invitation, 204)
+
+        del_inv = self._getEvent(self.client,
+                                 "%stest-delegation.ics"
+                                 % self.attendee1_delegate_calendar)
+        self._compareAttendees(del_inv, invitation)
+        org_inv = self._getEvent(self.client,
+                                 "%stest-delegation.ics" % self.user_calendar)
+        self._compareAttendees(org_inv, invitation)
+
+        # 3. delegate accepts
+        #    => org: 1 (updated), attendee: 1 (updated,pst=D),
+        #       delegate: 1 (accepted,pst=A)
+
+        invitation.add("method").value = "REPLY"
+        delegate.partstat_param = "ACCEPTED"
+        self._postEvent(self.client,
+                        self.attendee1_delegate_calendar, invitation,
+                        self.attendee1_delegate_email, [self.user_email, self.attendee1_email])
+        del invitation.method
+        self._putEvent(self.client,
+                       "%stest-delegation.ics" % self.attendee1_delegate_calendar,
+                       invitation, 204)
+
+        org_inv = self._getEvent(self.client,
+                                 "%stest-delegation.ics" % self.user_calendar)
+        self._compareAttendees(org_inv, invitation)
+        att_inv = self._getEvent(self.client,
+                                 "%stest-delegation.ics" % self.attendee1_calendar)
+        self._compareAttendees(att_inv, invitation)
+
+        # 4. attendee accepts
+        # => org: 1 (updated), attendee: 1 (updated,pst=A),
+        #    delegate: 0 (cancelled, deleted)
+
+        cancellation = vobject.iCalendar()
+        cancellation.copy(invitation)
+        cancellation.add("method").value = "CANCEL"
+        cancellation.vevent.sequence.value = "1"
+        self._postEvent(self.client,
+                        self.attendee1_calendar, cancellation,
+                        self.attendee1_email, [self.attendee1_delegate_email])
+
+        attendee1 = invitation.vevent.attendee
+        attendee1.partstat_param = "ACCEPTED"
+        del attendee1.delegated_to_param
+        invitation.add("method").value = "REPLY"
+        invitation.vevent.remove(delegate)
+        self._postEvent(self.client,
+                        self.attendee1_calendar, invitation,
+                        self.attendee1_email, [self.user_email])
+
+        del invitation.method
+        self._putEvent(self.client,
+                       "%stest-delegation.ics" % self.attendee1_calendar,
+                       invitation, 204)
+
+        org_inv = self._getEvent(self.client,
+                                 "%stest-delegation.ics" % self.user_calendar)
+        self._compareAttendees(org_inv, invitation)
+
+        del_inv = self._getEvent(self.client,
+                                 "%stest-delegation.ics" % self.attendee1_delegate_calendar, 404)
+
+        # 5. org updates inv.
+        #    => org: 1 (updated), attendee: 1 (updated), delegate: 0
+
+        invitation.add("method").value = "REQUEST"
+        invitation.vevent.summary.value = "Updated invitation"
+        invitation.vevent.sequence.value = "1"
+        attendee.partstat_param = "NEEDS-ACTION"
+        now = datetime.datetime.now()
+        invitation.vevent.last_modified.value = now
+        invitation.vevent.dtstamp.value = now
+
+        self._postEvent(self.client, self.user_calendar, invitation,
+                        self.user_email, [self.attendee1_email])
+
+        del invitation.method
+        self._putEvent(self.client,
+                       "%stest-delegation.ics" % self.user_calendar,
+                       invitation, 204)
+
+        att_inv = self._getEvent(self.client,
+                                 "%stest-delegation.ics" % self.attendee1_calendar)
+        self._compareAttendees(att_inv, invitation)
+
+        # 6. attendee delegates to delegate
+        #    => org: 1 (updated), attendee: 1 (updated), delegate: 1 (new)
+
+        invitation.add("method").value = "REQUEST"
+        attendee1.partstat_param = "DELEGATED"
+        attendee1.delegated_to_param = self.attendee1_delegate_email
+
+        delegate = invitation.vevent.add('attendee')
+        delegate.delegated_from_param = self.attendee1_email
+        delegate.cn_param = self.attendee1_delegate_name
+        delegate.rsvp_param = "TRUE"
+        delegate.partstat_param = "NEEDS-ACTION"
+        delegate.value = self.attendee1_delegate_email
+
+        self._postEvent(self.client,
+                        self.attendee1_calendar, invitation,
+                        self.attendee1_email, [self.attendee1_delegate_email])
+        invitation.method.value = "REPLY"
+        self._postEvent(self.client,
+                        self.attendee1_calendar, invitation,
+                        self.attendee1_email, [self.user_email])
+        del invitation.method
+        self._putEvent(self.client,
+                       "%stest-delegation.ics" % self.attendee1_calendar,
+                       invitation, 204)
+
+        org_inv = self._getEvent(self.client,
+                                 "%stest-delegation.ics" % self.user_calendar)
+        self._compareAttendees(org_inv, invitation)
+        del_inv = self._getEvent(self.client,
+                                 "%stest-delegation.ics"
+                                 % self.attendee1_delegate_calendar)
+        self._compareAttendees(del_inv, invitation)
+
+        # 7. delegate accepts
+        #    => org: 1 (updated), attendee: 1 (updated), delegate: 1 (accepted)
+
+        invitation.add("method").value = "REPLY"
+        delegate.partstat_param = "ACCEPTED"
+        self._postEvent(self.client,
+                        self.attendee1_delegate_calendar, invitation,
+                        self.attendee1_delegate_email, [self.user_email,
+                                                        self.attendee1_email])
+        del invitation.method
+        self._putEvent(self.client,
+                       "%stest-delegation.ics" % self.attendee1_delegate_calendar,
+                       invitation, 204)
+
+        org_inv = self._getEvent(self.client,
+                                 "%stest-delegation.ics" % self.user_calendar)
+        self._compareAttendees(org_inv, invitation)
+        att_inv = self._getEvent(self.client,
+                                 "%stest-delegation.ics" % self.attendee1_calendar)
+        self._compareAttendees(att_inv, invitation)
+
+        # 8. org updates inv.
+        #    => org: 1 (updated), attendee: 1 (updated,partstat unchanged),
+        #       delegate: 1 (updated,partstat reset)
+
+        invitation.add("method").value = "REQUEST"
+        now = datetime.datetime.now()
+        invitation.vevent.last_modified.value = now
+        invitation.vevent.dtstamp.value = now
+        invitation.vevent.summary.value = "Updated invitation (again)"
+        invitation.vevent.sequence.value = "2"
+        delegate.partstat_param = "NEEDS-ACTION"
+
+        self._postEvent(self.client, self.user_calendar, invitation,
+                        self.user_email, [self.attendee1_email, self.attendee1_delegate_email])
+
+        del invitation.method
+        self._putEvent(self.client,
+                       "%stest-delegation.ics" % self.user_calendar,
+                       invitation, 204)
+
+        att_inv = self._getEvent(self.client,
+                                 "%stest-delegation.ics" % self.attendee1_calendar)
+        self._compareAttendees(att_inv, invitation)
+        del_inv = self._getEvent(self.client,
+                                 "%stest-delegation.ics" % self.attendee1_calendar)
+        self._compareAttendees(del_inv, invitation)
+
+        # 9. org cancels invitation
+        #    => org: 1 (updated), attendee: 0 (cancelled, deleted),
+        #       delegate: 0 (cancelled, deleted)
+
+        invitation.add("method").value = "CANCEL"
+        now = datetime.datetime.now()
+        invitation.vevent.last_modified.value = now
+        invitation.vevent.dtstamp.value = now
+        invitation.vevent.summary.value = "Cancelled invitation (again)"
+        invitation.vevent.sequence.value = "3"
+
+        self._postEvent(self.client, self.user_calendar, invitation,
+                        self.user_email, [self.attendee1_email, self.attendee1_delegate_email])
+
+        del invitation.method
+        invitation.vevent.remove(attendee)
+        invitation.vevent.remove(delegate)
+        self._putEvent(self.client,
+                       "%stest-delegation.ics" % self.user_calendar,
+                       invitation, 204)
+
+        att_inv = self._getEvent(self.client,
+                                 "%stest-delegation.ics" % self.attendee1_calendar, 404)
+        del_inv = self._getEvent(self.client,
+                                 "%stest-delegation.ics" % self.attendee1_calendar, 404)
+
+if __name__ == "__main__":
+    unittest.main()

Tests/Integration/test-config.py

@@ -0,0 +1,54 @@
+#!/usr/bin/python
+
+from config import hostname, port, username, password, mailserver, subscriber_username, attendee1, attendee1_delegate
+
+import unittest
+
+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__":
+    unittest.main()

Tests/Integration/test-davacl.py

@@ -0,0 +1,631 @@
+#!/usr/bin/python
+
+from config import hostname, port, username, password, subscriber_username, subscriber_password
+
+import sys
+import unittest
+import webdavlib
+import time
+
+import utilities
+
+# TODO:
+# - cal: complete test for "modify": "respond to" causes a 204 but no actual
+#        modification should occur
+# - ab: testcase for addressbook-query, webdav-sync (no "calendar-data"
+#       equivalent)
+# - cal: testcase for "calendar-query"
+# - test rights validity:
+#   - send invalid rights to SOGo and expect failures
+#   - refetch the set of rights and make sure it matches what was set
+#     originally
+#   - test "current-user-acl-set"
+
+class DAVAclTest(unittest.TestCase):
+    resource = None
+
+    def __init__(self, arg):
+        self.client = webdavlib.WebDAVClient(hostname, port,
+                                             username, password)
+        unittest.TestCase.__init__(self, arg)
+
+    def setUp(self):
+        delete = webdavlib.WebDAVDELETE(self.resource)
+        self.client.execute(delete)
+        mkcol = webdavlib.WebDAVMKCOL(self.resource)
+        self.client.execute(mkcol)
+        self.assertEquals(mkcol.response["status"], 201,
+                          "preparation: failure creating collection"
+                          "(code = %d)" % mkcol.response["status"])
+        self.subscriber_client = webdavlib.WebDAVClient(hostname, port,
+                                                        subscriber_username,
+                                                        subscriber_password)
+
+    def tearDown(self):
+        delete = webdavlib.WebDAVDELETE(self.resource)
+        self.client.execute(delete)
+
+    def _versitLine(self, line):
+        key, value = line.split(":")
+        semicolon = key.find(";")
+        if semicolon > -1:
+            key = key[:semicolon]
+
+        return (key, value)
+
+    def versitDict(self, event):
+        versitStruct = {}
+        for line in event.splitlines():
+            (key, value) = self._versitLine(line)
+            if not (key == "BEGIN" or key == "END"):
+                versitStruct[key] = value
+
+        return versitStruct
+
+event_template = """BEGIN:VCALENDAR
+VERSION:2.0
+PRODID:-//Inverse//Event Generator//EN
+BEGIN:VEVENT
+SEQUENCE:0
+TRANSP:OPAQUE
+UID:12345-%(class)s-%(filename)s
+SUMMARY:%(class)s event (orig. title)
+DTSTART:20090805T100000Z
+DTEND:20090805T140000Z
+CLASS:%(class)s
+DESCRIPTION:%(class)s description
+LOCATION:location
+%(organizer_line)s%(attendee_line)sCREATED:20090805T100000Z
+DTSTAMP:20090805T100000Z
+END:VEVENT
+END:VCALENDAR"""
+
+class DAVCalendarAclTest(DAVAclTest):
+    resource = '/SOGo/dav/%s/Calendar/test-dav-acl/' % username
+    user_email = None
+
+    def __init__(self, arg):
+        DAVAclTest.__init__(self, arg)
+        self.acl_utility = utilities.TestCalendarACLUtility(self.client,
+                                                            self.resource)
+
+    def setUp(self):
+        DAVAclTest.setUp(self)
+        self.user_email = self.acl_utility.fetchUserInfo(username)[1]
+        self.classToICSClass = { "pu": "PUBLIC",
+                                 "pr": "PRIVATE",
+                                 "co": "CONFIDENTIAL" }
+        self._putEvent(self.client, "public.ics", "PUBLIC")
+        self._putEvent(self.client, "private.ics", "PRIVATE")
+        self._putEvent(self.client, "confidential.ics", "CONFIDENTIAL")
+
+    def testViewAllPublic(self):
+        """'view all' on a specific class (PUBLIC)"""
+        self._testRights({ "pu": "v" })
+
+    def testModifyPublicViewAllPrivateViewDConfidential(self):
+        """'modify' PUBLIC, 'view all' PRIVATE, 'view d&t' confidential"""
+        self._testRights({ "pu": "m", "pr": "v", "co": "d" })
+
+    def testCreateOnly(self):
+        """'create' only"""
+        self._testRights({ "c": True })
+
+    def testDeleteOnly(self):
+        """'delete' only"""
+        self._testRights({ "d": True })
+
+    def testCreateDeleteModifyPublicViewAllPrivateViewDConfidential(self):
+        """'create', 'delete', 'view d&t' PUBLIC, 'modify' PRIVATE"""
+        self._testRights({ "c": True, "d": True, "pu": "d", "pr": "m" })
+
+    def testCreateRespondToPublic(self):
+        """'create', 'respond to' PUBLIC"""
+        self._testRights({ "c": True, "pu": "r" })
+
+    def testNothing(self):
+        """no right given"""
+        self._testRights({})
+
+    def _putEvent(self, client, filename,
+                  event_class = "PUBLIC",
+                  exp_status = 201,
+                  organizer = None, attendee = None,
+                  partstat = "NEEDS-ACTION"):
+        url = "%s%s" % (self.resource, filename)
+        if organizer is not None:
+            organizer_line = "ORGANIZER:%s\n" % organizer
+        else:
+            organizer_line = ""
+        if attendee is not None:
+            attendee_line = "ATTENDEE;PARTSTAT=%s:%s\n" % (partstat, attendee)
+        else:
+            attendee_line = ""
+        event = event_template % { "class": event_class,
+                                   "filename": filename,
+                                   "organizer_line": organizer_line,
+                                   "attendee_line": attendee_line }
+        put = webdavlib.HTTPPUT(url, event)
+        put.content_type = "text/calendar; charset=utf-8"
+        client.execute(put)
+        self.assertEquals(put.response["status"], exp_status,
+                          "%s: event creation/modification:"
+                          " expected status code '%d' (received '%d')"
+                          % (filename, exp_status, put.response["status"]))
+
+    def _deleteEvent(self, client, filename, exp_status = 204):
+        url = "%s%s" % (self.resource, filename)
+        delete = webdavlib.WebDAVDELETE(url)
+        client.execute(delete)
+        self.assertEquals(delete.response["status"], exp_status,
+                          "%s: event deletion: expected status code '%d'"
+                          " (received '%d')"
+                          % (filename, exp_status, delete.response["status"]))
+
+    def _testRights(self, rights):
+        self.acl_utility.setupRights(subscriber_username, rights)
+        self._testCreate(rights)
+        self._testEventRight("pu", rights)
+        self._testEventRight("pr", rights)
+        self._testEventRight("co", rights)
+        self._testDelete(rights)
+
+    def _testCreate(self, rights):
+        if rights.has_key("c") and rights["c"]:
+            exp_code = 201
+        else:
+            exp_code = 403
+        self._putEvent(self.subscriber_client, "creation-test.ics", "PUBLIC",
+                       exp_code)
+
+    def _testDelete(self, rights):
+        if rights.has_key("d") and rights["d"]:
+            exp_code = 204
+        else:
+            exp_code = 403
+        self._deleteEvent(self.subscriber_client, "public.ics", exp_code)
+        self._deleteEvent(self.subscriber_client, "private.ics", exp_code)
+        self._deleteEvent(self.subscriber_client, "confidential.ics",
+                          exp_code)
+
+    def _testEventRight(self, event_class, rights):
+        if rights.has_key(event_class):
+            right = rights[event_class]
+        else:
+            right = None
+
+        event = self._getEvent(event_class)
+        self._checkViewEventRight("GET", event, event_class, right)
+        event = self._propfindEvent(event_class)
+        self._checkViewEventRight("PROPFIND", event, event_class, right)
+        event = self._multigetEvent(event_class)
+        self._checkViewEventRight("multiget", event, event_class, right)
+        event = self._webdavSyncEvent(event_class)
+        self._checkViewEventRight("webdav-sync", event, event_class, right)
+
+        self._testModify(event_class, right)
+        self._testRespondTo(event_class, right)
+
+    def _getEvent(self, event_class, is_invitation = False):
+        icsClass = self.classToICSClass[event_class]
+        if is_invitation:
+            filename = "invitation-%s" % icsClass.lower()
+        else:
+            filename = "%s" % icsClass.lower()
+        url = "%s%s.ics" % (self.resource, filename)
+        get = webdavlib.HTTPGET(url)
+        self.subscriber_client.execute(get)
+
+        if get.response["status"] == 200:
+            event = get.response["body"]
+        else:
+            event = None
+
+        return event
+
+    def _calendarDataInMultistatus(self, query, filename,
+                                   response_tag = "D:response"):
+        event = None
+
+        query.xpath_namespace = { "D": "DAV:",
+                                  "C": "urn:ietf:params:xml:ns:caldav" }
+        response_nodes = query.xpath_evaluate("/D:multistatus/%s" % response_tag)
+        for response_node in response_nodes:
+            href_node = query.xpath_evaluate("D:href", response_node)[0]
+            href = href_node.childNodes[0].nodeValue
+            if href.endswith(filename):
+                propstat_nodes = query.xpath_evaluate("D:propstat", response_node)
+                for propstat_node in propstat_nodes:
+                    status_node = query.xpath_evaluate("D:status",
+                                                       propstat_node)[0]
+                    status = status_node.childNodes[0].nodeValue
+                    data_nodes = query.xpath_evaluate("D:prop/C:calendar-data",
+                                                      propstat_node)
+                    if status.endswith("200 OK"):
+                        if (len(data_nodes) > 0
+                            and len(data_nodes[0].childNodes) > 0):
+                            event = data_nodes[0].childNodes[0].nodeValue
+                    else:
+                        if not (status.endswith("404 Resource Not Found")
+                                or status.endswith("404 Not Found")):
+                            self.fail("%s: unexpected status code: '%s'"
+                                      % (filename, status))
+
+        return event
+
+    def _propfindEvent(self, event_class):
+        event = None
+
+        icsClass = self.classToICSClass[event_class]
+        filename = "%s.ics" % icsClass.lower()
+        propfind = webdavlib.WebDAVPROPFIND(self.resource,
+                                            ["{urn:ietf:params:xml:ns:caldav}calendar-data"],
+                                            1)
+        self.subscriber_client.execute(propfind)
+        if propfind.response["status"] != 403:
+            event = self._calendarDataInMultistatus(propfind, filename)
+
+        return event
+
+    def _multigetEvent(self, event_class):
+        event = None
+
+        icsClass = self.classToICSClass[event_class]
+        url = "%s%s.ics" % (self.resource, icsClass.lower())
+        multiget = webdavlib.CalDAVCalendarMultiget(self.resource,
+                                                    ["{urn:ietf:params:xml:ns:caldav}calendar-data"],
+                                                    [ url ])
+        self.subscriber_client.execute(multiget)
+        if multiget.response["status"] != 403:
+            event = self._calendarDataInMultistatus(multiget, url)
+
+        return event
+
+    def _webdavSyncEvent(self, event_class):
+        event = None
+
+        icsClass = self.classToICSClass[event_class]
+        url = "%s%s.ics" % (self.resource, icsClass.lower())
+        sync_query = webdavlib.WebDAVSyncQuery(self.resource, None,
+                                               ["{urn:ietf:params:xml:ns:caldav}calendar-data"])
+        self.subscriber_client.execute(sync_query)
+        if sync_query.response["status"] != 403:
+            event = self._calendarDataInMultistatus(sync_query, url,
+                                                    "D:sync-response")
+
+        return event
+
+    def _checkViewEventRight(self, operation, event, event_class, right):
+        if right is None:
+            self.assertEquals(event, None,
+                              "None right expecting event invisibility for"
+                              " operation '%s'" % operation)
+        else:
+            self.assertTrue(event is not None,
+                            "no event returned during operation '%s'"
+                            " (right: %s)" % (operation, right))
+            if right == "v" or right == "r" or right == "m":
+                icsClass = self.classToICSClass[event_class]
+                complete_event = (event_template % { "class": icsClass,
+                                                     "filename": "%s.ics" % icsClass.lower(),
+                                                     "organizer_line": "",
+                                                     "attendee_line": ""})
+                self.assertTrue(event.strip() == complete_event.strip(),
+                                "Right '%s' should return complete event"
+                                " during operation '%s'"
+                                % (right, operation))
+            elif right == "d":
+                self._testEventIsSecureVersion(event_class, event)
+            else:
+                self.fail("Right '%s' is not supported" % right)
+
+    def _testEventIsSecureVersion(self, event_class, event):
+        icsClass = self.classToICSClass[event_class]
+        expected_dict = { "VERSION": "2.0",
+                          "PRODID": "-//Inverse//Event Generator//EN",
+                          "SEQUENCE": "0",
+                          "TRANSP": "OPAQUE",
+                          "UID": "12345-%s-%s.ics" % (icsClass,
+                                                      icsClass.lower()),
+                          "SUMMARY": "(%s event)" % icsClass.capitalize(),
+                          "DTSTART": "20090805T100000Z",
+                          "DTEND": "20090805T140000Z",
+                          "CLASS": icsClass,
+                          "CREATED": "20090805T100000Z",
+                          "DTSTAMP": "20090805T100000Z",
+                          "X-SOGO-SECURE": "YES" }
+        event_dict = self.versitDict(event)
+        for key in event_dict.keys():
+            self.assertTrue(expected_dict.has_key(key),
+                            "key '%s' of secure event not expected" % key)
+            self.assertTrue(expected_dict[key] == event_dict[key],
+                            "value for key '%s' of secure does not match"
+                            " (exp: '%s', obtained: '%s'"
+                            % (key, expected_dict[key], event_dict[key] ))
+
+        for key in expected_dict.keys():
+            self.assertTrue(event_dict.has_key(key),
+                            "expected key '%s' not found in secure event"
+                            % key)
+
+    def _testModify(self, event_class, right):
+        if right == "m" or right == "r":
+            exp_code = 204
+        else:
+            exp_code = 403
+        icsClass = self.classToICSClass[event_class]
+        filename = "%s.ics" % icsClass.lower()
+        self._putEvent(self.subscriber_client, filename, icsClass,
+                       exp_code)
+
+    def _testRespondTo(self, event_class, right):
+        icsClass = self.classToICSClass[event_class]
+        filename = "invitation-%s.ics" % icsClass.lower()
+        self._putEvent(self.client, filename, icsClass,
+                       201,
+                       "mailto:nobody@somewhere.com", self.user_email,
+                       "NEEDS-ACTION")
+
+        if right == "m" or right == "r":
+            exp_code = 204
+        else:
+            exp_code = 403
+
+        # here we only do 'passive' validation: if a user has a "respond to"
+        # right, only the attendee entry will me modified. The change of
+        # organizer must thus be silently ignored below.
+        self._putEvent(self.subscriber_client, filename, icsClass,
+                       exp_code, "mailto:someone@nowhere.com", self.user_email,
+                       "ACCEPTED")
+        if exp_code == 204:
+            att_line = "ATTENDEE;PARTSTAT=ACCEPTED:%s\n" % self.user_email
+            if right == "r":
+                exp_event = event_template % {"class": icsClass,
+                                              "filename": filename,
+                                              "organizer_line": "ORGANIZER:mailto:nobody@somewhere.com\n",
+                                              "attendee_line": att_line}
+            else:
+                exp_event = event_template % {"class": icsClass,
+                                              "filename": filename,
+                                              "organizer_line": "ORGANIZER:mailto:someone@nowhere.com\n",
+                                              "attendee_line": att_line}
+            event = self._getEvent(event_class, True).replace("\r", "")
+            self.assertEquals(exp_event.strip(), event.strip(),
+                              "'respond to' event does not match:\nreceived:\n"
+                              "/%s/\nexpected:\n/%s/" % (event, exp_event))
+
+class DAVAddressBookAclTest(DAVAclTest):
+    resource = '/SOGo/dav/%s/Contacts/test-dav-acl/' % username
+    cards = { "new.vcf": """BEGIN:VCARD
+VERSION:3.0
+PRODID:-//Inverse//Card Generator//EN
+UID:NEWTESTCARD
+N:New;Carte
+FN:Carte 'new'
+ORG:societe;service
+NICKNAME:surnom
+ADR;TYPE=work:adr2 societe;;adr societe;ville societe;etat soc;code soc;pays soc
+ADR;TYPE=home:rue perso 2;;rue perso;ville perso;etat perso;code post perso;pays perso
+TEL;TYPE=work:+1 514 123-3372
+TEL;TYPE=home:tel dom
+TEL;TYPE=cell:portable
+TEL;TYPE=fax:fax
+TEL;TYPE=pager:pager
+X-MOZILLA-HTML:FALSE
+EMAIL;TYPE=work:address.email@domaine.ca
+EMAIL;TYPE=home:address.email@domaine2.com
+URL;TYPE=home:web perso
+TITLE:fonction
+URL;TYPE=work:page soc
+CUSTOM1:divers1
+CUSTOM2:divers2
+CUSTOM3:divers3
+CUSTOM4:divers4
+NOTE:Remarque
+X-AIM:pseudo aim
+END:VCARD""",
+              "old.vcf": """BEGIN:VCARD
+VERSION:3.0
+PRODID:-//Inverse//Card Generator//EN
+UID:NEWTESTCARD
+N:Old;Carte
+FN:Carte 'old'
+ORG:societe;service
+NICKNAME:surnom
+ADR;TYPE=work:adr2 societe;;adr societe;ville societe;etat soc;code soc;pays soc
+ADR;TYPE=home:rue perso 2;;rue perso;ville perso;etat perso;code post perso;pays perso
+TEL;TYPE=work:+1 514 123-3372
+TEL;TYPE=home:tel dom
+TEL;TYPE=cell:portable
+TEL;TYPE=fax:fax
+TEL;TYPE=pager:pager
+X-MOZILLA-HTML:FALSE
+EMAIL;TYPE=work:address.email@domaine.ca
+EMAIL;TYPE=home:address.email@domaine2.com
+URL;TYPE=home:web perso
+TITLE:fonction
+URL;TYPE=work:page soc
+CUSTOM1:divers1
+CUSTOM2:divers2
+CUSTOM3:divers3
+CUSTOM4:divers4
+NOTE:Remarque
+X-AIM:pseudo aim
+END:VCARD""",
+              "new-modified.vcf": """BEGIN:VCARD
+VERSION:3.0
+PRODID:-//Inverse//Card Generator//EN
+UID:NEWTESTCARD
+N:New;Carte modifiee
+FN:Carte modifiee 'new'
+ORG:societe;service
+NICKNAME:surnom
+ADR;TYPE=work:adr2 societe;;adr societe;ville societe;etat soc;code soc;pays soc
+ADR;TYPE=home:rue perso 2;;rue perso;ville perso;etat perso;code post perso;pays perso
+TEL;TYPE=work:+1 514 123-3372
+TEL;TYPE=home:tel dom
+TEL;TYPE=cell:portable
+TEL;TYPE=fax:fax
+TEL;TYPE=pager:pager
+X-MOZILLA-HTML:FALSE
+EMAIL;TYPE=work:address.email@domaine.ca
+EMAIL;TYPE=home:address.email@domaine2.com
+URL;TYPE=home:web perso
+TITLE:fonction
+URL;TYPE=work:page soc
+CUSTOM1:divers1
+CUSTOM2:divers2
+CUSTOM3:divers3
+CUSTOM4:divers4
+NOTE:Remarque
+X-AIM:pseudo aim
+END:VCARD""",
+              "old-modified.vcf": """BEGIN:VCARD
+VERSION:3.0
+PRODID:-//Inverse//Card Generator//EN
+UID:NEWTESTCARD
+N:Old;Carte modifiee
+FN:Carte modifiee 'old'
+ORG:societe;service
+NICKNAME:surnom
+ADR;TYPE=work:adr2 societe;;adr societe;ville societe;etat soc;code soc;pays soc
+ADR;TYPE=home:rue perso 2;;rue perso;ville perso;etat perso;code post perso;pays perso
+TEL;TYPE=work:+1 514 123-3372
+TEL;TYPE=home:tel dom
+TEL;TYPE=cell:portable
+TEL;TYPE=fax:fax
+TEL;TYPE=pager:pager
+X-MOZILLA-HTML:FALSE
+EMAIL;TYPE=work:address.email@domaine.ca
+EMAIL;TYPE=home:address.email@domaine2.com
+URL;TYPE=home:web perso
+TITLE:fonction
+URL;TYPE=work:page soc
+CUSTOM1:divers1
+CUSTOM2:divers2
+CUSTOM3:divers3
+CUSTOM4:divers4
+NOTE:Remarque
+X-AIM:pseudo aim
+END:VCARD""" }
+
+    def __init__(self, arg):
+        DAVAclTest.__init__(self, arg)
+        self.acl_utility = utilities.TestAddressBookACLUtility(self.client,
+                                                               self.resource)
+
+    def setUp(self):
+        DAVAclTest.setUp(self)
+        self._putCard(self.client, "old.vcf", 201)
+
+    def testView(self):
+        """'view' only"""
+        self._testRights({ "v": True })
+
+    def testEdit(self):
+        """'edit' only"""
+        self._testRights({ "e": True })
+
+    def testCreateOnly(self):
+        """'create' only"""
+        self._testRights({ "c": True })
+
+    def testDeleteOnly(self):
+        """'delete' only"""
+        self._testRights({ "d": True })
+
+    def testCreateDelete(self):
+        """'create' only"""
+        self._testRights({ "c": True,
+                           "d": True })
+
+    def testViewCreate(self):
+        """'view' and 'create'"""
+        self._testRights({ "c": True,
+                           "v": True })
+
+    def testViewDelete(self):
+        """'view' and 'delete'"""
+        self._testRights({ "d": True,
+                           "v": True })
+
+    def testEditCreate(self):
+        """'edit' and 'create'"""
+        self._testRights({ "c": True,
+                           "e": True })
+
+    def testEditDelete(self):
+        """'edit' and 'delete'"""
+        self._testRights({ "d": True,
+                           "e": True })
+
+    def _testRights(self, rights):
+        self.acl_utility.setupRights(subscriber_username, rights)
+        self._testCreate(rights)
+        self._testView(rights)
+        self._testEdit(rights)
+        self._testDelete(rights)
+
+    def _putCard(self, client, filename, exp_status, real_card = None):
+        url = "%s%s" % (self.resource, filename)
+        if real_card is None:
+            real_card = filename
+        card = self.cards[real_card]
+        put = webdavlib.HTTPPUT(url, card)
+        put.content_type = "text/x-vcard; charset=utf-8"
+        client.execute(put)
+        self.assertEquals(put.response["status"], exp_status,
+                          "%s: card creation/modification:"
+                          " expected status code '%d' (received '%d')"
+                          % (filename, exp_status, put.response["status"]))
+
+    def _getCard(self, client, filename, exp_status):
+        url = "%s%s" % (self.resource, filename)
+        get = webdavlib.HTTPGET(url)
+        client.execute(get)
+        self.assertEquals(get.response["status"], exp_status,
+                          "%s: card get:"
+                          " expected status code '%d' (received '%d')"
+                          % (filename, exp_status, get.response["status"]))
+
+    def _deleteCard(self, client, filename, exp_status):
+        url = "%s%s" % (self.resource, filename)
+        delete = webdavlib.WebDAVDELETE(url)
+        client.execute(delete)
+        self.assertEquals(delete.response["status"], exp_status,
+                          "%s: card deletion:"
+                          " expected status code '%d' (received '%d')"
+                          % (filename, exp_status, delete.response["status"]))
+
+    def _testCreate(self, rights):
+        if rights.has_key("c") and rights["c"]:
+            exp_code = 201
+        else:
+            exp_code = 403
+        self._putCard(self.subscriber_client, "new.vcf", exp_code)
+
+    def _testView(self, rights):
+        if ((rights.has_key("v") and rights["v"])
+            or (rights.has_key("e") and rights["e"])):
+            exp_code = 200
+        else:
+            exp_code = 403
+        self._getCard(self.subscriber_client, "old.vcf", exp_code)
+
+    def _testEdit(self, rights):
+        if rights.has_key("e") and rights["e"]:
+            exp_code = 204
+        else:
+            exp_code = 403
+        self._putCard(self.subscriber_client, "old.vcf", exp_code, "old-modified.vcf")
+
+    def _testDelete(self, rights):
+        if rights.has_key("d") and rights["d"]:
+            exp_code = 204
+        else:
+            exp_code = 403
+        self._deleteCard(self.subscriber_client, "old.vcf", exp_code)
+
+if __name__ == "__main__":
+    unittest.main()
+

Tests/Integration/test-ical.py

@@ -0,0 +1,206 @@
+#!/usr/bin/python
+
+from config import hostname, port, username, password, subscriber_username
+
+import unittest
+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)
+        propfind.xpath_namespace = { "D": "DAV:" }
+        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)"
+        propfind.xpath_namespace = { "D": "DAV:" }
+        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-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, username, 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)"
+        proppatch.xpath_namespace = { "D": "DAV:" }
+        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, username, 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)"
+        propfind.xpath_namespace = { "D": "DAV:" }
+        client.execute(propfind)
+
+        hrefs = propfind.xpath_evaluate("/D:multistatus/D:response/D:propstat/D:prop/D:group-membership/D:href")
+        members = [x.childNodes[0].nodeValue 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, username, 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)"
+        propfind.xpath_namespace = { "D": "DAV:", "n1": "http://calendarserver.org/ns/" }
+        client.execute(propfind)
+
+        hrefs = propfind.xpath_evaluate("/D:multistatus/D:response/D:propstat/D:prop/n1:calendar-proxy-%s-for/D:href"
+                                        % perm)
+        members = [x.childNodes[0].nodeValue[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(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, username, 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(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(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(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(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__":
+    unittest.main()

Tests/Integration/test-maildav.py

@@ -0,0 +1,682 @@
+#!/usr/bin/python
+
+from config import hostname, port, username, password, mailserver, subscriber_username, subscriber_password
+
+import sys
+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)
+  propfind.xpath_namespace = { "D": "DAV:",
+                               "C": "urn:ietf:params:xml:ns:caldav" }
+  client.execute(propfind)
+  nodes = propfind.xpath_evaluate('/D:multistatus/D:response/D:propstat/D:prop/C:calendar-user-address-set/D: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 big 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(unittest.TestCase):
+  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("/D:multistatus/D:response/D:propstat/D: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"]))
+    query.xpath_namespace = { "D": "DAV:",
+                              "I": "urn:inverse:params:xml:ns:inverse-dav" }
+    response_nodes = query.xpath_evaluate("/D:multistatus/D:response")
+    for response_node in response_nodes:
+      href_node = query.xpath_evaluate("D: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"]))
+    query.xpath_namespace = { "D": "DAV:",
+                              "I": "urn:inverse:params:xml:ns:inverse-dav" }
+    response_nodes = query.xpath_evaluate("/D:multistatus/D:response")
+    for response_node in response_nodes:
+      href_node = query.xpath_evaluate("D:href", response_node)[0]
+      href = href_node.childNodes[0].nodeValue
+      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__":
+  unittest.main()

Tests/Integration/test-webdav.py

@@ -0,0 +1,154 @@
+#!/usr/bin/python
+
+from config import hostname, port, username, password
+
+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.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)
+        propfind.xpath_namespace = { "D": "DAV:" }
+        self.client.execute(propfind)
+        self.assertEquals(propfind.response["status"], 207)
+        nodes = propfind.xpath_evaluate('/D:multistatus/D:response/D:propstat/D:prop/D:principal-collection-set/D:href',
+                                        None)
+        responseHref = nodes[0].childNodes[0].nodeValue
+        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)
+        propfind.xpath_namespace = { "D": "DAV:" }
+        self.client.execute(propfind)
+        self.assertEquals(propfind.response["status"], 207)
+        nodes = propfind.xpath_evaluate('/D:multistatus/D:response/D:propstat/D:prop/D:principal-collection-set/D:href',
+                                        None)
+        responseHref = nodes[0].childNodes[0].nodeValue
+        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)
+        propfind.xpath_namespace = { "D": "DAV:" }
+        self.client.execute(propfind)
+        self.assertEquals(propfind.response["status"], 207)
+
+        nodes = propfind.xpath_evaluate('/D:multistatus/D:response',
+                                        None)
+        for node in nodes:
+            responseHref = propfind.xpath_evaluate('D:href', node)[0].childNodes[0].nodeValue
+            hasSlash = responseHref[-1] == '/'
+            resourcetypes = \
+                propfind.xpath_evaluate('D:propstat/D:prop/D:resourcetype',
+                                        node)[0].childNodes
+            isCollection = len(resourcetypes) > 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
+    # 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 = {"owner": { "href": resource,
+                                  "displayname": userInfo[0]},
+                       "principal-collection-set": { "href": "/SOGo/dav/",
+                                                     "displayname": "SOGo"}}
+        query = webdavlib.WebDAVExpandProperty(resource, query_props.keys(),
+                                               ["displayname"])
+        self.client.execute(query)
+        self.assertEquals(query.response["status"], 207)
+
+        topResponse = query.xpath_evaluate('/D:multistatus/D:response')[0]
+        topHref = query.xpath_evaluate('D:href', topResponse)[0]
+        self.assertEquals(resource, topHref.childNodes[0].nodeValue)
+        for query_prop in query_props.keys():
+            propResponse = query.xpath_evaluate('D:propstat/D:prop/D:%s'
+                                                % query_prop, topResponse)[0]
+
+
+# <?xml version="1.0" encoding="utf-8"?>
+# <D:multistatus xmlns:D="DAV:">
+#   <D:response>
+#     <D:href>/SOGo/dav/wsourdeau/</D:href>
+#     <D:propstat>
+#       <D:prop>
+#         <D:owner>
+#           <D:response>
+#             <D:href>/SOGo/dav/wsourdeau/</D:href>
+#             <D:propstat>
+#               <D:prop>
+#                 <D:displayname>Wolfgang Sourdeau</D:displayname>
+#               </D:prop>
+#               <D:status>HTTP/1.1 200 OK</D:status>
+#             </D:propstat>
+#           </D:response>
+#         </D:owner>
+            propHref = query.xpath_evaluate('D:response/D:href',
+                                            propResponse)[0]
+            self.assertEquals(query_props[query_prop]["href"],
+                              propHref.childNodes[0].nodeValue,
+                              "'%s', href mismatch: exp. '%s', got '%s'"
+                              % (query_prop,
+                                 query_props[query_prop]["href"],
+                                 propHref.childNodes[0].nodeValue))
+            propDisplayname = query.xpath_evaluate('D:response/D:propstat/D:prop/D:displayname',
+                                                   propResponse)[0]
+            self.assertEquals(query_props[query_prop]["displayname"],
+                              propDisplayname.childNodes[0].nodeValue,
+                              "'%s', displayname mismatch: exp. '%s', got '%s'"
+                              % (query_prop,
+                                 query_props[query_prop]["displayname"],
+                                 propDisplayname.nodeValue))
+
+if __name__ == "__main__":
+    unittest.main()

Tests/Integration/test-webdavlib.py

@@ -0,0 +1,50 @@
+#!/usr/bin/python
+
+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__":
+    unittest.main()

Tests/Integration/test-webdavsync.py

@@ -0,0 +1,63 @@
+#!/usr/bin/python
+
+from config import hostname, port, username, password
+
+import sys
+import unittest
+import webdavlib
+import time
+
+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):
+        # 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.xpath_evaluate("/D:multistatus/D:sync-token")[0]
+        # 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.childNodes[0].nodeValue)
+
+        self.assertTrue(token > 0)
+        self.assertTrue(token <= int(query1.start))
+
+        # we make sure that any token is invalid when the collection is empty
+        query2 = webdavlib.WebDAVSyncQuery(resource, "1234", [ "getetag" ])
+        self.client.execute(query2)
+        self.assertEquals(query2.response["status"], 403)
+        cond_nodes = query2.xpath_evaluate("/D:error/D:valid-sync-token")
+        self.assertTrue(len(cond_nodes) > 0,
+                        "expected 'valid-sync-token' condition error")
+
+if __name__ == "__main__":
+    unittest.main()

Tests/Integration/utilities.py

@@ -0,0 +1,128 @@
+#!/usr/bin/python
+
+import unittest
+import webdavlib
+
+class TestUtility(unittest.TestCase):
+    def __init__(self, client):
+        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)
+            propfind.xpath_namespace = { "D": "DAV:",
+                                         "C": "urn:ietf:params:xml:ns:caldav" }
+            self.client.execute(propfind)
+            assert(propfind.response["status"] == 207)
+            name_nodes = propfind.xpath_evaluate('/D:multistatus/D:response/D:propstat/D:prop/D:displayname',
+                                                 None)
+            email_nodes = propfind.xpath_evaluate('/D:multistatus/D:response/D:propstat/D:prop/C:calendar-user-address-set/D:href',
+                                                  None)
+            self.userInfo[login] = (name_nodes[0].childNodes[0].nodeValue,
+                                    email_nodes[0].childNodes[0].nodeValue)
+
+        return self.userInfo[login]
+
+class TestACLUtility(TestUtility):
+    def __init__(self, client, resource):
+        TestUtility.__init__(self, client)
+        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.assertEquals(post.response["status"], 204,
+                          "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):
+        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>" % (username,
+                                                                 rights_str)
+                    + "</acl-query>")
+
+        post = webdavlib.HTTPPOST(self.resource, aclQuery)
+        post.content_type = "application/xml; charset=\"utf-8\""
+        self.client.execute(post)
+        self.assertEquals(post.response["status"], 204,
+                          "rights modification: failure to set '%s' (status: %d)"
+                          % (rights_str, post.response["status"]))
+
+# 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

Tests/Integration/webdavlib.py

@@ -0,0 +1,500 @@
+import cStringIO
+import httplib
+import M2Crypto.httpslib
+import re
+import time
+import xml.sax.saxutils
+import xml.dom.ext.reader.Sax2
+import xml.xpath
+import sys
+
+xmlns_dav = "DAV:"
+xmlns_caldav = "urn:ietf:params:xml:ns:caldav"
+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, password, forcessl = False):
+        if port == "443" or forcessl:
+            self.conn = M2Crypto.httpslib.HTTPSConnection(hostname, int(port),
+                                                          True)
+        else:
+            self.conn = httplib.HTTPConnection(hostname, port, True)
+
+        self.simpleauth_hash = (("%s:%s" % (username, password))
+                                .encode('base64')[:-1])
+
+    def prepare_headers(self, query, body):
+        headers = { "User-Agent": self.user_agent,
+                    "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))
+        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"
+
+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):
+        HTTPQuery.__init__(self, url)
+        self.content = content
+
+    def render(self):
+        return self.content
+
+class HTTPPOST(HTTPPUT):
+    method = "POST"
+
+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
+        self.xml_response = None
+        self.xpath_namespace = { "D": xmlns_dav }
+
+    # 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):
+            reader = xml.dom.ext.reader.Sax2.Reader()
+            stream = cStringIO.StringIO(self.response["body"])
+            dom_response = reader.fromStream(stream)
+            self.response["document"] = dom_response.documentElement
+
+    def xpath_evaluate(self, query, top_node = None):
+        if top_node is None:
+            top_node = self.response["document"]
+        xpath_context = xml.xpath.CreateContext(top_node)
+        xpath_context.setNamespaces(self.xpath_namespace)
+        return xml.xpath.Evaluate(query, None, xpath_context)
+
+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 WebDAVPUT(WebDAVQuery):
+    method = "PUT"
+
+    def __init__(self, url, content):
+        WebDAVQuery.__init__(self, url)
+        self.content_type = "text/plain; charset=utf-8"
+        self.content = content
+    
+    def prepare_headers(self):
+        return WebDAVQuery.prepare_headers(self)
+
+    def render(self):
+        return self.content
+
+
+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):
+        WebDAVQuery.__init__(self, url)
+        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 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 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("")
+    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.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