feat(fusion_helpdesk): owner-approval engagement flow + AI summary + reporting
Ships the design spec at docs/superpowers/specs/2026-05-27-owner-approval-flow-design.md. What's new on central (fusion_helpdesk_central 19.0.1.2.0 -> 19.0.2.0.0): - Engagement model: 8 new fields on helpdesk.ticket (state, snapshotted owner email/name, single-use UUID4 token, sent/reminded/decided timestamps, AI summary, stored-computed turnaround hours). - Wizard: single + bulk modes on one fusion.helpdesk.engagement.wizard TransientModel with a child wizard.line for per-ticket bulk summaries. default_get pulls the OpenAI summary on open; AI fan-out for bulk is parallel via ThreadPoolExecutor (max 5 workers, 30s overall cap). - OpenAI client in utils.py — stdlib urllib, 15s per-call timeout, every failure collapses to '' so the wizard's manual-summary fallback fires. - Public portal: /fusion_helpdesk/engagement/<token>/<decision> GET + POST, four branded standalone QWeb pages (confirm/done/invalid/error). Token is single-use, cleared on confirm. Decision posts a public comment attributed to the resolved owner partner; chatter propagates to the employee's My Tickets thread per the "fully visible" UX choice. - Mail templates (single + bulk) with magic-link buttons. Bulk template renders one card per ticket, each with its own approve/reject URL. - Reminder cron: daily, single-shot per engagement, configurable via fusion_helpdesk_central.engagement_reminder_days ICP (default 3, 0 disables). - Reporting dashboard: pivot/graph/list/kanban over helpdesk.ticket filtered to engaged ones, with avg-turnaround measure. Menu lives under Helpdesk > Reporting > Owner Engagements. - Client_key extended with owner_email/owner_name fields; ticket.create upserts them from the client-side piggyback (no new sync endpoint). - 100% coverage on utils + integration tests on wizard, controllers, re-engagement, cron, computed turnaround. OpenAI mocked in CI. What's new on client (fusion_helpdesk 19.0.1.7.1 -> 19.0.2.0.0): - Two new ICP settings: fusion_helpdesk.owner_email / .owner_name with a new "Owner Approval" block in Settings > Fusion Helpdesk. - controllers/main.py::submit piggybacks both keys on every ticket payload so central keeps client_key.owner_email/name fresh automatically. Verified live end-to-end on entech -> nexa: payload upsert, wizard with mocked AI, action_send, portal GET/POST/GET-again cycle, second click hits the friendly invalid-token page. Token entropy = 122 bits (UUID4).
This commit is contained in:
gsinghpal
@@ -1,2 +1,4 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
from . import test_identity
|
||||
from . import test_utils
|
||||
from . import test_engagement
|
||||
|
||||
414
fusion_helpdesk_central/tests/test_engagement.py
Normal file
414
fusion_helpdesk_central/tests/test_engagement.py
Normal file
@@ -0,0 +1,414 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
# Copyright 2026 Nexa Systems Inc.
|
||||
# License OPL-1
|
||||
"""Integration-ish tests for the owner-approval engagement flow.
|
||||
|
||||
These need an env (helpdesk.ticket, client_key, wizard, portal controller),
|
||||
so they run as TransactionCase. OpenAI is mocked at the utils boundary —
|
||||
no live API calls in CI. HTTP requests use the standard Odoo test client.
|
||||
"""
|
||||
from datetime import timedelta
|
||||
from unittest.mock import patch
|
||||
|
||||
from odoo import fields
|
||||
from odoo.exceptions import UserError
|
||||
from odoo.tests import TransactionCase, HttpCase, tagged
|
||||
|
||||
|
||||
def _patch_openai(return_value='• summary bullet one\n• bullet two'):
|
||||
"""Mock the OpenAI client used by the wizard, returning a deterministic
|
||||
summary so tests don't depend on network or API keys."""
|
||||
return patch(
|
||||
'odoo.addons.fusion_helpdesk_central.models.engagement_wizard.'
|
||||
'call_openai_chat',
|
||||
return_value=return_value,
|
||||
)
|
||||
|
||||
|
||||
@tagged('post_install', '-at_install', 'fusion_helpdesk_central')
|
||||
class TestEngagementBase(TransactionCase):
|
||||
"""Shared fixtures: a client_key with an owner, an ENTECH ticket."""
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
super().setUpClass()
|
||||
# client_key.create() requires an actual bot user; set bot login
|
||||
# to the admin so the model can find it during setUp.
|
||||
cls.env['ir.config_parameter'].sudo().set_param(
|
||||
'fusion_helpdesk.bot_login', cls.env.user.login,
|
||||
)
|
||||
cls.client_key = cls.env['fusion.helpdesk.client.key'].create({
|
||||
'client_label': 'TESTCLIENT',
|
||||
'owner_email': 'owner@testclient.com',
|
||||
'owner_name': 'Test Owner',
|
||||
})
|
||||
cls.team = cls.env['helpdesk.team'].create({
|
||||
'name': 'Test Team for engagement',
|
||||
})
|
||||
|
||||
def _make_ticket(self, **overrides):
|
||||
vals = {
|
||||
'name': '[TESTCLIENT] Bug Report: Test ticket for engagement',
|
||||
'team_id': self.team.id,
|
||||
'x_fc_client_label': 'TESTCLIENT',
|
||||
'description': '<p>Steps to reproduce: do X, see Y.</p>',
|
||||
}
|
||||
vals.update(overrides)
|
||||
return self.env['helpdesk.ticket'].create(vals)
|
||||
|
||||
|
||||
@tagged('post_install', '-at_install', 'fusion_helpdesk_central')
|
||||
class TestOwnerContactSync(TestEngagementBase):
|
||||
|
||||
def test_sync_owner_contacts_from_payload(self):
|
||||
# Simulate the client-side submit piggyback: x_fc_owner_email +
|
||||
# x_fc_owner_name on create vals. Central must consume them
|
||||
# (pop from vals) and upsert the client_key row.
|
||||
self._make_ticket(
|
||||
x_fc_owner_email='newowner@testclient.com',
|
||||
x_fc_owner_name='New Owner',
|
||||
)
|
||||
self.client_key.invalidate_recordset()
|
||||
self.assertEqual(self.client_key.owner_email, 'newowner@testclient.com')
|
||||
self.assertEqual(self.client_key.owner_name, 'New Owner')
|
||||
|
||||
def test_sync_no_owner_payload_leaves_client_key_alone(self):
|
||||
# No piggyback keys → existing client_key contacts must NOT be
|
||||
# nuked. (We had a bug like this in the customer-followup ship.)
|
||||
original_email = self.client_key.owner_email
|
||||
self._make_ticket()
|
||||
self.client_key.invalidate_recordset()
|
||||
self.assertEqual(self.client_key.owner_email, original_email)
|
||||
|
||||
def test_sync_unknown_client_label_is_silently_skipped(self):
|
||||
# If a ticket arrives for a client_label we don't have a row for,
|
||||
# we must not create one (would bypass API-key issuance). Just
|
||||
# log and move on without raising.
|
||||
self._make_ticket(
|
||||
x_fc_client_label='UNKNOWN_CLIENT',
|
||||
x_fc_owner_email='wat@example.com',
|
||||
)
|
||||
# No client_key row was created for UNKNOWN_CLIENT
|
||||
unknown = self.env['fusion.helpdesk.client.key'].search(
|
||||
[('client_label', '=', 'UNKNOWN_CLIENT')])
|
||||
self.assertFalse(unknown)
|
||||
|
||||
|
||||
@tagged('post_install', '-at_install', 'fusion_helpdesk_central')
|
||||
class TestEngagementReset(TestEngagementBase):
|
||||
|
||||
def test_reset_engagement_sets_all_fields(self):
|
||||
t = self._make_ticket()
|
||||
t._fc_reset_engagement('o@x.com', 'Owner', 'Summary text')
|
||||
self.assertEqual(t.x_fc_engagement_state, 'pending')
|
||||
self.assertEqual(t.x_fc_engagement_email, 'o@x.com')
|
||||
self.assertEqual(t.x_fc_engagement_name, 'Owner')
|
||||
self.assertEqual(t.x_fc_ai_summary, 'Summary text')
|
||||
self.assertTrue(t.x_fc_engagement_token)
|
||||
self.assertTrue(t.x_fc_engagement_sent_at)
|
||||
self.assertFalse(t.x_fc_engagement_reminded_at)
|
||||
self.assertFalse(t.x_fc_engagement_decided_at)
|
||||
|
||||
def test_re_engagement_rotates_token_and_clears_decision(self):
|
||||
t = self._make_ticket()
|
||||
t._fc_reset_engagement('o@x.com', 'Owner', 'summary 1')
|
||||
original_token = t.x_fc_engagement_token
|
||||
# Simulate the owner having decided…
|
||||
t.write({
|
||||
'x_fc_engagement_state': 'rejected',
|
||||
'x_fc_engagement_decided_at': fields.Datetime.now(),
|
||||
'x_fc_engagement_reminded_at': fields.Datetime.now(),
|
||||
})
|
||||
# …then re-engage. State must reset, token must rotate.
|
||||
t._fc_reset_engagement('o@x.com', 'Owner', 'summary 2')
|
||||
self.assertEqual(t.x_fc_engagement_state, 'pending')
|
||||
self.assertNotEqual(t.x_fc_engagement_token, original_token)
|
||||
self.assertFalse(t.x_fc_engagement_reminded_at)
|
||||
self.assertFalse(t.x_fc_engagement_decided_at)
|
||||
|
||||
def test_token_is_unique_per_call(self):
|
||||
t = self._make_ticket()
|
||||
tokens = set()
|
||||
for _ in range(20):
|
||||
t._fc_reset_engagement('o@x.com', 'Owner', '')
|
||||
tokens.add(t.x_fc_engagement_token)
|
||||
self.assertEqual(len(tokens), 20)
|
||||
|
||||
def test_finalize_posts_chatter_and_clears_token(self):
|
||||
t = self._make_ticket()
|
||||
t._fc_reset_engagement('o@x.com', 'Owner', 's')
|
||||
partner = self.env['res.partner'].create({
|
||||
'name': 'Owner', 'email': 'o@x.com',
|
||||
})
|
||||
before_count = self.env['mail.message'].search_count(
|
||||
[('res_id', '=', t.id), ('model', '=', 'helpdesk.ticket')])
|
||||
t._fc_finalize_engagement('approved', partner, comment='LGTM')
|
||||
after_count = self.env['mail.message'].search_count(
|
||||
[('res_id', '=', t.id), ('model', '=', 'helpdesk.ticket')])
|
||||
self.assertGreater(after_count, before_count)
|
||||
self.assertEqual(t.x_fc_engagement_state, 'approved')
|
||||
self.assertFalse(t.x_fc_engagement_token)
|
||||
self.assertTrue(t.x_fc_engagement_decided_at)
|
||||
|
||||
def test_turnaround_hours_computed(self):
|
||||
t = self._make_ticket()
|
||||
now = fields.Datetime.now()
|
||||
t.write({
|
||||
'x_fc_engagement_sent_at': now - timedelta(hours=5),
|
||||
'x_fc_engagement_decided_at': now,
|
||||
})
|
||||
self.assertAlmostEqual(
|
||||
t.x_fc_engagement_turnaround_hours, 5.0, places=1,
|
||||
)
|
||||
|
||||
def test_turnaround_zero_when_not_decided(self):
|
||||
t = self._make_ticket()
|
||||
t.write({
|
||||
'x_fc_engagement_sent_at': fields.Datetime.now() - timedelta(hours=2),
|
||||
})
|
||||
self.assertEqual(t.x_fc_engagement_turnaround_hours, 0.0)
|
||||
|
||||
|
||||
@tagged('post_install', '-at_install', 'fusion_helpdesk_central')
|
||||
class TestEngagementWizard(TestEngagementBase):
|
||||
|
||||
def test_single_send_via_wizard(self):
|
||||
t = self._make_ticket()
|
||||
with _patch_openai():
|
||||
wizard = self.env['fusion.helpdesk.engagement.wizard'].with_context(
|
||||
default_ticket_id=t.id,
|
||||
active_id=t.id,
|
||||
active_model='helpdesk.ticket',
|
||||
).create({})
|
||||
self.assertEqual(wizard.mode, 'single')
|
||||
self.assertIn('summary bullet', wizard.ai_summary)
|
||||
wizard.personal_note = 'please review'
|
||||
result = wizard.action_send()
|
||||
# action returns the standard close-modal action
|
||||
self.assertEqual(result.get('type'), 'ir.actions.act_window_close')
|
||||
self.assertEqual(t.x_fc_engagement_state, 'pending')
|
||||
self.assertEqual(t.x_fc_engagement_email, 'owner@testclient.com')
|
||||
self.assertTrue(t.x_fc_engagement_token)
|
||||
|
||||
def test_single_send_uses_current_client_key_owner(self):
|
||||
# The wizard must read the FRESH owner contact from client_key,
|
||||
# not a stale snapshot — if the client_key is updated between
|
||||
# default_get and Send, Send wins.
|
||||
t = self._make_ticket()
|
||||
with _patch_openai():
|
||||
wizard = self.env['fusion.helpdesk.engagement.wizard'].with_context(
|
||||
default_ticket_id=t.id,
|
||||
).create({})
|
||||
self.client_key.owner_email = 'changed@testclient.com'
|
||||
wizard.action_send()
|
||||
self.assertEqual(t.x_fc_engagement_email, 'changed@testclient.com')
|
||||
|
||||
def test_wizard_rejects_ticket_without_client_label(self):
|
||||
t = self._make_ticket(x_fc_client_label=False)
|
||||
with _patch_openai(), self.assertRaises(UserError):
|
||||
self.env['fusion.helpdesk.engagement.wizard'].with_context(
|
||||
default_ticket_id=t.id,
|
||||
).create({})
|
||||
|
||||
def test_wizard_rejects_when_owner_contact_missing(self):
|
||||
self.client_key.write({'owner_email': False, 'owner_name': False})
|
||||
t = self._make_ticket()
|
||||
with _patch_openai(), self.assertRaises(UserError):
|
||||
self.env['fusion.helpdesk.engagement.wizard'].with_context(
|
||||
default_ticket_id=t.id,
|
||||
).create({})
|
||||
|
||||
def test_wizard_marks_ai_unavailable_when_summary_empty(self):
|
||||
t = self._make_ticket()
|
||||
with _patch_openai(return_value=''):
|
||||
wizard = self.env['fusion.helpdesk.engagement.wizard'].with_context(
|
||||
default_ticket_id=t.id,
|
||||
).create({})
|
||||
self.assertTrue(wizard.ai_unavailable)
|
||||
self.assertEqual(wizard.ai_summary, '')
|
||||
|
||||
def test_bulk_send_creates_one_engagement_per_ticket(self):
|
||||
ts = self.env['helpdesk.ticket']
|
||||
for i in range(3):
|
||||
ts |= self._make_ticket(name='[TESTCLIENT] Bug %s' % i)
|
||||
with _patch_openai():
|
||||
wizard = self.env['fusion.helpdesk.engagement.wizard'].with_context(
|
||||
default_ticket_ids=ts.ids,
|
||||
active_ids=ts.ids,
|
||||
active_model='helpdesk.ticket',
|
||||
fhc_bulk=True,
|
||||
).create({})
|
||||
self.assertEqual(wizard.mode, 'bulk')
|
||||
self.assertEqual(len(wizard.line_ids), 3)
|
||||
wizard.action_send()
|
||||
for t in ts:
|
||||
self.assertEqual(t.x_fc_engagement_state, 'pending')
|
||||
self.assertTrue(t.x_fc_engagement_token)
|
||||
# Each ticket must have its OWN token
|
||||
tokens = {t.x_fc_engagement_token for t in ts}
|
||||
self.assertEqual(len(tokens), 3)
|
||||
|
||||
def test_bulk_rejects_mixed_clients(self):
|
||||
t1 = self._make_ticket()
|
||||
# Need another client_key for the mix to be valid otherwise the
|
||||
# owner-contact check fires first.
|
||||
self.env['fusion.helpdesk.client.key'].create({
|
||||
'client_label': 'OTHERCLIENT',
|
||||
'owner_email': 'other@x.com', 'owner_name': 'Other',
|
||||
})
|
||||
t2 = self._make_ticket(
|
||||
name='[OTHERCLIENT] x', x_fc_client_label='OTHERCLIENT')
|
||||
with _patch_openai(), self.assertRaises(UserError):
|
||||
self.env['fusion.helpdesk.engagement.wizard'].with_context(
|
||||
default_ticket_ids=[t1.id, t2.id],
|
||||
fhc_bulk=True,
|
||||
).create({})
|
||||
|
||||
def test_bulk_rejects_already_pending_in_selection(self):
|
||||
t1 = self._make_ticket()
|
||||
t1._fc_reset_engagement('o@x.com', 'Owner', '') # already pending
|
||||
t2 = self._make_ticket(name='[TESTCLIENT] B')
|
||||
with _patch_openai(), self.assertRaises(UserError):
|
||||
self.env['fusion.helpdesk.engagement.wizard'].with_context(
|
||||
default_ticket_ids=[t1.id, t2.id],
|
||||
fhc_bulk=True,
|
||||
).create({})
|
||||
|
||||
|
||||
@tagged('post_install', '-at_install', 'fusion_helpdesk_central')
|
||||
class TestReminderCron(TestEngagementBase):
|
||||
|
||||
def test_reminder_fires_for_stale_pending_only(self):
|
||||
# 1 stale (should be reminded), 1 recent (no reminder), 1 already
|
||||
# reminded (no second reminder), 1 already-decided (no reminder).
|
||||
old = fields.Datetime.now() - timedelta(days=10)
|
||||
recent = fields.Datetime.now() - timedelta(hours=2)
|
||||
|
||||
stale = self._make_ticket()
|
||||
stale._fc_reset_engagement('o@x.com', 'Owner', '')
|
||||
stale.x_fc_engagement_sent_at = old
|
||||
|
||||
too_recent = self._make_ticket(name='[TESTCLIENT] too recent')
|
||||
too_recent._fc_reset_engagement('o@x.com', 'Owner', '')
|
||||
too_recent.x_fc_engagement_sent_at = recent
|
||||
|
||||
already_reminded = self._make_ticket(name='[TESTCLIENT] already')
|
||||
already_reminded._fc_reset_engagement('o@x.com', 'Owner', '')
|
||||
already_reminded.write({
|
||||
'x_fc_engagement_sent_at': old,
|
||||
'x_fc_engagement_reminded_at': old,
|
||||
})
|
||||
|
||||
decided = self._make_ticket(name='[TESTCLIENT] decided')
|
||||
decided._fc_reset_engagement('o@x.com', 'Owner', '')
|
||||
decided.write({
|
||||
'x_fc_engagement_sent_at': old,
|
||||
'x_fc_engagement_state': 'approved',
|
||||
})
|
||||
|
||||
# Default ICP is 3 days, so >=10 days qualifies.
|
||||
sent = self.env['helpdesk.ticket']._fc_send_engagement_reminders()
|
||||
self.assertEqual(sent, 1)
|
||||
self.assertTrue(stale.x_fc_engagement_reminded_at)
|
||||
self.assertFalse(too_recent.x_fc_engagement_reminded_at)
|
||||
# already_reminded's reminded_at must not have moved
|
||||
self.assertEqual(
|
||||
already_reminded.x_fc_engagement_reminded_at, old,
|
||||
)
|
||||
|
||||
def test_reminder_disabled_when_days_zero(self):
|
||||
self.env['ir.config_parameter'].sudo().set_param(
|
||||
'fusion_helpdesk_central.engagement_reminder_days', '0')
|
||||
t = self._make_ticket()
|
||||
t._fc_reset_engagement('o@x.com', 'Owner', '')
|
||||
t.x_fc_engagement_sent_at = fields.Datetime.now() - timedelta(days=30)
|
||||
sent = self.env['helpdesk.ticket']._fc_send_engagement_reminders()
|
||||
self.assertEqual(sent, 0)
|
||||
self.assertFalse(t.x_fc_engagement_reminded_at)
|
||||
|
||||
|
||||
@tagged('post_install', '-at_install', 'fusion_helpdesk_central')
|
||||
class TestEngagementPortal(HttpCase):
|
||||
"""HTTP-layer tests for the public approve/reject portal pages."""
|
||||
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
self.env['ir.config_parameter'].sudo().set_param(
|
||||
'fusion_helpdesk.bot_login', self.env.user.login,
|
||||
)
|
||||
self.client_key = self.env['fusion.helpdesk.client.key'].create({
|
||||
'client_label': 'PORTALCLIENT',
|
||||
'owner_email': 'owner@portalclient.com',
|
||||
'owner_name': 'Portal Owner',
|
||||
})
|
||||
self.team = self.env['helpdesk.team'].create({
|
||||
'name': 'Test team portal',
|
||||
})
|
||||
|
||||
def _make_pending_ticket(self):
|
||||
t = self.env['helpdesk.ticket'].create({
|
||||
'name': '[PORTALCLIENT] Bug Report: portal smoke',
|
||||
'team_id': self.team.id,
|
||||
'x_fc_client_label': 'PORTALCLIENT',
|
||||
'description': '<p>nothing fancy</p>',
|
||||
})
|
||||
t._fc_reset_engagement('owner@portalclient.com', 'Portal Owner', 'sm')
|
||||
# Make sure cursor sees it for the public request
|
||||
self.env.cr.commit()
|
||||
return t
|
||||
|
||||
def test_get_with_valid_token_renders_confirm(self):
|
||||
t = self._make_pending_ticket()
|
||||
try:
|
||||
r = self.url_open(
|
||||
'/fusion_helpdesk/engagement/%s/approve' % t.x_fc_engagement_token,
|
||||
timeout=10,
|
||||
)
|
||||
self.assertEqual(r.status_code, 200)
|
||||
self.assertIn('Confirm Approval', r.text)
|
||||
self.assertIn(t.name, r.text)
|
||||
finally:
|
||||
t.unlink()
|
||||
self.env.cr.commit()
|
||||
|
||||
def test_get_with_bad_token_renders_invalid(self):
|
||||
r = self.url_open(
|
||||
'/fusion_helpdesk/engagement/bogus-token/approve', timeout=10,
|
||||
)
|
||||
self.assertEqual(r.status_code, 200)
|
||||
self.assertIn('Link no longer valid', r.text)
|
||||
|
||||
def test_get_with_bad_decision_renders_invalid(self):
|
||||
t = self._make_pending_ticket()
|
||||
try:
|
||||
r = self.url_open(
|
||||
'/fusion_helpdesk/engagement/%s/sideways'
|
||||
% t.x_fc_engagement_token, timeout=10,
|
||||
)
|
||||
self.assertEqual(r.status_code, 200)
|
||||
self.assertIn('Link no longer valid', r.text)
|
||||
finally:
|
||||
t.unlink()
|
||||
self.env.cr.commit()
|
||||
|
||||
def test_post_records_decision_and_invalidates_token(self):
|
||||
t = self._make_pending_ticket()
|
||||
token = t.x_fc_engagement_token
|
||||
try:
|
||||
r = self.url_open(
|
||||
'/fusion_helpdesk/engagement/%s/approve' % token,
|
||||
data={'comment': 'looks good'}, timeout=10,
|
||||
)
|
||||
self.assertEqual(r.status_code, 200)
|
||||
self.assertIn('Approval recorded', r.text)
|
||||
t.invalidate_recordset()
|
||||
self.assertEqual(t.x_fc_engagement_state, 'approved')
|
||||
self.assertFalse(t.x_fc_engagement_token)
|
||||
# Second click on the same URL must now show the invalid page.
|
||||
r2 = self.url_open(
|
||||
'/fusion_helpdesk/engagement/%s/approve' % token, timeout=10,
|
||||
)
|
||||
self.assertIn('Link no longer valid', r2.text)
|
||||
finally:
|
||||
t.unlink()
|
||||
self.env.cr.commit()
|
||||
193
fusion_helpdesk_central/tests/test_utils.py
Normal file
193
fusion_helpdesk_central/tests/test_utils.py
Normal file
@@ -0,0 +1,193 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
# Copyright 2026 Nexa Systems Inc.
|
||||
# License OPL-1
|
||||
"""Pure-helper tests for fusion_helpdesk_central.utils.
|
||||
|
||||
No Odoo env, no network — these run in any environment with `markupsafe`
|
||||
installed. Mirrors the pattern in fusion_helpdesk/tests/test_utils.py.
|
||||
"""
|
||||
from unittest.mock import patch
|
||||
|
||||
from odoo.tests import TransactionCase, tagged
|
||||
|
||||
from odoo.addons.fusion_helpdesk_central.utils import (
|
||||
SUMMARY_PROMPT,
|
||||
OPENAI_PROMPT_MAX_CHARS,
|
||||
build_summary_prompt,
|
||||
call_openai_chat,
|
||||
format_engagement_chatter,
|
||||
truncate_for_openai,
|
||||
)
|
||||
|
||||
|
||||
@tagged('post_install', '-at_install', 'fusion_helpdesk_central')
|
||||
class TestBuildSummaryPrompt(TransactionCase):
|
||||
|
||||
def test_includes_title_description_and_messages(self):
|
||||
prompt = build_summary_prompt(
|
||||
'My ticket',
|
||||
'Detailed description here',
|
||||
[{'author': 'Alice', 'date': '2026-05-27 10:00:00',
|
||||
'body_plain': 'first reply'}],
|
||||
)
|
||||
self.assertIn('My ticket', prompt)
|
||||
self.assertIn('Detailed description here', prompt)
|
||||
self.assertIn('Alice', prompt)
|
||||
self.assertIn('first reply', prompt)
|
||||
|
||||
def test_empty_messages_becomes_explicit_marker(self):
|
||||
# The model needs to know "no replies yet" — a blank section would
|
||||
# invite hallucination.
|
||||
prompt = build_summary_prompt('t', 'd', [])
|
||||
self.assertIn('(no replies yet)', prompt)
|
||||
|
||||
def test_empty_title_and_description_use_fallbacks(self):
|
||||
prompt = build_summary_prompt('', '', [])
|
||||
self.assertIn('(untitled)', prompt)
|
||||
self.assertIn('(no description)', prompt)
|
||||
|
||||
def test_empty_message_body_is_marked_not_dropped(self):
|
||||
prompt = build_summary_prompt('t', 'd', [
|
||||
{'author': 'A', 'date': 'X', 'body_plain': ''},
|
||||
])
|
||||
# An empty body must still produce a line so author + date context
|
||||
# survives; '(empty)' marker keeps the model honest.
|
||||
self.assertIn('(empty)', prompt)
|
||||
|
||||
|
||||
@tagged('post_install', '-at_install', 'fusion_helpdesk_central')
|
||||
class TestTruncateForOpenAI(TransactionCase):
|
||||
|
||||
def test_no_truncation_when_under_limit(self):
|
||||
prompt = 'x' * 100
|
||||
self.assertEqual(truncate_for_openai(prompt, max_chars=200), prompt)
|
||||
|
||||
def test_truncation_appends_marker_and_respects_cap(self):
|
||||
prompt = 'x' * 500
|
||||
out = truncate_for_openai(prompt, max_chars=200)
|
||||
self.assertLessEqual(len(out), 200)
|
||||
self.assertIn('truncated', out.lower())
|
||||
|
||||
def test_default_cap_is_8000(self):
|
||||
# Regression guard — flipping this default has real $$ implications.
|
||||
self.assertEqual(OPENAI_PROMPT_MAX_CHARS, 8000)
|
||||
|
||||
|
||||
@tagged('post_install', '-at_install', 'fusion_helpdesk_central')
|
||||
class TestFormatEngagementChatter(TransactionCase):
|
||||
|
||||
def test_approve_without_comment(self):
|
||||
out = str(format_engagement_chatter('approved', 'Kris'))
|
||||
self.assertIn('Approved', out)
|
||||
self.assertIn('Kris', out)
|
||||
self.assertIn('✓', out)
|
||||
# No blockquote when no comment.
|
||||
self.assertNotIn('blockquote', out)
|
||||
|
||||
def test_reject_with_comment(self):
|
||||
out = str(format_engagement_chatter(
|
||||
'rejected', 'Kris', comment='not in scope this quarter'))
|
||||
self.assertIn('Rejected', out)
|
||||
self.assertIn('✗', out)
|
||||
self.assertIn('blockquote', out)
|
||||
self.assertIn('not in scope this quarter', out)
|
||||
|
||||
def test_comment_html_escaped(self):
|
||||
# XSS guard: a malicious comment must not inject script tags.
|
||||
out = str(format_engagement_chatter(
|
||||
'approved', 'Kris', comment='<script>alert(1)</script>'))
|
||||
self.assertNotIn('<script>', out)
|
||||
self.assertIn('<script>', out)
|
||||
|
||||
def test_owner_name_html_escaped(self):
|
||||
# Same XSS guard on the owner name path.
|
||||
out = str(format_engagement_chatter(
|
||||
'approved', '<b>Pwn</b>'))
|
||||
self.assertNotIn('<b>Pwn</b>', out)
|
||||
self.assertIn('<b>', out)
|
||||
|
||||
def test_missing_owner_name_falls_back(self):
|
||||
out = str(format_engagement_chatter('approved', None))
|
||||
self.assertIn('the owner', out)
|
||||
|
||||
def test_invalid_decision_raises(self):
|
||||
# Typo upstream should fail loud, not silently post a malformed line.
|
||||
with self.assertRaises(ValueError):
|
||||
format_engagement_chatter('maybe', 'Kris')
|
||||
|
||||
|
||||
@tagged('post_install', '-at_install', 'fusion_helpdesk_central')
|
||||
class TestCallOpenAIChat(TransactionCase):
|
||||
"""OpenAI client unit tests — mock urlopen, never hit the network."""
|
||||
|
||||
def test_empty_api_key_returns_empty_string(self):
|
||||
# No key configured? Don't even try to call OpenAI.
|
||||
self.assertEqual(call_openai_chat('', 'gpt-4o-mini', 'hi'), '')
|
||||
|
||||
def test_empty_prompt_returns_empty_string(self):
|
||||
self.assertEqual(call_openai_chat('sk-x', 'gpt-4o-mini', ''), '')
|
||||
|
||||
def test_happy_path_returns_message_content(self):
|
||||
fake_response = type('R', (), {})()
|
||||
fake_response.read = lambda: (
|
||||
b'{"choices":[{"message":{"content":"bullet one\\nbullet two"}}]}'
|
||||
)
|
||||
fake_response.__enter__ = lambda s: s
|
||||
fake_response.__exit__ = lambda *a: None
|
||||
with patch(
|
||||
'odoo.addons.fusion_helpdesk_central.utils.urllib.request.urlopen',
|
||||
return_value=fake_response,
|
||||
):
|
||||
out = call_openai_chat('sk-x', 'gpt-4o-mini', 'prompt')
|
||||
self.assertEqual(out, 'bullet one\nbullet two')
|
||||
|
||||
def test_network_error_returns_empty(self):
|
||||
# All exceptions must collapse to '' so the wizard's fallback fires.
|
||||
import urllib.error
|
||||
with patch(
|
||||
'odoo.addons.fusion_helpdesk_central.utils.urllib.request.urlopen',
|
||||
side_effect=urllib.error.URLError('boom'),
|
||||
):
|
||||
self.assertEqual(call_openai_chat('sk-x', 'm', 'p'), '')
|
||||
|
||||
def test_http_error_returns_empty(self):
|
||||
import urllib.error
|
||||
with patch(
|
||||
'odoo.addons.fusion_helpdesk_central.utils.urllib.request.urlopen',
|
||||
side_effect=urllib.error.HTTPError(
|
||||
'url', 429, 'rate limited', {}, None,
|
||||
),
|
||||
):
|
||||
self.assertEqual(call_openai_chat('sk-x', 'm', 'p'), '')
|
||||
|
||||
def test_malformed_json_returns_empty(self):
|
||||
fake_response = type('R', (), {})()
|
||||
fake_response.read = lambda: b'not json at all'
|
||||
fake_response.__enter__ = lambda s: s
|
||||
fake_response.__exit__ = lambda *a: None
|
||||
with patch(
|
||||
'odoo.addons.fusion_helpdesk_central.utils.urllib.request.urlopen',
|
||||
return_value=fake_response,
|
||||
):
|
||||
self.assertEqual(call_openai_chat('sk-x', 'm', 'p'), '')
|
||||
|
||||
def test_empty_choices_returns_empty(self):
|
||||
fake_response = type('R', (), {})()
|
||||
fake_response.read = lambda: b'{"choices":[]}'
|
||||
fake_response.__enter__ = lambda s: s
|
||||
fake_response.__exit__ = lambda *a: None
|
||||
with patch(
|
||||
'odoo.addons.fusion_helpdesk_central.utils.urllib.request.urlopen',
|
||||
return_value=fake_response,
|
||||
):
|
||||
self.assertEqual(call_openai_chat('sk-x', 'm', 'p'), '')
|
||||
|
||||
|
||||
@tagged('post_install', '-at_install', 'fusion_helpdesk_central')
|
||||
class TestPromptConstant(TransactionCase):
|
||||
|
||||
def test_summary_prompt_has_required_placeholders(self):
|
||||
# The wizard calls .format(name=, description_plain=, messages_plain=)
|
||||
# — silently dropping any of these would yield a useless prompt.
|
||||
for placeholder in ('{name}', '{description_plain}', '{messages_plain}'):
|
||||
self.assertIn(placeholder, SUMMARY_PROMPT)
|
||||
Reference in New Issue
Block a user