Coverage for adhoc-cicd-odoo-odoo / odoo / modules / db.py: 90%

1# Part of Odoo. See LICENSE file for full copyright and licensing details. 

2""" Initialize the database for module management and Odoo installation. """ 

3from __future__ import annotations 

4 

5import logging 

6import typing 

7from enum import IntEnum 

8 

9from psycopg2.extras import Json 

10 

11import odoo.modules 

12import odoo.tools 

13 

14if typing.TYPE_CHECKING: 

15 from odoo.sql_db import BaseCursor, Cursor 

16 

17_logger = logging.getLogger(__name__) 

18 

19 

20def is_initialized(cr: Cursor) -> bool: 

21 """ Check if a database has been initialized for the ORM. 

22 

23 The database can be initialized with the 'initialize' function below. 

24 

25 """ 

26 return odoo.tools.sql.table_exists(cr, 'ir_module_module') 

27 

28 

29def initialize(cr: Cursor) -> None: 

30 """ Initialize a database with for the ORM. 

31 

32 This executes base/data/base_data.sql, creates the ir_module_categories 

33 (taken from each module descriptor file), and creates the ir_module_module 

34 and ir_model_data entries. 

35 

36 """ 

37 try: 

38 f = odoo.tools.misc.file_path('base/data/base_data.sql') 

39 except FileNotFoundError: 

40 m = "File not found: 'base.sql' (provided by module 'base')." 

41 _logger.critical(m) 

42 raise OSError(m) 

43 

44 with odoo.tools.misc.file_open(f) as base_sql_file: 

45 cr.execute(base_sql_file.read()) # pylint: disable=sql-injection 

46 

47 for info in odoo.modules.Manifest.all_addon_manifests(): 

48 module_name = info.name 

49 categories = info['category'].split('/') 

50 category_id = create_categories(cr, categories) 

51 

52 if info['installable']: 

53 state = 'uninstalled' 

54 else: 

55 state = 'uninstallable' 

56 

57 cr.execute('INSERT INTO ir_module_module \ 

58 (author, website, name, shortdesc, description, \ 

59 category_id, auto_install, state, web, license, application, icon, sequence, summary) \ 

60 VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) RETURNING id', ( 

61 info['author'], 

62 info['website'], module_name, Json({'en_US': info['name']}), 

63 Json({'en_US': info['description']}), category_id, 

64 info['auto_install'] is not False, state, 

65 info['web'], 

66 info['license'], 

67 info['application'], info['icon'], 

68 info['sequence'], Json({'en_US': info['summary']}))) 

69 row = cr.fetchone() 

70 assert row is not None # for typing 

71 module_id = row[0] 

72 cr.execute( 

73 'INSERT INTO ir_model_data' 

74 '(name,model,module, res_id, noupdate) VALUES (%s,%s,%s,%s,%s)', 

75 ('module_' + module_name, 'ir.module.module', 'base', module_id, True), 

76 ) 

77 dependencies = info['depends'] 

78 for d in dependencies: 

79 cr.execute( 

80 'INSERT INTO ir_module_module_dependency (module_id, name, auto_install_required)' 

81 ' VALUES (%s, %s, %s)', 

82 (module_id, d, d in (info['auto_install'] or ())) 

83 ) 

84 

85 from odoo.tools import config # noqa: PLC0415 

86 if config.get('skip_auto_install'): 86 ↛ 88line 86 didn't jump to line 88 because the condition on line 86 was never true

87 # even if skip_auto_install is enabled we still want to have base 

88 cr.execute("""UPDATE ir_module_module SET state='to install' WHERE name = 'base'""") 

89 return 

90 

91 # Install recursively all auto-installing modules 

92 while True: 

93 # this selects all the auto_install modules whose auto_install_required 

94 # deps are marked as to install 

95 cr.execute(""" 

96 SELECT m.name FROM ir_module_module m 

97 WHERE m.auto_install 

98 AND state not in ('to install', 'uninstallable') 

99 AND NOT EXISTS ( 

100 SELECT 1 FROM ir_module_module_dependency d 

101 JOIN ir_module_module mdep ON (d.name = mdep.name) 

102 WHERE d.module_id = m.id 

103 AND d.auto_install_required 

104 AND mdep.state != 'to install' 

105 )""") 

106 to_auto_install = [x[0] for x in cr.fetchall()] 

107 # however if the module has non-required deps we need to install 

108 # those, so merge-in the modules which have a dependen*t* which is 

109 # *either* to_install or in to_auto_install and merge it in? 

110 cr.execute(""" 

111 SELECT d.name FROM ir_module_module_dependency d 

112 JOIN ir_module_module m ON (d.module_id = m.id) 

113 JOIN ir_module_module mdep ON (d.name = mdep.name) 

114 WHERE (m.state = 'to install' OR m.name = any(%s)) 

115 -- don't re-mark marked modules 

116 AND NOT (mdep.state = 'to install' OR mdep.name = any(%s)) 

117 """, [to_auto_install, to_auto_install]) 

118 to_auto_install.extend(x[0] for x in cr.fetchall()) 

119 

120 if not to_auto_install: 

121 break 

122 cr.execute("""UPDATE ir_module_module SET state='to install' WHERE name in %s""", (tuple(to_auto_install),)) 

123 

124 

125def create_categories(cr: Cursor, categories: list[str]) -> int | None: 

126 """ Create the ir_module_category entries for some categories. 

127 

128 categories is a list of strings forming a single category with its 

129 parent categories, like ['Grand Parent', 'Parent', 'Child']. 

130 

131 Return the database id of the (last) category. 

132 

133 """ 

134 p_id = None 

135 category = [] 

136 while categories: 

137 category.append(categories[0]) 

138 xml_id = 'module_category_' + ('_'.join(x.lower() for x in category)).replace('&', 'and').replace(' ', '_') 

139 # search via xml_id (because some categories are renamed) 

140 cr.execute("SELECT res_id FROM ir_model_data WHERE name=%s AND module=%s AND model=%s", 

141 (xml_id, "base", "ir.module.category")) 

142 

143 row = cr.fetchone() 

144 if not row: 

145 cr.execute('INSERT INTO ir_module_category \ 

146 (name, parent_id) \ 

147 VALUES (%s, %s) RETURNING id', (Json({'en_US': categories[0]}), p_id)) 

148 row = cr.fetchone() 

149 assert row is not None # for typing 

150 p_id = row[0] 

151 cr.execute('INSERT INTO ir_model_data (module, name, res_id, model, noupdate) \ 

152 VALUES (%s, %s, %s, %s, %s)', ('base', xml_id, p_id, 'ir.module.category', True)) 

153 else: 

154 p_id = row[0] 

155 assert isinstance(p_id, int) 

156 categories = categories[1:] 

157 return p_id 

158 

159 

160class FunctionStatus(IntEnum): 

161 MISSING = 0 # function is not present (falsy) 

162 PRESENT = 1 # function is present but not indexable (not immutable) 

163 INDEXABLE = 2 # function is present and indexable (immutable) 

164 

165 

166def has_unaccent(cr: BaseCursor) -> FunctionStatus: 

167 """ Test whether the database has function 'unaccent' and return its status. 

168 

169 The unaccent is supposed to be provided by the PostgreSQL unaccent contrib 

170 module but any similar function will be picked by OpenERP. 

171 

172 :rtype: FunctionStatus 

173 """ 

174 cr.execute(""" 

175 SELECT p.provolatile 

176 FROM pg_proc p 

177 WHERE p.proname = 'unaccent' 

178 AND p.pronamespace = current_schema::regnamespace 

179 AND p.pronargs = 1 

180 """) 

181 result = cr.fetchone() 

182 if not result: 182 ↛ 187line 182 didn't jump to line 187 because the condition on line 182 was always true

183 return FunctionStatus.MISSING 

184 # The `provolatile` of unaccent allows to know whether the unaccent function 

185 # can be used to create index (it should be 'i' - means immutable), see 

186 # https://www.postgresql.org/docs/current/catalog-pg-proc.html. 

187 return FunctionStatus.INDEXABLE if result[0] == 'i' else FunctionStatus.PRESENT 

188 

189 

190def has_trigram(cr: BaseCursor) -> bool: 

191 """ Test if the database has the a word_similarity function. 

192 

193 The word_similarity is supposed to be provided by the PostgreSQL built-in 

194 pg_trgm module but any similar function will be picked by Odoo. 

195 

196 """ 

197 cr.execute("SELECT proname FROM pg_proc WHERE proname='word_similarity'") 

198 return len(cr.fetchall()) > 0