Raphael Collet
63ee425b46
This aligns the behavior of Form with the web client form view. Part-of: odoo/odoo#116779
1054 lines
42 KiB
Python
1054 lines
42 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
The module :mod:`odoo.tests.form` provides an implementation of a client form
|
|
view for server-side unit tests.
|
|
"""
|
|
import collections
|
|
import json
|
|
import logging
|
|
import operator
|
|
import time
|
|
from datetime import datetime, date
|
|
from dateutil.relativedelta import relativedelta
|
|
|
|
from lxml import etree
|
|
|
|
import odoo
|
|
from odoo.models import BaseModel
|
|
from odoo.fields import Command
|
|
from odoo.osv import expression
|
|
from odoo.osv.expression import normalize_domain, TRUE_LEAF, FALSE_LEAF
|
|
from odoo.tools.safe_eval import safe_eval
|
|
|
|
_logger = logging.getLogger(__name__)
|
|
|
|
|
|
class Form:
|
|
""" Server-side form view implementation (partial)
|
|
|
|
Implements much of the "form view" manipulation flow, such that server-side
|
|
tests can more properly reflect the behaviour which would be observed when
|
|
manipulating the interface:
|
|
|
|
* call the relevant onchanges on "creation";
|
|
* call the relevant onchanges on setting fields;
|
|
* properly handle defaults & onchanges around x2many fields.
|
|
|
|
Saving the form returns the current record (which means the created record
|
|
if in creation mode). It can also be accessed as ``form.record``, but only
|
|
when the form has no pending changes.
|
|
|
|
Regular fields can just be assigned directly to the form. In the case
|
|
of :class:`~odoo.fields.Many2one` fields, one can assign a recordset::
|
|
|
|
# empty recordset => creation mode
|
|
f = Form(self.env['sale.order'])
|
|
f.partner_id = a_partner
|
|
so = f.save()
|
|
|
|
One can also use the form as a context manager to create or edit a record.
|
|
The changes are automatically saved at the end of the scope::
|
|
|
|
with Form(self.env['sale.order']) as f1:
|
|
f1.partner_id = a_partner
|
|
# f1 is saved here
|
|
|
|
# retrieve the created record
|
|
so = f1.record
|
|
|
|
# call Form on record => edition mode
|
|
with Form(so) as f2:
|
|
f2.payment_term_id = env.ref('account.account_payment_term_15days')
|
|
# f2 is saved here
|
|
|
|
For :class:`~odoo.fields.Many2many` fields, the field itself is a
|
|
:class:`~odoo.tests.common.M2MProxy` and can be altered by adding or
|
|
removing records::
|
|
|
|
with Form(user) as u:
|
|
u.groups_id.add(env.ref('account.group_account_manager'))
|
|
u.groups_id.remove(id=env.ref('base.group_portal').id)
|
|
|
|
Finally :class:`~odoo.fields.One2many` are reified as :class:`~O2MProxy`.
|
|
|
|
Because the :class:`~odoo.fields.One2many` only exists through its parent,
|
|
it is manipulated more directly by creating "sub-forms" with
|
|
the :meth:`~O2MProxy.new` and :meth:`~O2MProxy.edit` methods. These would
|
|
normally be used as context managers since they get saved in the parent
|
|
record::
|
|
|
|
with Form(so) as f3:
|
|
f.partner_id = a_partner
|
|
# add support
|
|
with f3.order_line.new() as line:
|
|
line.product_id = env.ref('product.product_product_2')
|
|
# add a computer
|
|
with f3.order_line.new() as line:
|
|
line.product_id = env.ref('product.product_product_3')
|
|
# we actually want 5 computers
|
|
with f3.order_line.edit(1) as line:
|
|
line.product_uom_qty = 5
|
|
# remove support
|
|
f3.order_line.remove(index=0)
|
|
# SO is saved here
|
|
|
|
:param record: empty or singleton recordset. An empty recordset will put
|
|
the view in "creation" mode from default values, while a
|
|
singleton will put it in "edit" mode and only load the
|
|
view's data.
|
|
:type record: odoo.models.Model
|
|
:param view: the id, xmlid or actual view object to use for onchanges and
|
|
view constraints. If none is provided, simply loads the
|
|
default view for the model.
|
|
:type view: int | str | odoo.model.Model
|
|
|
|
.. versionadded:: 12.0
|
|
"""
|
|
def __init__(self, record, view=None):
|
|
assert isinstance(record, BaseModel)
|
|
assert len(record) <= 1
|
|
|
|
# use object.__setattr__ to bypass Form's override of __setattr__
|
|
object.__setattr__(self, '_record', record)
|
|
object.__setattr__(self, '_env', record.env)
|
|
|
|
# determine view and process it
|
|
if isinstance(view, BaseModel):
|
|
assert view._name == 'ir.ui.view', "the view parameter must be a view id, xid or record, got %s" % view
|
|
view_id = view.id
|
|
elif isinstance(view, str):
|
|
view_id = record.env.ref(view).id
|
|
else:
|
|
view_id = view or False
|
|
|
|
views = record.get_views([(view_id, 'form')])
|
|
object.__setattr__(self, '_models_info', views['models'])
|
|
# self._models_info = {model_name: {field_name: field_info}}
|
|
tree = etree.fromstring(views['views']['form']['arch'])
|
|
view = self._process_view(tree, record)
|
|
object.__setattr__(self, '_view', view)
|
|
# self._view = {
|
|
# 'tree': view_arch_etree,
|
|
# 'fields': {field_name: field_info},
|
|
# 'modifiers': {field_name: {modifier: domain}},
|
|
# 'contexts': {field_name: field_context_str},
|
|
# 'onchange': onchange_spec,
|
|
# }
|
|
|
|
# determine record values
|
|
vals = dict.fromkeys(view['fields'], False)
|
|
object.__setattr__(self, '_values', vals)
|
|
object.__setattr__(self, '_changed', set())
|
|
if record:
|
|
assert record.id, "editing unstored records is not supported"
|
|
vals.update(read_record(record, self._view['fields']))
|
|
else:
|
|
self._init_from_defaults()
|
|
|
|
def _process_view(self, tree, model, level=2):
|
|
""" Post-processes to augment the view_get with:
|
|
* an id field (may not be present if not in the view but needed)
|
|
* pre-processed modifiers (map of modifier name to json-loaded domain)
|
|
* pre-processed onchanges list
|
|
"""
|
|
fields = {'id': {'type': 'id'}}
|
|
modifiers = {'id': {'required': [FALSE_LEAF], 'readonly': [TRUE_LEAF]}}
|
|
contexts = {}
|
|
view = {
|
|
'tree': tree,
|
|
'fields': fields,
|
|
'modifiers': modifiers,
|
|
'contexts': contexts,
|
|
}
|
|
# pre-resolve modifiers & bind to arch toplevel
|
|
eval_context = {
|
|
"uid": self._env.user.id,
|
|
"tz": self._env.user.tz,
|
|
"lang": self._env.user.lang,
|
|
"datetime": datetime,
|
|
"context_today": lambda: odoo.fields.Date.context_today(self._env.user),
|
|
"relativedelta": relativedelta,
|
|
"current_date": time.strftime("%Y-%m-%d"),
|
|
"allowed_company_ids": [self._env.user.company_id.id],
|
|
"context": {},
|
|
}
|
|
# retrieve <field> nodes at the current level
|
|
flevel = tree.xpath('count(ancestor::field)')
|
|
for node in tree.xpath(f'.//field[count(ancestor::field) = {flevel}]'):
|
|
field_name = node.get('name')
|
|
|
|
# add field_info into fields
|
|
field_info = self._models_info[model._name].get(field_name) or {'type': None}
|
|
fields[field_name] = field_info
|
|
|
|
# determine modifiers
|
|
field_modifiers = {}
|
|
for modifier, domain in json.loads(node.get('modifiers', '{}')).items():
|
|
if isinstance(domain, int):
|
|
domain = [TRUE_LEAF] if domain else [FALSE_LEAF]
|
|
elif isinstance(domain, str):
|
|
domain = safe_eval(domain, eval_context)
|
|
field_modifiers[modifier] = normalize_domain(domain)
|
|
|
|
# Combine the field modifiers with its ancestor modifiers with an
|
|
# OR: A field is invisible if its own invisible modifier is True OR
|
|
# if one of its ancestor invisible modifier is True
|
|
for ancestor in node.xpath(f'ancestor::*[@modifiers][count(ancestor::field) = {flevel}]'):
|
|
ancestor_modifiers = json.loads(ancestor.get('modifiers'))
|
|
if 'invisible' in ancestor_modifiers:
|
|
domain = ancestor_modifiers['invisible']
|
|
if isinstance(domain, int):
|
|
domain = [TRUE_LEAF] if domain else [FALSE_LEAF]
|
|
elif isinstance(domain, str):
|
|
domain = safe_eval(domain, eval_context)
|
|
domain = normalize_domain(domain)
|
|
field_modifiers['invisible'] = expression.OR([
|
|
domain,
|
|
field_modifiers.get('invisible', [FALSE_LEAF]),
|
|
])
|
|
|
|
# merge field_modifiers into modifiers[field_name]
|
|
if field_name in modifiers:
|
|
# The field is several times in the view, combine the modifier
|
|
# domains with an AND: a field is X if all occurences of the
|
|
# field in the view are X.
|
|
for modifier in field_modifiers.keys() | modifiers[field_name].keys():
|
|
field_modifiers[modifier] = expression.AND([
|
|
modifiers[field_name].get(modifier, [FALSE_LEAF]),
|
|
field_modifiers.get(modifier, [FALSE_LEAF]),
|
|
])
|
|
|
|
modifiers[field_name] = field_modifiers
|
|
|
|
# determine context
|
|
ctx = node.get('context')
|
|
if ctx:
|
|
contexts[field_name] = ctx
|
|
|
|
# FIXME: better widgets support
|
|
# NOTE: selection breaks because of m2o widget=selection
|
|
if node.get('widget') in ['many2many']:
|
|
field_info['type'] = node.get('widget')
|
|
|
|
# determine subview to use for edition
|
|
if level and field_info['type'] == 'one2many':
|
|
field_info['invisible'] = field_modifiers.get('invisible') == [TRUE_LEAF]
|
|
field_info['edition_view'] = self._get_one2many_edition_view(field_info, node, level)
|
|
|
|
view['onchange'] = model._onchange_spec({'arch': etree.tostring(tree)})
|
|
|
|
return view
|
|
|
|
def _get_one2many_edition_view(self, field_info, node, level):
|
|
""" Return a suitable view for editing records into a one2many field. """
|
|
submodel = self._env[field_info['relation']]
|
|
|
|
# by simplicity, ensure we always have tree and form views
|
|
views = {
|
|
view.tag: view for view in node.xpath('./*[descendant::field]')
|
|
}
|
|
for view_type in ['tree', 'form']:
|
|
if view_type in views:
|
|
continue
|
|
if field_info['invisible']:
|
|
# add an empty view
|
|
views[view_type] = etree.Element(view_type)
|
|
continue
|
|
refs = self._env['ir.ui.view']._get_view_refs(node)
|
|
subviews = submodel.with_context(**refs).get_views([(None, view_type)])
|
|
subnode = etree.fromstring(subviews['views'][view_type]['arch'])
|
|
views[view_type] = subnode
|
|
node.append(subnode)
|
|
for model_name, fields in subviews['models'].items():
|
|
self._models_info.setdefault(model_name, {}).update(fields)
|
|
|
|
# pick the first editable subview
|
|
view_type = next(
|
|
vtype for vtype in node.get('mode', 'tree').split(',') if vtype != 'form'
|
|
)
|
|
if not (view_type == 'tree' and views['tree'].get('editable')):
|
|
view_type = 'form'
|
|
|
|
# don't recursively process o2ms in o2ms
|
|
return self._process_view(views[view_type], submodel, level=level-1)
|
|
|
|
def __str__(self):
|
|
return f"<{type(self).__name__} {self._record}>"
|
|
|
|
def _init_from_defaults(self):
|
|
""" Initialize the form for a new record. """
|
|
vals = self._values
|
|
vals.clear()
|
|
vals['id'] = False
|
|
|
|
# call onchange with no field; this retrieves default values, applies
|
|
# onchanges and return the result
|
|
self._perform_onchange()
|
|
# fill in whatever fields are still missing with falsy values
|
|
vals.update({
|
|
field_name: _cleanup_from_default(field_info['type'], False)
|
|
for field_name, field_info in self._view['fields'].items()
|
|
if field_name not in vals
|
|
})
|
|
# mark all fields as modified
|
|
self._changed.update(self._view['fields'])
|
|
|
|
def __getattr__(self, field_name):
|
|
""" Return the current value of the given field. """
|
|
field_info = self._view['fields'].get(field_name)
|
|
assert field_info is not None, f"{field_name!r} was not found in the view"
|
|
|
|
value = self._values[field_name]
|
|
if field_info['type'] == 'many2one':
|
|
Model = self._env[field_info['relation']]
|
|
return Model.browse(value)
|
|
elif field_info['type'] == 'one2many':
|
|
return O2MProxy(self, field_name)
|
|
elif field_info['type'] == 'many2many':
|
|
return M2MProxy(self, field_name)
|
|
return value
|
|
|
|
def __setattr__(self, field_name, value):
|
|
""" Set the given field to the given value, and proceed with the expected onchanges. """
|
|
field_info = self._view['fields'].get(field_name)
|
|
assert field_info is not None, f"{field_name!r} was not found in the view"
|
|
assert field_info['type'] not in ('one2many', 'many2many'), "Can't set an x2many field directly, use its proxy instead"
|
|
assert not self._get_modifier(field_name, 'readonly'), f"can't write on readonly field {field_name!r}"
|
|
assert not self._get_modifier(field_name, 'invisible'), f"can't write on invisible field {field_name!r}"
|
|
|
|
if field_info['type'] == 'many2one':
|
|
assert isinstance(value, BaseModel) and value._name == field_info['relation']
|
|
value = value.id
|
|
|
|
self._values[field_name] = value
|
|
self._perform_onchange(field_name)
|
|
|
|
def _get_modifier(self, field_name, modifier, *, view=None, vals=None):
|
|
if view is None:
|
|
view = self._view
|
|
|
|
domain = view['modifiers'][field_name].get(modifier, False)
|
|
if isinstance(domain, bool):
|
|
return domain
|
|
|
|
if vals is None:
|
|
vals = self._values
|
|
|
|
stack = []
|
|
for it in reversed(domain):
|
|
if it == '!':
|
|
stack.append(not stack.pop())
|
|
elif it == '&':
|
|
e1 = stack.pop()
|
|
e2 = stack.pop()
|
|
stack.append(e1 and e2)
|
|
elif it == '|':
|
|
e1 = stack.pop()
|
|
e2 = stack.pop()
|
|
stack.append(e1 or e2)
|
|
elif isinstance(it, tuple):
|
|
if it == TRUE_LEAF:
|
|
stack.append(True)
|
|
continue
|
|
if it == FALSE_LEAF:
|
|
stack.append(False)
|
|
continue
|
|
left, operator, right = it
|
|
# hack-ish handling of parent.<field> modifiers
|
|
if left.startswith('parent.'):
|
|
left_value = vals['•parent•'][left[7:]]
|
|
else:
|
|
left_value = vals[left]
|
|
# apparent artefact of JS data representation: m2m field
|
|
# values are assimilated to lists of ids?
|
|
# FIXME: SSF should do that internally, but the requirement
|
|
# of recursively post-processing to generate lists of
|
|
# commands on save (e.g. m2m inside an o2m) means the
|
|
# data model needs proper redesign
|
|
# we're looking up the "current view" so bits might be
|
|
# missing when processing o2ms in the parent (see
|
|
# values_to_save:1450 or so)
|
|
left_field = view['fields'].get(left, {'type': None})
|
|
if left_field['type'] == 'many2many':
|
|
# field value should be [(6, _, ids)], we want just the ids
|
|
left_value = left_value[0][2] if left_value else []
|
|
stack.append(self._OPS[operator](left_value, right))
|
|
else:
|
|
raise ValueError(f"Unknown domain element {it!r}")
|
|
assert len(stack) == 1
|
|
return stack[0]
|
|
|
|
_OPS = {
|
|
'=': operator.eq,
|
|
'==': operator.eq,
|
|
'!=': operator.ne,
|
|
'<': operator.lt,
|
|
'<=': operator.le,
|
|
'>=': operator.ge,
|
|
'>': operator.gt,
|
|
'in': lambda a, b: (a in b) if isinstance(b, (tuple, list)) else (b in a),
|
|
'not in': lambda a, b: (a not in b) if isinstance(b, (tuple, list)) else (b not in a),
|
|
'like': lambda a, b: a and b and isinstance(a, str) and isinstance(b, str) and a in b,
|
|
'ilike': lambda a, b: a and b and isinstance(a, str) and isinstance(b, str) and a.lower() in b.lower(),
|
|
'not like': lambda a, b: a and b and isinstance(a, str) and isinstance(b, str) and a not in b,
|
|
'not ilike': lambda a, b: a and b and isinstance(a, str) and isinstance(b, str) and a.lower() not in b.lower(),
|
|
}
|
|
|
|
def _get_context(self, field_name):
|
|
""" Return the context of a given field. """
|
|
context_str = self._view['contexts'].get(field_name)
|
|
if not context_str:
|
|
return {}
|
|
eval_context = self._get_eval_context()
|
|
return safe_eval(context_str, eval_context)
|
|
|
|
def _get_eval_context(self):
|
|
""" Return the context dict to eval something. """
|
|
context = {
|
|
'id': self._record.id,
|
|
'active_id': self._record.id,
|
|
'active_ids': self._record.ids,
|
|
'active_model': self._record._name,
|
|
'current_date': date.today().strftime("%Y-%m-%d"),
|
|
**self._env.context,
|
|
}
|
|
return {
|
|
**context,
|
|
'context': context,
|
|
**self._values_to_save(all_fields=True),
|
|
}
|
|
|
|
def __enter__(self):
|
|
""" This makes the Form usable as a context manager. """
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc_value, traceback):
|
|
if not exc_type:
|
|
self.save()
|
|
|
|
def save(self):
|
|
""" Save the form (if necessary) and return the current record:
|
|
|
|
* does not save ``readonly`` fields;
|
|
* does not save unmodified fields (during edition) — any assignment
|
|
or onchange return marks the field as modified, even if set to its
|
|
current value.
|
|
|
|
When nothing must be saved, it simply returns the current record.
|
|
|
|
:raises AssertionError: if the form has any unfilled required field
|
|
"""
|
|
values = self._values_to_save()
|
|
if self._record:
|
|
if values:
|
|
self._record.write(values)
|
|
else:
|
|
object.__setattr__(self, '_record', self._record.create(values))
|
|
# reload the record
|
|
self._values.update(read_record(self._record, self._view['fields']))
|
|
self._changed.clear()
|
|
self._env.flush_all()
|
|
self._env.clear() # discard cache and pending recomputations
|
|
return self._record
|
|
|
|
@property
|
|
def record(self):
|
|
""" Return the record being edited by the form. This attribute is
|
|
readonly and can only be accessed when the form has no pending changes.
|
|
"""
|
|
assert not self._changed
|
|
return self._record
|
|
|
|
def _values_to_save(self, all_fields=False):
|
|
""" Validates values and returns only fields modified since
|
|
load/save
|
|
|
|
:param bool all_fields: if False (the default), checks for required
|
|
fields and only save fields which are changed
|
|
and not readonly
|
|
"""
|
|
view = self._view
|
|
fields = self._view['fields']
|
|
values = self._values
|
|
changed = self._changed
|
|
return self._values_to_save_(values, fields, view, changed, all_fields)
|
|
|
|
def _values_to_save_(self, values, fields, view, changed, all_fields=False,
|
|
modifiers_values=None, parent_link=None):
|
|
""" Validates & extracts values to save, recursively in order to handle
|
|
o2ms properly
|
|
|
|
:param dict values: values of the record to extract
|
|
:param dict fields: fields_get result ``{field_name: field_info}``
|
|
:param view: view tree
|
|
:param set changed: set of fields which have been modified (since last save)
|
|
:param bool all_fields:
|
|
whether to ignore normal filtering and just return everything
|
|
:param dict modifiers_values:
|
|
defaults to ``values``, but o2ms need some additional
|
|
massaging
|
|
"""
|
|
modifiers_values = modifiers_values or values
|
|
|
|
result = {}
|
|
for field_name, field_info in fields.items():
|
|
if field_name == 'id':
|
|
continue
|
|
|
|
value = values[field_name]
|
|
|
|
# note: maybe `invisible` should not skip `required` if model attribute
|
|
if (
|
|
value is False
|
|
and not all_fields
|
|
and field_name != parent_link
|
|
and field_info['type'] != 'boolean'
|
|
and not self._get_modifier(field_name, 'invisible', view=view, vals=modifiers_values)
|
|
and not self._get_modifier(field_name, 'column_invisible', view=view, vals=modifiers_values)
|
|
and self._get_modifier(field_name, 'required', view=view, vals=modifiers_values)
|
|
):
|
|
raise AssertionError(f"{field_name} is a required field ({view['modifiers'][field_name]})")
|
|
|
|
# skip unmodified fields unless all_fields
|
|
if not (all_fields or field_name in changed):
|
|
continue
|
|
|
|
if not all_fields and self._get_modifier(field_name, 'readonly', view=view, vals=modifiers_values):
|
|
field_node = next(
|
|
node
|
|
for node in view['tree'].iter('field')
|
|
if node.get('name') == field_name
|
|
)
|
|
if not field_node.get('force_save'):
|
|
continue
|
|
|
|
if field_info['type'] == 'one2many':
|
|
subview = field_info['edition_view']
|
|
subfields = subview['fields']
|
|
res = []
|
|
for (cmd, rid, vs) in value:
|
|
if cmd == Command.UPDATE and not vs:
|
|
cmd, vs = Command.LINK, False
|
|
elif cmd in (Command.CREATE, Command.UPDATE):
|
|
vs = vs or {}
|
|
|
|
missing = subfields.keys() - vs.keys()
|
|
# FIXME: maybe do this during initial loading instead?
|
|
if missing:
|
|
comodel = self._env[field_info['relation']]
|
|
if cmd == Command.CREATE:
|
|
vs.update(dict.fromkeys(missing, False))
|
|
vs.update({
|
|
key: _cleanup_from_default(subfields[key], val)
|
|
for key, val in comodel.default_get(list(missing)).items()
|
|
})
|
|
else:
|
|
vs.update(read_record(
|
|
comodel.browse(rid),
|
|
{key: val for key, val in subfields.items() if key not in vs},
|
|
))
|
|
vs = self._values_to_save_(
|
|
vs, subfields, subview,
|
|
vs._changed if isinstance(vs, UpdateDict) else vs.keys(),
|
|
all_fields,
|
|
modifiers_values={'id': False, **vs, '•parent•': values},
|
|
# related o2m don't have a relation_field
|
|
parent_link=field_info.get('relation_field'),
|
|
)
|
|
res.append((cmd, rid, vs))
|
|
value = res
|
|
|
|
result[field_name] = value
|
|
return result
|
|
|
|
def _perform_onchange(self, field_name=None):
|
|
assert field_name is None or isinstance(field_name, str)
|
|
|
|
# marks onchange source as changed
|
|
if field_name:
|
|
self._changed.add(field_name)
|
|
|
|
# skip calling onchange() if there's no on_change on the field
|
|
spec = self._view['onchange']
|
|
if field_name and not spec[field_name]:
|
|
return
|
|
|
|
record = self._record
|
|
|
|
# if the onchange is triggered by a field, add the context of that field
|
|
if field_name:
|
|
context = self._get_context(field_name)
|
|
if context:
|
|
record = record.with_context(**context)
|
|
|
|
result = record.onchange(self._onchange_values(), field_name, spec)
|
|
self._env.flush_all()
|
|
self._env.clear() # discard cache and pending recomputations
|
|
|
|
if result.get('warning'):
|
|
_logger.getChild('onchange').warning("%(title)s %(message)s", result['warning'])
|
|
|
|
# mark onchange output as changed
|
|
fields = self._view['fields']
|
|
values = {
|
|
key: self._cleanup_onchange(fields[key], val, self._values.get(key))
|
|
for key, val in result.get('value', {}).items()
|
|
if key in fields
|
|
}
|
|
self._values.update(values)
|
|
self._changed.update(values)
|
|
return result
|
|
|
|
def _onchange_values(self):
|
|
return self._onchange_values_(self._view['fields'], self._values)
|
|
|
|
def _onchange_values_(self, fields, values):
|
|
""" Recursively cleanup o2m values for onchanges:
|
|
|
|
* if an o2m command is a 1 (UPDATE) and there is nothing to update, send
|
|
a 4 instead (LINK_TO) instead as that's what the webclient sends for
|
|
unmodified rows
|
|
* if an o2m command is a 1 (UPDATE) and only a subset of its fields have
|
|
been modified, only send the modified ones
|
|
|
|
This needs to be recursive as there are people who put invisible o2ms
|
|
inside their o2ms.
|
|
"""
|
|
result = {}
|
|
for key, val in values.items():
|
|
if fields[key]['type'] == 'one2many':
|
|
subfields = fields[key]['edition_view']['fields']
|
|
result[key] = []
|
|
for (cmd, rid, vs) in val:
|
|
if cmd == Command.UPDATE and isinstance(vs, UpdateDict):
|
|
vs = dict(vs.changed_items())
|
|
if cmd == Command.UPDATE and not vs:
|
|
result[key].append((Command.LINK, rid, False))
|
|
elif cmd in (Command.CREATE, Command.UPDATE):
|
|
result[key].append((cmd, rid, self._onchange_values_(subfields, vs)))
|
|
else:
|
|
result[key].append((cmd, rid, vs))
|
|
else:
|
|
result[key] = val
|
|
return result
|
|
|
|
def _cleanup_onchange(self, field_info, value, current):
|
|
""" Transform the value returned by onchange() into the expected format
|
|
for self._values.
|
|
"""
|
|
if field_info['type'] == 'many2one':
|
|
return value[0] if value else False
|
|
|
|
if field_info['type'] == 'one2many':
|
|
# ignore o2ms nested in o2ms
|
|
if not field_info['edition_view']:
|
|
return []
|
|
|
|
if current is None:
|
|
current = []
|
|
|
|
result = []
|
|
current_ids = {rid for cmd, rid, vs in current if cmd in (1, 2)}
|
|
current_values = {rid: vs for cmd, rid, vs in current if cmd == 1}
|
|
# which view should this be???
|
|
subfields = field_info['edition_view']['fields']
|
|
# TODO: simplistic, unlikely to work if e.g. there's a 5 inbetween other commands
|
|
for command in value:
|
|
if command[0] == Command.CREATE:
|
|
result.append((Command.CREATE, 0, {
|
|
key: self._cleanup_onchange(subfields[key], val, None)
|
|
for key, val in command[2].items()
|
|
if key in subfields
|
|
}))
|
|
elif command[0] == Command.UPDATE:
|
|
record_id = command[1]
|
|
current_ids.discard(record_id)
|
|
stored = current_values.get(record_id)
|
|
if stored is None:
|
|
record = self._env[field_info['relation']].browse(record_id)
|
|
stored = UpdateDict(read_record(record, subfields))
|
|
for key, val in command[2].items():
|
|
if key in subfields:
|
|
val = self._cleanup_onchange(subfields[key], val, stored.get(key))
|
|
# if there are values from the onchange which differ
|
|
# from current values, update & mark field as changed
|
|
if stored.get(key, val) != val:
|
|
stored[key] = val
|
|
stored._changed.add(key)
|
|
result.append((Command.UPDATE, record_id, stored))
|
|
elif command[0] == Command.DELETE:
|
|
current_ids.discard(command[1])
|
|
result.append((Command.DELETE, command[1], False))
|
|
elif command[0] == Command.LINK:
|
|
current_ids.discard(command[1])
|
|
result.append((Command.UPDATE, command[1], None))
|
|
elif command[0] == Command.CLEAR:
|
|
result = []
|
|
# explicitly mark all non-relinked (or modified) records as deleted
|
|
for id_ in current_ids:
|
|
result.append((Command.DELETE, id_, False))
|
|
return result
|
|
|
|
if field_info['type'] == 'many2many':
|
|
# onchange result is a bunch of commands, normalize to single 6
|
|
ids = [] if current is None else list(current[0][2])
|
|
for command in value:
|
|
if command[0] == Command.UPDATE:
|
|
ids.append(command[1])
|
|
elif command[0] == Command.UNLINK:
|
|
ids.remove(command[1])
|
|
elif command[0] == Command.LINK:
|
|
ids.append(command[1])
|
|
elif command[0] == Command.CLEAR:
|
|
del ids[:]
|
|
elif command[0] == Command.SET:
|
|
ids[:] = command[2]
|
|
else:
|
|
raise ValueError(f"Unsupported M2M command {command[0]}")
|
|
return [(Command.SET, False, ids)]
|
|
|
|
return value
|
|
|
|
|
|
class O2MForm(Form):
|
|
# noinspection PyMissingConstructor
|
|
# pylint: disable=super-init-not-called
|
|
def __init__(self, proxy, index=None):
|
|
model = proxy._model
|
|
object.__setattr__(self, '_proxy', proxy)
|
|
object.__setattr__(self, '_index', index)
|
|
|
|
object.__setattr__(self, '_record', model)
|
|
object.__setattr__(self, '_env', model.env)
|
|
|
|
object.__setattr__(self, '_models_info', proxy._form._models_info)
|
|
tree = proxy._field_info['edition_view']['tree']
|
|
view = self._process_view(tree, model)
|
|
object.__setattr__(self, '_view', view)
|
|
|
|
vals = dict.fromkeys(view['fields'], False)
|
|
object.__setattr__(self, '_values', vals)
|
|
object.__setattr__(self, '_changed', set())
|
|
if index is None:
|
|
self._init_from_defaults()
|
|
else:
|
|
vals = proxy._records[index]
|
|
self._values.update(vals)
|
|
if hasattr(vals, '_changed'):
|
|
self._changed.update(vals._changed)
|
|
if vals.get('id'):
|
|
object.__setattr__(self, '_record', model.browse(vals['id']))
|
|
|
|
def _get_modifier(self, field_name, modifier, *, view=None, vals=None):
|
|
if vals is None:
|
|
vals = {**self._values, '•parent•': self._proxy._form._values}
|
|
return super()._get_modifier(field_name, modifier, view=view, vals=vals)
|
|
|
|
def _get_eval_context(self):
|
|
eval_context = super()._get_eval_context()
|
|
eval_context['parent'] = Dotter(self._proxy._form._values)
|
|
return eval_context
|
|
|
|
def _onchange_values(self):
|
|
values = super()._onchange_values()
|
|
# computed o2m may not have a relation_field(?)
|
|
field_info = self._proxy._field_info
|
|
if 'relation_field' in field_info: # note: should be fine because not recursive
|
|
values[field_info['relation_field']] = self._proxy._form._onchange_values()
|
|
return values
|
|
|
|
def save(self):
|
|
proxy = self._proxy
|
|
field_value = proxy._form._values[proxy._field]
|
|
values = self._values_to_save()
|
|
if self._index is None:
|
|
field_value.append((Command.CREATE, 0, values))
|
|
else:
|
|
index = proxy._command_index(self._index)
|
|
(cmd, id_, vs) = field_value[index]
|
|
if cmd == Command.CREATE:
|
|
vs.update(values)
|
|
elif cmd == Command.UPDATE:
|
|
if vs is None:
|
|
vs = UpdateDict()
|
|
assert isinstance(vs, UpdateDict), type(vs)
|
|
vs.update(values)
|
|
field_value[index] = (Command.UPDATE, id_, vs)
|
|
else:
|
|
raise AssertionError(f"Expected command 0 or 1, found {cmd!r}")
|
|
|
|
proxy._form._perform_onchange(proxy._field)
|
|
|
|
def _values_to_save(self, all_fields=False):
|
|
""" Validates values and returns only fields modified since
|
|
load/save
|
|
"""
|
|
values = UpdateDict(self._values)
|
|
values._changed.update(self._changed)
|
|
if all_fields:
|
|
return values
|
|
|
|
for field_name in self._view['fields']:
|
|
if self._get_modifier(field_name, 'required') and not (
|
|
self._get_modifier(field_name, 'column_invisible')
|
|
or self._get_modifier(field_name, 'invisible')
|
|
):
|
|
assert values[field_name] is not False, "{fname!r} is a required field"
|
|
|
|
return values
|
|
|
|
|
|
class UpdateDict(dict):
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
self._changed = set()
|
|
if args and isinstance(args[0], UpdateDict):
|
|
self._changed.update(args[0]._changed)
|
|
|
|
def changed_items(self):
|
|
return (
|
|
(k, v) for k, v in self.items()
|
|
if k in self._changed
|
|
)
|
|
|
|
def update(self, *args, **kw):
|
|
super().update(*args, **kw)
|
|
if args and isinstance(args[0], UpdateDict):
|
|
self._changed.update(args[0]._changed)
|
|
|
|
|
|
class X2MProxy:
|
|
""" A proxy represents the value of an x2many field, but not directly.
|
|
Instead, it provides an API to add, remove or edit records in the value.
|
|
"""
|
|
_form = None # Form containing the corresponding x2many field
|
|
_field = None # name of the x2many field
|
|
_field_info = None # field info
|
|
|
|
def __init__(self, form, field_name):
|
|
self._form = form
|
|
self._field = field_name
|
|
self._field_info = self._form._view['fields'][field_name]
|
|
|
|
def _assert_editable(self):
|
|
assert not self._form._get_modifier(self._field, 'readonly'), f'field {self._field!r} is not editable'
|
|
assert not self._form._get_modifier(self._field, 'invisible'), f'field {self._field!r} is not visible'
|
|
|
|
|
|
class O2MProxy(X2MProxy):
|
|
""" Proxy object for editing the value of a one2many field. """
|
|
def __init__(self, form, field_name):
|
|
super().__init__(form, field_name)
|
|
# reify records to a list so they can be manipulated easily?
|
|
self._records = []
|
|
model = self._model
|
|
fields = self._field_info['edition_view']['fields']
|
|
for (command, rid, values) in self._form._values[self._field]:
|
|
if command == Command.CREATE:
|
|
self._records.append(values)
|
|
elif command == Command.UPDATE:
|
|
if values is None:
|
|
# read based on view info
|
|
r = model.browse(rid)
|
|
values = UpdateDict(read_record(r, fields))
|
|
self._records.append(values)
|
|
elif command == Command.DELETE:
|
|
pass
|
|
else:
|
|
raise AssertionError("O2M proxy only supports commands 0, 1 and 2, found %s" % command)
|
|
|
|
def __len__(self):
|
|
return len(self._records)
|
|
|
|
@property
|
|
def _model(self):
|
|
model = self._form._env[self._field_info['relation']]
|
|
context = self._form._get_context(self._field)
|
|
if context:
|
|
model = model.with_context(**context)
|
|
return model
|
|
|
|
def new(self):
|
|
""" Returns a :class:`Form` for a new
|
|
:class:`~odoo.fields.One2many` record, properly initialised.
|
|
|
|
The form is created from the list view if editable, or the field's
|
|
form view otherwise.
|
|
|
|
:raises AssertionError: if the field is not editable
|
|
"""
|
|
self._assert_editable()
|
|
return O2MForm(self)
|
|
|
|
def edit(self, index):
|
|
""" Returns a :class:`Form` to edit the pre-existing
|
|
:class:`~odoo.fields.One2many` record.
|
|
|
|
The form is created from the list view if editable, or the field's
|
|
form view otherwise.
|
|
|
|
:raises AssertionError: if the field is not editable
|
|
"""
|
|
self._assert_editable()
|
|
return O2MForm(self, index)
|
|
|
|
def remove(self, index):
|
|
""" Removes the record at ``index`` from the parent form.
|
|
|
|
:raises AssertionError: if the field is not editable
|
|
"""
|
|
self._assert_editable()
|
|
# remove reified record from local list & either remove 0 from
|
|
# commands list or replace 1 (update) by 2 (remove)
|
|
cidx = self._command_index(index)
|
|
commands = self._form._values[self._field]
|
|
(command, rid, _) = commands[cidx]
|
|
if command == Command.CREATE:
|
|
# record not saved yet -> just remove the command
|
|
del commands[cidx]
|
|
elif command == Command.UPDATE:
|
|
# record already saved, replace by 2
|
|
commands[cidx] = (Command.DELETE, rid, 0)
|
|
else:
|
|
raise AssertionError("Expected command 0 or 1, got %s" % commands[cidx])
|
|
# remove reified record
|
|
del self._records[index]
|
|
self._form._perform_onchange(self._field)
|
|
|
|
def _command_index(self, for_record):
|
|
""" Takes a record index and finds the corresponding record index
|
|
(skips all 2s, basically)
|
|
|
|
:param int for_record:
|
|
"""
|
|
commands = self._form._values[self._field]
|
|
return next(
|
|
cidx
|
|
for ridx, cidx in enumerate(
|
|
cidx for cidx, (c, _1, _2) in enumerate(commands)
|
|
if c in (Command.CREATE, Command.UPDATE)
|
|
)
|
|
if ridx == for_record
|
|
)
|
|
|
|
|
|
class M2MProxy(X2MProxy, collections.abc.Sequence):
|
|
""" Proxy object for editing the value of a many2many field.
|
|
|
|
Behaves as a :class:`~collection.Sequence` of recordsets, can be
|
|
indexed or sliced to get actual underlying recordsets.
|
|
"""
|
|
def __getitem__(self, index):
|
|
comodel_name = self._field_info['relation']
|
|
return self._form._env[comodel_name].browse(self._get_ids()[index])
|
|
|
|
def __len__(self):
|
|
return len(self._get_ids())
|
|
|
|
def __iter__(self):
|
|
comodel_name = self._field_info['relation']
|
|
records = self._form._env[comodel_name].browse(self._get_ids())
|
|
return iter(records)
|
|
|
|
def __contains__(self, record):
|
|
comodel_name = self._field_info['relation']
|
|
assert isinstance(record, BaseModel) and record._name == comodel_name
|
|
return record.id in self._get_ids()
|
|
|
|
def _get_ids(self):
|
|
return self._form._values[self._field][0][2]
|
|
|
|
def add(self, record):
|
|
""" Adds ``record`` to the field, the record must already exist.
|
|
|
|
The addition will only be finalized when the parent record is saved.
|
|
"""
|
|
self._assert_editable()
|
|
parent = self._form
|
|
comodel_name = self._field_info['relation']
|
|
assert isinstance(record, BaseModel) and record._name == comodel_name, \
|
|
f"trying to assign a {record._name!r} object to a {comodel_name!r} field"
|
|
self._get_ids().append(record.id)
|
|
|
|
parent._perform_onchange(self._field)
|
|
|
|
# pylint: disable=redefined-builtin
|
|
def remove(self, id=None, index=None):
|
|
""" Removes a record at a certain index or with a provided id from
|
|
the field.
|
|
"""
|
|
self._assert_editable()
|
|
assert (id is None) ^ (index is None), "can remove by either id or index"
|
|
if id is None:
|
|
# remove by index
|
|
del self._get_ids()[index]
|
|
else:
|
|
self._get_ids().remove(id)
|
|
self._form._perform_onchange(self._field)
|
|
|
|
def clear(self):
|
|
""" Removes all existing records in the m2m
|
|
"""
|
|
self._assert_editable()
|
|
self._get_ids()[:] = []
|
|
self._form._perform_onchange(self._field)
|
|
|
|
|
|
def read_record(record, fields):
|
|
""" Read the given fields (field descriptions) on the given record. """
|
|
result = {}
|
|
# don't read the id explicitly, not sure why but if any of the "magic" hr
|
|
# field is read alongside `id` then it blows up e.g.
|
|
# james.read(['barcode']) works fine but james.read(['id', 'barcode'])
|
|
# triggers an ACL error on barcode, likewise km_home_work or
|
|
# emergency_contact or whatever. Since we always get the id anyway, just
|
|
# remove it from the fields to read
|
|
field_names = [fname for fname in fields if fname != 'id']
|
|
if not field_names:
|
|
return result
|
|
for fname, value in record.read(field_names)[0].items():
|
|
field_info = fields[fname]
|
|
if field_info['type'] == 'many2one':
|
|
value = value and value[0]
|
|
elif field_info['type'] == 'many2many':
|
|
value = [(Command.SET, 0, value or [])]
|
|
elif field_info['type'] == 'one2many':
|
|
value = [(Command.UPDATE, result, None) for result in value or []]
|
|
elif field_info['type'] == 'datetime' and isinstance(value, datetime):
|
|
value = odoo.fields.Datetime.to_string(value)
|
|
elif field_info['type'] == 'date' and isinstance(value, date):
|
|
value = odoo.fields.Date.to_string(value)
|
|
result[fname] = value
|
|
return result
|
|
|
|
|
|
def _cleanup_from_default(type_, value):
|
|
if not value:
|
|
if type_ == 'many2many':
|
|
return [(Command.SET, False, [])]
|
|
elif type_ == 'one2many':
|
|
return []
|
|
elif type_ in ('integer', 'float'):
|
|
return 0
|
|
return value
|
|
|
|
if type_ == 'one2many':
|
|
return [cmd for cmd in value if cmd[0] != Command.SET]
|
|
elif type_ == 'datetime' and isinstance(value, datetime):
|
|
return odoo.fields.Datetime.to_string(value)
|
|
elif type_ == 'date' and isinstance(value, date):
|
|
return odoo.fields.Date.to_string(value)
|
|
return value
|
|
|
|
|
|
class Dotter:
|
|
""" Simple wrapper for a dict where keys are accessed as readonly attributes. """
|
|
__slots__ = ['__values']
|
|
|
|
def __init__(self, values):
|
|
self.__values = values
|
|
|
|
def __getattr__(self, key):
|
|
val = self.__values[key]
|
|
return Dotter(val) if isinstance(val, dict) else val
|