fix(shopfloor): Phase C review findings — lock_session closes unlock event + cron test

Important 1: lock_session now closes the original unlock event's
session_ended_at via the same parameterized-SQL bypass pattern used
by the force-lock cron. Without this, every Hand-Off click became
a duplicate force_lock event 8 hours later (cron saw the unlock still
open and re-processed).

Important 2: test_unlock_lock_session_endpoints setUp now
unconditionally overrides the kiosk password (was gated on
'if not get_param(...)' which broke on entech where the post-migrate
hook already generated a random password — tests failed against the
real value). HttpCase rolls back per test so no persistence.

Minor 4: _cron_force_lock_stale_sessions now routes the force_lock
create through write_event helper for consistency (single audit-write
path; helper captures acting_uid/ip/ua uniformly).

Minor 5: Hoisted local imports inside method bodies to top-of-file
in tablet_controller.py (AccessDenied, _tablet_session_audit) and
fp_tablet_session_event.py (timedelta, write_event).

Minor 6: New test_force_lock_cron.py with 3 tests: stale session
emits force_lock + closes original; recent session unaffected;
already-closed session not re-processed. Would have caught
Important 1 if it had existed during Phase C review.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

fusion_plating/fusion_plating_shopfloor/tests/__init__.py

@@ -7,3 +7,4 @@ from . import test_kiosk_user_acl
 from . import test_tablet_session_event_model
 from . import test_tablet_pin_auth_manager
 from . import test_unlock_lock_session_endpoints
+from . import test_force_lock_cron

fusion_plating/fusion_plating_shopfloor/tests/test_force_lock_cron.py

@@ -0,0 +1,95 @@
+from datetime import timedelta
+
+from odoo import fields
+from odoo.tests.common import TransactionCase, tagged
+
+
+@tagged('-at_install', 'post_install', 'fp_tablet')
+class TestForceLockCron(TransactionCase):
+
+    def setUp(self):
+        super().setUp()
+        Users = self.env['res.users'].with_context(no_reset_password=True)
+        self.tech = Users.create({
+            'login': 'forcelock_tech@example.com', 'name': 'ForceLock Tech',
+            'email': 'forcelock_tech@example.com',
+            'group_ids': [(6, 0, [
+                self.env.ref('fusion_plating.group_fp_technician').id
+            ])],
+        })
+        self.SessionEvent = self.env['fp.tablet.session.event']
+
+    def _make_unlock(self, started_at):
+        """Create an unlock event manually (bypassing the helper) so we
+        control session_started_at."""
+        return self.SessionEvent.sudo().create({
+            'event_type': 'unlock',
+            'user_id': self.tech.id,
+            'session_id_hash': 'deadbeef' * 8,  # 64 hex chars
+            'session_started_at': started_at,
+        })
+
+    def test_stale_session_emits_force_lock(self):
+        # An unlock event 9 hours ago (past the default 8-hour ceiling).
+        nine_hours_ago = fields.Datetime.now() - timedelta(hours=9)
+        event = self._make_unlock(nine_hours_ago)
+
+        self.SessionEvent._cron_force_lock_stale_sessions()
+
+        # A force_lock event should now exist for this user.
+        force_locks = self.SessionEvent.sudo().search([
+            ('event_type', '=', 'force_lock'),
+            ('user_id', '=', self.tech.id),
+        ])
+        self.assertGreater(len(force_locks), 0,
+            'Cron did not emit force_lock for stale session')
+
+        # The original unlock event should now be closed
+        # (session_ended_at set).
+        event.invalidate_recordset()  # force re-read after SQL UPDATE
+        self.assertIsNotNone(event.session_ended_at,
+            'Cron did not close the original unlock event')
+
+    def test_recent_session_not_force_locked(self):
+        # An unlock event 1 hour ago (well under the 8-hour ceiling).
+        one_hour_ago = fields.Datetime.now() - timedelta(hours=1)
+        self._make_unlock(one_hour_ago)
+
+        before = self.SessionEvent.sudo().search_count([
+            ('event_type', '=', 'force_lock'),
+            ('user_id', '=', self.tech.id),
+        ])
+        self.SessionEvent._cron_force_lock_stale_sessions()
+        after = self.SessionEvent.sudo().search_count([
+            ('event_type', '=', 'force_lock'),
+            ('user_id', '=', self.tech.id),
+        ])
+
+        self.assertEqual(before, after,
+            'Cron incorrectly force-locked a recent session')
+
+    def test_closed_session_not_reprocessed(self):
+        # An unlock event 9 hours ago, but already closed.
+        nine_hours_ago = fields.Datetime.now() - timedelta(hours=9)
+        event = self._make_unlock(nine_hours_ago)
+        # Close it via SQL bypass (mimicking what lock_session does).
+        now = fields.Datetime.now()
+        self.env.cr.execute(
+            """UPDATE fp_tablet_session_event
+               SET session_ended_at = %s, duration_seconds = %s
+             WHERE id = %s""",
+            (now, 1000, event.id),
+        )
+
+        before = self.SessionEvent.sudo().search_count([
+            ('event_type', '=', 'force_lock'),
+            ('user_id', '=', self.tech.id),
+        ])
+        self.SessionEvent._cron_force_lock_stale_sessions()
+        after = self.SessionEvent.sudo().search_count([
+            ('event_type', '=', 'force_lock'),
+            ('user_id', '=', self.tech.id),
+        ])
+
+        self.assertEqual(before, after,
+            'Cron re-processed an already-closed session')

fusion_plating/fusion_plating_shopfloor/tests/test_unlock_lock_session_endpoints.py

@@ -19,10 +19,13 @@ class TestUnlockLockSessionEndpoints(HttpCase):
         self.tech.sudo().set_tablet_pin('1234')
         # Make sure the kiosk password is set so lock_session can re-auth.
         ICP = self.env['ir.config_parameter'].sudo()
-        if not ICP.get_param('fp.tablet.kiosk_password'):
-            ICP.set_param('fp.tablet.kiosk_password', 'test_kiosk_pwd')
-            kiosk = self.env.ref('fusion_plating_shopfloor.user_fp_tablet_kiosk')
-            kiosk.sudo().password = 'test_kiosk_pwd'
+        # Always override — HttpCase rolls back per test, so we never persist
+        # the test password. Was previously gated on "if not get_param(...)"
+        # which broke against entech where the post-migrate hook already
+        # generated a random kiosk password.
+        ICP.set_param('fp.tablet.kiosk_password', 'test_kiosk_pwd')
+        kiosk = self.env.ref('fusion_plating_shopfloor.user_fp_tablet_kiosk')
+        kiosk.sudo().password = 'test_kiosk_pwd'
 
     def _jsonrpc(self, route, params):
         return self.url_open(