TechnoLibre/bemade-addons
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

First commit for caldav_sync

Browse source
This commit is contained in:
Marc Durepos 2024-05-23 08:50:09 -04:00
parent ca348b11a2
commit a610f1ba69
12 changed files with 421 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
caldav_sync/__init__.py Normal file
View file

@ -0,0 +1,3 @@
# -*- coding: utf-8 -*-
from . import models
from . import controllers

25
caldav_sync/__manifest__.py Normal file
View file

@ -0,0 +1,25 @@
#
# Bemade Inc.
#
# Copyright (C) 2023-June Bemade Inc. (<https://www.bemade.org>).
# Author: Marc Durepos (Contact : mdurepos@durpro.com)
#
# This program is under the terms of the GNU Lesser General Public License (LGPL-3)
# For details, visit https://www.gnu.org/licenses/lgpl-3.0.en.html
{
'name': 'CalDAV Sync',
'version': '1.0',
'category': 'Calendar',
'summary': 'Synchronize Odoo Calendar with a CalDAV Server',
'description': """
This module allows Odoo to synchronize calendar events with a CalDAV server.
""",
'author': 'Your Name',
'depends': ['base', 'calendar'],
'data': [
'data/caldav_sync_data.xml',
],
'installable': True,
'application': True,
}

2
caldav_sync/controllers/__init__.py Normal file
View file

@ -0,0 +1,2 @@
# -*- coding: utf-8 -*-
from . import main

56
caldav_sync/controllers/main.py Normal file
View file

@ -0,0 +1,56 @@
from odoo import http
from odoo.http import request
import logging
import caldav
from caldav.elements import dav, cdav
from datetime import datetime
_logger = logging.getLogger(__name__)
class CaldavController(http.Controller):
@http.route('/caldav_sync/sync', type='json', auth='user')
def sync(self, **kwargs):
# Fetch user credentials and server settings from Odoo
user = request.env.user
server_url = user.caldav_server_url
username = user.caldav_username
password = user.caldav_password
# Connect to the CalDAV server
client = caldav.DAVClient(url=server_url, username=username, password=password)
principal = client.principal()
calendars = principal.calendars()
for calendar in calendars:
events = calendar.events()
for event in events:
ical = event.icalendar()
self.sync_event(ical)
return {'status': 'success', 'message': 'Synchronization completed'}
def sync_event(self, ical):
event = ical.subcomponents[0]
uid = event['UID']
start = event['DTSTART'].dt
end = event['DTEND'].dt
summary = event['SUMMARY']
description = event.get('DESCRIPTION', '')
location = event.get('LOCATION', '')
# Search for an existing event in Odoo
odoo_event = request.env['calendar.event'].search([('caldav_uid', '=', uid)], limit=1)
values = {
'name': summary,
'start': start,
'stop': end,
'description': description,
'location': location,
}
if odoo_event:
odoo_event.write(values)
else:
values['caldav_uid'] = uid
request.env['calendar.event'].create(values)

15
caldav_sync/data/caldav_sync_data.xml Normal file
View file

@ -0,0 +1,15 @@
<odoo>
<data noupdate="1">
<record id="ir_cron_caldav_sync" model="ir.cron">
<field name="name">CalDAV Sync</field>
<field name="model_id" ref="model_calendar_event"/>
<field name="state">code</field>
<field name="code">model.poll_caldav_server()</field>
<field name="interval_number">1</field>
<field name="interval_type">minutes</field>
<field name="numbercall">-1</field>
<field name="doall" eval="False"/>
<field name="active" eval="True"/>
</record>
</data>
</odoo>

8
caldav_sync/data/neutralize.sql Normal file
View file

@ -0,0 +1,8 @@
-- Neutralize CalDAV synchronization by setting credentials to NULL
UPDATE res_users
SET caldav_server_url = NULL,
caldav_username = NULL,
caldav_password = NULL
WHERE caldav_server_url IS NOT NULL
OR caldav_username IS NOT NULL
OR caldav_password IS NOT NULL;

3
caldav_sync/models/__init__.py Normal file
View file

@ -0,0 +1,3 @@
# -*- coding: utf-8 -*-
from . import calendar_event
from . import res_users

134
caldav_sync/models/calendar_event.py Normal file
View file

@ -0,0 +1,134 @@
from odoo import models, fields, api
import caldav
import logging
_logger = logging.getLogger(__name__)
class CalendarEvent(models.Model):
_inherit = 'calendar.event'
caldav_uid = fields.Char('CalDAV UID')
@api.model
def create(self, values):
event = super(CalendarEvent, self).create(values)
if event._is_caldav_enabled():
try:
event.sync_to_caldav()
except Exception as e:
_logger.error(f"Failed to sync event to CalDAV: {e}")
return event
def write(self, values):
result = super(CalendarEvent, self).write(values)
if self._is_caldav_enabled():
try:
self.sync_to_caldav()
except Exception as e:
_logger.error(f"Failed to sync event to CalDAV: {e}")
return result
def unlink(self):
if self._is_caldav_enabled():
try:
self.remove_from_caldav()
except Exception as e:
_logger.error(f"Failed to remove event from CalDAV: {e}")
return super(CalendarEvent, self).unlink()
def sync_to_caldav(self):
for event in self:
if event._is_caldav_enabled():
client = event._get_caldav_client()
calendar = client.principal().calendars()[0] # Assuming the first calendar
vevent = calendar.add_event(event._get_icalendar())
event.caldav_uid = vevent.vobject_instance.vevent.uid.value
def remove_from_caldav(self):
for event in self:
if event.caldav_uid and event._is_caldav_enabled():
client = event._get_caldav_client()
calendar = client.principal().calendars()[0] # Assuming the first calendar
caldav_event = calendar.event_by_uid(event.caldav_uid)
if caldav_event:
caldav_event.delete()
@api.model
def poll_caldav_server(self):
"""Poll the CalDAV server and synchronize events for all users"""
_logger.info('Polling CalDAV server for updates...')
users = self.env['res.users'].search([])
for user in users:
if user._is_caldav_enabled():
try:
_logger.info(f'Polling CalDAV server for user {user.name}...')
client = user._get_caldav_client()
calendar = client.principal().calendars()[0] # Assuming the first calendar
events = calendar.events()
# Collect all current CalDAV UIDs for this user
current_uids = set(event.caldav_uid for event in self.search([('caldav_uid', '!=', False), ('create_uid', '=', user.id)]))
for event in events:
ical = event.icalendar()
uid = ical.subcomponents[0]['UID']
current_uids.discard(uid) # Remove from the set of current UIDs
self.sync_event_from_ical(ical, user)
# Any UIDs remaining in current_uids are events that have been deleted on the CalDAV server
for uid in current_uids:
odoo_event = self.search([('caldav_uid', '=', uid), ('create_uid', '=', user.id)], limit=1)
if odoo_event:
odoo_event.unlink()
except Exception as e:
_logger.error(f"Failed to poll CalDAV server for user {user.name}: {e}")
def sync_event_from_ical(self, ical, user):
event = ical.subcomponents[0]
uid = event['UID']
start = event['DTSTART'].dt
end = event['DTEND'].dt
summary = event['SUMMARY']
description = event.get('DESCRIPTION', '')
location = event.get('LOCATION', '')
odoo_event = self.search([('caldav_uid', '=', uid), ('create_uid', '=', user.id)], limit=1)
values = {
'name': summary,
'start': start,
'stop': end,
'description': description,
'location': location,
'create_uid': user.id,
}
if odoo_event:
odoo_event.write(values)
else:
values['caldav_uid'] = uid
self.create(values)
def _get_caldav_client(self):
user = self.env.user
return caldav.DAVClient(url=user.caldav_server_url, username=user.caldav_username, password=user.caldav_password)
def _is_caldav_enabled(self):
user = self.env.user
return bool(user.caldav_server_url and user.caldav_username and user.caldav_password)
def _get_icalendar(self):
vevent = f"""
BEGIN:VCALENDAR
VERSION:2.0
BEGIN:VEVENT
UID:{self.caldav_uid}
DTSTAMP:{self.start.strftime('%Y%m%dT%H%M%SZ')}
DTSTART:{self.start.strftime('%Y%m%dT%H%M%SZ')}
DTEND:{self.stop.strftime('%Y%m%dT%H%M%SZ')}
SUMMARY:{self.name}
DESCRIPTION:{self.description or ''}
LOCATION:{self.location or ''}
END:VEVENT
END:VCALENDAR
"""
return vevent

15
caldav_sync/models/res_users.py Normal file
View file

@ -0,0 +1,15 @@
from odoo import models, fields
import caldav
class ResUsers(models.Model):
_inherit = 'res.users'
caldav_server_url = fields.Char('CalDAV Server URL')
caldav_username = fields.Char('CalDAV Username')
caldav_password = fields.Char('CalDAV Password')
def _get_caldav_client(self):
return caldav.DAVClient(url=self.caldav_server_url, username=self.caldav_username, password=self.caldav_password)
def _is_caldav_enabled(self):
return bool(self.caldav_server_url and self.caldav_username and self.caldav_password)

1
caldav_sync/tests/__init__.py Normal file
View file

@ -0,0 +1 @@
from . import test_caldav_sync

142
caldav_sync/tests/test_caldav_sync.py Normal file
View file

@ -0,0 +1,142 @@
from odoo.tests.common import TransactionCase
from unittest.mock import patch, MagicMock
import caldav
class TestCaldavSync(TransactionCase):
def setUp(self):
super(TestCaldavSync, self).setUp()
self.user = self.env['res.users'].create({
'name': 'Test User',
'login': 'testuser',
'caldav_server_url': 'http://testserver/caldav',
'caldav_username': 'testuser',
'caldav_password': 'password',
})
self.env = self.env(context=dict(self.env.context, no_reset_password=True))
def test_create_caldav_event(self):
with patch('caldav.DAVClient') as MockClient:
mock_client = MockClient.return_value
mock_principal = mock_client.principal.return_value
mock_calendar = mock_principal.calendars.return_value[0]
mock_event = MagicMock()
mock_event.vobject_instance.vevent.uid.value = 'test-uid-12345'
mock_calendar.add_event.return_value = mock_event
event = self.env['calendar.event'].create({
'name': 'Test Event',
'start': '2024-05-22 10:00:00',
'stop': '2024-05-22 11:00:00',
'description': 'This is a test event',
'location': 'Test Location',
'create_uid': self.user.id,
})
self.assertEqual(event.name, 'Test Event')
self.assertEqual(event.caldav_uid, 'test-uid-12345')
def test_update_caldav_event(self):
with patch('caldav.DAVClient') as MockClient:
mock_client = MockClient.return_value
mock_principal = mock_client.principal.return_value
mock_calendar = mock_principal.calendars.return_value[0]
mock_event = MagicMock()
mock_event.vobject_instance.vevent.uid.value = 'test-uid-12345'
mock_calendar.add_event.return_value = mock_event
event = self.env['calendar.event'].create({
'name': 'Test Event',
'start': '2024-05-22 10:00:00',
'stop': '2024-05-22 11:00:00',
'description': 'This is a test event',
'location': 'Test Location',
'create_uid': self.user.id,
})
with patch.object(event, 'sync_to_caldav') as mock_sync_to_caldav:
event.write({
'name': 'Updated Test Event',
'start': '2024-05-22 12:00:00',
'stop': '2024-05-22 13:00:00',
})
mock_sync_to_caldav.assert_called_once()
self.assertEqual(event.name, 'Updated Test Event')
self.assertEqual(event.start, '2024-05-22 12:00:00')
def test_delete_caldav_event(self):
with patch('caldav.DAVClient') as MockClient:
mock_client = MockClient.return_value
mock_principal = mock_client.principal.return_value
mock_calendar = mock_principal.calendars.return_value[0]
mock_event = MagicMock()
mock_event.vobject_instance.vevent.uid.value = 'test-uid-12345'
mock_calendar.add_event.return_value = mock_event
event = self.env['calendar.event'].create({
'name': 'Test Event',
'start': '2024-05-22 10:00:00',
'stop': '2024-05-22 11:00:00',
'description': 'This is a test event',
'location': 'Test Location',
'create_uid': self.user.id,
})
with patch.object(event, 'remove_from_caldav') as mock_remove_from_caldav:
event.unlink()
mock_remove_from_caldav.assert_called_once()
def test_poll_caldav_server(self):
with patch('caldav.DAVClient') as MockClient:
mock_client = MockClient.return_value
mock_principal = mock_client.principal.return_value
mock_calendar = mock_principal.calendars.return_value[0]
mock_event = MagicMock()
mock_event.icalendar.return_value = """
BEGIN:VCALENDAR
VERSION:2.0
BEGIN:VEVENT
UID:test-uid-12345
DTSTAMP:20240522T100000Z
DTSTART:20240522T100000Z
DTEND:20240522T110000Z
SUMMARY:Polled Event
DESCRIPTION:This event was polled from CalDAV
LOCATION:Polled Location
END:VEVENT
END:VCALENDAR
"""
mock_calendar.events.return_value = [mock_event]
with patch.object(self.env['calendar.event'], 'sync_event_from_ical') as mock_sync_event_from_ical:
self.env['calendar.event'].poll_caldav_server()
mock_sync_event_from_ical.assert_called_once()
def test_poll_caldav_server_with_exception(self):
with patch('caldav.DAVClient') as MockClient:
mock_client = MockClient.return_value
mock_principal = mock_client.principal.return_value
mock_calendar = mock_principal.calendars.return_value[0]
mock_event = MagicMock()
mock_event.icalendar.return_value = """
BEGIN:VCALENDAR
VERSION:2.0
BEGIN:VEVENT
UID:test-uid-12345
DTSTAMP:20240522T100000Z
DTSTART:20240522T100000Z
DTEND:20240522T110000Z
SUMMARY:Polled Event
DESCRIPTION:This event was polled from CalDAV
LOCATION:Polled Location
END:VEVENT
END:VCALENDAR
"""
mock_calendar.events.return_value = [mock_event]
mock_client.principal.side_effect = Exception('Invalid credentials')
with patch.object(self.env['calendar.event'], '_logger') as mock_logger:
self.env['calendar.event'].poll_caldav_server()
mock_logger.error.assert_any_call('Failed to poll CalDAV server for user Test User: Invalid credentials')

17
caldav_sync/views/res_users_views.xml Normal file
View file

@ -0,0 +1,17 @@
<?xml version="1.0" encoding="UTF-8" ?>
<odoo>
<record id="view_users_form_caldav_sync" model="ir.ui.view">
<field name="name">res.users.form.caldav.sync</field>
<field name="model">res.users</field>
<field name="inherit_id" ref="base.view_users_form"/>
<field name="arch" type="xml">
<xpath expr="//group[@name='preferences']" position="inside">
<group string="CalDAV Sync">
<field name="caldav_server_url"/>
<field name="caldav_username"/>
<field name="caldav_password"/>
</group>
</xpath>
</field>
</record>
</odoo>