diff --git a/CHANGELOG b/CHANGELOG index 108181ad6e..382c4dfd30 100644 --- a/CHANGELOG +++ b/CHANGELOG @@ -1,3 +1,6 @@ +* Modify ondelete attribute of Many2One according to required attribute value +* Refactor import_data of ModelStorage +* Raise exception when search function is missing on Function fields * Add LRU memoization, use it to cache compiled code for safe_eval. * Add PySON to replace python string evaluation on client side * Add JSON-RPC diff --git a/COPYRIGHT b/COPYRIGHT index aa03494999..bab90c9e3e 100644 --- a/COPYRIGHT +++ b/COPYRIGHT @@ -1,7 +1,7 @@ Copyright (C) 2004-2008 Tiny SPRL. -Copyright (C) 2007-2009 Cédric Krier. -Copyright (C) 2007-2008 Bertrand Chenal. -Copyright (C) 2008-2009 B2CK SPRL. +Copyright (C) 2007-2010 Cédric Krier. +Copyright (C) 2007-2010 Bertrand Chenal. +Copyright (C) 2008-2010 B2CK SPRL. This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by diff --git a/trytond/backend/postgresql/database.py b/trytond/backend/postgresql/database.py index dad35e6b2f..7be319b01f 100644 --- a/trytond/backend/postgresql/database.py +++ b/trytond/backend/postgresql/database.py @@ -389,7 +389,7 @@ class Cursor(CursorInterface): def limit_clause(self, select, limit=None, offset=None): if limit is not None: select += ' LIMIT %d' % limit - if offset is not None: + if offset is not None and offset != 0: select += ' OFFSET %d' % offset return select diff --git a/trytond/backend/postgresql/table.py b/trytond/backend/postgresql/table.py index d05f5cdcbb..a2dbdc7433 100644 --- a/trytond/backend/postgresql/table.py +++ b/trytond/backend/postgresql/table.py @@ -14,7 +14,7 @@ class TableHandler(TableHandlerInterface): self._columns = {} self._constraints = [] self._fk_deltypes = {} - self._indexes = {} + self._indexes = [] self._field2module = {} # Create sequence if necessary @@ -136,7 +136,7 @@ class TableHandler(TableHandlerInterface): "size": size, "typname": typname} - # fetch constrains for the table + # fetch constraints for the table self.cursor.execute("SELECT co.contype, co.confdeltype, at.attname, "\ "cl2.relname, co.conname "\ "FROM pg_constraint co "\ diff --git a/trytond/backend/sqlite/database.py b/trytond/backend/sqlite/database.py index d522a599ce..6d11388fe3 100644 --- a/trytond/backend/sqlite/database.py +++ b/trytond/backend/sqlite/database.py @@ -54,7 +54,7 @@ def extract(lookup_type, date): elif lookup_type.lower() == 'millennium': return date.year / 1000 + (date.year % 1000 and 1 or 0) elif lookup_type.lower() == 'milliseconds': - return a.microsecond / 1000 + return date.microsecond / 1000 elif lookup_type.lower() == 'quarter': return date.month / 4 + 1 elif lookup_type.lower() == 'week': @@ -65,15 +65,16 @@ def date_trunc(_type, date): if _type == 'second': return date try: - time.strptime(date, '%Y-%m-%d %H:%M:%S') + tm_tuple = time.strptime(date, '%Y-%m-%d %H:%M:%S') except: return None if _type == 'year': - return "%i-01-01 00:00:00" % dt.year + return "%i-01-01 00:00:00" % tm_tuple.tm_year elif _type == 'month': - return "%i-%02i-01 00:00:00" % (dt.year, dt.month) + return "%i-%02i-01 00:00:00" % (tm_tuple.tm_year, tm_tuple.tm_mon) elif _type == 'day': - return "%i-%02i-%02i 00:00:00" % (dt.year, dt.month, dt.day) + return "%i-%02i-%02i 00:00:00" % (tm_tuple.tm_year, tm_tuple.tm_mon, + tm_tuple.tm_mday) def split_part(text, delimiter, count): return (text.split(delimiter) + [''] * (count - 1))[count - 1] diff --git a/trytond/ir/translation.py b/trytond/ir/translation.py index f7d17d6123..524e88aa0d 100644 --- a/trytond/ir/translation.py +++ b/trytond/ir/translation.py @@ -891,6 +891,11 @@ class TranslationClean(Wizard): 'delete_xml_record', 'xml_record_desc', 'write_xml_record', + 'not_found_in_selection', + 'relation_not_found', + 'too_many_relations_found', + 'xml_id_syntax_error', + 'reference_syntax_error', 'delete_workflow_record', 'domain_validation_record', 'required_validation_record', @@ -901,6 +906,7 @@ class TranslationClean(Wizard): 'required_field', 'foreign_model_missing', 'foreign_model_exist', + 'search_function_missing', ): continue if model_name in self.pool.object_name_list(): diff --git a/trytond/ir/translation.xml b/trytond/ir/translation.xml index 9a95fd10ba..08afa35524 100644 --- a/trytond/ir/translation.xml +++ b/trytond/ir/translation.xml @@ -97,6 +97,51 @@ this repository contains the full copyright notices and license terms. --> ir + + not_found_in_selection + en_US + error + Key %r not found in selection field %r + Key %r not found in selection field %r + ir + + + + relation_not_found + en_US + error + Relation not found: %r in %s + Relation not found: %r in %s + ir + + + + too_many_relations_found + en_US + error + Too many relations found: %r in %s + Too many relations found: %r in %s + ir + + + + xml_id_syntax_error + en_US + error + Syntax error for XML id %r in %s + Syntax error for XML id %r in %s + ir + + + + reference_syntax_error + en_US + error + Syntax error for reference %r in %s + Syntax error for reference %r in %s + ir + + delete_workflow_record en_US @@ -193,6 +238,15 @@ this repository contains the full copyright notices and license terms. --> ir + + search_function_missing + en_US + error + Missing search function on field "%s". + Missing search function on field "%s". + ir + + ir.translation.set_report.init diff --git a/trytond/ir/ui/view.py b/trytond/ir/ui/view.py index b455d84cb9..508f1c9c65 100644 --- a/trytond/ir/ui/view.py +++ b/trytond/ir/ui/view.py @@ -154,9 +154,9 @@ class View(ModelSQL, ModelView): if seqmatch.ratio() > 0.6: cursor.execute('UPDATE ir_translation ' \ 'SET src = %s, ' \ - 'fuzzy = True ' \ + 'fuzzy = %s ' \ 'WHERE id = %s ', - (string, trans_views[string_trans]['id'])) + (string, True, trans_views[string_trans]['id'])) del trans_views[string_trans] done = True break diff --git a/trytond/model/fields/function.py b/trytond/model/fields/function.py index 16b911f173..2d3320df25 100644 --- a/trytond/model/fields/function.py +++ b/trytond/model/fields/function.py @@ -92,12 +92,14 @@ class Function(Field): :param cursor: The database cursor. :param user: The user id. :param model: The model. + :param name: The name of the field. :param args: The search domain. See ModelStorage.search :param context: The context. :return: New list of domain. ''' if not self.fnct_search: - return [] + model.raise_user_error(cursor, 'search_function_missing', + name, context=context) return getattr(model, self.fnct_search)(cursor, user, name, args, context=context) diff --git a/trytond/model/fields/many2one.py b/trytond/model/fields/many2one.py index 8787fe40be..8e04f94f63 100644 --- a/trytond/model/fields/many2one.py +++ b/trytond/model/fields/many2one.py @@ -23,9 +23,14 @@ class Many2One(Field): :param right: The name of the field to store the right value. See left :param ondelete: Define the behavior of the record when the target record is deleted. (``CASCADE``, ``RESTRICT``, ``SET NULL``) + ``SET NULL`` will be changed into ``RESTRICT`` if required is set. :param datetime_field: The name of the field that contains the datetime value to read the target record. ''' + self.__required = required + if ondelete not in ('CASCADE', 'RESTRICT', 'SET NULL'): + raise Exception('Bad arguments') + self.ondelete = ondelete if datetime_field: if depends: depends.append(datetime_field) @@ -40,8 +45,16 @@ class Many2One(Field): self.model_name = model_name self.left = left self.right = right - if ondelete not in ('CASCADE', 'RESTRICT', 'SET NULL'): - raise Exception('Bad arguments') - self.ondelete = ondelete self.datetime_field = datetime_field + + def __get_required(self): + return self.__required + + def __set_required(self, value): + self.__required = value + if value and self.ondelete == 'SET NULL': + self.ondelete = 'RESTRICT' + + required = property(__get_required, __set_required) + __init__.__doc__ += Field.__init__.__doc__ diff --git a/trytond/model/modelsingleton.py b/trytond/model/modelsingleton.py index 07da6d3656..4a9e79d394 100644 --- a/trytond/model/modelsingleton.py +++ b/trytond/model/modelsingleton.py @@ -29,9 +29,11 @@ class ModelSingleton(ModelStorage): if not singleton_id: res = self.default_get(cursor, user, fields_names, context=context, with_rec_name=False) - res['id'] = 1 if not isinstance(ids, (int, long)): + res['id'] = ids[0] res = [res] + else: + res['id'] = ids return res if isinstance(ids, (int, long)): ids2 = singleton_id diff --git a/trytond/model/modelstorage.py b/trytond/model/modelstorage.py index b17952b640..810130a9e9 100644 --- a/trytond/model/modelstorage.py +++ b/trytond/model/modelstorage.py @@ -1,5 +1,6 @@ #This file is part of Tryton. The COPYRIGHT file at the top level of #this repository contains the full copyright notices and license terms. + from trytond.model import Model from trytond.model import fields from trytond.model.browse import BrowseRecordList, BrowseRecord, BrowseRecordNull @@ -12,6 +13,11 @@ import time from decimal import Decimal import logging from itertools import chain +try: + import cStringIO as StringIO +except ImprotError: + import StringIO +import csv class ModelStorage(Model): @@ -509,8 +515,8 @@ class ModelStorage(Model): if isinstance(ids, (int, long)): return BrowseRecord(cursor, user, ids, self, context=context) local_cache = {} - return BrowseRecordList([BrowseRecord(cursor, user, x, self, - local_cache=local_cache, context=context) for x in ids], + return BrowseRecordList((BrowseRecord(cursor, user, x, self, + local_cache=local_cache, context=context) for x in ids), context=context) def __export_row(self, cursor, user, row, fields_names, context=None): @@ -584,186 +590,244 @@ class ModelStorage(Model): - the exception if failed - the warning if failed ''' - if context is None: - context = {} - fields_names = [x.split('/') for x in fields_names] - logger = logging.getLogger('import') + def process_lines(self, datas, prefix, fields_def, position=0): + + def warn(msgname, *args): + msg = self.raise_user_error(cursor, msgname, args, + raise_exception=False, context=context) + warnings.warn(msg) + + def get_selection(selection, value): + res = False + if not isinstance(selection, (tuple, list)): + selection = getattr(self, selection)(cursor, user, + context=context) + for key, val in selection: + if str(key) == value: + res = key + break + if value and not res: + warn('not_found_in_selection', value, '/'.join(field)) + return res + + def get_many2one(relation, value): + if not value: + return False + relation_obj = self.pool.get(relation) + res = relation_obj.search(cursor, user, [ + ('rec_name', '=', value), + ], limit=2, context=context) + if len(res) < 1: + warn('relation_not_found', value, relation) + res = False + elif len(res) > 1: + warn('too_many_relations_found', value, relation) + res = False + else: + res = res[0] + return res + + def get_many2many(relation, value): + if not value: + return False + res = [] + relation_obj = self.pool.get(relation) + for word in csv.reader(StringIO.StringIO(value), delimiter=',', + quoting=csv.QUOTE_NONE, escapechar='\\').next(): + res2 = relation_obj.search(cursor, user, [ + ('rec_name', '=', word), + ], limit=2, context=context) + if len(res2) < 1: + warn('relation_not_found', word, relation) + elif len(res2) > 1: + warn('too_many_relations_found', word, relation) + else: + res.extend(res2) + if len(res): + res = [('set', res)] + return res + + def get_reference(value): + if not value: + return False + try: + relation, value = value.split(',', 1) + except: + warn('reference_syntax_error', value, '/'.join(field)) + return False + relation_obj = self.pool.get(relation) + res = relation_obj.search(cursor, user, [ + ('rec_name', '=', value), + ], limit=2, context=context) + if len(res) < 1: + warn('relation_not_found', value, relation) + res = False + elif len(res) > 1: + warn('too_many_relations_found', value, relation) + res = False + else: + res = '%s,%s' % (relation, str(res[0])) + return res + + def get_by_id(value): + if not value: + return False + relation = None + type = fields_def[field[-1][:-3]]['type'] + if type == 'many2many': + value = csv.reader(StringIO.StringIO(value), delimiter=',', + quoting=csv.QUOTE_NONE, escapechar='\\').next() + elif type == 'reference': + try: + relation, value = value.split(',', 1) + except: + warn('reference_syntax_error', value, '/'.join(field)) + return False + value = [value] + else: + value = [value] + res_ids = [] + for word in value: + try: + module, xml_id = word.rsplit('.', 1) + except: + warn('xml_id_syntax_error', word, '/'.join(field)) + continue + db_id = ir_model_data_obj.get_id(cursor, user, + module, xml_id) + res_ids.append(db_id) + if type == 'many2many' and res_ids: + return [('set', res_ids)] + elif type == 'reference' and res_ids: + return '%s,%s' % (relation, str(res_ids[0])) + return res_ids and res_ids[0] or False - def process_liness(self, datas, prefix, fields_def, position=0): line = datas[position] row = {} translate = {} - todo = [] - warning = '' - + todo = set() + prefix_len = len(prefix) # Import normal fields_names - for i in range(len(fields_names)): + for i, field in enumerate(fields_names): if i >= len(line): raise Exception('ImportError', 'Please check that all your lines have %d cols.' % \ (len(fields_names),)) - field = fields_names[i] - if (len(field) == len(prefix) + 1) \ - and field[len(prefix)].endswith(':id'): - res_id = False - if line[i]: - if fields_def[field[len(prefix)][:-3]]['type'] \ - == 'many2many': - res_id = [] - for word in line[i].split(','): - module, xml_id = word.rsplit('.', 1) - ir_model_data_obj = \ - self.pool.get('ir.model.data') - new_id = ir_model_data_obj.get_id(cursor, - user, module, xml_id) - res_id2 = ir_model_data_obj.read(cursor, user, - [new_id], ['res_id'])[0]['res_id'] - if res_id2: - res_id.append(res_id2) - if len(res_id): - res_id = [('set', res_id)] + is_prefix_len = (len(field) == (prefix_len + 1)) + value = line[i] + if is_prefix_len and field[-1].endswith(':id'): + row[field[0][:-3]] = get_by_id(value) + elif is_prefix_len and ':lang=' in field[-1]: + field_name, lang = field[-1].split(':lang=') + translate.setdefault(lang, {})[field_name] = value or False + elif is_prefix_len and prefix == field[:-1]: + this_field_def = fields_def[field[-1]] + field_type = this_field_def['type'] + res = False + if field_type == 'boolean': + if value.lower() == 'true': + res = True + elif value.lower() == 'false': + res = False else: - module, xml_id = line[i].rsplit('.', 1) - ir_model_data_obj = self.pool.get('ir.model.data') - new_id = ir_model_data_obj.get_id(cursor, user, - module, xml_id) - res_id = ir_model_data_obj.read(cursor, user, - [new_id], ['res_id'])[0]['res_id'] - row[field[0][:-3]] = res_id or False - continue - if (len(field) == len(prefix)+1) and \ - len(field[len(prefix)].split(':lang=')) == 2: - field, lang = field[len(prefix)].split(':lang=') - translate.setdefault(lang, {})[field]=line[i] or False - continue - if (len(field) == len(prefix)+1) and \ - (prefix == field[0:len(prefix)]): - if fields_def[field[len(prefix)]]['type'] == 'integer': - res = line[i] and int(line[i]) - elif fields_def[field[len(prefix)]]['type'] == 'float': - res = line[i] and float(line[i]) - elif fields_def[field[len(prefix)]]['type'] == 'selection': - res = False - if isinstance( - fields_def[field[len(prefix)]]['selection'], - (tuple, list)): - sel = fields_def[field[len(prefix)]]['selection'] - else: - sel = getattr(self, fields_def[field[len(prefix)]]\ - ['selection'])(cursor, user, context) - for key, val in sel: - if str(key) == line[i]: - res = key - if line[i] and not res: - logger.warning("key '%s' not found " \ - "in selection field '%s'" % \ - (line[i], field[len(prefix)])) - elif fields_def[field[len(prefix)]]['type'] == 'many2one': - res = False - if line[i]: - relation = \ - fields_def[field[len(prefix)]]['relation'] - relation_obj = self.pool.get(relation) - res = relation_obj.search(cursor, user, [ - ('rec_name', '=', line[i]), - ], limit=1, context=context) - if not res: - warning += ('Relation not found: ' + line[i] + \ - ' on ' + relation + ' !\n') - logger.warning( - 'Relation not found: ' + line[i] + \ - ' on ' + relation + ' !\n') - res = False - else: - res = res[0] - elif fields_def[field[len(prefix)]]['type'] == 'many2many': - res = [] - if line[i]: - relation = \ - fields_def[field[len(prefix)]]['relation'] - for word in line[i].split(','): - relation_obj = self.pool.get(relation) - res2 = relation_obj.search(cursor, user, [ - ('rec_name', '=', word), - ], limit=1, context=context) - if not res2: - warning += ('Relation not found: ' + \ - word + ' on '+relation + ' !\n') - logger.warning( - 'Relation not found: ' + word + \ - ' on '+relation + ' !\n') - else: - res.extend(res2) - if len(res): - res = [('set', res)] + res = value and bool(int(value)) + elif field_type == 'integer': + res = value and int(value) + elif field_type == 'float': + res = value and float(value) + elif field_type == 'numeric': + res = value and Decimal(value) + elif field_type == 'date': + res = value and datetime.date(*time.strptime(value, + '%Y-%m-%d')[:3]) + elif field_type == 'datetime': + res = value and datetime.datetime(*time.strptime(value, + '%Y-%m-%d %H:%M:%S')[:6]) + elif field_type == 'selection': + res = get_selection(this_field_def['selection'], value) + elif field_type == 'many2one': + res = get_many2one(this_field_def['relation'], value) + elif field_type == 'many2many': + res = get_many2many(this_field_def['relation'], value) + elif field_type == 'reference': + res = get_reference(value) else: - res = line[i] or False - row[field[len(prefix)]] = res - elif (prefix==field[0:len(prefix)]): - if field[0] not in todo: - todo.append(field[len(prefix)]) - + res = value or False + row[field[-1]] = res + elif prefix == field[0:prefix_len]: + todo.add(field[prefix_len]) # Import one2many fields nbrmax = 1 for field in todo: newfd = self.pool.get(fields_def[field]['relation']).fields_get( cursor, user, context=context) - res = process_liness(self, datas, prefix + [field], newfd, + res = process_lines(self, datas, prefix + [field], newfd, position) - (newrow, max2, warning2, translate2) = res + (newrow, max2, translate2) = res nbrmax = max(nbrmax, max2) - warning = warning + warning2 reduce(lambda x, y: x and y, newrow) - row[field] = (reduce(lambda x, y: x or y, newrow.values()) and \ - [('create', newrow)]) or [] + row[field] = (reduce(lambda x, y: x or y, newrow.values()) and + [('create', newrow)]) or [] i = max2 - while (position+i) + + + + + + Test + + + + Duplicate + + + + Duplicate + + + + Test 1 + + + + Test 2 + + + + Test, comma + + + + Duplicate + + + + Duplicate + + + + Test + + + + Duplicate + + + + Duplicate + + + + diff --git a/trytond/tests/test_tryton.py b/trytond/tests/test_tryton.py index e4e0c8d9ba..549157d153 100644 --- a/trytond/tests/test_tryton.py +++ b/trytond/tests/test_tryton.py @@ -2913,6 +2913,335 @@ class MPTTTestCase(unittest.TestCase): self.CheckTree() +class ImportDataTestCase(unittest.TestCase): + ''' + Test import_data. + ''' + + def setUp(self): + install_module('tests') + self.boolean = RPCProxy('tests.import_data.boolean') + self.integer = RPCProxy('tests.import_data.integer') + self.float = RPCProxy('tests.import_data.float') + self.numeric = RPCProxy('tests.import_data.numeric') + self.char = RPCProxy('tests.import_data.char') + self.text = RPCProxy('tests.import_data.text') + self.sha = RPCProxy('tests.import_data.sha') + self.date = RPCProxy('tests.import_data.date') + self.datetime = RPCProxy('tests.import_data.datetime') + self.selection = RPCProxy('tests.import_data.selection') + self.many2one = RPCProxy('tests.import_data.many2one') + self.many2many = RPCProxy('tests.import_data.many2many') + self.one2many = RPCProxy('tests.import_data.one2many') + self.reference = RPCProxy('tests.import_data.reference') + + def test0010boolean(self): + ''' + Test boolean. + ''' + self.assert_(self.boolean.import_data(['boolean'], [['True']], CONTEXT) + == (1, 0, 0, 0)) + + self.assert_(self.boolean.import_data(['boolean'], [['1']], CONTEXT) + == (1, 0, 0, 0)) + + self.assert_(self.boolean.import_data(['boolean'], [['False']], CONTEXT) + == (1, 0, 0, 0)) + + self.assert_(self.boolean.import_data(['boolean'], [['0']], CONTEXT) + == (1, 0, 0, 0)) + + self.assert_(self.boolean.import_data(['boolean'], [['']], CONTEXT) + == (1, 0, 0, 0)) + + self.assert_(self.boolean.import_data(['boolean'], + [['True'], ['False']], CONTEXT) == (2, 0, 0, 0)) + + self.assert_(self.boolean.import_data(['boolean'], [['foo']], + CONTEXT)[0] == -1) + + def test0020integer(self): + ''' + Test integer. + ''' + self.assert_(self.integer.import_data(['integer'], [['1']], CONTEXT) + == (1, 0, 0, 0)) + + self.assert_(self.integer.import_data(['integer'], [['-1']], CONTEXT) + == (1, 0, 0, 0)) + + self.assert_(self.integer.import_data(['integer'], [['']], CONTEXT) + == (1, 0, 0, 0)) + + self.assert_(self.integer.import_data(['integer'], + [['1'], ['2']], CONTEXT) == (2, 0, 0, 0)) + + self.assert_(self.integer.import_data(['integer'], [['1.1']], + CONTEXT)[0] == -1) + + self.assert_(self.integer.import_data(['integer'], [['-1.1']], + CONTEXT)[0] == -1) + + self.assert_(self.integer.import_data(['integer'], [['foo']], + CONTEXT)[0] == -1) + + def test0030float(self): + ''' + Test float. + ''' + self.assert_(self.float.import_data(['float'], [['1.1']], CONTEXT) + == (1, 0, 0, 0)) + + self.assert_(self.float.import_data(['float'], [['-1.1']], CONTEXT) + == (1, 0, 0, 0)) + + self.assert_(self.float.import_data(['float'], [['1']], CONTEXT) + == (1, 0, 0, 0)) + + self.assert_(self.float.import_data(['float'], [['']], CONTEXT) + == (1, 0, 0, 0)) + + self.assert_(self.float.import_data(['float'], + [['1.1'], ['2.2']], CONTEXT) == (2, 0, 0, 0)) + + self.assert_(self.float.import_data(['float'], [['foo']], CONTEXT)[0] + == -1) + + def test0040numeric(self): + ''' + Test numeric. + ''' + self.assert_(self.numeric.import_data(['numeric'], [['1.1']], CONTEXT) + == (1, 0, 0, 0)) + + self.assert_(self.numeric.import_data(['numeric'], [['-1.1']], CONTEXT) + == (1, 0, 0, 0)) + + self.assert_(self.numeric.import_data(['numeric'], [['1']], CONTEXT) + == (1, 0, 0, 0)) + + self.assert_(self.numeric.import_data(['numeric'], [['']], CONTEXT) + == (1, 0, 0, 0)) + + self.assert_(self.numeric.import_data(['numeric'], + [['1.1'], ['2.2']], CONTEXT) == (2, 0, 0, 0)) + + self.assert_(self.numeric.import_data(['numeric'], [['foo']], + CONTEXT)[0] == -1) + + def test0050char(self): + ''' + Test char. + ''' + self.assert_(self.char.import_data(['char'], [['test']], CONTEXT) + == (1, 0, 0, 0)) + + self.assert_(self.char.import_data(['char'], [['']], CONTEXT) + == (1, 0, 0, 0)) + + self.assert_(self.char.import_data(['char'], + [['test'], ['foo'], ['bar']], CONTEXT) == (3, 0, 0, 0)) + + def test0060text(self): + ''' + Test text. + ''' + self.assert_(self.text.import_data(['text'], [['test']], CONTEXT) + == (1, 0, 0, 0)) + + self.assert_(self.text.import_data(['text'], [['']], CONTEXT) + == (1, 0, 0, 0)) + + self.assert_(self.text.import_data(['text'], + [['test'], ['foo'], ['bar']], CONTEXT) == (3, 0, 0, 0)) + + def test0070sha(self): + ''' + Test sha. + ''' + self.assert_(self.sha.import_data(['sha'], [['test']], CONTEXT) + == (1, 0, 0, 0)) + + self.assert_(self.sha.import_data(['sha'], [['']], CONTEXT) + == (1, 0, 0, 0)) + + self.assert_(self.sha.import_data(['sha'], + [['test'], ['foo']], CONTEXT) == (2, 0, 0, 0)) + + def test0080date(self): + ''' + Test date. + ''' + self.assert_(self.date.import_data(['date'], [['2010-01-01']], CONTEXT) + == (1, 0, 0, 0)) + + self.assert_(self.date.import_data(['date'], [['']], CONTEXT) + == (1, 0, 0, 0)) + + self.assert_(self.date.import_data(['date'], [ + ['2010-01-01'], ['2010-02-01']], CONTEXT) == (2, 0, 0, 0)) + + self.assert_(self.date.import_data(['date'], [['foo']], CONTEXT)[0] + == -1) + + def test0090datetime(self): + ''' + Test datetime. + ''' + self.assert_(self.datetime.import_data(['datetime'], + [['2010-01-01 12:00:00']], CONTEXT) + == (1, 0, 0, 0)) + + self.assert_(self.datetime.import_data(['datetime'], [['']], CONTEXT) + == (1, 0, 0, 0)) + + self.assert_(self.datetime.import_data(['datetime'], + [['2010-01-01 12:00:00'], ['2010-01-01 13:30:00']], CONTEXT) + == (2, 0, 0, 0)) + + self.assert_(self.datetime.import_data(['datetime'], [['foo']], + CONTEXT)[0] == -1) + + def test0100selection(self): + ''' + Test selection. + ''' + self.assert_(self.selection.import_data(['selection'], [['select1']], + CONTEXT) == (1, 0, 0, 0)) + + self.assert_(self.selection.import_data(['selection'], [['']], + CONTEXT) == (1, 0, 0, 0)) + + self.assert_(self.selection.import_data(['selection'], + [['select1'], ['select2']], CONTEXT) == (2, 0, 0, 0)) + + self.assert_(self.selection.import_data(['selection'], [['foo']], + CONTEXT)[0] == -1) + + def test0110many2one(self): + ''' + Test many2one. + ''' + self.assert_(self.many2one.import_data(['many2one'], [['Test']], + CONTEXT) == (1, 0, 0, 0)) + + self.assert_(self.many2one.import_data(['many2one:id'], + [['tests.import_data_many2one_target_test']], CONTEXT) + == (1, 0, 0, 0)) + + self.assert_(self.many2one.import_data(['many2one'], [['']], + CONTEXT) == (1, 0, 0, 0)) + + self.assert_(self.many2one.import_data(['many2one'], + [['Test'], ['Test']], CONTEXT) == (2, 0, 0, 0)) + + self.assert_(self.many2one.import_data(['many2one'], [['foo']], + CONTEXT)[0] == -1) + + self.assert_(self.many2one.import_data(['many2one'], [['Duplicate']], + CONTEXT)[0] == -1) + + self.assert_(self.many2one.import_data(['many2one:id'], [['foo']], + CONTEXT)[0] == -1) + + self.assert_(self.many2one.import_data(['many2one:id'], + [['tests.foo']], CONTEXT)[0] == -1) + + def test0120many2many(self): + ''' + Test many2many. + ''' + self.assert_(self.many2many.import_data(['many2many'], [['Test 1']], + CONTEXT) == (1, 0, 0, 0)) + + self.assert_(self.many2many.import_data(['many2many:id'], + [['tests.import_data_many2many_target_test1']], CONTEXT) + == (1, 0, 0, 0)) + + self.assert_(self.many2many.import_data(['many2many'], + [['Test 1,Test 2']], CONTEXT) == (1, 0, 0, 0)) + + self.assert_(self.many2many.import_data(['many2many:id'], + [['tests.import_data_many2many_target_test1,' \ + 'tests.import_data_many2many_target_test2']], CONTEXT) + == (1, 0, 0, 0)) + + self.assert_(self.many2many.import_data(['many2many'], + [['Test\, comma']], CONTEXT) == (1, 0, 0, 0)) + + self.assert_(self.many2many.import_data(['many2many'], + [['Test\, comma,Test 1']], CONTEXT) == (1, 0, 0, 0)) + + self.assert_(self.many2many.import_data(['many2many'], [['']], + CONTEXT) == (1, 0, 0, 0)) + + self.assert_(self.many2many.import_data(['many2many'], + [['Test 1'], ['Test 2']], CONTEXT) == (2, 0, 0, 0)) + + self.assert_(self.many2many.import_data(['many2many'], [['foo']], + CONTEXT)[0] == -1) + + self.assert_(self.many2many.import_data(['many2many'], [['Test 1,foo']], + CONTEXT)[0] == -1) + + self.assert_(self.many2many.import_data(['many2many'], [['Duplicate']], + CONTEXT)[0] == -1) + + self.assert_(self.many2many.import_data(['many2many'], + [['Test 1,Duplicate']], CONTEXT)[0] == -1) + + def test0130one2many(self): + ''' + Test one2many. + ''' + self.assert_(self.one2many.import_data(['name', 'one2many/name'], + [['Test', 'Test 1']], CONTEXT) == (1, 0, 0, 0)) + + self.assert_(self.one2many.import_data(['name', 'one2many/name'], + [['Test', 'Test 1'], ['', 'Test 2']], CONTEXT) == (1, 0, 0, 0)) + + self.assert_(self.one2many.import_data(['name', 'one2many/name'], + [['Test 1', 'Test 1'], ['', 'Test 2'], + ['Test 2', 'Test 1']], CONTEXT) == (2, 0, 0, 0)) + + def test0140reference(self): + ''' + Test reference. + ''' + self.assert_(self.reference.import_data(['reference'], + [['tests.import_data.reference.selection,Test']], + CONTEXT) == (1, 0, 0, 0)) + + self.assert_(self.reference.import_data(['reference:id'], + [['tests.import_data.reference.selection,'\ + 'tests.import_data_reference_selection_test']], CONTEXT) + == (1, 0, 0, 0)) + + self.assert_(self.reference.import_data(['reference'], [['']], + CONTEXT) == (1, 0, 0, 0)) + + self.assert_(self.reference.import_data(['reference'], + [['tests.import_data.reference.selection,Test'], + ['tests.import_data.reference.selection,Test']], + CONTEXT) == (2, 0, 0, 0)) + + self.assert_(self.reference.import_data(['reference'], + [['tests.import_data.reference.selection,foo']], + CONTEXT)[0] == -1) + + self.assert_(self.reference.import_data(['reference'], + [['tests.import_data.reference.selection,Duplicate']], + CONTEXT)[0] == -1) + + self.assert_(self.reference.import_data(['reference:id'], + [['tests.import_data.reference.selection,foo']], + CONTEXT)[0] == -1) + + self.assert_(self.reference.import_data(['reference:id'], + [['tests.import_data.reference.selection,tests.foo']], + CONTEXT)[0] == -1) + + class RPCProxy(object): def __init__(self, name): @@ -3008,5 +3337,7 @@ if __name__ == '__main__': suite.addTests(unittest.TestLoader().loadTestsFromTestCase(ModelSingletonTestCase)) suite.addTests(unittest.TestLoader().loadTestsFromTestCase(ModelViewTestCase)) suite.addTests(unittest.TestLoader().loadTestsFromTestCase(MPTTTestCase)) + suite.addTests(unittest.TestLoader().loadTestsFromTestCase( + ImportDataTestCase)) unittest.TextTestRunner(verbosity=2).run(suite) SOCK.disconnect() diff --git a/trytond/tools/misc.py b/trytond/tools/misc.py index 21ecac11d6..50bff0c056 100644 --- a/trytond/tools/misc.py +++ b/trytond/tools/misc.py @@ -67,10 +67,8 @@ def exec_pg_command_pipe(name, *args): child_env = dict(os.environ) if CONFIG['db_password']: child_env['PGPASSWORD'] = CONFIG['db_password'] - pipe = subprocess.Popen(cmd, shell=True, - stdin=subprocess.PIPE, stdout=subprocess.PIPE, - close_fds=True, - env=child_env) + pipe = subprocess.Popen(cmd, shell=True, stdin=subprocess.PIPE, + stdout=subprocess.PIPE, env=child_env) return pipe def exec_command_pipe(name, *args):