Coverage for adhoc-cicd-odoo-odoo / odoo / orm / fields_reference.py: 81%

1 

2from collections import defaultdict 

3from operator import attrgetter 

4 

5from odoo.tools import OrderedSet, unique 

6from odoo.tools.sql import pg_varchar 

7 

8from .fields import Field 

9from .fields_numeric import Integer 

10from .fields_selection import Selection 

11from .models import BaseModel 

12 

13 

14class Reference(Selection): 

15 """ Pseudo-relational field (no FK in database). 

16 

17 The field value is stored as a :class:`string <str>` following the pattern 

18 ``"res_model,res_id"`` in database. 

19 """ 

20 type = 'reference' 

21 

22 _column_type = ('varchar', pg_varchar()) 

23 

24 def convert_to_column(self, value, record, values=None, validate=True): 

25 return Field.convert_to_column(self, value, record, values, validate) 

26 

27 def convert_to_cache(self, value, record, validate=True): 

28 # cache format: str ("model,id") or None 

29 if isinstance(value, BaseModel): 29 ↛ 30line 29 didn't jump to line 30 because the condition on line 29 was never true

30 if not validate or (value._name in self.get_values(record.env) and len(value) <= 1): 

31 return "%s,%s" % (value._name, value.id) if value else None 

32 elif isinstance(value, str): 32 ↛ 39line 32 didn't jump to line 39 because the condition on line 32 was always true

33 res_model, res_id = value.split(',') 

34 if not validate or res_model in self.get_values(record.env): 34 ↛ 41line 34 didn't jump to line 41 because the condition on line 34 was always true

35 if record.env[res_model].browse(int(res_id)).exists(): 35 ↛ 38line 35 didn't jump to line 38 because the condition on line 35 was always true

36 return value 

37 else: 

38 return None 

39 elif not value: 

40 return None 

41 raise ValueError("Wrong value for %s: %r" % (self, value)) 

42 

43 def convert_to_record(self, value, record): 

44 if value: 

45 res_model, res_id = value.split(',') 

46 return record.env[res_model].browse(int(res_id)) 

47 return None 

48 

49 def convert_to_read(self, value, record, use_display_name=True): 

50 return "%s,%s" % (value._name, value.id) if value else False 

51 

52 def convert_to_export(self, value, record): 

53 return value.display_name if value else '' 

54 

55 def convert_to_display_name(self, value, record): 

56 return value.display_name if value else False 

57 

58 

59class Many2oneReference(Integer): 

60 """ Pseudo-relational field (no FK in database). 

61 

62 The field value is stored as an :class:`integer <int>` id in database. 

63 

64 Contrary to :class:`Reference` fields, the model has to be specified 

65 in a :class:`Char` field, whose name has to be specified in the 

66 `model_field` attribute for the current :class:`Many2oneReference` field. 

67 

68 :param str model_field: name of the :class:`Char` where the model name is stored. 

69 """ 

70 type = 'many2one_reference' 

71 

72 model_field = None 

73 aggregator = None 

74 

75 _related_model_field = property(attrgetter('model_field')) 

76 

77 _description_model_field = property(attrgetter('model_field')) 

78 

79 def convert_to_cache(self, value, record, validate=True): 

80 # cache format: id or None 

81 if isinstance(value, BaseModel): 81 ↛ 82line 81 didn't jump to line 82 because the condition on line 81 was never true

82 value = value._ids[0] if value._ids else None 

83 return super().convert_to_cache(value, record, validate) 

84 

85 def _update_inverses(self, records: BaseModel, value): 

86 """ Add `records` to the cached values of the inverse fields of `self`. """ 

87 if not value: 

88 return 

89 model_ids = self._record_ids_per_res_model(records) 

90 

91 for invf in records.pool.field_inverses[self]: 

92 records = records.browse(model_ids[invf.model_name]) 

93 if not records: 

94 continue 

95 corecord = records.env[invf.model_name].browse(value) 

96 records = records.filtered_domain(invf.get_comodel_domain(corecord)) 

97 if not records: 

98 continue 

99 ids0 = invf._get_cache(corecord.env).get(corecord.id) 

100 # if the value for the corecord is not in cache, but this is a new 

101 # record, assign it anyway, as you won't be able to fetch it from 

102 # database (see `test_sale_order`) 

103 if ids0 is not None or not corecord.id: 

104 ids1 = tuple(unique((ids0 or ()) + records._ids)) 

105 invf._update_cache(corecord, ids1) 

106 

107 def _record_ids_per_res_model(self, records: BaseModel) -> dict[str, OrderedSet]: 

108 model_ids = defaultdict(OrderedSet) 

109 for record in records: 

110 model = record[self.model_field] 

111 if not model and record._fields[self.model_field].compute: 

112 # fallback when the model field is computed :-/ 

113 record._fields[self.model_field].compute_value(record) 

114 model = record[self.model_field] 

115 if not model: 115 ↛ 116line 115 didn't jump to line 116 because the condition on line 115 was never true

116 continue 

117 model_ids[model].add(record.id) 

118 return model_ids