tryton-do/tryton
2
0 Fork 0
Code Issues Pull Requests 2 Actions 2 Packages Projects Releases Wiki Activity

merge

Browse Source
This commit is contained in:
Cédric Krier committed 2010-02-11 11:12:33 +01:00
2 parents 0eea3a9afb 6997fd60b2
commit 5a2baacef4
19 files changed
+884 -161
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

CHANGELOG
+3
View File
@@ -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
COPYRIGHT
+3 -3
View File
@@ -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
trytond/backend/postgresql/database.py
+1 -1
View File
@@ -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
trytond/backend/postgresql/table.py
+2 -2
View File
@@ -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 "\
trytond/backend/sqlite/database.py
+6 -5
View File
@@ -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]
trytond/ir/translation.py
+6
View File
@@ -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():
trytond/ir/translation.xml
+54
View File
@@ -97,6 +97,51 @@ this repository contains the full copyright notices and license terms. -->
<field name="module">ir</field>
<field name="fuzzy" eval="False"/>
</record>
<record model="ir.translation" id="translation_not_found_in_selection">
<field name="name">not_found_in_selection</field>
<field name="lang">en_US</field>
<field name="type">error</field>
<field name="src">Key %r not found in selection field %r</field>
<field name="value">Key %r not found in selection field %r</field>
<field name="module">ir</field>
<field name="fuzzy" eval="False"/>
</record>
<record model="ir.translation" id="translation_relation_not_found">
<field name="name">relation_not_found</field>
<field name="lang">en_US</field>
<field name="type">error</field>
<field name="src">Relation not found: %r in %s</field>
<field name="value">Relation not found: %r in %s</field>
<field name="module">ir</field>
<field name="fuzzy" eval="False"/>
</record>
<record model="ir.translation" id="translation_too_many_relations_found">
<field name="name">too_many_relations_found</field>
<field name="lang">en_US</field>
<field name="type">error</field>
<field name="src">Too many relations found: %r in %s</field>
<field name="value">Too many relations found: %r in %s</field>
<field name="module">ir</field>
<field name="fuzzy" eval="False"/>
</record>
<record model="ir.translation" id="translation_xml_id_syntax_error">
<field name="name">xml_id_syntax_error</field>
<field name="lang">en_US</field>
<field name="type">error</field>
<field name="src">Syntax error for XML id %r in %s</field>
<field name="value">Syntax error for XML id %r in %s</field>
<field name="module">ir</field>
<field name="fuzzy" eval="False"/>
</record>
<record model="ir.translation" id="translation_reference_syntax_error">
<field name="name">reference_syntax_error</field>
<field name="lang">en_US</field>
<field name="type">error</field>
<field name="src">Syntax error for reference %r in %s</field>
<field name="value">Syntax error for reference %r in %s</field>
<field name="module">ir</field>
<field name="fuzzy" eval="False"/>
</record>
<record model="ir.translation" id="translation_delete_workflow_record">
<field name="name">delete_workflow_record</field>
<field name="lang">en_US</field>
@@ -193,6 +238,15 @@ this repository contains the full copyright notices and license terms. -->
<field name="module">ir</field>
<field name="fuzzy" eval="False"/>
</record>
<record model="ir.translation" id="translation_search_function_missing">
<field name="name">search_function_missing</field>
<field name="lang">en_US</field>
<field name="type">error</field>
<field name="src">Missing search function on field "%s".</field>
<field name="value">Missing search function on field "%s".</field>
<field name="module">ir</field>
<field name="fuzzy" eval="False"/>
</record>
<record model="ir.ui.view" id="act_translation_set_report_init_view_form">
<field name="model">ir.translation.set_report.init</field>
trytond/ir/ui/view.py
+2 -2
View File
@@ -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
trytond/model/fields/function.py
+3 -1
View File
@@ -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)
trytond/model/fields/many2one.py
+16 -3
View File
@@ -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__
trytond/model/modelsingleton.py
+3 -1
View File
@@ -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
trytond/model/modelstorage.py
+200 -136
View File
@@ -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)<len(datas):
while (position + i) < len(datas):
test = True
for j in range(len(fields_names)):
field2 = fields_names[j]
if (len(field2) <= (len(prefix)+1)) \
and datas[position+i][j]:
for j, field2 in enumerate(fields_names):
if len(field2) <= (prefix_len + 1) \
and datas[position + i][j]:
test = False
break
if not test:
break
(newrow, max2, warning2, translate2) = \
process_liness(self, datas, prefix+[field], newfd,
(newrow, max2, translate2) = \
process_lines(self, datas, prefix + [field], newfd,
position + i)
warning = warning + warning2
if reduce(lambda x, y: x or y, newrow.values()):
row[field].append(('create', newrow))
i += max2
nbrmax = max(nbrmax, i)
if len(prefix) == 0:
for i in range(max(nbrmax, 1)):
if prefix_len == 0:
for i in xrange(max(nbrmax, 1)):
datas.pop(0)
result = (row, nbrmax, warning, translate)
return result
return (row, nbrmax, translate)
ir_model_data_obj = self.pool.get('ir.model.data')
# logger for collecting warnings for the client
warnings = logging.Logger("import")
warning_stream = StringIO.StringIO()
warnings.addHandler(logging.StreamHandler(warning_stream))
if context is None:
context = {}
len_fields_names = len(fields_names)
for data in datas:
assert len(data) == len_fields_names
fields_names = [x.split('/') for x in fields_names]
fields_def = self.fields_get(cursor, user, context=context)
done = 0
warning = ''
while len(datas):
res = {}
warning = ''
try:
(res, other, warning, translate) = \
process_liness(self, datas, [], fields_def)
(res, other, translate) = \
process_lines(self, datas, [], fields_def)
warning = warning_stream.getvalue()
if warning:
cursor.rollback()
return (-1, res, warning, '')
new_id = self.create(cursor, user, res, context=context)
trans_ctx = context.copy()
for lang in translate:
context2 = context.copy()
context2['language'] = lang
trans_ctx['language'] = lang
self.write(cursor, user, new_id, translate[lang],
context=context2)
context=trans_ctx)
except Exception, exp:
logger = logging.getLogger('import')
logger.error(exp)
cursor.rollback()
return (-1, res, exp[1], warning)
warning = '\n'.join(map(str, exp[1:]) + [warning])
return (-1, res, exp[0], warning)
done += 1
return (done, 0, 0, 0)
trytond/model/modelview.py
+1 -2
View File
@@ -91,6 +91,7 @@ class ModelView(Model):
super(ModelView, self).__init__()
self._rpc['fields_view_get'] = False
@Cache('modelview.fields_view_get')
def fields_view_get(self, cursor, user, view_id=None, view_type='form',
context=None, toolbar=False, hexmd5=None):
'''
@@ -248,8 +249,6 @@ class ModelView(Model):
return True
return result
fields_view_get = Cache('modelview.fields_view_get')(fields_view_get)
def view_header_get(self, cursor, user, value, view_type='form',
context=None):
"""
trytond/tests/__init__.py
+1
View File
@@ -4,3 +4,4 @@
from tests import *
from model import *
from mptt import *
from import_data import *
trytond/tests/__tryton__.py
+2 -1
View File
@@ -7,7 +7,8 @@
'active': True,
'depends': ['ir', 'res'],
'xml': [
],
'import_data.xml',
],
'translation': [
],
}
trytond/tests/import_data.py
+185
View File
@@ -0,0 +1,185 @@
#This file is part of Tryton. The COPYRIGHT file at the top level of
#this repository contains the full copyright notices and license terms.
"Test for import_data"
from trytond.model import ModelSQL, fields
class ImportDataBoolean(ModelSQL):
"Import Data Boolean"
_name = 'tests.import_data.boolean'
_description = __doc__
boolean = fields.Boolean('Boolean')
ImportDataBoolean()
class ImportDataInteger(ModelSQL):
"Import Data Integer"
_name = 'tests.import_data.integer'
_description = __doc__
integer = fields.Integer('Integer')
ImportDataInteger()
class ImportDataFloat(ModelSQL):
"Import Data Float"
_name = 'tests.import_data.float'
_description = __doc__
float = fields.Float('Float')
ImportDataFloat()
class ImportDataNumeric(ModelSQL):
"Import Data Numeric"
_name = 'tests.import_data.numeric'
_description = __doc__
numeric = fields.Numeric('Numeric')
ImportDataNumeric()
class ImportDataChar(ModelSQL):
"Import Data Char"
_name = 'tests.import_data.char'
_description = __doc__
char = fields.Char('Char')
ImportDataChar()
class ImportDataText(ModelSQL):
"Import Data Text"
_name = 'tests.import_data.text'
_description = __doc__
text = fields.Text('Text')
ImportDataText()
class ImportDataSha(ModelSQL):
"Import Data Sha"
_name = 'tests.import_data.sha'
_description = __doc__
sha = fields.Sha('Sha')
ImportDataSha()
class ImportDataDate(ModelSQL):
"Import Data Date"
_name = 'tests.import_data.date'
_description = __doc__
date = fields.Date('Date')
ImportDataDate()
class ImportDataDateTime(ModelSQL):
"Import Data DateTime"
_name = 'tests.import_data.datetime'
_description = __doc__
datetime = fields.DateTime('DateTime')
ImportDataDateTime()
class ImportDataSelection(ModelSQL):
"Import Data Selection"
_name = 'tests.import_data.selection'
_description = __doc__
selection = fields.Selection([
('select1', 'Select 1'),
('select2', 'Select 2'),
], 'Selection')
ImportDataSelection()
class ImportDataMany2OneTarget(ModelSQL):
"Import Data Many2One Target"
_name = 'tests.import_data.many2one.target'
_description = __doc__
name = fields.Char('Name')
ImportDataMany2OneTarget()
class ImportDataMany2One(ModelSQL):
"Import Data Many2One"
_name = 'tests.import_data.many2one'
_description = __doc__
many2one = fields.Many2One('tests.import_data.many2one.target',
'Many2One')
ImportDataMany2One()
class ImportDataMany2ManyTarget(ModelSQL):
"Import Data Many2Many Target"
_name = 'tests.import_data.many2many.target'
_description = __doc__
name = fields.Char('Name')
ImportDataMany2ManyTarget()
class ImportDataMany2Many(ModelSQL):
"Import Data Many2Many"
_name = 'tests.import_data.many2many'
_description = __doc__
many2many = fields.Many2Many('tests.import_data.many2many.relation',
'many2many', 'target', 'Many2Many')
ImportDataMany2Many()
class ImportDataMany2ManyRelation(ModelSQL):
"Import Data Many2Many Relation"
_name = 'tests.import_data.many2many.relation'
many2many = fields.Many2One('tests.import_data.many2many', 'Many2One')
target = fields.Many2One('tests.import_data.many2many.target', 'Target')
ImportDataMany2ManyRelation()
class ImportDataOne2Many(ModelSQL):
"Import Data One2Many"
_name = 'tests.import_data.one2many'
_description = __doc__
name = fields.Char('Name')
one2many = fields.One2Many('tests.import_data.one2many.target', 'one2many',
'One2Many')
ImportDataOne2Many()
class ImportDataOne2ManyTarget(ModelSQL):
"Import Data One2Many Target"
_name = 'tests.import_data.one2many.target'
_description = __doc__
name = fields.Char('Name')
one2many = fields.Many2One('tests.import_data.one2many', 'One2Many')
ImportDataOne2ManyTarget()
class ImportDataReferenceSelection(ModelSQL):
"Import Data Reference Selection"
_name = 'tests.import_data.reference.selection'
_description = __doc__
name = fields.Char('Name')
ImportDataReferenceSelection()
class ImportDataReference(ModelSQL):
"Import Data Reference"
_name = 'tests.import_data.reference'
reference = fields.Reference('Reference', [
('tests.import_data.reference.selection', 'Test'),
])
ImportDataReference()
trytond/tests/import_data.xml
+63
View File
@@ -0,0 +1,63 @@
<?xml version="1.0"?>
<!-- This file is part of Tryton. The COPYRIGHT file at the top level of
this repository contains the full copyright notices and license terms. -->
<tryton>
<data>
<record model="tests.import_data.many2one.target"
id="import_data_many2one_target_test">
<field name="name">Test</field>
</record>
<record model="tests.import_data.many2one.target"
id="import_data_many2one_target_duplicate1">
<field name="name">Duplicate</field>
</record>
<record model="tests.import_data.many2one.target"
id="import_data_many2one_target_duplicate2">
<field name="name">Duplicate</field>
</record>
<record model="tests.import_data.many2many.target"
id="import_data_many2many_target_test1">
<field name="name">Test 1</field>
</record>
<record model="tests.import_data.many2many.target"
id="import_data_many2many_target_test2">
<field name="name">Test 2</field>
</record>
<record model="tests.import_data.many2many.target"
id="import_data_many2many_target_test_comma">
<field name="name">Test, comma</field>
</record>
<record model="tests.import_data.many2many.target"
id="import_data_many2many_target_duplicate1">
<field name="name">Duplicate</field>
</record>
<record model="tests.import_data.many2many.target"
id="import_data_many2many_target_duplicate2">
<field name="name">Duplicate</field>
</record>
<record model="tests.import_data.reference.selection"
id="import_data_reference_selection_test">
<field name="name">Test</field>
</record>
<record model="tests.import_data.reference.selection"
id="import_data_reference_selection_duplicate1">
<field name="name">Duplicate</field>
</record>
<record model="tests.import_data.reference.selection"
id="import_data_reference_selection_duplicate2">
<field name="name">Duplicate</field>
</record>
</data>
</tryton>
trytond/tests/test_tryton.py
+331
View File
@@ -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()
trytond/tools/misc.py
+2 -4
View File
@@ -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):