# This file is part of Radicale Server - Calendar Server # Copyright © 2012-2016 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 . """ Radicale tests with simple requests. """ import base64 import logging import os import posixpath import shutil import tempfile import pytest from radicale import Application, config from . import BaseTest from .helpers import get_file_content class BaseRequestsMixIn: """Tests with simple requests.""" def test_root(self): """GET request at "/".""" status, headers, answer = self.request("GET", "/") assert status == 200 assert "Radicale works!" in answer # Test the creation of the collection self.request("MKCOL", "/calendar.ics/") self.request( "PUT", "/calendar.ics/", "BEGIN:VCALENDAR\r\nEND:VCALENDAR") status, headers, answer = self.request("GET", "/calendar.ics/") assert "BEGIN:VCALENDAR" in answer assert "END:VCALENDAR" in answer def test_add_event(self): """Add an event.""" self.request("MKCOL", "/calendar.ics/") self.request( "PUT", "/calendar.ics/", "BEGIN:VCALENDAR\r\nEND:VCALENDAR") event = get_file_content("event1.ics") path = "/calendar.ics/event1.ics" status, headers, answer = self.request("PUT", path, event) assert status == 201 status, headers, answer = self.request("GET", path) assert "ETag" in headers.keys() assert status == 200 assert "VEVENT" in answer assert "Event" in answer assert "UID:event" in answer def test_add_todo(self): """Add a todo.""" self.request("MKCOL", "/calendar.ics/") self.request( "PUT", "/calendar.ics/", "BEGIN:VCALENDAR\r\nEND:VCALENDAR") todo = get_file_content("todo1.ics") path = "/calendar.ics/todo1.ics" status, headers, answer = self.request("PUT", path, todo) assert status == 201 status, headers, answer = self.request("GET", path) assert "ETag" in headers.keys() assert "VTODO" in answer assert "Todo" in answer assert "UID:todo" in answer def test_update(self): """Update an event.""" self.request("MKCOL", "/calendar.ics/") self.request( "PUT", "/calendar.ics/", "BEGIN:VCALENDAR\r\nEND:VCALENDAR") event = get_file_content("event1.ics") path = "/calendar.ics/event1.ics" status, headers, answer = self.request("PUT", path, event) assert status == 201 status, headers, answer = self.request("GET", path) assert "ETag" in headers.keys() assert status == 200 assert "VEVENT" in answer assert "Event" in answer assert "UID:event" in answer assert "DTSTART;TZID=Europe/Paris:20130901T180000" in answer assert "DTEND;TZID=Europe/Paris:20130901T190000" in answer # Then we send another PUT request event = get_file_content("event1-prime.ics") status, headers, answer = self.request("PUT", path, event) assert status == 201 status, headers, answer = self.request("GET", "/calendar.ics/") assert answer.count("BEGIN:VEVENT") == 1 status, headers, answer = self.request("GET", path) assert "ETag" in headers.keys() assert status == 200 assert "VEVENT" in answer assert "Event" in answer assert "UID:event" in answer assert "DTSTART;TZID=Europe/Paris:20130901T180000" not in answer assert "DTEND;TZID=Europe/Paris:20130901T190000" not in answer assert "DTSTART;TZID=Europe/Paris:20140901T180000" in answer assert "DTEND;TZID=Europe/Paris:20140901T210000" in answer def test_put_whole_collection(self): """Create and overwrite a whole collection.""" event = get_file_content("event1.ics") status, headers, answer = self.request("PUT", "/calendar.ics/", event) assert status == 201 status, headers, answer = self.request( "PUT", "/calendar.ics/event1.ics", event) assert status == 201 # Overwrite status, headers, answer = self.request("PUT", "/calendar.ics/", event) assert status == 201 status, headers, answer = self.request( "GET", "/calendar.ics/event1.ics") assert status == 404 def test_delete(self): """Delete an event.""" self.request("MKCOL", "/calendar.ics/") self.request( "PUT", "/calendar.ics/", "BEGIN:VCALENDAR\r\nEND:VCALENDAR") event = get_file_content("event1.ics") path = "/calendar.ics/event1.ics" status, headers, answer = self.request("PUT", path, event) # Then we send a DELETE request status, headers, answer = self.request("DELETE", path) assert status == 200 assert "href>%s/calendar.ics///%s%s%s#BADA55 """) assert answer.count("") == 1 status, headers, answer = self.request("GET", "/calendar.ics/") assert answer.count("BEGIN:VEVENT") == 2 def _test_filter(self, filters, kind="event", items=1): filters_text = "".join( "%s" % filter_ for filter_ in filters) self.request("MKCOL", "/calendar.ics/") self.request( "PUT", "/calendar.ics/", "BEGIN:VCALENDAR\r\nEND:VCALENDAR") for i in range(items): filename = "{}{}.ics".format(kind, i + 1) event = get_file_content(filename) self.request("PUT", "/calendar.ics/{}".format(filename), event) status, headers, answer = self.request( "REPORT", "/calendar.ics", """ %s """ % filters_text) return answer def test_calendar_empty_filter(self): self._test_filter([""]) def test_calendar_tag_filter(self): """Report request with tag-based filter on calendar.""" assert "href>/calendar.ics/event1.ics"""]) def test_item_tag_filter(self): """Report request with tag-based filter on an item.""" assert "href>/calendar.ics/event1.ics """]) assert "href>/calendar.ics/event1.ics """]) def test_item_not_tag_filter(self): """Report request with tag-based is-not filter on an item.""" assert "href>/calendar.ics/event1.ics """]) assert "href>/calendar.ics/event1.ics """]) def test_item_prop_filter(self): """Report request with prop-based filter on an item.""" assert "href>/calendar.ics/event1.ics """]) assert "href>/calendar.ics/event1.ics """]) def test_item_not_prop_filter(self): """Report request with prop-based is-not filter on an item.""" assert "href>/calendar.ics/event1.ics """]) assert "href>/calendar.ics/event1.ics """]) def test_mutiple_filters(self): """Report request with multiple filters on an item.""" assert "href>/calendar.ics/event1.ics """, """ """]) assert "href>/calendar.ics/event1.ics """, """ """]) assert "href>/calendar.ics/event1.ics """]) def test_text_match_filter(self): """Report request with text-match filter on calendar.""" assert "href>/calendar.ics/event1.ics event """]) assert "href>/calendar.ics/event1.ics event """]) assert "href>/calendar.ics/event1.ics unknown """]) assert "href>/calendar.ics/event1.ics event """]) def test_param_filter(self): """Report request with param-filter on calendar.""" assert "href>/calendar.ics/event1.ics ACCEPTED """]) assert "href>/calendar.ics/event1.ics UNKNOWN """]) assert "href>/calendar.ics/event1.ics """]) assert "href>/calendar.ics/event1.ics """]) def test_time_range_filter_events(self): """Report request with time-range filter on events.""" answer = self._test_filter([""" """], "event", items=5) assert "href>/calendar.ics/event1.ics/calendar.ics/event2.ics/calendar.ics/event3.ics/calendar.ics/event4.ics/calendar.ics/event5.ics """], items=5) assert "href>/calendar.ics/event1.ics/calendar.ics/event2.ics/calendar.ics/event3.ics/calendar.ics/event4.ics/calendar.ics/event5.ics """], items=5) assert "href>/calendar.ics/event1.ics/calendar.ics/event2.ics/calendar.ics/event3.ics/calendar.ics/event4.ics/calendar.ics/event5.ics """], items=5) assert "href>/calendar.ics/event1.ics/calendar.ics/event2.ics/calendar.ics/event3.ics/calendar.ics/event4.ics/calendar.ics/event5.ics """], items=5) assert "href>/calendar.ics/event1.ics/calendar.ics/event2.ics/calendar.ics/event3.ics/calendar.ics/event4.ics/calendar.ics/event5.ics """], items=5) assert "href>/calendar.ics/event1.ics/calendar.ics/event2.ics/calendar.ics/event3.ics/calendar.ics/event4.ics/calendar.ics/event5.ics """], "event", items=2) assert "href>/calendar.ics/event1.ics/calendar.ics/event2.ics """], "event", items=2) assert "href>/calendar.ics/event1.ics/calendar.ics/event2.ics """], "event", items=2) assert "href>/calendar.ics/event1.ics/calendar.ics/event2.ics """], "event", items=2) assert "href>/calendar.ics/event1.ics/calendar.ics/event2.ics """], "todo", items=8) assert "href>/calendar.ics/todo1.ics/calendar.ics/todo2.ics/calendar.ics/todo3.ics/calendar.ics/todo4.ics/calendar.ics/todo5.ics/calendar.ics/todo6.ics/calendar.ics/todo7.ics/calendar.ics/todo8.ics """], "todo", items=8) assert "href>/calendar.ics/todo1.ics/calendar.ics/todo2.ics/calendar.ics/todo3.ics/calendar.ics/todo4.ics/calendar.ics/todo5.ics/calendar.ics/todo6.ics/calendar.ics/todo7.ics/calendar.ics/todo8.ics """], "todo", items=8) assert "href>/calendar.ics/todo2.ics """], "todo", items=8) assert "href>/calendar.ics/todo2.ics """], "todo", items=8) assert "href>/calendar.ics/todo3.ics """], "todo", items=8) assert "href>/calendar.ics/todo7.ics """], "todo", items=2) assert "href>/calendar.ics/todo1.ics/calendar.ics/todo2.ics """], "todo", items=2) assert "href>/calendar.ics/todo1.ics/calendar.ics/todo2.ics """], "todo", items=2) assert "href>/calendar.ics/todo1.ics/calendar.ics/todo2.ics """], "todo", items=2) assert "href>/calendar.ics/todo1.ics/calendar.ics/todo2.ics """], "journal", items=3) assert "href>/calendar.ics/journal1.ics/calendar.ics/journal2.ics/calendar.ics/journal3.ics """], "journal", items=3) assert "href>/calendar.ics/journal1.ics/calendar.ics/journal2.ics/calendar.ics/journal3.ics """], "journal", items=3) assert "href>/calendar.ics/journal1.ics/calendar.ics/journal2.ics/calendar.ics/journal3.ics """], "journal", items=3) assert "href>/calendar.ics/journal1.ics/calendar.ics/journal2.ics/calendar.ics/journal3.ics """], "journal", items=3) assert "href>/calendar.ics/journal1.ics/calendar.ics/journal2.ics/calendar.ics/journal3.ics """], "journal", items=2) assert "href>/calendar.ics/journal1.ics/calendar.ics/journal2.ics """], "journal", items=2) assert "href>/calendar.ics/journal1.ics/calendar.ics/journal2.ics """], "journal", items=2) assert "href>/calendar.ics/journal1.ics/calendar.ics/journal2.ics """) assert status == 207 assert "href>%s<" % event_path in answer def test_authorization(self): authorization = "Basic " + base64.b64encode(b"user:").decode() status, headers, answer = self.request( "PROPFIND", "/", """ """, HTTP_AUTHORIZATION=authorization) assert status == 207 assert "href>/user/<" in answer def test_authentication(self): """Test if server sends authentication request.""" self.configuration.set("rights", "type", "owner_only") self.application = Application(self.configuration, self.logger) status, headers, answer = self.request("MKCOL", "/user/") assert status in (401, 403) assert headers.get("WWW-Authenticate") def test_principal_collection_creation(self): """Verify existence of the principal collection.""" status, headers, answer = self.request( "PROPFIND", "/user/", REMOTE_USER="user") assert status == 207 def test_existence_of_root_collections(self): """Verify that the root collection always exists.""" # Use PROPFIND because GET returns message status, headers, answer = self.request("PROPFIND", "/") assert status == 207 # it should still exist after deletion self.request("DELETE", "/") status, headers, answer = self.request("PROPFIND", "/") assert status == 207 def test_fsync(self): """Create a directory and file with syncing enabled.""" self.configuration.set("storage", "fsync", "True") status, headers, answer = self.request("MKCALENDAR", "/calendar.ics/") assert status == 201 def test_hook(self): """Run hook.""" self.configuration.set( "storage", "hook", "mkdir %s" % os.path.join("collection-root", "created_by_hook")) status, headers, answer = self.request("MKCOL", "/calendar.ics/") assert status == 201 status, headers, answer = self.request("PROPFIND", "/created_by_hook/") assert status == 207 def test_hook_read_access(self): """Verify that hook is not run for read accesses.""" self.configuration.set( "storage", "hook", "mkdir %s" % os.path.join("collection-root", "created_by_hook")) status, headers, answer = self.request("GET", "/") assert status == 200 status, headers, answer = self.request("GET", "/created_by_hook/") assert status == 404 @pytest.mark.skipif(os.system("type flock") != 0, reason="flock command not found") def test_hook_storage_locked(self): """Verify that the storage is locked when the hook runs.""" self.configuration.set( "storage", "hook", "flock -n .Radicale.lock || exit 0; exit 1") status, headers, answer = self.request("MKCOL", "/calendar.ics/") assert status == 201 def test_hook_principal_collection_creation(self): """Verify that the hooks runs when a new user is created.""" self.configuration.set( "storage", "hook", "mkdir %s" % os.path.join("collection-root", "created_by_hook")) status, headers, answer = self.request("GET", "/", REMOTE_USER="user") assert status == 200 status, headers, answer = self.request("PROPFIND", "/created_by_hook/") assert status == 207 def test_hook_fail(self): """Verify that a request fails if the hook fails.""" self.configuration.set("storage", "hook", "exit 1") try: status, headers, answer = self.request("MKCOL", "/calendar.ics/") assert status != 201 except Exception: pass class BaseFileSystemTest(BaseTest): """Base class for filesystem backend tests.""" storage_type = None def setup(self): self.configuration = config.load() self.configuration.set("storage", "type", self.storage_type) self.logger = logging.getLogger("radicale_test") self.colpath = tempfile.mkdtemp() self.configuration.set("storage", "filesystem_folder", self.colpath) # Disable syncing to disk for better performance self.configuration.set("storage", "fsync", "False") # Required on Windows, doesn't matter on Unix self.configuration.set("storage", "close_lock_file", "True") self.application = Application(self.configuration, self.logger) def teardown(self): shutil.rmtree(self.colpath) class TestMultiFileSystem(BaseFileSystemTest, BaseRequestsMixIn): """Test BaseRequests on multifilesystem.""" storage_type = "multifilesystem" class TestCustomStorageSystem(BaseFileSystemTest): """Test custom backend loading.""" storage_type = "tests.custom.storage" def test_root(self): """A simple test to verify that the custom backend works.""" BaseRequestsMixIn.test_root(self)