import importlib.util import os import sys import tempfile import types import unittest from configparser import ConfigParser from unittest.mock import Mock, patch PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) SETUP_PATH = os.path.join(PROJECT_ROOT, "setup.py") def load_setup_module(): setuphelpers = types.ModuleType("setuphelpers") setuphelpers.makepath = lambda *parts: os.path.join(*parts) setuphelpers.isfile = os.path.isfile setuphelpers.WAPT = types.SimpleNamespace(private_dir=tempfile.gettempdir()) sys.modules["setuphelpers"] = setuphelpers spec = importlib.util.spec_from_file_location("ticketing_setup", SETUP_PATH) module = importlib.util.module_from_spec(spec) sys.modules["ticketing_setup"] = module spec.loader.exec_module(module) return module ticketing = load_setup_module() def make_config(): config = ConfigParser() config.add_section("odoo") config.set("odoo", "url", "https://odoo.example.test") config.set("odoo", "database", "test-db") config.set("odoo", "username", "test@example.test") config.set("odoo", "api_key", "test-api-key") config.set("odoo", "model", "helpdesk.ticket") config.set("odoo", "limit", "0") config.set("odoo", "order", "id asc") config.add_section("rocket") config.set("rocket", "webhook_url", "https://rocket.example.test/webhook-test") config.add_section("http") config.set("http", "timeout", "5") config.add_section("notification") config.set("notification", "max_message_chars", "3500") return config class TicketingStateTest(unittest.TestCase): def test_build_tickets_state_uses_string_ids_for_json_keys(self): state = ticketing.build_tickets_state([ {"id": 12, "write_date": "2026-07-24 10:00:00"}, {"id": 34, "write_date": "2026-07-24 11:00:00"}, ]) self.assertEqual(state, { "12": "2026-07-24 10:00:00", "34": "2026-07-24 11:00:00", }) def test_get_updated_ticket_ids_detects_new_and_changed_tickets(self): previous_state = { "1": "2026-07-24 09:00:00", "2": "2026-07-24 09:30:00", } tickets = [ {"id": 1, "write_date": "2026-07-24 09:00:00"}, {"id": 2, "write_date": "2026-07-24 10:00:00"}, {"id": 3, "write_date": "2026-07-24 10:30:00"}, ] self.assertEqual(ticketing.get_updated_ticket_ids(tickets, previous_state), [2, 3]) def test_build_notification_message_is_verbose_and_limited(self): previous_state = { "1": "2026-07-24 09:00:00", } tickets = [ { "id": 1, "name": "Probleme imprimante", "stage_id": [4, "En cours"], "partner_id": [8, "Comitari"], "user_id": [9, "Support"], "write_date": "2026-07-24 10:00:00", }, { "id": 2, "name": "Nouveau ticket", "stage_id": False, "partner_id": False, "user_id": False, "write_date": "2026-07-24 11:00:00", }, ] message = ticketing.build_notification_message( tickets, previous_state, 1, 3500, "{odoo_url}/web#id={id}&model={model}&view_type=form", "https://odoo.example.test", "helpdesk.ticket", ) self.assertIn("Les tickets suivants ont ete mis a jour depuis la derniere verification :", message) self.assertIn( "1 : Sujet : Probleme imprimante gere par Support : " "[cliquer ici pour ouvrir le ticket]" "(https://odoo.example.test/web#id=1&model=helpdesk.ticket&view_type=form)", message, ) self.assertIn("1 ticket(s) supplementaire(s) non affiche(s) sur 2 au total", message) self.assertNotIn("Nouveau ticket", message) def test_build_notification_message_is_limited_by_message_size(self): tickets = [ {"id": 1, "name": "Premier ticket avec sujet assez long", "user_id": False}, {"id": 2, "name": "Deuxieme ticket avec sujet assez long", "user_id": False}, ] message = ticketing.build_notification_message( tickets, {}, 50, 260, "{odoo_url}/web#id={id}&model={model}&view_type=form", "https://odoo.example.test", "helpdesk.ticket", ) self.assertIn("1 : Sujet : Premier ticket", message) self.assertNotIn("2 : Sujet : Deuxieme ticket", message) self.assertIn("1 ticket(s) supplementaire(s) non affiche(s) sur 2 au total", message) def test_format_ticket_line_uses_personne_when_no_assignee(self): line = ticketing.format_ticket_line( {"id": 2, "name": "Nouveau ticket", "user_id": False}, "{odoo_url}/web#id={id}&model={model}&view_type=form", "https://odoo.example.test", "helpdesk.ticket", ) self.assertEqual( line, "2 : Sujet : Nouveau ticket gere par Personne : " "[cliquer ici pour ouvrir le ticket]" "(https://odoo.example.test/web#id=2&model=helpdesk.ticket&view_type=form)", ) def test_load_tickets_state_returns_empty_dict_when_file_is_missing(self): missing_path = os.path.join(tempfile.gettempdir(), "missing-ticket-state.json") self.assertEqual(ticketing.load_tickets_state(missing_path), {}) def test_save_and_load_tickets_state_roundtrip(self): with tempfile.TemporaryDirectory() as tmpdir: state_path = os.path.join(tmpdir, "tickets_state.json") expected_state = {"42": "2026-07-24 12:00:00"} ticketing.save_tickets_state(state_path, expected_state) self.assertEqual(ticketing.load_tickets_state(state_path), expected_state) @patch("ticketing_setup.requests.post") def test_fetch_odoo_tickets_authenticates_and_returns_result(self, post): auth_response = Mock() auth_response.json.return_value = {"result": 7} tickets_response = Mock() tickets_response.json.return_value = { "result": [{"id": 42, "write_date": "2026-07-24 12:00:00"}], } post.side_effect = [auth_response, tickets_response] tickets = ticketing.fetch_odoo_tickets(make_config()) self.assertEqual(tickets, [{"id": 42, "write_date": "2026-07-24 12:00:00"}]) auth_response.raise_for_status.assert_called_once_with() tickets_response.raise_for_status.assert_called_once_with() self.assertEqual(post.call_count, 2) self.assertEqual(post.call_args.kwargs["timeout"], 5) self.assertEqual(post.call_args.kwargs["json"]["params"]["service"], "object") self.assertEqual(post.call_args.kwargs["json"]["params"]["method"], "execute_kw") self.assertEqual( post.call_args.kwargs["json"]["params"]["args"][6]["fields"], ["id", "name", "stage_id", "partner_id", "user_id", "write_date"], ) self.assertEqual(post.call_args.kwargs["json"]["params"]["args"][6]["limit"], 0) self.assertEqual(post.call_args.kwargs["json"]["params"]["args"][6]["order"], "id asc") @patch("ticketing_setup.requests.post") def test_fetch_odoo_tickets_rejects_jsonrpc_error(self, post): response = Mock() response.json.return_value = {"error": {"message": "Access denied"}} post.return_value = response with self.assertRaisesRegex(Exception, "Erreur Odoo JSON-RPC"): ticketing.fetch_odoo_tickets(make_config()) @patch("ticketing_setup.requests.post") def test_fetch_odoo_tickets_can_use_configured_uid(self, post): config = make_config() config.set("odoo", "uid", "12") response = Mock() response.json.return_value = {"result": []} post.return_value = response self.assertEqual(ticketing.fetch_odoo_tickets(config), []) self.assertEqual(post.call_count, 1) self.assertEqual(post.call_args.kwargs["json"]["params"]["args"][1], 12) @patch("ticketing_setup.requests.post") def test_send_to_rocket_raises_for_http_errors(self, post): response = Mock() response.status_code = 200 post.return_value = response ticketing.send_to_rocket("message de test", conf_wapt=make_config()) response.raise_for_status.assert_called_once_with() self.assertEqual(post.call_args.kwargs["timeout"], 5) self.assertEqual(post.call_args.kwargs["json"], {"text": "message de test"}) if __name__ == "__main__": unittest.main()