admin/Odoo-Modules
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

feat(plating-migration): dry-run + Owner-approval workflow

Browse Source 
Phase H of permissions overhaul (LAST subagent phase).

New models:
- fp.migration.preview (state: pending/approved/cancelled/rolled_back)
- fp.migration.preview.line (one per active internal user)

On -u, post_init_hook creates a preview in 'pending' state, walks all
active non-share users through the 12-rule mapping predicate chain
(first match wins, highest precedence first), and schedules a
mail.activity on every Owner.

Mapping table (per spec Section 5):
  uid 1/2 / Administrator   -> owner
  CGP DO (existing)          -> owner + res.company DO field set
  CGP Officer                -> quality_manager
  Manager / Shop Mgr (old)   -> manager
  Accounting                 -> manager
  Estimator-without-Manager  -> sales_rep (flagged: loses confirm)
  Supervisor / Receiving     -> shop_manager
  Operator                   -> technician
  catchall                   -> 'no'

Owner clicks 'Approve & Run' on the preview form -> sudo write removes
old plating groups, adds new role's group, posts Markup chatter audit.
Optionally sets res.company.x_fc_cgp_designated_official_id for the DO.

30-day rollback window via JSON snapshot of groups_id per line. Daily
cron (Fusion Plating: Purge Expired Role Migrations) clears snapshots
+ unlinks old [DEPRECATED] groups after 30 days.

ACL: fp.migration.preview + .line both Owner-only (CRUD).
Menu: Plating > Configuration > Role Migrations (Owner-only).

Tests cover: only-Owner-can-approve, approve advances state, cancel
blocks after approval, rollback restores groups_id, Estimator warning
flagged, uid 2 maps to owner, rollback blocked after 30 days.

Per CLAUDE.md: ir.cron uses only Odoo-19-valid fields (no numbercall,
no doall). Post-init hook is idempotent — won't double-create previews
or re-fire if all users already migrated.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
gsinghpal
2026-05-24 02:21:43 -04:00
parent de3ec7d97a
commit 5cc1117f75
10 changed files with 597 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
fusion_plating/fusion_plating/models/__init__.py
View File

@@ -49,3 +49,9 @@ from . import fp_job_step_move
# Phase 1 — Plating landing-page resolver
from . import fp_landing
# Phase H — dry-run + Owner-approval role migration workflow.
# fp_role_constants MUST be imported before fp_migration (the latter
# imports the predicate chain + xmlid maps from the former).
from . import fp_role_constants
from . import fp_migration

248
fusion_plating/fusion_plating/models/fp_migration.py Normal file
View File

@@ -0,0 +1,248 @@
# -*- coding: utf-8 -*-
"""Phase H — dry-run + Owner-approval migration workflow."""
import json
import logging
from datetime import timedelta
from markupsafe import Markup
from odoo import _, api, fields, models
from odoo.exceptions import UserError
from .fp_role_constants import (
_FP_OLD_GROUP_XMLIDS,
_NEW_ROLE_XMLID,
fp_resolve_target_role,
)
_logger = logging.getLogger(__name__)
_ROLE_SELECTION = [
('no', 'No'),
('technician', 'Technician'),
('sales_rep', 'Sales Representative'),
('shop_manager', 'Shop Manager'),
('sales_manager', 'Sales Manager'),
('manager', 'Manager'),
('quality_manager', 'Quality Manager'),
('owner', 'Owner'),
]
class FpMigrationPreview(models.Model):
_name = 'fp.migration.preview'
_description = 'Fusion Plating Role Migration Preview'
_inherit = ['mail.thread']
_order = 'create_date desc'
name = fields.Char(
default=lambda s: _('Migration %s') % fields.Datetime.now(),
tracking=True,
)
state = fields.Selection(
[
('pending', 'Pending Review'),
('approved', 'Approved & Applied'),
('cancelled', 'Cancelled'),
('rolled_back', 'Rolled Back'),
],
default='pending',
required=True,
tracking=True,
)
line_ids = fields.One2many('fp.migration.preview.line', 'preview_id')
user_count = fields.Integer(compute='_compute_counts', store=True)
warning_count = fields.Integer(compute='_compute_counts', store=True)
approved_by_id = fields.Many2one('res.users', readonly=True)
approved_at = fields.Datetime(readonly=True)
rollback_deadline = fields.Datetime(compute='_compute_rollback_deadline')
@api.depends('line_ids', 'line_ids.warning')
def _compute_counts(self):
for rec in self:
rec.user_count = len(rec.line_ids)
rec.warning_count = sum(1 for ln in rec.line_ids if ln.warning)
@api.depends('approved_at')
def _compute_rollback_deadline(self):
for rec in self:
rec.rollback_deadline = (
rec.approved_at + timedelta(days=30) if rec.approved_at else False
)
def _fp_build_lines(self):
"""Walk all active internal users; one line per user with the
proposed role + capability_delta."""
self.ensure_one()
Line = self.env['fp.migration.preview.line']
users = self.env['res.users'].search([
('share', '=', False),
('active', '=', True),
])
vals_list = []
for user in users:
role, delta = fp_resolve_target_role(user)
vals_list.append({
'preview_id': self.id,
'user_id': user.id,
'proposed_role': role,
'capability_delta': delta or '',
'warning': bool(delta),
})
if vals_list:
Line.create(vals_list)
def _fp_notify_owners(self):
"""Schedule a 'Review Fusion Plating role migration' activity on
every Owner user. Idempotent — won't double-schedule."""
self.ensure_one()
owner_grp = self.env.ref('fusion_plating.group_fp_owner', raise_if_not_found=False)
if not owner_grp:
return
owners = owner_grp.users.filtered(lambda u: u.active and not u.share)
if not owners:
_logger.warning('Fusion Plating migration preview %s: no Owner users to notify', self.id)
return
activity_type = self.env.ref('mail.mail_activity_data_todo')
for owner in owners:
existing = self.env['mail.activity'].search([
('res_model_id', '=', self.env.ref('fusion_plating.model_fp_migration_preview').id),
('res_id', '=', self.id),
('user_id', '=', owner.id),
], limit=1)
if existing:
continue
self.env['mail.activity'].create({
'res_model_id': self.env.ref('fusion_plating.model_fp_migration_preview').id,
'res_id': self.id,
'activity_type_id': activity_type.id,
'summary': _('Review Fusion Plating role migration'),
'note': _('%(n)d users affected, %(w)d with capability changes.') % {
'n': self.user_count,
'w': self.warning_count,
},
'user_id': owner.id,
'date_deadline': fields.Date.today(),
})
def action_approve_and_run(self):
self.ensure_one()
if not self.env.user.has_group('fusion_plating.group_fp_owner'):
raise UserError(_('Only Owners can approve role migrations.'))
if self.state != 'pending':
raise UserError(_(
'Migration is no longer pending - current state: %s'
) % self.state)
# Resolve old group ids once
old_group_ids = []
for xmlid in _FP_OLD_GROUP_XMLIDS:
g = self.env.ref(xmlid, raise_if_not_found=False)
if g:
old_group_ids.append(g.id)
for line in self.line_ids:
user = line.user_id
# Snapshot current groups_id for rollback
line.applied_groups_snapshot = json.dumps(user.groups_id.ids)
# Remove old plating-role groups
if old_group_ids:
user.sudo().write({
'groups_id': [(3, gid) for gid in old_group_ids]
})
# Add the new role group (no-op for 'no')
target_xmlid = _NEW_ROLE_XMLID.get(line.proposed_role)
if target_xmlid:
target = self.env.ref(target_xmlid, raise_if_not_found=False)
if target:
user.sudo().write({'groups_id': [(4, target.id)]})
# Audit chatter on the user
user.message_post(
body=Markup(_(
'Plating role assigned by migration: <b>%s</b>'
)) % line.proposed_role,
message_type='notification',
)
# Special: CGP DO becomes a res.company field, not a role
if line.capability_delta and 'CGP DO' in line.capability_delta:
user.company_id.x_fc_cgp_designated_official_id = user.id
self.write({
'state': 'approved',
'approved_by_id': self.env.user.id,
'approved_at': fields.Datetime.now(),
})
def action_cancel(self):
self.ensure_one()
if self.state != 'pending':
raise UserError(_('Only pending migrations can be cancelled.'))
self.state = 'cancelled'
def action_rollback(self):
self.ensure_one()
if self.state != 'approved':
raise UserError(_('Only approved migrations can be rolled back.'))
if self.rollback_deadline and fields.Datetime.now() > self.rollback_deadline:
raise UserError(_(
'Rollback window has expired (30 days after approval). '
'Restore from pg_dump backup instead.'
))
for line in self.line_ids:
if line.applied_groups_snapshot:
old_ids = json.loads(line.applied_groups_snapshot)
line.user_id.sudo().write({'groups_id': [(6, 0, old_ids)]})
self.state = 'rolled_back'
@api.model
def _cron_purge_expired_migrations(self):
"""After 30 days, clear snapshots + unlink old plating groups.
Runs daily via fp_migration_cron.xml."""
deadline = fields.Datetime.now() - timedelta(days=30)
expired = self.search([
('state', '=', 'approved'),
('approved_at', '<', deadline),
])
if not expired:
return
# Clear snapshots (no more rollback possible)
for preview in expired:
preview.line_ids.write({'applied_groups_snapshot': False})
# Unlink old plating groups (now confirmed unused — every user is
# on the new groups; backward-compat implied_ids chains can drop)
old_group_ids = []
for xmlid in _FP_OLD_GROUP_XMLIDS:
g = self.env.ref(xmlid, raise_if_not_found=False)
if g:
old_group_ids.append(g.id)
if old_group_ids:
self.env['res.groups'].browse(old_group_ids).unlink()
_logger.info('Fusion Plating migration: purged %d expired old plating groups', len(old_group_ids))
class FpMigrationPreviewLine(models.Model):
_name = 'fp.migration.preview.line'
_description = 'Migration Preview Line'
preview_id = fields.Many2one('fp.migration.preview', required=True, ondelete='cascade')
user_id = fields.Many2one('res.users', required=True, ondelete='cascade')
current_groups = fields.Char(compute='_compute_current_groups')
proposed_role = fields.Selection(_ROLE_SELECTION)
capability_delta = fields.Char()
warning = fields.Boolean()
notes = fields.Text(help='Owner may annotate before approving')
applied_groups_snapshot = fields.Text(help='JSON of pre-migration groups_id for rollback')
@api.depends('user_id', 'user_id.groups_id')
def _compute_current_groups(self):
for line in self:
if line.user_id:
line.current_groups = ', '.join(line.user_id.groups_id.mapped('name'))
else:
line.current_groups = ''

86
fusion_plating/fusion_plating/models/fp_role_constants.py Normal file
View File

@@ -0,0 +1,86 @@
# -*- coding: utf-8 -*-
"""Single source of truth for migration mapping rules + old-group xmlids.
The mapping predicates are evaluated against res.users records. First match
wins (highest-precedence first). See spec Section 5 + plan Phase H.
"""
# Every plating role group xmlid that exists BEFORE the migration (deprecated
# but still defined for backward-compat during 30-day rollback window).
_FP_OLD_GROUP_XMLIDS = (
'fusion_plating.group_fusion_plating_operator',
'fusion_plating.group_fusion_plating_supervisor',
'fusion_plating.group_fusion_plating_manager',
'fusion_plating.group_fusion_plating_admin',
'fusion_plating_configurator.group_fp_estimator',
'fusion_plating_configurator.group_fp_shop_manager',
'fusion_plating_invoicing.group_fp_accounting',
'fusion_plating_receiving.group_fp_receiving',
'fusion_plating_cgp.group_fusion_plating_cgp_officer',
'fusion_plating_cgp.group_fusion_plating_cgp_designated_official',
'fusion_plating_jobs.group_fusion_plating_legacy_menus',
)
# New role -> the group xmlid to add when migration assigns this role.
# 'no' maps to None (no plating group added; old ones still get removed).
_NEW_ROLE_XMLID = {
'no': None,
'technician': 'fusion_plating.group_fp_technician',
'sales_rep': 'fusion_plating.group_fp_sales_rep',
'shop_manager': 'fusion_plating.group_fp_shop_manager_v2',
'sales_manager': 'fusion_plating.group_fp_sales_manager',
'manager': 'fusion_plating.group_fp_manager',
'quality_manager': 'fusion_plating.group_fp_quality_manager',
'owner': 'fusion_plating.group_fp_owner',
}
# Mapping rules: (label, predicate, new_role, capability_delta_or_None)
# Highest precedence first; first match wins.
# Predicate is a callable taking a res.users record; returns bool.
_FP_ROLE_MAPPING_RULES = [
('uid_1_or_2',
lambda u: u.id in (1, 2),
'owner', None),
('admin',
lambda u: u.has_group('fusion_plating.group_fusion_plating_admin'),
'owner', None),
('cgp_designated_official',
lambda u: u.has_group('fusion_plating_cgp.group_fusion_plating_cgp_designated_official'),
'owner', 'Was CGP DO; field set on res.company'),
('cgp_officer',
lambda u: u.has_group('fusion_plating_cgp.group_fusion_plating_cgp_officer'),
'quality_manager', None),
('manager',
lambda u: u.has_group('fusion_plating.group_fusion_plating_manager'),
'manager', None),
('shop_manager_old',
lambda u: u.has_group('fusion_plating_configurator.group_fp_shop_manager'),
'manager', None),
('accounting',
lambda u: u.has_group('fusion_plating_invoicing.group_fp_accounting'),
'manager', None),
('estimator_alone',
lambda u: (u.has_group('fusion_plating_configurator.group_fp_estimator')
and not u.has_group('fusion_plating.group_fusion_plating_manager')),
'sales_rep', 'Loses order-confirm authority'),
('supervisor',
lambda u: u.has_group('fusion_plating.group_fusion_plating_supervisor'),
'shop_manager', None),
('receiving',
lambda u: u.has_group('fusion_plating_receiving.group_fp_receiving'),
'shop_manager', None),
('operator',
lambda u: u.has_group('fusion_plating.group_fusion_plating_operator'),
'technician', None),
('catchall',
lambda u: True,
'no', None),
]
def fp_resolve_target_role(user):
"""Returns (role_key, capability_delta_or_None). First predicate match wins."""
for _label, predicate, role, delta in _FP_ROLE_MAPPING_RULES:
if predicate(user):
return role, delta
return 'no', None