[ADD] base_dav

base_dav/tests/__init__.py

@@ -0,0 +1,3 @@
+# Copyright 2018 Therp BV <https://therp.nl>
+# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl.html).
+from . import test_base_dav, test_collection

base_dav/tests/test_base_dav.py

@@ -0,0 +1,117 @@
+# Copyright 2018 Therp BV <https://therp.nl>
+# Copyright 2019-2020 initOS GmbH <https://initos.com>
+# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl.html).
+
+from base64 import b64encode
+from unittest import mock
+from urllib.parse import urlparse
+
+from odoo.tests.common import TransactionCase
+
+from ..controllers.main import PREFIX
+from ..controllers.main import Main as Controller
+
+MODULE_PATH = "odoo.addons.base_dav"
+CONTROLLER_PATH = MODULE_PATH + ".controllers.main"
+RADICALE_PATH = MODULE_PATH + ".radicale"
+
+ADMIN_PASSWORD = "RadicalePa$$word"
+
+
+@mock.patch(CONTROLLER_PATH + ".request")
+@mock.patch(RADICALE_PATH + ".auth.request")
+@mock.patch(RADICALE_PATH + ".collection.request")
+class TestBaseDav(TransactionCase):
+    def setUp(self):
+        super().setUp()
+
+        self.collection = self.env["dav.collection"].create({
+            "name": "Test Collection",
+            "dav_type": "calendar",
+            "model_id": self.env.ref("base.model_res_users").id,
+            "domain": "[]",
+        })
+
+        self.dav_path = urlparse(self.collection.url).path.replace(PREFIX, '')
+
+        self.controller = Controller()
+        self.env.user.password_crypt = ADMIN_PASSWORD
+
+        self.test_user = self.env["res.users"].create({
+            "login": "tester",
+            "name": "tester",
+        })
+
+        self.auth_owner = self.auth_string(self.env.user, ADMIN_PASSWORD)
+        self.auth_tester = self.auth_string(self.test_user, ADMIN_PASSWORD)
+
+        patcher = mock.patch('odoo.http.request')
+        self.addCleanup(patcher.stop)
+        patcher.start()
+
+    def auth_string(self, user, password):
+        return b64encode(
+            ("%s:%s" % (user.login, password)).encode()
+        ).decode()
+
+    def init_mocks(self, coll_mock, auth_mock, req_mock):
+        req_mock.env = self.env
+        req_mock.httprequest.environ = {
+            "HTTP_AUTHORIZATION": "Basic %s" % self.auth_owner,
+            "REQUEST_METHOD": "PROPFIND",
+            "HTTP_X_SCRIPT_NAME": PREFIX,
+        }
+
+        auth_mock.env["res.users"]._login.return_value = self.env.uid
+        coll_mock.env = self.env
+
+    def check_status_code(self, response, forbidden):
+        if forbidden:
+            self.assertNotEqual(response.status_code, 403)
+        else:
+            self.assertEqual(response.status_code, 403)
+
+    def check_access(self, environ, auth_string, read, write):
+        environ.update({
+            "REQUEST_METHOD": "PROPFIND",
+            "HTTP_AUTHORIZATION": "Basic %s" % auth_string,
+        })
+        response = self.controller.handle_dav_request(self.dav_path)
+        self.check_status_code(response, read)
+
+        environ["REQUEST_METHOD"] = "PUT"
+        response = self.controller.handle_dav_request(self.dav_path)
+        self.check_status_code(response, write)
+
+    def test_well_known(self, coll_mock, auth_mock, req_mock):
+        req_mock.env = self.env
+
+        response = self.controller.handle_well_known_request()
+        self.assertEqual(response.status_code, 301)
+
+    def test_authenticated(self, coll_mock, auth_mock, req_mock):
+        self.init_mocks(coll_mock, auth_mock, req_mock)
+        environ = req_mock.httprequest.environ
+
+        self.collection.rights = "authenticated"
+
+        self.check_access(environ, self.auth_owner, read=True, write=True)
+        self.check_access(environ, self.auth_tester, read=True, write=True)
+
+    def test_owner_only(self, coll_mock, auth_mock, req_mock):
+        self.init_mocks(coll_mock, auth_mock, req_mock)
+        environ = req_mock.httprequest.environ
+
+        self.collection.rights = "owner_only"
+
+        self.check_access(environ, self.auth_owner, read=True, write=True)
+        self.check_access(environ, self.auth_tester, read=False, write=False)
+
+    def test_owner_write_only(self, coll_mock, auth_mock, req_mock):
+        self.init_mocks(coll_mock, auth_mock, req_mock)
+        environ = req_mock.httprequest.environ
+
+        self.collection.rights = "owner_write_only"
+
+        self.check_access(environ, self.auth_owner, read=True, write=True)
+        self.check_access(environ, self.auth_tester, read=True, write=False)

base_dav/tests/test_collection.py

@@ -0,0 +1,118 @@
+# Copyright 2019-2020 initOS GmbH <https://initos.com>
+# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl.html).
+from datetime import datetime, timedelta
+from unittest import mock
+
+from odoo.exceptions import MissingError
+from odoo.tests.common import TransactionCase
+from odoo.tools import DEFAULT_SERVER_DATETIME_FORMAT
+
+from ..radicale.collection import Collection
+
+
+class TestCalendar(TransactionCase):
+    def setUp(self):
+        super().setUp()
+
+        self.collection = self.env["dav.collection"].create({
+            "name": "Test Collection",
+            "dav_type": "calendar",
+            "model_id": self.env.ref("base.model_res_users").id,
+            "domain": "[]",
+        })
+
+        self.create_field_mapping(
+            "login", "base.field_res_users_login",
+            excode="result = record.login",
+            imcode="result = item.value",
+        )
+        self.create_field_mapping(
+            "name", "base.field_res_users_name",
+        )
+        self.create_field_mapping(
+            "dtstart", "base.field_res_users_create_date",
+        )
+        self.create_field_mapping(
+            "dtend", "base.field_res_users_write_date",
+        )
+
+        start = datetime.now()
+        stop = start + timedelta(hours=1)
+        self.record = self.env["res.users"].create({
+            "login": "tester",
+            "name": "Test User",
+            "create_date": start.strftime(DEFAULT_SERVER_DATETIME_FORMAT),
+            "write_date": stop.strftime(DEFAULT_SERVER_DATETIME_FORMAT),
+        })
+
+    def create_field_mapping(self, name, field_ref, imcode=None, excode=None):
+        return self.env["dav.collection.field_mapping"].create({
+            "collection_id": self.collection.id,
+            "name": name,
+            "field_id": self.env.ref(field_ref).id,
+            "mapping_type": "code" if imcode or excode else "simple",
+            "import_code": imcode,
+            "export_code": excode,
+        })
+
+    def compare_record(self, vobj, rec=None):
+        tmp = self.collection.from_vobject(vobj)
+
+        self.assertEqual((rec or self.record).login, tmp["login"])
+        self.assertEqual((rec or self.record).name, tmp["name"])
+        self.assertEqual((rec or self.record).create_date, tmp["create_date"])
+        self.assertEqual((rec or self.record).write_date, tmp["write_date"])
+
+    def test_import_export(self):
+        # Exporting and importing should result in the same record
+        vobj = self.collection.to_vobject(self.record)
+        self.compare_record(vobj)
+
+    def test_get_record(self):
+        rec = self.collection.get_record([self.record.id])
+        self.assertEqual(rec, self.record)
+
+        self.collection.field_uuid = self.env.ref(
+            "base.field_res_users_login",
+        ).id
+        rec = self.collection.get_record([self.record.login])
+        self.assertEqual(rec, self.record)
+
+    @mock.patch("odoo.addons.base_dav.radicale.collection.request")
+    def test_collection(self, request_mock):
+        request_mock.env = self.env
+        collection_url = "/%s/%s" % (self.env.user.login, self.collection.id)
+        collection = list(Collection.discover(collection_url))[0]
+
+        # Try to get the test record
+        record_url = "%s/%s" % (collection_url, self.record.id)
+        self.assertIn(record_url, collection.list())
+
+        # Get the test record using the URL and compare it
+        item = collection.get(record_url)
+        self.compare_record(item.item)
+        self.assertEqual(item.href, record_url)
+
+        # Get a non-existing record
+        self.assertFalse(collection.get(record_url + "0"))
+
+        # Get the record and alter it later
+        item = self.collection.to_vobject(self.record)
+        self.record.login = "different"
+        with self.assertRaises(AssertionError):
+            self.compare_record(item)
+
+        # Restore the record
+        item = collection.upload(record_url, item)
+        self.compare_record(item.item)
+
+        # Delete an record
+        collection.delete(item.href)
+        with self.assertRaises(MissingError):
+            self.record.name
+
+        # Create a new record
+        item = collection.upload(record_url + "0", item)
+        record = self.collection.get_record(collection._split_path(item.href))
+        self.assertNotEqual(record, self.record)
+        self.compare_record(item.item, record)