diff --git a/fusion_plating/fusion_plating_shopfloor/controllers/tablet_controller.py b/fusion_plating/fusion_plating_shopfloor/controllers/tablet_controller.py index f22bd824..be0748b3 100644 --- a/fusion_plating/fusion_plating_shopfloor/controllers/tablet_controller.py +++ b/fusion_plating/fusion_plating_shopfloor/controllers/tablet_controller.py @@ -16,9 +16,11 @@ import logging from datetime import timedelta from odoo import _, fields, http -from odoo.exceptions import UserError +from odoo.exceptions import AccessDenied, UserError from odoo.http import request +from ._tablet_session_audit import write_event, _sha256_session_sid + _logger = logging.getLogger(__name__) @@ -226,8 +228,6 @@ class FpTabletController(http.Controller): Spec: docs/superpowers/specs/2026-05-24-tablet-pin-session-redesign-design.md """ - from odoo.exceptions import AccessDenied - from ._tablet_session_audit import write_event, _sha256_session_sid env = request.env Users = env['res.users'].sudo() target = Users.browse(int(user_id)) @@ -336,7 +336,6 @@ class FpTabletController(http.Controller): ceiling -> ceiling_lock Anything else falls back to manual_lock. """ - from ._tablet_session_audit import write_event, _sha256_session_sid env = request.env now = fields.Datetime.now() sid = request.session.sid @@ -376,6 +375,18 @@ class FpTabletController(http.Controller): session_started_at=session_started_at, session_ended_at=now, duration_seconds=duration_seconds) + # Close the original unlock event (mirror what the cron does + # for stale sessions) so the cron doesn't re-process it later. + # The Python write override blocks updates without the explicit + # admin context flag, so use parameterized SQL (consistent with + # _cron_force_lock_stale_sessions in fp_tablet_session_event.py). + if open_event: + self.env.cr.execute( + """UPDATE fp_tablet_session_event + SET session_ended_at = %s, duration_seconds = %s + WHERE id = %s""", + (now, duration_seconds or 0, open_event.id), + ) _logger.info( 'Tablet locked (reason=%s) for uid %s (sid %s..) duration=%ss', reason, tech_id, sid[:8] if sid else '', duration_seconds, diff --git a/fusion_plating/fusion_plating_shopfloor/models/fp_tablet_session_event.py b/fusion_plating/fusion_plating_shopfloor/models/fp_tablet_session_event.py index d87cbb13..5143f410 100644 --- a/fusion_plating/fusion_plating_shopfloor/models/fp_tablet_session_event.py +++ b/fusion_plating/fusion_plating_shopfloor/models/fp_tablet_session_event.py @@ -7,8 +7,11 @@ write/unlink/edit are forbidden to anyone except root via direct SQL. Spec section 4: docs/superpowers/specs/2026-05-24-tablet-pin-session-redesign-design.md """ +from datetime import timedelta + from odoo import _, api, fields, models from odoo.exceptions import AccessError +from odoo.addons.fusion_plating_shopfloor.controllers._tablet_session_audit import write_event class FpTabletSessionEvent(models.Model): @@ -124,7 +127,6 @@ class FpTabletSessionEvent(models.Model): Runs every 5 minutes per fp_tablet_cron.xml. """ - from datetime import timedelta ceiling_hours = int(self.env['ir.config_parameter'].sudo().get_param( 'fp.tablet.session_ceiling_hours', 8)) cutoff = fields.Datetime.now() - timedelta(hours=ceiling_hours) @@ -136,15 +138,15 @@ class FpTabletSessionEvent(models.Model): now = fields.Datetime.now() for event in stale: duration = int((now - event.session_started_at).total_seconds()) - self.sudo().create({ - 'event_type': 'force_lock', - 'user_id': event.user_id.id, - 'session_id_hash': event.session_id_hash, - 'session_started_at': event.session_started_at, - 'session_ended_at': now, - 'duration_seconds': duration, - 'notes': 'Cron force-lock: session exceeded %d-hour ceiling' % ceiling_hours, - }) + write_event(self.env, + event_type='force_lock', + user_id=event.user_id.id, + session_id_hash=event.session_id_hash, + session_started_at=event.session_started_at, + session_ended_at=now, + duration_seconds=duration, + notes='Cron force-lock: session exceeded %d-hour ceiling' % ceiling_hours, + ) # Mark the original unlock event closed so it's not reprocessed # next tick. write() is blocked by the model override — use # direct SQL bypass (this is the documented escape hatch for diff --git a/fusion_plating/fusion_plating_shopfloor/tests/__init__.py b/fusion_plating/fusion_plating_shopfloor/tests/__init__.py index 3cf4b4ed..ecf22138 100644 --- a/fusion_plating/fusion_plating_shopfloor/tests/__init__.py +++ b/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 diff --git a/fusion_plating/fusion_plating_shopfloor/tests/test_force_lock_cron.py b/fusion_plating/fusion_plating_shopfloor/tests/test_force_lock_cron.py new file mode 100644 index 00000000..2d1f61b5 --- /dev/null +++ b/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') diff --git a/fusion_plating/fusion_plating_shopfloor/tests/test_unlock_lock_session_endpoints.py b/fusion_plating/fusion_plating_shopfloor/tests/test_unlock_lock_session_endpoints.py index f1417e9f..b55c35d7 100644 --- a/fusion_plating/fusion_plating_shopfloor/tests/test_unlock_lock_session_endpoints.py +++ b/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(