chore: secure ticketing configuration

tests/test_ticketing.py

@@ -0,0 +1,121 @@
+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", "api_key", "test-api-key")
+    config.set("odoo", "model", "helpdesk.ticket")
+    config.add_section("rocket")
+    config.set("rocket", "webhook_url", "https://rocket.example.test/webhook-test")
+    config.add_section("http")
+    config.set("http", "timeout", "5")
+    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_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_uses_timeout_and_returns_result(self, post):
+        response = Mock()
+        response.json.return_value = {
+            "result": [{"id": 42, "write_date": "2026-07-24 12:00:00"}],
+        }
+        post.return_value = response
+
+        tickets = ticketing.fetch_odoo_tickets(make_config())
+
+        self.assertEqual(tickets, [{"id": 42, "write_date": "2026-07-24 12:00:00"}])
+        response.raise_for_status.assert_called_once_with()
+        post.assert_called_once()
+        self.assertEqual(post.call_args.kwargs["timeout"], 5)
+
+    @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_send_to_rocket_raises_for_http_errors(self, post):
+        response = Mock()
+        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()