feat(quality_dashboard): populate section items (Task 3)
_fetch_section_items pulls top-5 open records per type, ranked overdue-first by oldest create_date. _build_item shapes each row with id/name/customer/subtitle/urgency/open_action. _resolve_partner defensively walks partner_id -> job_id.partner_id -> ncr_id.partner_id per type. _build_subtitle generates the human-readable second line. Tests cover empty list, 5-cap on 8-record set, and required item keys (id/name/customer/subtitle/urgency/open_action). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -114,14 +114,13 @@ class FpQualityDashboardSnapshot:
|
||||
Model = self.env[cfg['model']].sudo()
|
||||
open_count = Model.search_count(state_dom)
|
||||
overdue_count = self._overdue_count(type_code, Model)
|
||||
# Section items — fully populated in Task 3; empty for now
|
||||
return {
|
||||
'type': type_code,
|
||||
'label': cfg['label'],
|
||||
'icon': cfg['icon'],
|
||||
'open': open_count,
|
||||
'overdue': overdue_count,
|
||||
'items': [],
|
||||
'items': self._fetch_section_items(type_code, Model),
|
||||
'open_kanban_xmlid': cfg['kanban_xmlid'],
|
||||
}
|
||||
|
||||
@@ -137,6 +136,87 @@ class FpQualityDashboardSnapshot:
|
||||
dom += [('create_date', '<', cutoff)]
|
||||
return Model.search_count(dom)
|
||||
|
||||
def _fetch_section_items(self, type_code, Model):
|
||||
"""Top-N open records for the type, ordered by urgency
|
||||
(overdue first by oldest create_date, then by create_date asc).
|
||||
Returns list of item dicts in the snapshot shape.
|
||||
"""
|
||||
state_dom = list(OVERDUE_THRESHOLDS[type_code]['state_domain'])
|
||||
# We need urgency labels per record — fetch open set, sort in
|
||||
# Python by (overdue_flag, age) so the top-N reflects urgency.
|
||||
recs = Model.search(state_dom, limit=200) # safety cap
|
||||
if not recs:
|
||||
return []
|
||||
overdue_ids = set(self._overdue_ids(type_code, Model))
|
||||
# Sort: overdue first (by oldest create_date), then non-overdue
|
||||
# by oldest create_date.
|
||||
def rank_key(r):
|
||||
is_od = r.id in overdue_ids
|
||||
return (0 if is_od else 1, r.create_date or self.now)
|
||||
recs_sorted = sorted(recs, key=rank_key)
|
||||
top = recs_sorted[:self.SECTION_TOP_N]
|
||||
return [self._build_item(type_code, r, r.id in overdue_ids)
|
||||
for r in top]
|
||||
|
||||
def _overdue_ids(self, type_code, Model):
|
||||
"""IDs of overdue records for the type — reuses _overdue_count
|
||||
domain logic."""
|
||||
cfg = OVERDUE_THRESHOLDS[type_code]
|
||||
dom = list(cfg['state_domain'])
|
||||
if cfg['use_due_date']:
|
||||
dom += [('due_date', '<', self.today),
|
||||
('due_date', '!=', False)]
|
||||
else:
|
||||
cutoff = self.now - timedelta(days=cfg['days'])
|
||||
dom += [('create_date', '<', cutoff)]
|
||||
return Model.search(dom).ids
|
||||
|
||||
def _build_item(self, type_code, rec, is_overdue):
|
||||
"""Shape one record into the snapshot item dict."""
|
||||
cfg = TYPE_CONFIG[type_code]
|
||||
# Customer name — partner_id direct, OR via job_id for check
|
||||
partner = self._resolve_partner(rec)
|
||||
return {
|
||||
'id': rec.id,
|
||||
'name': rec.display_name or rec.name or f'#{rec.id}',
|
||||
'customer': partner.name if partner else '',
|
||||
'subtitle': self._build_subtitle(type_code, rec, is_overdue),
|
||||
'urgency': 'overdue' if is_overdue else 'normal',
|
||||
'critical_badge': None, # populated in Task 4
|
||||
'open_action': {
|
||||
'res_model': cfg['model'],
|
||||
'res_id': rec.id,
|
||||
},
|
||||
}
|
||||
|
||||
def _resolve_partner(self, rec):
|
||||
"""Walk to partner: direct partner_id, or via job_id (checks),
|
||||
or via ncr_id (capa). Defensive on missing fields."""
|
||||
if 'partner_id' in rec._fields and rec.partner_id:
|
||||
return rec.partner_id
|
||||
if 'job_id' in rec._fields and rec.job_id \
|
||||
and 'partner_id' in rec.job_id._fields:
|
||||
return rec.job_id.partner_id
|
||||
if 'ncr_id' in rec._fields and rec.ncr_id \
|
||||
and 'partner_id' in rec.ncr_id._fields:
|
||||
return rec.ncr_id.partner_id
|
||||
return False
|
||||
|
||||
def _build_subtitle(self, type_code, rec, is_overdue):
|
||||
"""Second-line text per type. Implementation-phase choice
|
||||
per spec open question 1."""
|
||||
# Age in human terms — hours if < 24h, days otherwise
|
||||
if rec.create_date:
|
||||
age = self.now - rec.create_date
|
||||
hours = int(age.total_seconds() / 3600)
|
||||
age_str = (f'{hours}h' if hours < 24
|
||||
else f'{hours // 24}d')
|
||||
else:
|
||||
age_str = ''
|
||||
if is_overdue:
|
||||
return f'{age_str} overdue' if age_str else 'overdue'
|
||||
return age_str
|
||||
|
||||
|
||||
class FpQualityDashboardController(http.Controller):
|
||||
|
||||
|
||||
@@ -52,3 +52,60 @@ class TestDashboardSnapshotShape(TransactionCase):
|
||||
self.assertIsInstance(section['items'], list)
|
||||
self.assertIsInstance(section['open'], int)
|
||||
self.assertIsInstance(section['overdue'], int)
|
||||
|
||||
|
||||
class TestDashboardSnapshotItems(TransactionCase):
|
||||
"""Per-section items list — ranking + cap + shape."""
|
||||
|
||||
def _build(self):
|
||||
from odoo.addons.fusion_plating_quality.controllers.fp_quality_dashboard \
|
||||
import FpQualityDashboardSnapshot
|
||||
return FpQualityDashboardSnapshot(self.env).build()
|
||||
|
||||
def _get_section(self, snap, type_code):
|
||||
for s in snap['sections']:
|
||||
if s['type'] == type_code:
|
||||
return s
|
||||
return None
|
||||
|
||||
def test_section_items_empty_when_no_records(self):
|
||||
snap = self._build()
|
||||
cert_sec = self._get_section(snap, 'cert')
|
||||
if cert_sec is None:
|
||||
self.skipTest('fp.certificate not installed')
|
||||
self.assertEqual(cert_sec['items'], [])
|
||||
|
||||
def test_section_items_capped_at_5(self):
|
||||
if 'fusion.plating.quality.hold' not in self.env:
|
||||
self.skipTest('fusion.plating.quality.hold not installed')
|
||||
partner = self.env['res.partner'].create({'name': 'Cust'})
|
||||
# Create 8 holds in the same open state
|
||||
Hold = self.env['fusion.plating.quality.hold']
|
||||
for i in range(8):
|
||||
Hold.create({
|
||||
'partner_id': partner.id,
|
||||
'state': 'on_hold',
|
||||
'reason': f'test hold {i}',
|
||||
})
|
||||
snap = self._build()
|
||||
sec = self._get_section(snap, 'hold')
|
||||
self.assertEqual(len(sec['items']), 5)
|
||||
self.assertEqual(sec['open'], 8)
|
||||
|
||||
def test_item_has_required_keys(self):
|
||||
if 'fusion.plating.quality.hold' not in self.env:
|
||||
self.skipTest('fusion.plating.quality.hold not installed')
|
||||
partner = self.env['res.partner'].create({'name': 'Cust'})
|
||||
hold = self.env['fusion.plating.quality.hold'].create({
|
||||
'partner_id': partner.id, 'state': 'on_hold', 'reason': 'x',
|
||||
})
|
||||
snap = self._build()
|
||||
sec = self._get_section(snap, 'hold')
|
||||
self.assertEqual(len(sec['items']), 1)
|
||||
item = sec['items'][0]
|
||||
for k in ('id', 'name', 'customer', 'subtitle',
|
||||
'urgency', 'open_action'):
|
||||
self.assertIn(k, item, f'item missing key {k!r}')
|
||||
self.assertEqual(item['open_action']['res_model'],
|
||||
'fusion.plating.quality.hold')
|
||||
self.assertEqual(item['open_action']['res_id'], hold.id)
|
||||
|
||||
Reference in New Issue
Block a user