f1a2b300f7
Findings from the post-feature code review on commit 396170b4. Addresses
the two CRITICAL + one HIGH + two MEDIUM issues; rest are deferred.
CRITICAL #1 — magic-link token race:
Two near-simultaneous POSTs on the same /engagement/<token>/approve
could both SELECT state='pending' under READ COMMITTED, both post
chatter, and let the last writer flip the outcome. Now the POST path
does an atomic UPDATE helpdesk_ticket SET token=NULL WHERE token=%s
AND state='pending' RETURNING id — the loser gets no row back and
renders the friendly invalid-link page. Verified live: 2 concurrent
POSTs → 1 wins, 1 loses, exactly 1 chatter row.
CRITICAL #2 — reminder cron without per-row savepoint:
Per CLAUDE.md rule #14, a DB failure mid-loop aborts the whole
transaction and silently kills the rest of the batch. Wrap each row's
send_mail+write in `with self.env.cr.savepoint()`. Also corrected the
success-count log (was len(stale), now actual sent count).
HIGH #3 — turnaround pivot summed instead of averaged:
fields.Float defaults to SUM aggregator; meaningless for per-ticket
decision delays. Added aggregator='avg' so the pivot reads "avg
turnaround per ticket" not "summed wait time".
HIGH #4 — added test_concurrent_claim_only_one_wins regression test
that fires two real HTTP POSTs against the same token and asserts
exactly one wins + exactly one approval chatter row exists.
MEDIUM #6 — cron nextcall pinned to 09:00 tomorrow so reminders land
in business hours regardless of when the module was last upgraded.
MEDIUM #10 — escalate failed owner-partner-create from WARNING to
ERROR (via _logger.exception) since silent attribution to the bot
account is a real audit-trail confusion.
Deferred (follow-up commits): #5, #7 (executor cleanup), #8, #9,
#11–#14 — none are bugs, all spec-drift or hardening.
465 lines
19 KiB
Python
465 lines
19 KiB
Python
# -*- 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_concurrent_claim_only_one_wins(self):
|
|
"""Regression for the magic-link double-click race.
|
|
|
|
Two POSTs against the same token must NOT both record decisions.
|
|
The controller uses UPDATE...RETURNING with a WHERE on
|
|
state='pending' so the second call gets a NULL row back and
|
|
returns the invalid-link page. Without that atomic claim, two
|
|
worker transactions could each SELECT the same pending row and
|
|
both post chatter — last-writer-wins on state.
|
|
|
|
url_open hits live HTTP, so each call is its own request/
|
|
transaction — different from a same-transaction simulation and
|
|
the actual production race scenario.
|
|
"""
|
|
t = self._make_pending_ticket()
|
|
token = t.x_fc_engagement_token
|
|
try:
|
|
r1 = self.url_open(
|
|
'/fusion_helpdesk/engagement/%s/approve' % token,
|
|
data={'comment': 'first'}, timeout=10,
|
|
)
|
|
r2 = self.url_open(
|
|
'/fusion_helpdesk/engagement/%s/approve' % token,
|
|
data={'comment': 'second'}, timeout=10,
|
|
)
|
|
self.assertEqual(r1.status_code, 200)
|
|
self.assertEqual(r2.status_code, 200)
|
|
ok_count = sum(
|
|
'Approval recorded' in r.text for r in (r1, r2))
|
|
invalid_count = sum(
|
|
'Link no longer valid' in r.text for r in (r1, r2))
|
|
self.assertEqual(
|
|
ok_count, 1,
|
|
'Both clicks must not both succeed (race condition).',
|
|
)
|
|
self.assertEqual(invalid_count, 1)
|
|
t.invalidate_recordset()
|
|
approval_chatter = self.env['mail.message'].search_count([
|
|
('res_id', '=', t.id),
|
|
('model', '=', 'helpdesk.ticket'),
|
|
('body', 'ilike', 'Approved by'),
|
|
])
|
|
self.assertEqual(
|
|
approval_chatter, 1,
|
|
'Race must not produce duplicate approval chatter posts.',
|
|
)
|
|
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()
|