From 2421da1d63cddd4615ff3d4ce1dc84cddd8e7548 Mon Sep 17 00:00:00 2001 From: "I. Alcaraz Salazar" Date: Sun, 15 Feb 2026 09:04:21 +0000 Subject: [PATCH] feat: add Odoo JSON-RPC client with projects, tasks, and calendar methods Co-Authored-By: Claude Opus 4.6 --- backend/modules/odoo_client.py | 91 ++++++++++++++++++++++ backend/tests/test_odoo_client.py | 124 ++++++++++++++++++++++++++++++ 2 files changed, 215 insertions(+) create mode 100644 backend/modules/odoo_client.py create mode 100644 backend/tests/test_odoo_client.py diff --git a/backend/modules/odoo_client.py b/backend/modules/odoo_client.py new file mode 100644 index 0000000..7e2e1b0 --- /dev/null +++ b/backend/modules/odoo_client.py @@ -0,0 +1,91 @@ +from typing import Any + +import httpx + + +class OdooClient: + def __init__(self, url: str, database: str, username: str, password: str): + self.url = url.rstrip("/") + self.database = database + self.username = username + self.password = password + self.uid: int | None = None + + async def _jsonrpc(self, endpoint: str, params: dict[str, Any]) -> Any: + payload = { + "jsonrpc": "2.0", + "method": "call", + "params": params, + "id": 1, + } + async with httpx.AsyncClient() as client: + response = await client.post( + f"{self.url}{endpoint}", + json=payload, + timeout=30.0, + ) + response.raise_for_status() + result = response.json() + if "error" in result: + raise Exception(f"Odoo error: {result['error']}") + return result.get("result") + + async def authenticate(self) -> int: + self.uid = await self._jsonrpc("/web/session/authenticate", { + "db": self.database, + "login": self.username, + "password": self.password, + }) + return self.uid + + async def search_read( + self, + model: str, + domain: list | None = None, + fields: list[str] | None = None, + limit: int = 0, + order: str = "", + ) -> list[dict[str, Any]]: + return await self._jsonrpc("/web/dataset/call_kw", { + "model": model, + "method": "search_read", + "args": [domain or []], + "kwargs": { + "fields": fields or [], + "limit": limit, + "order": order, + }, + }) + + async def get_projects(self) -> list[dict[str, Any]]: + return await self.search_read( + model="project.project", + domain=[("active", "=", True)], + fields=["name", "task_count", "color"], + ) + + async def get_tasks(self, project_id: int | None = None) -> list[dict[str, Any]]: + domain: list = [("active", "=", True)] + if project_id: + domain.append(("project_id", "=", project_id)) + return await self.search_read( + model="project.task", + domain=domain, + fields=[ + "name", "project_id", "stage_id", "user_ids", + "priority", "date_deadline", "kanban_state", + ], + ) + + async def get_calendar_events( + self, date_from: str, date_to: str + ) -> list[dict[str, Any]]: + return await self.search_read( + model="calendar.event", + domain=[ + ("start", ">=", f"{date_from} 00:00:00"), + ("start", "<=", f"{date_to} 23:59:59"), + ], + fields=["name", "start", "stop", "location", "description", "partner_ids"], + order="start asc", + ) diff --git a/backend/tests/test_odoo_client.py b/backend/tests/test_odoo_client.py new file mode 100644 index 0000000..409502d --- /dev/null +++ b/backend/tests/test_odoo_client.py @@ -0,0 +1,124 @@ +import json +import pytest +from unittest.mock import AsyncMock, patch, MagicMock +from modules.odoo_client import OdooClient + + +@pytest.fixture +def odoo(): + return OdooClient( + url="http://localhost:8069", + database="test_db", + username="admin", + password="admin", + ) + + +@pytest.mark.asyncio +async def test_authenticate(odoo): + mock_response = MagicMock() + mock_response.json.return_value = {"jsonrpc": "2.0", "result": 2} + mock_response.raise_for_status = MagicMock() + + with patch("httpx.AsyncClient.post", new_callable=AsyncMock, return_value=mock_response): + uid = await odoo.authenticate() + assert uid == 2 + assert odoo.uid == 2 + + +@pytest.mark.asyncio +async def test_search_read(odoo): + odoo.uid = 2 + + mock_response = MagicMock() + mock_response.json.return_value = { + "jsonrpc": "2.0", + "result": [ + {"id": 1, "name": "Project A"}, + {"id": 2, "name": "Project B"}, + ], + } + mock_response.raise_for_status = MagicMock() + + with patch("httpx.AsyncClient.post", new_callable=AsyncMock, return_value=mock_response): + results = await odoo.search_read( + model="project.project", + domain=[("active", "=", True)], + fields=["name"], + ) + assert len(results) == 2 + assert results[0]["name"] == "Project A" + + +@pytest.mark.asyncio +async def test_get_projects(odoo): + odoo.uid = 2 + + mock_response = MagicMock() + mock_response.json.return_value = { + "jsonrpc": "2.0", + "result": [ + {"id": 1, "name": "Web App", "task_count": 5}, + ], + } + mock_response.raise_for_status = MagicMock() + + with patch("httpx.AsyncClient.post", new_callable=AsyncMock, return_value=mock_response): + projects = await odoo.get_projects() + assert len(projects) == 1 + assert projects[0]["name"] == "Web App" + + +@pytest.mark.asyncio +async def test_get_tasks(odoo): + odoo.uid = 2 + + mock_response = MagicMock() + mock_response.json.return_value = { + "jsonrpc": "2.0", + "result": [ + { + "id": 1, + "name": "Implement login", + "project_id": [1, "Web App"], + "stage_id": [1, "In Progress"], + "user_ids": [2], + "priority": "1", + "date_deadline": "2026-02-20", + }, + ], + } + mock_response.raise_for_status = MagicMock() + + with patch("httpx.AsyncClient.post", new_callable=AsyncMock, return_value=mock_response): + tasks = await odoo.get_tasks() + assert len(tasks) == 1 + assert tasks[0]["name"] == "Implement login" + + +@pytest.mark.asyncio +async def test_get_calendar_events(odoo): + odoo.uid = 2 + + mock_response = MagicMock() + mock_response.json.return_value = { + "jsonrpc": "2.0", + "result": [ + { + "id": 1, + "name": "Standup", + "start": "2026-02-15 09:00:00", + "stop": "2026-02-15 09:30:00", + "location": "Sala de juntas", + }, + ], + } + mock_response.raise_for_status = MagicMock() + + with patch("httpx.AsyncClient.post", new_callable=AsyncMock, return_value=mock_response): + events = await odoo.get_calendar_events( + date_from="2026-02-15", + date_to="2026-02-22", + ) + assert len(events) == 1 + assert events[0]["name"] == "Standup"