141 lines
5.2 KiB
Python
141 lines
5.2 KiB
Python
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.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_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")
|
|
|
|
@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()
|
|
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()
|