Odoo-Modules/fusion_helpdesk_central/tests/test_engagement.py

# -*- 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 TestOwnerAsFollower(TestEngagementBase):
    """The one-click 'add owner as follower' button — independent of the
    engagement flow, just loops the owner into the chatter."""

    def test_owner_display_renders_name_and_email(self):
        t = self._make_ticket()
        self.assertEqual(
            t.x_fc_owner_display, 'Test Owner <owner@testclient.com>',
        )
        self.assertEqual(t.x_fc_owner_email_resolved, 'owner@testclient.com')
        self.assertFalse(t.x_fc_owner_is_follower)

    def test_owner_display_blank_when_no_client_label(self):
        t = self._make_ticket(x_fc_client_label=False)
        self.assertFalse(t.x_fc_owner_display)
        self.assertFalse(t.x_fc_owner_email_resolved)

    def test_owner_display_blank_when_client_key_has_no_owner(self):
        self.client_key.write({'owner_email': False, 'owner_name': False})
        t = self._make_ticket()
        self.assertFalse(t.x_fc_owner_display)

    def test_action_creates_partner_and_subscribes(self):
        t = self._make_ticket()
        # Pre-condition: no res.partner with that email exists.
        existing = self.env['res.partner'].search(
            [('email', '=ilike', 'owner@testclient.com')])
        existing.unlink()
        t.action_add_owner_as_follower()
        # Partner created
        partner = self.env['res.partner'].search(
            [('email', '=ilike', 'owner@testclient.com')], limit=1)
        self.assertTrue(partner)
        self.assertEqual(partner.email, 'owner@testclient.com')
        # And subscribed
        self.assertIn(partner.id, t.message_partner_ids.ids)
        self.assertTrue(t.x_fc_owner_is_follower)

    def test_action_reuses_existing_partner(self):
        t = self._make_ticket()
        existing = self.env['res.partner'].create({
            'name': 'Pre-existing Owner',
            'email': 'owner@testclient.com',
        })
        t.action_add_owner_as_follower()
        # No duplicate partner created
        count = self.env['res.partner'].search_count(
            [('email', '=ilike', 'owner@testclient.com')])
        self.assertEqual(count, 1)
        self.assertIn(existing.id, t.message_partner_ids.ids)

    def test_action_is_idempotent_when_already_following(self):
        t = self._make_ticket()
        t.action_add_owner_as_follower()
        followers_before = t.message_partner_ids.ids
        t.action_add_owner_as_follower()
        followers_after = t.message_partner_ids.ids
        # Second call must not duplicate or re-trigger the subscribe
        self.assertEqual(followers_before, followers_after)

    def test_action_raises_when_no_owner_configured(self):
        self.client_key.write({'owner_email': False})
        t = self._make_ticket()
        with self.assertRaises(UserError):
            t.action_add_owner_as_follower()


@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()