diff --git a/.gitignore b/.gitignore
index 0d20b64..0d4670c 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1 +1,6 @@
*.pyc
+build/
+*~
+.project
+.pydevproject
+.settings/
diff --git a/config b/config
index cc72b87..7b5702a 100644
--- a/config
+++ b/config
@@ -80,9 +80,12 @@ courier_socket =
[rights]
# Rights management method
-# Value: None | owner_only | owner_write
+# Value: None | owner_only | owner_write | from_file
type = None
+# File for rights management from_file
+file = ~/.config/radicale/rights
+
[storage]
# Storage backend
diff --git a/radicale/__init__.py b/radicale/__init__.py
index 3bda1a6..0713ac3 100644
--- a/radicale/__init__.py
+++ b/radicale/__init__.py
@@ -51,9 +51,18 @@ from . import auth, config, ical, log, rights, storage, xmlutils
VERSION = "git"
-# Standard "not allowed" response
+# Standard "not allowed" response that is returned when an authenticated
+# user tries to access information they don't have rights to.
NOT_ALLOWED = (
client.FORBIDDEN,
+ {},
+ None)
+
+# Standard "authenticate" response that is returned when a
+# user tries to access non-public information w/o submitting
+# proper authentication credentials
+WRONG_CREDENTIALS = (
+ client.UNAUTHORIZED,
{"WWW-Authenticate": "Basic realm=\"Radicale - Password Required\""},
None)
@@ -176,6 +185,58 @@ class Application(object):
trailing_slash = "" if uri == "/" else trailing_slash
return uri + trailing_slash
+
+ def collect_allowed_items(self, items, user):
+ """ Collect those items from the request that the user
+ is actually allowed to access """
+
+ read_last_collection_allowed = None
+ write_last_collection_allowed = None
+ read_allowed_items = []
+ write_allowed_items = []
+ for item in items:
+ if isinstance(item, ical.Collection):
+ if rights.read_authorized(user, item):
+ log.LOGGER.debug("%s has read access to collection %s" % (user, item.url or "/"))
+ read_last_collection_allowed = True
+ read_allowed_items.append(item)
+ else:
+ log.LOGGER.debug("%s has NO read access to collection %s" % (user, item.url or "/"))
+ read_last_collection_allowed = False
+
+ if rights.write_authorized(user, item):
+ log.LOGGER.debug("%s has write access to collection %s" % (user, item.url or "/"))
+ write_last_collection_allowed = True
+ write_allowed_items.append(item)
+ else:
+ log.LOGGER.debug("%s has NO write access to collection %s" % (user, item.url or "/"))
+ write_last_collection_allowed = False
+ # item is not a collection, it's the child of the last
+ # collection we've met in the loop. Only add this item
+ # if this last collection was allowed.
+ else:
+ if read_last_collection_allowed:
+ log.LOGGER.debug("%s has read access to item %s" % (user, item.name or "/"))
+ read_allowed_items.append(item)
+ if write_last_collection_allowed:
+ log.LOGGER.debug("%s has write access to item %s" % (user, item.name or "/"))
+ write_allowed_items.append(item)
+
+ if (not write_last_collection_allowed) and (not read_last_collection_allowed):
+ log.LOGGER.info("%s has NO access to item %s" % (user, item.name or "/"))
+
+ return read_allowed_items, write_allowed_items
+
+
+ def _union(self, list1, list2):
+ out = []
+ out.extend(list1)
+ for thing in list2:
+ if not thing in list1:
+ list1.append(thing)
+ return out
+
+
def __call__(self, environ, start_response):
"""Manage a request."""
log.LOGGER.info("%s request at %s received" % (
@@ -218,58 +279,21 @@ class Application(object):
if not items or function == self.options or \
auth.is_authenticated(user, password):
- last_collection_allowed = None
- allowed_items = []
- for item in items:
- if isinstance(item, ical.Collection):
- if rights.read_authorized(user, item) or \
- rights.write_authorized(user, item):
- log.LOGGER.info("%s has access to collection %s" % (
- user, item.name or "/"))
- last_collection_allowed = True
- allowed_items.append(item)
- else:
- log.LOGGER.info("%s has NO access to collection %s" % (
- user, item.name or "/"))
- last_collection_allowed = False
- else:
- # item is not a collection, it's the child of the last
- # collection we've met in the loop. Only add this item
- # if this last collection was allowed.
- if last_collection_allowed:
- log.LOGGER.info("%s has access to item %s" % (
- user, item.name or "/"))
- allowed_items.append(item)
- else:
- log.LOGGER.info("%s has NO access to item %s" % (
- user, item.name or "/"))
- if allowed_items or function == self.options:
+ read_allowed_items, write_allowed_items = self.collect_allowed_items(items, user)
+
+ if read_allowed_items or write_allowed_items or function == self.options:
# Collections found
status, headers, answer = function(
- environ, allowed_items, content, user)
+ environ, read_allowed_items, write_allowed_items, content, user)
else:
- # Good user and no collections found, redirect user to home
- location = "/%s/" % str(quote(user))
- if path == location:
- # Send answer anyway since else we're getting into a
- # redirect loop
- status, headers, answer = function(
- environ, allowed_items, content, user)
- else:
- log.LOGGER.info("redirecting to %s" % location)
- status = client.FOUND
- headers = {"Location": location}
- answer = "Redirecting to %s" % location
+ # Good user but has no rights to any of the given collections
+ status, headers, answer = NOT_ALLOWED
else:
# Unknown or unauthorized user
log.LOGGER.info(
"%s refused" % (user or "Anonymous user"))
- status = client.UNAUTHORIZED
- headers = {
- "WWW-Authenticate":
- "Basic realm=\"Radicale Server - Password Required\""}
- answer = None
+ status, headers, answer = WRONG_CREDENTIALS
# Set content length
if answer:
@@ -288,9 +312,12 @@ class Application(object):
# All these functions must have the same parameters, some are useless
# pylint: disable=W0612,W0613,R0201
- def delete(self, environ, collections, content, user):
+ def delete(self, environ, read_collections, write_collections, content, user):
"""Manage DELETE request."""
- collection = collections[0]
+ if not len(write_collections):
+ return client.PRECONDITION_FAILED, {}, None
+
+ collection = write_collections[0]
if collection.path == environ["PATH_INFO"].strip("/"):
# Path matching the collection, the collection must be deleted
@@ -305,16 +332,13 @@ class Application(object):
etag = environ.get("HTTP_IF_MATCH", item.etag).replace("\\", "")
if etag == item.etag:
# No ETag precondition or precondition verified, delete item
- if rights.write_authorized(user, collection):
- answer = xmlutils.delete(environ["PATH_INFO"], collection)
- return client.OK, {}, answer
- else:
- return NOT_ALLOWED
+ answer = xmlutils.delete(environ["PATH_INFO"], collection)
+ return client.OK, {}, answer
# No item or ETag precondition not verified, do not delete item
return client.PRECONDITION_FAILED, {}, None
- def get(self, environ, collections, content, user):
+ def get(self, environ, read_collections, write_collections, content, user):
"""Manage GET request.
In Radicale, GET requests create collections when the URL is not
@@ -328,36 +352,37 @@ class Application(object):
answer = b"\nRadicaleRadicale works!"
return client.OK, headers, answer
- collection = collections[0]
+ if not len(read_collections):
+ return NOT_ALLOWED
+
+ collection = read_collections[0]
+
item_name = xmlutils.name_from_path(environ["PATH_INFO"], collection)
if item_name:
# Get collection item
item = collection.get_item(item_name)
if item:
- if rights.read_authorized(user, collection):
- items = collection.timezones
- items.append(item)
- answer_text = ical.serialize(
- collection.tag, collection.headers, items)
- etag = item.etag
- else:
- return NOT_ALLOWED
+ items = collection.timezones
+ items.append(item)
+ answer_text = ical.serialize(
+ collection.tag, collection.headers, items)
+ etag = item.etag
else:
return client.GONE, {}, None
else:
# Create the collection if it does not exist
- if not collection.exists and \
- rights.write_authorized(user, collection):
- log.LOGGER.debug("Creating collection %s" % collection.name)
- collection.write()
+ if not collection.exists:
+ if collection in write_collections:
+ log.LOGGER.debug("Creating collection %s" % collection.name)
+ collection.write()
+ else:
+ log.LOGGER.debug("Collection %s not available and could not be created due to missing write rights" % collection.name)
+ return NOT_ALLOWED
- if rights.read_authorized(user, collection):
- # Get whole collection
- answer_text = collection.text
- etag = collection.etag
- else:
- return NOT_ALLOWED
+ # Get whole collection
+ answer_text = collection.text
+ etag = collection.etag
headers = {
"Content-Type": collection.mimetype,
@@ -366,44 +391,50 @@ class Application(object):
answer = answer_text.encode(self.encoding)
return client.OK, headers, answer
- def head(self, environ, collections, content, user):
+ def head(self, environ, read_collections, write_collections, content, user):
"""Manage HEAD request."""
- status, headers, answer = self.get(environ, collections, content, user)
+ status, headers, answer = self.get(environ, read_collections, write_collections, content, user)
return status, headers, None
- def mkcalendar(self, environ, collections, content, user):
+ def mkcalendar(self, environ, read_collections, write_collections, content, user):
"""Manage MKCALENDAR request."""
- collection = collections[0]
- if rights.write_authorized(user, collection):
- props = xmlutils.props_from_request(content)
- timezone = props.get("C:calendar-timezone")
- if timezone:
- collection.replace("", timezone)
- del props["C:calendar-timezone"]
- with collection.props as collection_props:
- for key, value in props.items():
- collection_props[key] = value
- collection.write()
- return client.CREATED, {}, None
- else:
+ if not len(write_collections):
return NOT_ALLOWED
-
- def mkcol(self, environ, collections, content, user):
- """Manage MKCOL request."""
- collection = collections[0]
- if rights.write_authorized(user, collection):
- props = xmlutils.props_from_request(content)
- with collection.props as collection_props:
- for key, value in props.items():
- collection_props[key] = value
+
+ collection = write_collections[0]
+
+ props = xmlutils.props_from_request(content)
+ timezone = props.get("C:calendar-timezone")
+ if timezone:
+ collection.replace("", timezone)
+ del props["C:calendar-timezone"]
+ with collection.props as collection_props:
+ for key, value in props.items():
+ collection_props[key] = value
collection.write()
- return client.CREATED, {}, None
- else:
- return NOT_ALLOWED
+ return client.CREATED, {}, None
- def move(self, environ, collections, content, user):
+ def mkcol(self, environ, read_collections, write_collections, content, user):
+ """Manage MKCOL request."""
+ if not len(write_collections):
+ return NOT_ALLOWED
+
+ collection = write_collections[0]
+
+ props = xmlutils.props_from_request(content)
+ with collection.props as collection_props:
+ for key, value in props.items():
+ collection_props[key] = value
+ collection.write()
+ return client.CREATED, {}, None
+
+ def move(self, environ, read_collections, write_collections, content, user):
"""Manage MOVE request."""
- from_collection = collections[0]
+ if not len(write_collections):
+ return NOT_ALLOWED
+
+ from_collection = write_collections[0]
+
from_name = xmlutils.name_from_path(
environ["PATH_INFO"], from_collection)
if from_name:
@@ -416,8 +447,7 @@ class Application(object):
to_path, to_name = to_url.rstrip("/").rsplit("/", 1)
to_collection = ical.Collection.from_path(
to_path, depth="0")[0]
- if rights.write_authorized(user, to_collection) and \
- rights.write_authorized(user.from_collection):
+ if to_collection in write_collections:
to_collection.append(to_name, item.text)
from_collection.remove(from_name)
return client.CREATED, {}, None
@@ -433,7 +463,7 @@ class Application(object):
# Moving collections, not supported
return client.FORBIDDEN, {}, None
- def options(self, environ, collections, content, user):
+ def options(self, environ, read_collections, write_collections, content, user):
"""Manage OPTIONS request."""
headers = {
"Allow": ("DELETE, HEAD, GET, MKCALENDAR, MKCOL, MOVE, "
@@ -441,32 +471,38 @@ class Application(object):
"DAV": "1, 2, 3, calendar-access, addressbook, extended-mkcol"}
return client.OK, headers, None
- def propfind(self, environ, collections, content, user):
+ def propfind(self, environ, read_collections, write_collections, content, user):
"""Manage PROPFIND request."""
# Rights is handled by collection in xmlutils.propfind
headers = {
"DAV": "1, 2, 3, calendar-access, addressbook, extended-mkcol",
"Content-Type": "text/xml"}
+ collections = self._union(read_collections, write_collections)
answer = xmlutils.propfind(
environ["PATH_INFO"], content, collections, user)
return client.MULTI_STATUS, headers, answer
- def proppatch(self, environ, collections, content, user):
+ def proppatch(self, environ, read_collections, write_collections, content, user):
"""Manage PROPPATCH request."""
- collection = collections[0]
- if rights.write_authorized(user, collection):
- answer = xmlutils.proppatch(
- environ["PATH_INFO"], content, collection)
- headers = {
- "DAV": "1, 2, 3, calendar-access, addressbook, extended-mkcol",
- "Content-Type": "text/xml"}
- return client.MULTI_STATUS, headers, answer
- else:
+ if not len(write_collections):
return NOT_ALLOWED
+
+ collection = write_collections[0]
+
+ answer = xmlutils.proppatch(
+ environ["PATH_INFO"], content, collection)
+ headers = {
+ "DAV": "1, 2, 3, calendar-access, addressbook, extended-mkcol",
+ "Content-Type": "text/xml"}
+ return client.MULTI_STATUS, headers, answer
- def put(self, environ, collections, content, user):
+ def put(self, environ, read_collections, write_collections, content, user):
"""Manage PUT request."""
- collection = collections[0]
+ if not len(write_collections):
+ return NOT_ALLOWED
+
+ collection = write_collections[0]
+
collection.set_mimetype(environ.get("CONTENT_TYPE"))
headers = {}
item_name = xmlutils.name_from_path(environ["PATH_INFO"], collection)
@@ -481,31 +517,30 @@ class Application(object):
# Case 1: No item and no ETag precondition: Add new item
# Case 2: Item and ETag precondition verified: Modify item
# Case 3: Item and no Etag precondition: Force modifying item
- if rights.write_authorized(user, collection):
- xmlutils.put(environ["PATH_INFO"], content, collection)
- status = client.CREATED
- # Try to return the etag in the header.
- # If the added item does't have the same name as the one given
- # by the client, then there's no obvious way to generate an
- # etag, we can safely ignore it.
- new_item = collection.get_item(item_name)
- if new_item:
- headers["ETag"] = new_item.etag
- else:
- return NOT_ALLOWED
+ xmlutils.put(environ["PATH_INFO"], content, collection)
+ status = client.CREATED
+ # Try to return the etag in the header.
+ # If the added item does't have the same name as the one given
+ # by the client, then there's no obvious way to generate an
+ # etag, we can safely ignore it.
+ new_item = collection.get_item(item_name)
+ if new_item:
+ headers["ETag"] = new_item.etag
else:
# PUT rejected in all other cases
status = client.PRECONDITION_FAILED
return status, headers, None
- def report(self, environ, collections, content, user):
+ def report(self, environ, read_collections, write_collections, content, user):
"""Manage REPORT request."""
- collection = collections[0]
- headers = {"Content-Type": "text/xml"}
- if rights.read_authorized(user, collection):
- answer = xmlutils.report(environ["PATH_INFO"], content, collection)
- return client.MULTI_STATUS, headers, answer
- else:
+ if not len(read_collections):
return NOT_ALLOWED
+
+ collection = read_collections[0]
+
+ headers = {"Content-Type": "text/xml"}
+
+ answer = xmlutils.report(environ["PATH_INFO"], content, collection)
+ return client.MULTI_STATUS, headers, answer
# pylint: enable=W0612,W0613,R0201
diff --git a/radicale/config.py b/radicale/config.py
index 29c3b94..c94930f 100644
--- a/radicale/config.py
+++ b/radicale/config.py
@@ -67,7 +67,8 @@ INITIAL_CONFIG = {
"pam_group_membership": "",
"courier_socket": ""},
"rights": {
- "type": "None"},
+ "type": "None",
+ "file" : "None"},
"storage": {
"type": "filesystem",
"filesystem_folder": os.path.expanduser(
diff --git a/radicale/rights/from_file.py b/radicale/rights/from_file.py
new file mode 100644
index 0000000..5a0a2ed
--- /dev/null
+++ b/radicale/rights/from_file.py
@@ -0,0 +1,176 @@
+# -*- coding: utf-8 -*-
+#
+# This file is part of Radicale Server - Calendar Server
+# Copyright © 2012 Guillaume Ayoub
+#
+# This library is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This library is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with Radicale. If not, see .
+
+"""
+File-based rights.
+
+The owner is implied to have all rights on their collections.
+
+Rights are read from a file whose name is specified in the config
+(section "right", key "file").
+
+The file's format is per line:
+
+collectionpath ":" principal " " rights {", " principal " " rights}*
+
+collectionpath is the path part of the collection's url
+
+principal is a user name (no whitespace allowed)
+rights is a string w/o whitespace that contains "r" for reading rights,
+"w" for writing rights and a combination of these for all rights.
+
+Empty lines are ignored. Lines starting with "#" (hash sign) are comments.
+
+Example:
+
+# This means user1 may read, user2 may write, user3 has full access
+/user0/calendar : user1 r, user2 w, user3 rw
+# user0 can read /user1/cal
+/user1/cal : user0 r
+
+If a collection /a/b is shared and other users than the owner are
+supposed to find the collection in a propfind request, an additional
+line for /a has to be in the defintions. E.g.:
+
+/user0/cal: user
+
+"""
+
+from radicale import config, log
+from radicale.rights import owner_only
+
+
+READ_AUTHORIZED = None
+WRITE_AUTHORIZED = None
+
+
+class ParsingError(BaseException):
+ """Raised if the file cannot be parsed"""
+
+
+def read_authorized(user, collection):
+ """Check if the user is allowed to read the collection."""
+ if owner_only.read_authorized(user, collection):
+ return True
+
+ curl = _normalize_trail_slash(collection.url)
+
+ return _dict_knows(READ_AUTHORIZED, curl, user)
+
+
+
+def write_authorized(user, collection):
+ """Check if the user is allowed to write the collection."""
+ if owner_only.read_authorized(user, collection):
+ return True
+
+ curl = _normalize_trail_slash(collection.url)
+
+ return _dict_knows(WRITE_AUTHORIZED, curl, user)
+
+
+
+def _dict_knows(adict, url, user):
+ return adict.has_key(url) and adict.get(url).count(user) != 0
+
+
+
+def _load():
+ read = {}
+ write = {}
+ file_name = config.get("rights", "file")
+ if file_name == "None":
+ log.LOGGER.error("No file name configured for rights type 'from_file'")
+ return
+
+ log.LOGGER.debug("Reading rights from file %s" % file_name)
+
+ lines = open(file_name, "r").readlines()
+
+ for line in lines:
+ _process(line, read, write)
+
+ global READ_AUTHORIZED, WRITE_AUTHORIZED
+ READ_AUTHORIZED = read
+ WRITE_AUTHORIZED = write
+
+
+
+def _process(line, read, write):
+ line = line.strip()
+ if line == "":
+ """Empty line"""
+ return
+
+ if line.startswith("#"):
+ """Comment"""
+ return
+
+ collection, sep, rights_part = line.partition(":")
+
+ rights_part = rights_part.strip()
+
+ if rights_part == "":
+ return
+
+ collection = collection.strip()
+
+ if collection == "":
+ raise ParsingError
+
+ collection = _normalize_trail_slash(collection)
+
+ rights = rights_part.split(",")
+ for right in rights:
+ user, sep, right_defs = right.strip().partition(" ")
+
+ if user == "" or right_defs == "":
+ raise ParsingError
+
+ user = user.strip()
+ right_defs = right_defs.strip()
+
+ for right_def in list(right_defs):
+
+ if right_def == 'r':
+ _append(read, collection, user)
+ elif right_def == 'w':
+ _append(write, collection, user)
+ else:
+ raise ParsingError
+
+
+
+def _append(rdict, key, value):
+ if rdict.has_key(key):
+ rlist = rdict[key]
+ rlist.append(value)
+ else:
+ rlist = [value]
+ rdict[key] = rlist
+
+
+
+def _normalize_trail_slash(s):
+ """Removes a maybe existing trailing slash"""
+ if s != "/" and s.endswith("/"):
+ s, sep, empty = s.rpartition("/")
+ return s
+
+
+_load()
diff --git a/radicale/xmlutils.py b/radicale/xmlutils.py
index 012213f..d204289 100644
--- a/radicale/xmlutils.py
+++ b/radicale/xmlutils.py
@@ -35,7 +35,7 @@ except ImportError:
import re
import xml.etree.ElementTree as ET
-from . import client, config, ical, rights
+from . import client, config, ical
NAMESPACES = {
@@ -188,6 +188,10 @@ def propfind(path, xml_request, collections, user=None):
"""Read and answer PROPFIND requests.
Read rfc4918-9.1 for info.
+
+ The collections parameter is a list of collections that are
+ to be included in the output. Rights checking has to be done
+ by the caller.
"""
# Reading request
@@ -200,9 +204,8 @@ def propfind(path, xml_request, collections, user=None):
multistatus = ET.Element(_tag("D", "multistatus"))
for collection in collections:
- if rights.read_authorized(user, collection):
- response = _propfind_response(path, collection, props, user)
- multistatus.append(response)
+ response = _propfind_response(path, collection, props, user)
+ multistatus.append(response)
return _pretty_xml(multistatus)
diff --git a/setup.py b/setup.py
index 6ae1dc3..21d539f 100755
--- a/setup.py
+++ b/setup.py
@@ -36,9 +36,23 @@ For further information, please visit the `Radicale Website
"""
-from distutils.core import setup
-
+from distutils.core import setup, Command
+import unittest
import radicale
+import sys
+
+
+class RunTests(Command):
+ user_options = []
+ def initialize_options(self):
+ pass
+ def finalize_options(self):
+ pass
+ def run(self):
+ tests = unittest.defaultTestLoader.discover("test/python")
+ result = unittest.TextTestRunner(stream=sys.stdout, verbosity=99)._makeResult()
+ tests.run(result)
+
# When the version is updated, ``radicale.VERSION`` must be modified.
@@ -59,6 +73,7 @@ setup(
"radicale", "radicale.auth", "radicale.rights", "radicale.storage"],
provides=["radicale"],
scripts=["bin/radicale"],
+ cmdclass={'test': RunTests},
keywords=["calendar", "addressbook", "CalDAV", "CardDAV"],
classifiers=[
"Development Status :: 4 - Beta",
diff --git a/test/python/rights/__init__.py b/test/python/rights/__init__.py
new file mode 100644
index 0000000..9c4b447
--- /dev/null
+++ b/test/python/rights/__init__.py
@@ -0,0 +1,7 @@
+'''
+Created on 09.08.2012
+
+Tests for rights-related code.
+
+@author: mj
+'''
diff --git a/test/python/rights/test_from_file.py b/test/python/rights/test_from_file.py
new file mode 100644
index 0000000..94bef23
--- /dev/null
+++ b/test/python/rights/test_from_file.py
@@ -0,0 +1,163 @@
+"""
+
+Unit test for radicale.rights.from_file.
+
+Tests reading the file. The processing is untested, yet.
+
+"""
+
+
+from radicale.rights import from_file
+import unittest
+
+
+
+class Test1(unittest.TestCase):
+
+ def testProcessEmptyLine(self):
+ """ Line with a comment """
+
+ # Input values
+ line = " "
+ read = {}
+ write = {}
+
+ try:
+ # Call SUT
+ from_file._process(line, read, write)
+ except from_file.ParsingError:
+ self.assertTrue(False)
+
+ self.assertTrue(len(read.keys()) == 0)
+ self.assertTrue(len(write.keys()) == 0)
+
+
+ def testProcessComment(self):
+ """ Line with a comment """
+
+ # Input values
+ line = "# some comment"
+ read = {}
+ write = {}
+
+ try:
+ # Call SUT
+ from_file._process(line, read, write)
+ except from_file.ParsingError:
+ self.assertTrue(False)
+
+ self.assertTrue(len(read.keys()) == 0)
+ self.assertTrue(len(write.keys()) == 0)
+
+
+ def testProcess0a(self):
+ """ Pointless line: no rights definitions """
+
+ # Input values
+ line = "/user1/collection1 : "
+ read = {}
+ write = {}
+
+ try:
+ # Call SUT
+ from_file._process(line, read, write)
+ except from_file.ParsingError:
+ self.fail("Unexpected exception")
+
+ self.assertTrue(len(read.keys()) == 0)
+ self.assertTrue(len(write.keys()) == 0)
+
+
+ def testProcess1a(self):
+ """ Malformed line: no collection definitions """
+
+ # Input values
+ line = " : a b"
+ read = {}
+ write = {}
+
+ try:
+ # Call SUT
+ from_file._process(line, read, write)
+ except from_file.ParsingError:
+ """Exception expected"""
+ else:
+ self.fail("Expected exception not raised")
+
+
+
+ def testProcess1b(self):
+ """ Malformed line: right "b" unknown """
+
+ # Input values
+ line = "/user1/collection1 : a b"
+ read = {}
+ write = {}
+
+ try:
+ # Call SUT
+ from_file._process(line, read, write)
+ except from_file.ParsingError:
+ """Exception expected"""
+ else:
+ self.fail("Expected exception not raised")
+
+
+ def testProcess1c(self):
+ """ Malformed line: user/right empty """
+
+ # Input values
+ line = "/user1/collection1 : a"
+ read = {}
+ write = {}
+
+ try:
+ # Call SUT
+ from_file._process(line, read, write)
+ except from_file.ParsingError:
+ """Exception expected"""
+ else:
+ self.fail("Expected exception not raised")
+
+
+ def testProcess2(self):
+ """Actual sensible input all of which means the same"""
+
+ lines = [
+ "/user1/collection1 : other1 r, other2 w, other6 rw",
+ "/user1/collection1/ : other1 r, other2 w, other6 rw",
+ "/user1/collection1: other1 r, other2 w, other6 rw",
+ "/user1/collection1/: other1 r, other2 w, other6 rw",
+ "/user1/collection1: other1 r, other2 w,other6 rw",
+ "/user1/collection1 :other1 r,other2 w, other6 rw",
+ "/user1/collection1\t:other1 r,\tother2 w,\tother6 rw",
+ ]
+
+ for line in lines:
+ # Input values
+ read = {}
+ write = {}
+
+ try:
+ # Call SUT
+ from_file._process(line, read, write)
+ except:
+ self.fail("unexpected exception for input %s" % line)
+
+ # Check
+ self.assertEquals(len(read.keys()), 1, "keys in %s" % line)
+ self.assertEquals(len(read.get("/user1/collection1")), 2, "rights in %s" % line)
+ self.assertTrue(read.get("/user1/collection1").count("other1"), "other1 read in %s" % line)
+ self.assertTrue(read.get("/user1/collection1").count("other6"), "other6 read in %s" % line)
+
+ self.assertEquals(len(write.keys()), 1, "keys in %s" % line)
+ self.assertEquals(len(write.get("/user1/collection1")), 2, "rights in %s" % line)
+ self.assertTrue(write.get("/user1/collection1").count("other2"), "other2 write in %s" % line)
+ self.assertTrue(write.get("/user1/collection1").count("other6"), "other6 write in %s" % line)
+
+
+
+
+
+if __name__ == "__main__":
+ unittest.main()
\ No newline at end of file