# -*- 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': '
Steps to reproduce: do X, see Y.
', } 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': 'nothing fancy
', }) 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()