315 lines
12 KiB
Python
315 lines
12 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.set("odoo", "limit", "0")
|
|
config.set("odoo", "order", "id asc")
|
|
config.set("odoo", "track_chatter", "true")
|
|
config.set("odoo", "chatter_model", "mail.message")
|
|
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",
|
|
"_last_chatter_update": "2026-07-24 10:05:00|99",
|
|
},
|
|
{"id": 34, "write_date": "2026-07-24 11:00:00"},
|
|
])
|
|
|
|
self.assertEqual(state, {
|
|
"12": {
|
|
"write_date": "2026-07-24 10:00:00",
|
|
"last_chatter_update": "2026-07-24 10:05:00|99",
|
|
},
|
|
"34": {
|
|
"write_date": "2026-07-24 11:00:00",
|
|
"last_chatter_update": None,
|
|
},
|
|
})
|
|
|
|
def test_get_updated_ticket_ids_detects_new_and_changed_tickets(self):
|
|
previous_state = {
|
|
"1": {
|
|
"write_date": "2026-07-24 09:00:00",
|
|
"last_chatter_update": None,
|
|
},
|
|
"2": {
|
|
"write_date": "2026-07-24 09:30:00",
|
|
"last_chatter_update": None,
|
|
},
|
|
}
|
|
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_get_updated_ticket_ids_detects_chatter_change(self):
|
|
previous_state = {
|
|
"260": {
|
|
"write_date": "2026-07-24 09:00:00",
|
|
"last_chatter_update": "2026-07-24 09:15:00|1200",
|
|
},
|
|
}
|
|
tickets = [{
|
|
"id": 260,
|
|
"write_date": "2026-07-24 09:00:00",
|
|
"_last_chatter_update": "2026-07-24 10:00:00|1201",
|
|
}]
|
|
|
|
self.assertEqual(ticketing.get_updated_ticket_ids(tickets, previous_state), [260])
|
|
|
|
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": {
|
|
"write_date": "2026-07-24 12:00:00",
|
|
"last_chatter_update": None,
|
|
},
|
|
}
|
|
|
|
ticketing.save_tickets_state(state_path, expected_state)
|
|
|
|
self.assertEqual(ticketing.load_tickets_state(state_path), expected_state)
|
|
|
|
def test_should_send_notification_always_sends_when_tickets_changed(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
state_path = os.path.join(tmpdir, "no_change_notification_state.json")
|
|
ticket = {"id": 42, "write_date": "2026-07-24 12:00:00"}
|
|
|
|
self.assertTrue(ticketing.should_send_notification([ticket], state_path))
|
|
|
|
@patch("ticketing_setup.get_today", return_value="2026-07-27")
|
|
def test_should_send_no_change_notification_once_per_day(self, get_today):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
state_path = os.path.join(tmpdir, "no_change_notification_state.json")
|
|
|
|
self.assertTrue(ticketing.should_send_notification([], state_path))
|
|
|
|
ticketing.record_no_change_notification(state_path)
|
|
|
|
self.assertFalse(ticketing.should_send_notification([], state_path))
|
|
get_today.assert_called()
|
|
|
|
@patch("ticketing_setup.get_today", return_value="2026-07-28")
|
|
def test_should_send_no_change_notification_again_next_day(self, get_today):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
state_path = os.path.join(tmpdir, "no_change_notification_state.json")
|
|
ticketing.save_json_state(state_path, {
|
|
"last_no_change_notification_date": "2026-07-27",
|
|
})
|
|
|
|
self.assertTrue(ticketing.should_send_notification([], state_path))
|
|
|
|
@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"}],
|
|
}
|
|
chatter_response = Mock()
|
|
chatter_response.json.return_value = {
|
|
"result": [{
|
|
"id": 100,
|
|
"res_id": 42,
|
|
"write_date": "2026-07-24 12:05:00",
|
|
"date": "2026-07-24 12:04:00",
|
|
}],
|
|
}
|
|
post.side_effect = [auth_response, tickets_response, chatter_response]
|
|
|
|
tickets = ticketing.fetch_odoo_tickets(make_config())
|
|
|
|
self.assertEqual(tickets, [{
|
|
"id": 42,
|
|
"write_date": "2026-07-24 12:00:00",
|
|
"_last_chatter_update": "2026-07-24 12:05:00|100",
|
|
}])
|
|
auth_response.raise_for_status.assert_called_once_with()
|
|
tickets_response.raise_for_status.assert_called_once_with()
|
|
chatter_response.raise_for_status.assert_called_once_with()
|
|
self.assertEqual(post.call_count, 3)
|
|
ticket_call = post.call_args_list[1]
|
|
self.assertEqual(ticket_call.kwargs["timeout"], 5)
|
|
self.assertEqual(ticket_call.kwargs["json"]["params"]["service"], "object")
|
|
self.assertEqual(ticket_call.kwargs["json"]["params"]["method"], "execute_kw")
|
|
self.assertEqual(
|
|
ticket_call.kwargs["json"]["params"]["args"][6]["fields"],
|
|
["id", "name", "stage_id", "partner_id", "user_id", "write_date"],
|
|
)
|
|
self.assertEqual(ticket_call.kwargs["json"]["params"]["args"][6]["limit"], 0)
|
|
self.assertEqual(ticket_call.kwargs["json"]["params"]["args"][6]["order"], "id asc")
|
|
chatter_call = post.call_args_list[2]
|
|
self.assertEqual(chatter_call.kwargs["json"]["params"]["args"][3], "mail.message")
|
|
self.assertEqual(chatter_call.kwargs["json"]["params"]["args"][6]["limit"], 0)
|
|
|
|
@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()
|