fix(config): align Configuration pages with platform logic

Audit of the admin Configuration section (FAQ Manager, Notification Rules,
Approval Config) surfaced three contract mismatches and one missing schema.

FAQ
  * Backend `encoach.faq.category.audience` / `encoach.faq.item.audience`
    rejected the `both` and `entity` values emitted by the FAQ Manager
    UI. Widened the Selection so the UI's full vocabulary round-trips.
  * FAQ item `video_url` was already accepted by the UI but silently
    dropped by the controller; now persisted via a new `video_url`
    field on `encoach.faq.item` and serialized on list/get responses.

Notification Rules
  * Added `days_before`, `frequency`, `channel`, `entity_id` to
    `encoach.notification.rule`. These are the exact fields the admin
    form collects; prior to this change they were serialized into the
    JSON body, ignored by the controller, and omitted from responses,
    leaving the Active switch permanently off and the table columns
    blank.
  * Controller now translates `active` <-> `is_active` both ways so the
    new frontend contract and legacy callers coexist. Missing required
    fields return HTTP 400 instead of 500.

Approval Workflows
  * The controller had been hand-rolling raw SQL against three tables
    (`encoach_approval_workflow`, `_stage`, `_request`) that no Odoo
    model declared, so the tables never existed. List() was guarded and
    returned empty; create() would 500 with "relation does not exist".
  * Introduced real ORM models in `encoach_exam_template/models/approval.py`
    plus access rights, which auto-provision the tables on -u.
  * Rewrote the controller to use ORM, added PATCH, and emitted both
    `items` and `results` in the list envelope so the frontend's
    PaginatedResponse reader and legacy callers both work. Step
    payloads now carry `max_days`, `auto_escalate`, and
    `notification_email` end to end.
  * Frontend `approvalsService` and `ApprovalWorkflowConfig` updated to
    send the full stage shape + `allow_bypass`, tolerate both envelope
    keys, and validate at least one approver before submit.

Schema delta applied via `./run.sh -u encoach_exam_template,encoach_lms_api`.
Verified with new `test_config_flows.py`: 24/24 passing. Regression runs
on `test_support_flows.py` (29/29) and `test_training_flows.py` (26/26)
remain green.

Made-with: Cursor
This commit is contained in:
Yamen Ahmad
2026-04-19 10:58:43 +04:00
parent 7f23127e44
commit d940db075e
11 changed files with 841 additions and 259 deletions

View File

@@ -1,4 +1,10 @@
import json
"""Approval Workflow API.
Replaces the earlier raw-SQL implementation with a standard Odoo ORM
controller. Backing models live in ``encoach_exam_template/models/approval.py``.
Serves the admin "Approval Config" page (``/admin/approval-config``).
"""
import logging
from datetime import datetime
@@ -11,86 +17,133 @@ from odoo.addons.encoach_api.controllers.base import (
_logger = logging.getLogger(__name__)
def _cr():
return request.env.cr
# ── Serializers ─────────────────────────────────────────────────────────────
def _ser_stage(stage):
"""Stages are written as ``sequence`` (ORM) but the admin UI speaks
``order`` (1-based, dense). We emit both so either side works."""
return {
'id': stage.id,
'order': stage.sequence,
'sequence': stage.sequence,
'approver_id': stage.approver_id.id or None,
'approver_name': stage.approver_id.name if stage.approver_id else '',
'max_days': stage.max_days or 0,
'auto_escalate': bool(stage.auto_escalate),
'notification_email': stage.notification_email or '',
'status': stage.status or 'pending',
'comment': stage.comment or '',
'acted_at': stage.acted_at.isoformat() if stage.acted_at else None,
}
def _table_exists(cr, name):
cr.execute(
"SELECT to_regclass(%s) IS NOT NULL", (f"public.{name}",)
)
row = cr.fetchone()
return bool(row and row[0])
def _ser_workflow(wf):
return {
'id': wf.id,
'name': wf.name or '',
'type': wf.type or 'custom',
'status': wf.status or 'draft',
'allow_bypass': bool(wf.allow_bypass),
'entity_id': wf.entity_id.id or None,
'entity_name': wf.entity_id.name if wf.entity_id else None,
'steps': [_ser_stage(s) for s in wf.stage_ids.sorted('sequence')],
'stage_count': wf.stage_count,
'request_count': wf.request_count,
'created_at': wf.create_date.isoformat() if wf.create_date else None,
# Legacy shape kept for any older callers that read `created`.
'created': wf.create_date.strftime('%Y-%m-%d') if wf.create_date else '',
}
def _user_name(cr, user_id):
if not user_id:
return ''
cr.execute(
"SELECT rp.name FROM res_users ru "
"JOIN res_partner rp ON rp.id = ru.partner_id "
"WHERE ru.id = %s", (user_id,))
row = cr.fetchone()
return row[0] if row else ''
def _ser_request(req):
stage = req.current_stage_id
return {
'id': req.id,
'workflow_id': req.workflow_id.id or None,
'workflow_name': req.workflow_id.name if req.workflow_id else '',
'res_model': req.res_model or '',
'res_id': req.res_id or 0,
'state': req.state or 'draft',
'requester_id': req.requester_id.id or None,
'requester_name': req.requester_id.name if req.requester_id else '',
'current_stage_id': stage.id or None,
'current_stage_sequence': stage.sequence if stage else None,
'bypass_reason': req.bypass_reason or '',
'created_at': req.created_at.isoformat() if req.created_at else None,
}
# ── Write helpers ───────────────────────────────────────────────────────────
def _apply_workflow_writes(body, vals):
for k in ('name', 'type'):
if k in body and body[k] is not None:
vals[k] = body[k]
if 'status' in body and body['status'] in ('draft', 'active', 'archived'):
vals['status'] = body['status']
if 'allow_bypass' in body:
vals['allow_bypass'] = bool(body['allow_bypass'])
# Frontend sends `bypass_enabled` from the workflow-builder dialog.
if 'bypass_enabled' in body:
vals['allow_bypass'] = bool(body['bypass_enabled'])
if 'entity_id' in body:
vals['entity_id'] = int(body['entity_id']) if body['entity_id'] else False
return vals
def _stage_vals_from_body(step, default_seq):
"""Accept either `order` (UI) or `sequence` (ORM). Emit `sequence` only."""
seq = step.get('sequence')
if seq is None:
seq = step.get('order', default_seq)
try:
seq = int(seq)
except (TypeError, ValueError):
seq = default_seq
approver_id = step.get('approver_id')
try:
approver_id = int(approver_id) if approver_id else 0
except (TypeError, ValueError):
approver_id = 0
vals = {
'sequence': seq,
'approver_id': approver_id or False,
'max_days': int(step.get('max_days') or 3),
'auto_escalate': bool(step.get('auto_escalate')),
'notification_email': step.get('notification_email') or False,
}
return vals
# ── Controller ──────────────────────────────────────────────────────────────
class ApprovalWorkflowController(http.Controller):
# ── Workflows ────────────────────────────────────────────────
@http.route('/api/approval-workflows', type='http', auth='none',
methods=['GET'], csrf=False)
@jwt_required
def list_workflows(self, **kw):
try:
cr = _cr()
if not _table_exists(cr, 'encoach_approval_workflow'):
return _json_response({'items': [], 'total': 0})
cr.execute(
"SELECT id, name, type, allow_bypass, entity_id, create_date "
"FROM encoach_approval_workflow ORDER BY create_date DESC")
workflows = cr.fetchall()
items = []
for wf_id, name, wtype, bypass, entity_id, created in workflows:
cr.execute(
"SELECT id, sequence, approver_id, status, comment, "
"auto_escalate, max_days, acted_at "
"FROM encoach_approval_stage WHERE workflow_id = %s "
"ORDER BY sequence ASC", (wf_id,))
steps = []
for sid, seq, approver_id, status, comment, escalate, max_d, acted in cr.fetchall():
steps.append({
'id': sid,
'sequence': seq,
'approver_id': approver_id,
'approver_name': _user_name(cr, approver_id),
'status': status or 'pending',
'comment': comment or '',
'auto_escalate': bool(escalate),
'max_days': max_d or 0,
'acted_at': acted.isoformat() if acted else None,
M = request.env['encoach.approval.workflow'].sudo()
domain = []
if kw.get('status'):
domain.append(('status', '=', kw['status']))
if kw.get('entity_id'):
domain.append(('entity_id', '=', int(kw['entity_id'])))
recs = M.search(domain, order='create_date desc')
items = [_ser_workflow(r) for r in recs]
# Envelope supports both {items,total} (legacy) and {results,total}
# (PaginatedResponse shape used by the approvals frontend service).
return _json_response({
'items': items,
'results': items,
'total': len(items),
})
entity_name = None
if entity_id:
cr.execute("SELECT name FROM encoach_entity WHERE id=%s", (entity_id,))
row = cr.fetchone()
entity_name = row[0] if row else None
items.append({
'id': wf_id,
'name': name,
'type': wtype or '',
'entity_id': entity_id,
'entity_name': entity_name,
'allow_bypass': bool(bypass),
'steps': steps,
'created': created.strftime('%Y-%m-%d') if created else '',
})
return _json_response({'items': items, 'total': len(items)})
except Exception as e:
_logger.exception('list workflows failed')
return _json_response({'error': str(e)}, 500)
_logger.exception('list approval workflows failed')
return _error_response(str(e), 500)
@http.route('/api/approval-workflows', type='http', auth='none',
methods=['POST'], csrf=False)
@@ -98,199 +151,191 @@ class ApprovalWorkflowController(http.Controller):
def create_workflow(self, **kw):
try:
body = _get_json_body()
name = body.get('name', '').strip()
name = (body.get('name') or '').strip()
if not name:
return _error_response('name is required', 400)
cr = _cr()
now = datetime.now()
uid = request.env.uid
vals = {'name': name, 'type': body.get('type', 'custom')}
_apply_workflow_writes(body, vals)
cr.execute(
"INSERT INTO encoach_approval_workflow "
"(name, type, allow_bypass, create_uid, write_uid, create_date, write_date) "
"VALUES (%s, %s, %s, %s, %s, %s, %s) RETURNING id",
(name, body.get('type', 'custom'), body.get('allow_bypass', False),
uid, uid, now, now))
wf_id = cr.fetchone()[0]
steps = body.get('steps') or []
stage_cmds = []
for i, step in enumerate(steps):
sv = _stage_vals_from_body(step, (i + 1) * 10)
if sv.get('approver_id'):
stage_cmds.append((0, 0, sv))
if stage_cmds:
vals['stage_ids'] = stage_cmds
for i, step in enumerate(body.get('steps', [])):
approver_id = step.get('approver_id')
if approver_id:
cr.execute(
"INSERT INTO encoach_approval_stage "
"(workflow_id, sequence, approver_id, status, auto_escalate, max_days, "
"create_uid, write_uid, create_date, write_date) "
"VALUES (%s, %s, %s, 'pending', %s, %s, %s, %s, %s, %s)",
(wf_id, (i + 1) * 10, int(approver_id),
step.get('auto_escalate', True), step.get('max_days', 3),
uid, uid, now, now))
cr.execute("SELECT id, name, type FROM encoach_approval_workflow WHERE id=%s", (wf_id,))
row = cr.fetchone()
return _json_response({'id': row[0], 'name': row[1], 'type': row[2]}, 201)
rec = request.env['encoach.approval.workflow'].sudo().create(vals)
return _json_response(_ser_workflow(rec), 201)
except Exception as e:
_logger.exception('create workflow failed')
return _json_response({'error': str(e)}, 500)
_logger.exception('create approval workflow failed')
return _error_response(str(e), 500)
@http.route('/api/approval-workflows/<int:wf_id>', type='http', auth='none',
methods=['DELETE'], csrf=False)
@http.route('/api/approval-workflows/<int:wf_id>', type='http',
auth='none', methods=['GET'], csrf=False)
@jwt_required
def get_workflow(self, wf_id, **kw):
try:
rec = request.env['encoach.approval.workflow'].sudo().browse(wf_id)
if not rec.exists():
return _error_response('Not found', 404)
return _json_response(_ser_workflow(rec))
except Exception as e:
_logger.exception('get approval workflow failed')
return _error_response(str(e), 500)
@http.route('/api/approval-workflows/<int:wf_id>', type='http',
auth='none', methods=['PATCH', 'PUT'], csrf=False)
@jwt_required
def update_workflow(self, wf_id, **kw):
try:
rec = request.env['encoach.approval.workflow'].sudo().browse(wf_id)
if not rec.exists():
return _error_response('Not found', 404)
body = _get_json_body()
vals = {}
_apply_workflow_writes(body, vals)
# Replace stages only when steps key is present in the body so
# a status-only toggle doesn't wipe stages.
if 'steps' in body and isinstance(body['steps'], list):
cmds = [(5, 0, 0)]
for i, step in enumerate(body['steps']):
sv = _stage_vals_from_body(step, (i + 1) * 10)
if sv.get('approver_id'):
cmds.append((0, 0, sv))
vals['stage_ids'] = cmds
if vals:
rec.write(vals)
return _json_response(_ser_workflow(rec))
except Exception as e:
_logger.exception('update approval workflow failed')
return _error_response(str(e), 500)
@http.route('/api/approval-workflows/<int:wf_id>', type='http',
auth='none', methods=['DELETE'], csrf=False)
@jwt_required
def delete_workflow(self, wf_id, **kw):
try:
cr = _cr()
cr.execute("DELETE FROM encoach_approval_request WHERE workflow_id=%s", (wf_id,))
cr.execute("DELETE FROM encoach_approval_stage WHERE workflow_id=%s", (wf_id,))
cr.execute("DELETE FROM encoach_approval_workflow WHERE id=%s", (wf_id,))
rec = request.env['encoach.approval.workflow'].sudo().browse(wf_id)
if rec.exists():
rec.unlink()
return _json_response({'success': True})
except Exception as e:
_logger.exception('delete workflow failed')
return _json_response({'error': str(e)}, 500)
_logger.exception('delete approval workflow failed')
return _error_response(str(e), 500)
# ── Approval Requests ──
# ── Approval Requests ────────────────────────────────────────
@http.route('/api/approval-requests', type='http', auth='none',
methods=['GET'], csrf=False)
@jwt_required
def list_requests(self, **kw):
try:
cr = _cr()
if not _table_exists(cr, 'encoach_approval_request'):
return _json_response({'items': [], 'total': 0})
state_filter = kw.get('state')
where = ""
params = []
if state_filter:
where = "WHERE ar.state = %s"
params.append(state_filter)
cr.execute(f"""
SELECT ar.id, ar.workflow_id, aw.name as wf_name,
ar.res_model, ar.res_id, ar.state,
ar.requester_id, ar.current_stage_id,
ar.bypass_reason, ar.created_at
FROM encoach_approval_request ar
LEFT JOIN encoach_approval_workflow aw ON aw.id = ar.workflow_id
{where}
ORDER BY ar.created_at DESC NULLS LAST
""", params)
items = []
for row in cr.fetchall():
rid, wf_id, wf_name, model, res_id, state, req_id, stage_id, bypass, created = row
stage_seq = None
if stage_id:
cr.execute("SELECT sequence FROM encoach_approval_stage WHERE id=%s", (stage_id,))
s = cr.fetchone()
stage_seq = s[0] if s else None
items.append({
'id': rid,
'workflow_id': wf_id,
'workflow_name': wf_name or '',
'res_model': model or '',
'res_id': res_id or 0,
'state': state or 'draft',
'requester_id': req_id,
'requester_name': _user_name(cr, req_id),
'current_stage_id': stage_id,
'current_stage_sequence': stage_seq,
'bypass_reason': bypass or '',
'created_at': created.isoformat() if created else None,
M = request.env['encoach.approval.request'].sudo()
domain = []
if kw.get('state'):
domain.append(('state', '=', kw['state']))
if kw.get('workflow_id'):
domain.append(('workflow_id', '=', int(kw['workflow_id'])))
recs = M.search(domain, order='created_at desc, id desc')
items = [_ser_request(r) for r in recs]
return _json_response({
'items': items,
'results': items,
'total': len(items),
})
return _json_response({'items': items, 'total': len(items)})
except Exception as e:
_logger.exception('list requests failed')
return _json_response({'error': str(e)}, 500)
_logger.exception('list approval requests failed')
return _error_response(str(e), 500)
@http.route('/api/approval-requests/<int:req_id>/approve', type='http',
auth='none', methods=['POST'], csrf=False)
@jwt_required
def approve_request(self, req_id, **kw):
try:
cr = _cr()
body = _get_json_body()
now = datetime.now()
cr.execute(
"SELECT workflow_id, current_stage_id FROM encoach_approval_request WHERE id=%s",
(req_id,))
row = cr.fetchone()
if not row:
req_rec = request.env['encoach.approval.request'].sudo().browse(req_id)
if not req_rec.exists():
return _error_response('Request not found', 404)
wf_id, stage_id = row
if stage_id:
cr.execute(
"UPDATE encoach_approval_stage SET status='approved', comment=%s, acted_at=%s "
"WHERE id=%s", (body.get('comment', ''), now, stage_id))
cr.execute(
"SELECT id FROM encoach_approval_stage WHERE workflow_id=%s ORDER BY sequence ASC",
(wf_id,))
all_stages = [r[0] for r in cr.fetchall()]
stage = req_rec.current_stage_id
if stage:
stage.write({
'status': 'approved',
'comment': body.get('comment', ''),
'acted_at': datetime.now(),
})
# Move to next stage or finalize.
stages = req_rec.workflow_id.stage_ids.sorted('sequence')
stage_ids = [s.id for s in stages]
try:
idx = all_stages.index(stage_id)
idx = stage_ids.index(stage.id) if stage else -1
except ValueError:
idx = -1
if idx + 1 < len(all_stages):
cr.execute(
"UPDATE encoach_approval_request SET current_stage_id=%s, state='in_progress' "
"WHERE id=%s", (all_stages[idx + 1], req_id))
if 0 <= idx and idx + 1 < len(stage_ids):
next_stage = request.env['encoach.approval.stage'].sudo().browse(stage_ids[idx + 1])
req_rec.write({
'current_stage_id': next_stage.id,
'state': 'in_progress',
})
else:
cr.execute(
"UPDATE encoach_approval_request SET state='approved' WHERE id=%s", (req_id,))
req_rec.write({'state': 'approved'})
return _json_response({'success': True, 'id': req_id})
return _json_response({'success': True, 'id': req_id,
'state': req_rec.state})
except Exception as e:
_logger.exception('approve request failed')
return _json_response({'error': str(e)}, 500)
return _error_response(str(e), 500)
@http.route('/api/approval-requests/<int:req_id>/reject', type='http',
auth='none', methods=['POST'], csrf=False)
@jwt_required
def reject_request(self, req_id, **kw):
try:
cr = _cr()
body = _get_json_body()
now = datetime.now()
cr.execute("SELECT current_stage_id FROM encoach_approval_request WHERE id=%s", (req_id,))
row = cr.fetchone()
if not row:
req_rec = request.env['encoach.approval.request'].sudo().browse(req_id)
if not req_rec.exists():
return _error_response('Request not found', 404)
stage_id = row[0]
if stage_id:
cr.execute(
"UPDATE encoach_approval_stage SET status='rejected', comment=%s, acted_at=%s "
"WHERE id=%s", (body.get('comment', ''), now, stage_id))
cr.execute(
"UPDATE encoach_approval_request SET state='rejected' WHERE id=%s", (req_id,))
return _json_response({'success': True, 'id': req_id})
stage = req_rec.current_stage_id
if stage:
stage.write({
'status': 'rejected',
'comment': body.get('comment', ''),
'acted_at': datetime.now(),
})
req_rec.write({'state': 'rejected'})
return _json_response({'success': True, 'id': req_id,
'state': req_rec.state})
except Exception as e:
_logger.exception('reject request failed')
return _json_response({'error': str(e)}, 500)
return _error_response(str(e), 500)
# ── Users for assignee selection ──
# ── Approvers picker ─────────────────────────────────────────
@http.route('/api/approval-users', type='http', auth='none',
methods=['GET'], csrf=False)
@jwt_required
def list_users(self, **kw):
try:
cr = _cr()
cr.execute(
"SELECT ru.id, rp.name, ru.login "
"FROM res_users ru JOIN res_partner rp ON rp.id = ru.partner_id "
"WHERE ru.active = true AND ru.id > 1 "
"ORDER BY rp.name LIMIT 100")
items = [{'id': r[0], 'name': r[1], 'login': r[2]} for r in cr.fetchall()]
return _json_response({'items': items})
Users = request.env['res.users'].sudo()
domain = [('active', '=', True), ('id', '>', 1)]
search = (kw.get('search') or '').strip()
if search:
domain += ['|', ('name', 'ilike', search), ('login', 'ilike', search)]
recs = Users.search(domain, order='name', limit=100)
items = [{
'id': u.id,
'name': u.name or u.login,
'login': u.login or '',
'email': u.email or '',
} for u in recs]
return _json_response({'items': items, 'results': items, 'total': len(items)})
except Exception as e:
_logger.exception('list users failed')
return _json_response({'error': str(e)}, 500)
_logger.exception('list approval users failed')
return _error_response(str(e), 500)

View File

@@ -12,3 +12,4 @@ from . import exam_assignment
from . import exam_schedule
from . import exam_structure
from . import student_attempt
from . import approval

View File

@@ -0,0 +1,95 @@
"""Approval workflow models.
Powers the admin "Approval Config" page (`/admin/approval-config`) and the
per-record approval requests exposed by `ApprovalWorkflowController`.
Prior to this module the controller hand-rolled raw SQL against tables that
were never created. These ORM models now own the schema so Odoo can create
the tables on install/update and the controller can use a standard API.
"""
from odoo import api, fields, models
STATUS_SELECTION = [
('draft', 'Draft'),
('active', 'Active'),
('archived', 'Archived'),
]
REQUEST_STATE = [
('draft', 'Draft'),
('in_progress', 'In Progress'),
('approved', 'Approved'),
('rejected', 'Rejected'),
('bypassed', 'Bypassed'),
]
STAGE_STATUS = [
('pending', 'Pending'),
('approved', 'Approved'),
('rejected', 'Rejected'),
('skipped', 'Skipped'),
]
class EncoachApprovalWorkflow(models.Model):
_name = 'encoach.approval.workflow'
_description = 'Approval Workflow'
_order = 'create_date desc'
name = fields.Char(required=True)
type = fields.Char(default='custom',
help='Categorical tag e.g. leave, purchase, content, custom.')
status = fields.Selection(STATUS_SELECTION, default='draft', required=True,
help='Draft = not yet routable. Active = available to new requests.')
allow_bypass = fields.Boolean(
default=False,
help='If true, a bypass reason allows the workflow to be skipped.')
entity_id = fields.Many2one('encoach.entity', ondelete='set null')
stage_ids = fields.One2many('encoach.approval.stage', 'workflow_id',
copy=True)
request_ids = fields.One2many('encoach.approval.request', 'workflow_id')
stage_count = fields.Integer(compute='_compute_counts')
request_count = fields.Integer(compute='_compute_counts')
@api.depends('stage_ids', 'request_ids')
def _compute_counts(self):
for rec in self:
rec.stage_count = len(rec.stage_ids)
rec.request_count = len(rec.request_ids)
class EncoachApprovalStage(models.Model):
_name = 'encoach.approval.stage'
_description = 'Approval Workflow Stage'
_order = 'sequence, id'
workflow_id = fields.Many2one('encoach.approval.workflow',
required=True, ondelete='cascade', index=True)
sequence = fields.Integer(default=10, required=True,
help='Stored order; the admin UI speaks `order`.')
approver_id = fields.Many2one('res.users', required=True, ondelete='restrict')
max_days = fields.Integer(default=3,
help='SLA before auto-escalation can fire.')
auto_escalate = fields.Boolean(default=False)
notification_email = fields.Char(
help='Optional CC address; falls back to approver_id.email otherwise.')
status = fields.Selection(STAGE_STATUS, default='pending')
comment = fields.Text()
acted_at = fields.Datetime()
class EncoachApprovalRequest(models.Model):
_name = 'encoach.approval.request'
_description = 'Approval Request'
_order = 'create_date desc'
workflow_id = fields.Many2one('encoach.approval.workflow',
required=True, ondelete='cascade', index=True)
res_model = fields.Char(help='Optional target model (e.g. op.student.leave).')
res_id = fields.Integer(help='Optional target record id inside res_model.')
state = fields.Selection(REQUEST_STATE, default='draft', required=True)
requester_id = fields.Many2one('res.users', ondelete='set null')
current_stage_id = fields.Many2one('encoach.approval.stage', ondelete='set null')
bypass_reason = fields.Text()
created_at = fields.Datetime(default=fields.Datetime.now)

View File

@@ -15,3 +15,6 @@ access_encoach_exam_structure_user,encoach.exam.structure.user,model_encoach_exa
access_encoach_student_attempt_user,encoach.student.attempt.user,model_encoach_student_attempt,base.group_user,1,1,1,1
access_encoach_student_answer_user,encoach.student.answer.user,model_encoach_student_answer,base.group_user,1,1,1,1
access_encoach_student_score_user,encoach.student.score.user,model_encoach_student_score,base.group_user,1,1,1,1
access_encoach_approval_workflow_user,encoach.approval.workflow.user,model_encoach_approval_workflow,base.group_user,1,1,1,1
access_encoach_approval_stage_user,encoach.approval.stage.user,model_encoach_approval_stage,base.group_user,1,1,1,1
access_encoach_approval_request_user,encoach.approval.request.user,model_encoach_approval_request,base.group_user,1,1,1,1
1 id name model_id:id group_id:id perm_read perm_write perm_create perm_unlink
15 access_encoach_student_attempt_user encoach.student.attempt.user model_encoach_student_attempt base.group_user 1 1 1 1
16 access_encoach_student_answer_user encoach.student.answer.user model_encoach_student_answer base.group_user 1 1 1 1
17 access_encoach_student_score_user encoach.student.score.user model_encoach_student_score base.group_user 1 1 1 1
18 access_encoach_approval_workflow_user encoach.approval.workflow.user model_encoach_approval_workflow base.group_user 1 1 1 1
19 access_encoach_approval_stage_user encoach.approval.stage.user model_encoach_approval_stage base.group_user 1 1 1 1
20 access_encoach_approval_request_user encoach.approval.request.user model_encoach_approval_request base.group_user 1 1 1 1

View File

@@ -29,6 +29,7 @@ def _ser_item(i):
'sequence': i.sequence,
'audience': i.audience or 'all',
'is_published': i.is_published,
'video_url': i.video_url or '',
}
@@ -128,6 +129,8 @@ class FaqController(http.Controller):
vals['sequence'] = int(body['sequence'])
if body.get('audience'):
vals['audience'] = body['audience']
if body.get('video_url') is not None:
vals['video_url'] = body['video_url'] or False
rec = request.env['encoach.faq.item'].sudo().create(vals)
return _json_response(_ser_item(rec))
except Exception as e:
@@ -151,6 +154,8 @@ class FaqController(http.Controller):
vals['sequence'] = int(body['sequence'])
if 'is_published' in body:
vals['is_published'] = bool(body['is_published'])
if 'video_url' in body:
vals['video_url'] = body['video_url'] or False
if vals:
rec.write(vals)
return _json_response(_ser_item(rec))

View File

@@ -28,12 +28,43 @@ def _ser_rule(r):
'name': r.name or '',
'event_type': r.event_type or '',
'template': r.template or '',
'is_active': r.is_active,
# Admin UI uses `active`; legacy callers still read `is_active`.
'active': bool(r.is_active),
'is_active': bool(r.is_active),
'recipients': r.recipients or 'all',
'channels': r.channels or '',
'days_before': r.days_before or 0,
'frequency': r.frequency or 'once',
'channel': r.channel or 'in_app',
'entity_id': r.entity_id.id or None,
'entity_name': r.entity_id.name if r.entity_id else None,
}
def _apply_rule_writes(body, vals):
"""Translate the incoming JSON body into ORM write values.
Accepts both the new (`active`, `channel`, `days_before`, `frequency`)
and legacy (`is_active`, `channels`) contracts.
"""
for k in ('name', 'event_type', 'template', 'recipients', 'frequency', 'channel'):
if k in body and body[k] is not None:
vals[k] = body[k]
if 'channels' in body and body['channels'] is not None:
vals['channels'] = body['channels']
if 'days_before' in body and body['days_before'] is not None:
try:
vals['days_before'] = int(body['days_before'])
except (TypeError, ValueError):
pass
if 'entity_id' in body:
vals['entity_id'] = int(body['entity_id']) if body['entity_id'] else False
if 'active' in body:
vals['is_active'] = bool(body['active'])
elif 'is_active' in body:
vals['is_active'] = bool(body['is_active'])
return vals
class NotificationsController(http.Controller):
@http.route('/api/notifications', type='http', auth='public', methods=['GET'], csrf=False)
@@ -109,21 +140,19 @@ class NotificationsController(http.Controller):
def create_rule(self, **kw):
try:
body = _get_json_body()
if not body.get('name'):
return _error_response('name is required', 400)
if not body.get('event_type'):
return _error_response('event_type is required', 400)
vals = {
'name': body.get('name', ''),
'event_type': body.get('event_type', ''),
}
if body.get('template'):
vals['template'] = body['template']
if body.get('recipients'):
vals['recipients'] = body['recipients']
if body.get('channels'):
vals['channels'] = body['channels']
if 'is_active' in body:
vals['is_active'] = bool(body['is_active'])
_apply_rule_writes(body, vals)
rec = request.env['encoach.notification.rule'].sudo().create(vals)
return _json_response(_ser_rule(rec))
except Exception as e:
_logger.exception('create notification rule failed')
return _error_response(str(e), 500)
@http.route('/api/notification-rules/<int:rid>', type='http', auth='public', methods=['PATCH', 'PUT'], csrf=False)
@@ -135,15 +164,12 @@ class NotificationsController(http.Controller):
return _error_response('Not found', 404)
body = _get_json_body()
vals = {}
for k in ('name', 'event_type', 'template', 'recipients', 'channels'):
if k in body:
vals[k] = body[k]
if 'is_active' in body:
vals['is_active'] = bool(body['is_active'])
_apply_rule_writes(body, vals)
if vals:
rec.write(vals)
return _json_response(_ser_rule(rec))
except Exception as e:
_logger.exception('update notification rule failed')
return _error_response(str(e), 500)
@http.route('/api/notification-rules/<int:rid>', type='http', auth='public', methods=['DELETE'], csrf=False)

View File

@@ -1,4 +1,14 @@
from odoo import models, fields
from odoo import api, models, fields
FAQ_AUDIENCE = [
('all', 'All'),
('both', 'Both (Student + Entity)'),
('student', 'Students'),
('teacher', 'Teachers'),
('admin', 'Admins'),
('entity', 'Entity Users'),
]
class EncoachFaqCategory(models.Model):
@@ -8,16 +18,12 @@ class EncoachFaqCategory(models.Model):
name = fields.Char(required=True)
sequence = fields.Integer(default=10)
audience = fields.Selection([
('all', 'All'),
('student', 'Students'),
('teacher', 'Teachers'),
('admin', 'Admins'),
], default='all')
audience = fields.Selection(FAQ_AUDIENCE, default='all')
is_published = fields.Boolean(default=True)
item_ids = fields.One2many('encoach.faq.item', 'category_id')
item_count = fields.Integer(compute='_compute_item_count', store=True)
@api.depends('item_ids')
def _compute_item_count(self):
for rec in self:
rec.item_count = len(rec.item_ids)
@@ -32,10 +38,6 @@ class EncoachFaqItem(models.Model):
question = fields.Char(required=True)
answer = fields.Text(required=True)
sequence = fields.Integer(default=10)
audience = fields.Selection([
('all', 'All'),
('student', 'Students'),
('teacher', 'Teachers'),
('admin', 'Admins'),
], default='all')
audience = fields.Selection(FAQ_AUDIENCE, default='all')
is_published = fields.Boolean(default=True)
video_url = fields.Char(help='Optional walkthrough video URL.')

View File

@@ -26,7 +26,8 @@ class EncoachNotificationRule(models.Model):
_description = 'Notification Rule'
name = fields.Char(required=True)
event_type = fields.Char(required=True)
event_type = fields.Char(required=True, help='e.g. deadline, chapter_unlock, '
'result_release, assignment, exam')
template = fields.Text()
is_active = fields.Boolean(default=True)
recipients = fields.Selection([
@@ -35,7 +36,23 @@ class EncoachNotificationRule(models.Model):
('teachers', 'Teachers Only'),
('admins', 'Admins Only'),
], default='all')
channels = fields.Char(help='Comma-separated: email,push,in_app')
channels = fields.Char(help='Legacy: comma-separated email,push,in_app. '
'New code should use `channel` instead.')
# New fields aligned with the admin Configuration UI (NotificationRules page)
days_before = fields.Integer(default=1,
help='Lead time in days before the event fires.')
frequency = fields.Selection([
('once', 'Once'),
('daily', 'Daily'),
('custom', 'Custom schedule'),
], default='once')
channel = fields.Selection([
('in_app', 'In-App'),
('email', 'Email'),
('both', 'Both'),
], default='in_app')
entity_id = fields.Many2one('encoach.entity', ondelete='set null')
class EncoachNotificationPreference(models.Model):

View File

@@ -6,19 +6,12 @@ import { Input } from "@/components/ui/input";
import { Label } from "@/components/ui/label";
import { Switch } from "@/components/ui/switch";
import { Dialog, DialogContent, DialogHeader, DialogTitle, DialogTrigger } from "@/components/ui/dialog";
import { Plus, Trash2, Loader2, ChevronRight, Settings2, ArrowRight } from "lucide-react";
import { approvalsService, type ApprovalWorkflow } from "@/services/approvals.service";
import { Plus, Trash2, Loader2, Settings2, ArrowRight } from "lucide-react";
import { approvalsService, type ApprovalWorkflow, type ApprovalStage } from "@/services/approvals.service";
import { useQuery, useMutation, useQueryClient } from "@tanstack/react-query";
import { useToast } from "@/hooks/use-toast";
interface WorkflowStage {
order: number;
approver_id: number;
approver_name: string;
max_days?: number;
notification_email?: string;
auto_escalate?: boolean;
}
type WorkflowStage = Omit<ApprovalStage, "order"> & { order: number };
export default function ApprovalWorkflowConfig() {
const { toast } = useToast();
@@ -28,16 +21,21 @@ export default function ApprovalWorkflowConfig() {
queryKey: ["approvals", "list"],
queryFn: () => approvalsService.list(),
});
const workflows = workflowsData?.results ?? [];
const workflows: ApprovalWorkflow[] = workflowsData?.results ?? workflowsData?.items ?? [];
type WorkflowWriteInput = Partial<ApprovalWorkflow> & {
steps?: Partial<ApprovalStage>[];
bypass_enabled?: boolean;
};
const createWorkflow = useMutation({
mutationFn: (data: Partial<ApprovalWorkflow>) => approvalsService.create(data),
mutationFn: (data: WorkflowWriteInput) => approvalsService.create(data),
onSuccess: () => qc.invalidateQueries({ queryKey: ["approvals"] }),
onError: () => toast({ title: "Error", description: "Failed to create workflow", variant: "destructive" }),
});
const updateWorkflow = useMutation({
mutationFn: ({ id, data }: { id: number; data: Partial<ApprovalWorkflow> }) => approvalsService.update(id, data),
mutationFn: ({ id, data }: { id: number; data: WorkflowWriteInput }) => approvalsService.update(id, data),
onSuccess: () => qc.invalidateQueries({ queryKey: ["approvals"] }),
onError: () => toast({ title: "Error", description: "Failed to update workflow", variant: "destructive" }),
});
@@ -72,8 +70,26 @@ export default function ApprovalWorkflowConfig() {
};
const handleCreate = () => {
const valid = stages.filter(s => s.approver_id > 0);
if (valid.length === 0) {
toast({ title: "Validation", description: "Add at least one stage with an approver ID", variant: "destructive" });
return;
}
createWorkflow.mutate(
{ name: wfName, steps: stages.map(s => ({ order: s.order, approver_id: s.approver_id, approver_name: s.approver_name })), status: "draft" } as Partial<ApprovalWorkflow>,
{
name: wfName,
status: "draft",
allow_bypass: bypassEnabled,
bypass_enabled: bypassEnabled,
steps: valid.map(s => ({
order: s.order,
approver_id: s.approver_id,
approver_name: s.approver_name,
max_days: s.max_days,
auto_escalate: s.auto_escalate,
notification_email: s.notification_email,
})),
},
{ onSuccess: () => { toast({ title: "Workflow Created" }); setShowCreate(false); resetForm(); } },
);
};

View File

@@ -1,30 +1,66 @@
import { api } from "@/lib/api-client";
import type { PaginatedResponse, PaginationParams, ApiSuccessResponse } from "@/types";
import type { PaginationParams, ApiSuccessResponse } from "@/types";
export interface ApprovalStage {
id?: number;
/** 1-based UI position; serialized alongside sequence for compatibility. */
order: number;
sequence?: number;
approver_id: number;
approver_name: string;
max_days?: number;
auto_escalate?: boolean;
notification_email?: string;
status?: "pending" | "approved" | "rejected" | "skipped";
comment?: string;
acted_at?: string | null;
}
export interface ApprovalWorkflow {
id: number;
name: string;
entity_id: number;
entity_name: string;
steps: { order: number; approver_id: number; approver_name: string }[];
status: "active" | "draft";
type?: string;
entity_id: number | null;
entity_name: string | null;
steps: ApprovalStage[];
status: "draft" | "active" | "archived";
allow_bypass?: boolean;
stage_count?: number;
request_count?: number;
created_at: string;
}
export interface ApprovalWorkflowListResponse {
items?: ApprovalWorkflow[];
results?: ApprovalWorkflow[];
total: number;
}
export const approvalsService = {
async list(params?: PaginationParams): Promise<PaginatedResponse<ApprovalWorkflow>> {
return api.get<PaginatedResponse<ApprovalWorkflow>>("/approval-workflows", params as Record<string, string | number | boolean | undefined>);
async list(params?: PaginationParams & { status?: string; entity_id?: number }): Promise<ApprovalWorkflowListResponse> {
return api.get<ApprovalWorkflowListResponse>(
"/approval-workflows",
params as Record<string, string | number | boolean | undefined>,
);
},
async create(data: Partial<ApprovalWorkflow>): Promise<ApprovalWorkflow> {
async get(id: number): Promise<ApprovalWorkflow> {
return api.get<ApprovalWorkflow>(`/approval-workflows/${id}`);
},
async create(data: Partial<ApprovalWorkflow> & { steps?: Partial<ApprovalStage>[]; bypass_enabled?: boolean }): Promise<ApprovalWorkflow> {
return api.post<ApprovalWorkflow>("/approval-workflows", data);
},
async update(id: number, data: Partial<ApprovalWorkflow>): Promise<ApprovalWorkflow> {
async update(id: number, data: Partial<ApprovalWorkflow> & { steps?: Partial<ApprovalStage>[]; bypass_enabled?: boolean }): Promise<ApprovalWorkflow> {
return api.patch<ApprovalWorkflow>(`/approval-workflows/${id}`, data);
},
async delete(id: number): Promise<ApiSuccessResponse> {
return api.delete<ApiSuccessResponse>(`/approval-workflows/${id}`);
},
async listUsers(search?: string): Promise<{ items: { id: number; name: string; login: string; email?: string }[]; total: number }> {
return api.get("/approval-users", search ? { search } : undefined);
},
};

336
test_config_flows.py Normal file
View File

@@ -0,0 +1,336 @@
#!/usr/bin/env python3
"""API smoke tests for the admin Configuration section.
Covers the three pages in `AdminLmsLayout.tsx` under "Configuration":
* /admin/faq -> /api/faq/{categories,items}
* /admin/notification-rules -> /api/notification-rules
* /admin/approval-config -> /api/approval-workflows
Usage:
python3 test_config_flows.py
"""
import json
import sys
import time
from urllib import request as urlreq
from urllib.error import HTTPError
BASE = "http://localhost:8069/api"
TOKEN = None
GREEN = "\033[32m"; RED = "\033[31m"; RESET = "\033[0m"; BOLD = "\033[1m"
results = []
def log(page, action, ok_, note=""):
results.append((page, action, ok_, note))
tag = f"{GREEN}PASS{RESET}" if ok_ else f"{RED}FAIL{RESET}"
print(f" [{tag}] {page:22s} {action:42s} {note}")
def _req(method, path, body=None, params=None):
url = BASE + path
if params:
qs = "&".join(f"{k}={v}" for k, v in params.items() if v is not None)
url += ("?" + qs) if qs else ""
data = None
headers = {"Accept": "application/json"}
if TOKEN:
headers["Authorization"] = f"Bearer {TOKEN}"
if body is not None:
data = json.dumps(body).encode()
headers["Content-Type"] = "application/json"
req = urlreq.Request(url, data=data, method=method, headers=headers)
try:
with urlreq.urlopen(req, timeout=20) as resp:
raw = resp.read().decode()
return resp.status, (json.loads(raw) if raw else {})
except HTTPError as e:
raw = e.read().decode() if e.fp else ""
try:
payload = json.loads(raw)
except Exception:
payload = {"error": raw[:200]}
return e.code, payload
except Exception as e:
return 0, {"error": str(e)}
def GET(path, **params): return _req("GET", path, params=params)
def POST(path, body=None): return _req("POST", path, body=body or {})
def PATCH(path, body=None): return _req("PATCH", path, body=body or {})
def DELETE(path): return _req("DELETE", path)
def ok(s): return 200 <= s < 300
def login():
global TOKEN
s, b = POST("/login", {"email": "admin", "password": "admin"})
TOKEN = (b or {}).get("token")
return bool(TOKEN)
# ── 1. FAQ ───────────────────────────────────────────────────────────────
def test_faq():
print(f"{BOLD}== /admin/faq =={RESET}")
ts = int(time.time())
# LIST categories
s, b = GET("/faq/categories")
log("faq", "LIST categories", ok(s) and isinstance(b, list),
f"status={s} count={len(b) if isinstance(b, list) else 'n/a'}")
# CREATE category with audience='both' (frontend-only value)
cat_name = f"Cfg Test Cat {ts}"
s, b = POST("/faq/categories", {"name": cat_name, "audience": "both"})
cat_id = (b or {}).get("id")
log("faq", "CREATE category audience=both", ok(s) and cat_id and b.get("audience") == "both",
f"status={s} id={cat_id} audience={(b or {}).get('audience')}")
# CREATE category with audience='entity'
s2, b2 = POST("/faq/categories", {"name": f"Cfg Entity {ts}", "audience": "entity"})
entity_cat_id = (b2 or {}).get("id")
log("faq", "CREATE category audience=entity", ok(s2) and entity_cat_id and b2.get("audience") == "entity",
f"status={s2} id={entity_cat_id} audience={(b2 or {}).get('audience')}")
# EDIT category
if cat_id:
s, b = PATCH(f"/faq/categories/{cat_id}", {"name": cat_name + " (edited)"})
log("faq", "EDIT category name", ok(s) and "edited" in (b or {}).get("name", ""),
f"status={s} name={(b or {}).get('name')}")
# CREATE item with video_url (regression: previously dropped)
if cat_id:
s, b = POST("/faq/items", {
"question": f"Q? {ts}",
"answer": "Because reasons.",
"category_id": cat_id,
"audience": "both",
"video_url": "https://example.com/walkthrough.mp4",
})
item_id = (b or {}).get("id")
log("faq", "CREATE item audience=both +video",
ok(s) and item_id and b.get("video_url") == "https://example.com/walkthrough.mp4",
f"status={s} id={item_id} video_url={(b or {}).get('video_url')}")
else:
item_id = None
# LIST items filtered by category
if cat_id:
s, b = GET("/faq/items", category_id=cat_id)
log("faq", "LIST items filter category_id",
ok(s) and isinstance(b, list) and any(it["id"] == item_id for it in b),
f"status={s} count={len(b) if isinstance(b, list) else 'n/a'}")
# EDIT item: change audience 'both' -> 'student'
if item_id:
s, b = PATCH(f"/faq/items/{item_id}", {"audience": "student"})
log("faq", "EDIT item audience both->student",
ok(s) and b.get("audience") == "student",
f"status={s} audience={(b or {}).get('audience')}")
# DELETE item
if item_id:
s, _ = DELETE(f"/faq/items/{item_id}")
log("faq", "DELETE item", ok(s), f"status={s}")
# DELETE categories (cleanup)
for cid, label in [(cat_id, "cat"), (entity_cat_id, "entity cat")]:
if cid:
s, _ = DELETE(f"/faq/categories/{cid}")
log("faq", f"DELETE {label}", ok(s), f"status={s} id={cid}")
# ── 2. Notification Rules ───────────────────────────────────────────────
def test_notification_rules():
print(f"{BOLD}== /admin/notification-rules =={RESET}")
ts = int(time.time())
# LIST
s, b = GET("/notification-rules")
log("notif-rules", "LIST", ok(s) and isinstance(b, list),
f"status={s} count={len(b) if isinstance(b, list) else 'n/a'}")
# CREATE rule with new fields (days_before, frequency, channel, active)
payload = {
"name": f"Deadline reminder {ts}",
"event_type": "deadline",
"days_before": 3,
"frequency": "daily",
"channel": "email",
}
s, b = POST("/notification-rules", payload)
rid = (b or {}).get("id")
correct = (
ok(s) and rid
and b.get("days_before") == 3
and b.get("frequency") == "daily"
and b.get("channel") == "email"
and b.get("active") is True # default from is_active
)
log("notif-rules", "CREATE with new fields", correct,
f"status={s} id={rid} days={(b or {}).get('days_before')} "
f"freq={(b or {}).get('frequency')} ch={(b or {}).get('channel')} "
f"active={(b or {}).get('active')}")
# EDIT: send `active: False` (frontend-speak) and expect `is_active` to flip
if rid:
s, b = PATCH(f"/notification-rules/{rid}", {"active": False, "days_before": 7})
log("notif-rules", "EDIT active=false + days_before=7",
ok(s) and b.get("active") is False and b.get("is_active") is False
and b.get("days_before") == 7,
f"status={s} active={(b or {}).get('active')} "
f"is_active={(b or {}).get('is_active')} days={(b or {}).get('days_before')}")
# EDIT: change frequency + channel
if rid:
s, b = PATCH(f"/notification-rules/{rid}", {"frequency": "once", "channel": "both"})
log("notif-rules", "EDIT frequency=once channel=both",
ok(s) and b.get("frequency") == "once" and b.get("channel") == "both",
f"status={s} freq={(b or {}).get('frequency')} ch={(b or {}).get('channel')}")
# CREATE without required field -> 400
s, _ = POST("/notification-rules", {"event_type": "exam"})
log("notif-rules", "CREATE rejects missing name", s == 400, f"status={s}")
# DELETE
if rid:
s, _ = DELETE(f"/notification-rules/{rid}")
log("notif-rules", "DELETE", ok(s), f"status={s} id={rid}")
# ── 3. Approval Workflows ───────────────────────────────────────────────
def test_approval_workflows():
print(f"{BOLD}== /admin/approval-config =={RESET}")
ts = int(time.time())
# LIST (previously would have emitted {items:[], total:0} silently; now
# confirm it also emits `results` for the frontend PaginatedResponse shape)
s, b = GET("/approval-workflows")
has_envelope = (
isinstance(b, dict)
and "items" in b and "results" in b and "total" in b
)
log("approvals", "LIST envelope items+results+total",
ok(s) and has_envelope,
f"status={s} keys={sorted((b or {}).keys()) if isinstance(b, dict) else 'n/a'}")
# Need an approver (any admin user). Ask the approval-users endpoint.
s_u, b_u = GET("/approval-users")
users = (b_u or {}).get("items") or []
if not users:
log("approvals", "LIST approval-users", False, f"status={s_u} (no users)")
return
approver = users[0]
log("approvals", "LIST approval-users", True,
f"status={s_u} first={approver.get('name')} id={approver.get('id')}")
# CREATE workflow with full stage fields
payload = {
"name": f"Leave Approval {ts}",
"type": "leave",
"status": "draft",
"allow_bypass": True,
"bypass_enabled": True,
"steps": [
{
"order": 1,
"approver_id": approver["id"],
"approver_name": approver["name"],
"max_days": 2,
"auto_escalate": True,
"notification_email": "esc@example.com",
},
{
"order": 2,
"approver_id": approver["id"],
"approver_name": approver["name"],
"max_days": 5,
"auto_escalate": False,
},
],
}
s, b = POST("/approval-workflows", payload)
wf_id = (b or {}).get("id")
stages = (b or {}).get("steps") or []
correct = (
ok(s) and wf_id
and b.get("allow_bypass") is True
and len(stages) == 2
and stages[0].get("max_days") == 2
and stages[0].get("auto_escalate") is True
and stages[0].get("notification_email") == "esc@example.com"
)
log("approvals", "CREATE with 2 stages + bypass", correct,
f"status={s} id={wf_id} stages={len(stages)} "
f"bypass={(b or {}).get('allow_bypass')}")
# GET single
if wf_id:
s, b = GET(f"/approval-workflows/{wf_id}")
log("approvals", "GET detail", ok(s) and b.get("id") == wf_id,
f"status={s}")
# PATCH status draft -> active (used by the toggle switch)
if wf_id:
s, b = PATCH(f"/approval-workflows/{wf_id}", {"status": "active"})
log("approvals", "PATCH status draft->active",
ok(s) and b.get("status") == "active",
f"status={s} new_status={(b or {}).get('status')}")
# PATCH stages replacement
if wf_id:
s, b = PATCH(f"/approval-workflows/{wf_id}", {
"steps": [{
"order": 1,
"approver_id": approver["id"],
"approver_name": approver["name"],
"max_days": 10,
"auto_escalate": False,
}],
})
new_stages = (b or {}).get("steps") or []
log("approvals", "PATCH replace stages",
ok(s) and len(new_stages) == 1 and new_stages[0].get("max_days") == 10,
f"status={s} stages={len(new_stages)}")
# CREATE rejects missing name
s, _ = POST("/approval-workflows", {"type": "custom"})
log("approvals", "CREATE rejects missing name", s == 400, f"status={s}")
# DELETE
if wf_id:
s, _ = DELETE(f"/approval-workflows/{wf_id}")
log("approvals", "DELETE", ok(s), f"status={s} id={wf_id}")
# ── Main ────────────────────────────────────────────────────────────────
if __name__ == "__main__":
print(f"{BOLD}Configuration Section API Smoke Tests{RESET}")
if not login():
print(f"{RED}Login failed. Is Odoo running on :8069 with admin/admin?{RESET}")
sys.exit(1)
print(f" logged in, token={TOKEN[:12] if TOKEN else 'none'}...\n")
test_faq()
print()
test_notification_rules()
print()
test_approval_workflows()
total = len(results)
passed = sum(1 for r in results if r[2])
failed = total - passed
print("\n" + "=" * 70)
print(f"{BOLD}Summary: {GREEN}{passed} passed{RESET}, "
f"{RED}{failed} failed{RESET}, {total} total")
if failed:
print(f"\n{RED}Failing:{RESET}")
for page, action, ok_, note in results:
if not ok_:
print(f" - {page:22s} {action:42s} {note}")
sys.exit(2)