[REM] models: remove deprecated method BaseModel.import_data()
This commit is contained in:
Raphael Collet
@@ -501,7 +501,7 @@ class ir_import(orm.TransientModel):
|
||||
def _convert_import_data(self, record, fields, options, context=None):
|
||||
""" Extracts the input browse_record and fields list (with
|
||||
``False``-y placeholders for fields to *not* import) into a
|
||||
format Model.import_data can use: a fields list without holes
|
||||
format Model.load can use: a fields list without holes
|
||||
and the precisely matching data matrix
|
||||
|
||||
:param browse_record record:
|
||||
|
||||
@@ -358,7 +358,7 @@ class test_preview(TransactionCase):
|
||||
|
||||
class test_convert_import_data(TransactionCase):
|
||||
""" Tests conversion of base_import.import input into data which
|
||||
can be fed to Model.import_data
|
||||
can be fed to Model.load
|
||||
"""
|
||||
def test_all(self):
|
||||
Import = self.registry('base_import.import')
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
from . import test_export, test_import, test_load
|
||||
from . import test_export, test_load
|
||||
|
||||
|
||||
@@ -1,877 +0,0 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
import openerp.modules.registry
|
||||
import openerp
|
||||
|
||||
from openerp.tests import common
|
||||
from openerp.tools.misc import mute_logger
|
||||
|
||||
def ok(n):
|
||||
""" Successful import of ``n`` records
|
||||
|
||||
:param int n: number of records which should have been imported
|
||||
"""
|
||||
return n, 0, 0, 0
|
||||
|
||||
def error(row, message, record=None, **kwargs):
|
||||
""" Failed import of the record ``record`` at line ``row``, with the error
|
||||
message ``message``
|
||||
|
||||
:param str message:
|
||||
:param dict record:
|
||||
"""
|
||||
return (
|
||||
-1, dict(record or {}, **kwargs),
|
||||
"Line %d : %s" % (row, message),
|
||||
'')
|
||||
|
||||
def values(seq, field='value'):
|
||||
return [item[field] for item in seq]
|
||||
|
||||
class ImporterCase(common.TransactionCase):
|
||||
model_name = False
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
super(ImporterCase, self).__init__(*args, **kwargs)
|
||||
self.model = None
|
||||
|
||||
def setUp(self):
|
||||
super(ImporterCase, self).setUp()
|
||||
self.model = self.registry(self.model_name)
|
||||
|
||||
def import_(self, fields, rows, context=None):
|
||||
return self.model.import_data(
|
||||
self.cr, openerp.SUPERUSER_ID, fields, rows, context=context)
|
||||
def read(self, fields=('value',), domain=(), context=None):
|
||||
return self.model.read(
|
||||
self.cr, openerp.SUPERUSER_ID,
|
||||
self.model.search(self.cr, openerp.SUPERUSER_ID, domain, context=context),
|
||||
fields=fields, context=context)
|
||||
def browse(self, domain=(), context=None):
|
||||
return self.model.browse(
|
||||
self.cr, openerp.SUPERUSER_ID,
|
||||
self.model.search(self.cr, openerp.SUPERUSER_ID, domain, context=context),
|
||||
context=context)
|
||||
|
||||
def xid(self, record):
|
||||
ModelData = self.registry('ir.model.data')
|
||||
|
||||
ids = ModelData.search(
|
||||
self.cr, openerp.SUPERUSER_ID,
|
||||
[('model', '=', record._name), ('res_id', '=', record.id)])
|
||||
if ids:
|
||||
d = ModelData.read(
|
||||
self.cr, openerp.SUPERUSER_ID, ids, ['name', 'module'])[0]
|
||||
if d['module']:
|
||||
return '%s.%s' % (d['module'], d['name'])
|
||||
return d['name']
|
||||
|
||||
name = record.name_get()[0][1]
|
||||
# fix dotted name_get results, otherwise xid lookups blow up
|
||||
name = name.replace('.', '-')
|
||||
ModelData.create(self.cr, openerp.SUPERUSER_ID, {
|
||||
'name': name,
|
||||
'model': record._name,
|
||||
'res_id': record.id,
|
||||
'module': '__test__'
|
||||
})
|
||||
return '__test__.' + name
|
||||
|
||||
class test_ids_stuff(ImporterCase):
|
||||
model_name = 'export.integer'
|
||||
|
||||
def test_create_with_id(self):
|
||||
self.assertEqual(
|
||||
self.import_(['.id', 'value'], [['42', '36']]),
|
||||
error(1, u"Unknown database identifier '42'"))
|
||||
def test_create_with_xid(self):
|
||||
self.assertEqual(
|
||||
self.import_(['id', 'value'], [['somexmlid', '42']]),
|
||||
ok(1))
|
||||
self.assertEqual(
|
||||
'somexmlid',
|
||||
self.xid(self.browse()[0]))
|
||||
|
||||
def test_update_with_id(self):
|
||||
id = self.model.create(self.cr, openerp.SUPERUSER_ID, {'value': 36})
|
||||
self.assertEqual(
|
||||
36,
|
||||
self.model.browse(self.cr, openerp.SUPERUSER_ID, id).value)
|
||||
|
||||
self.assertEqual(
|
||||
self.import_(['.id', 'value'], [[str(id), '42']]),
|
||||
ok(1))
|
||||
self.assertEqual(
|
||||
[42], # updated value to imported
|
||||
values(self.read()))
|
||||
|
||||
def test_update_with_xid(self):
|
||||
self.import_(['id', 'value'], [['somexmlid', '36']])
|
||||
self.assertEqual([36], values(self.read()))
|
||||
|
||||
self.import_(['id', 'value'], [['somexmlid', '1234567']])
|
||||
self.assertEqual([1234567], values(self.read()))
|
||||
|
||||
def test_wrong_format(self):
|
||||
self.assertEqual(
|
||||
self.import_(['value'], [['50%']]),
|
||||
error(1, u"'50%' does not seem to be an integer for field 'unknown'"))
|
||||
|
||||
|
||||
class test_boolean_field(ImporterCase):
|
||||
model_name = 'export.boolean'
|
||||
|
||||
def test_empty(self):
|
||||
self.assertEqual(
|
||||
self.import_(['value'], []),
|
||||
ok(0))
|
||||
|
||||
def test_exported(self):
|
||||
self.assertEqual(
|
||||
self.import_(['value'], [
|
||||
['False'],
|
||||
['True'],
|
||||
]),
|
||||
ok(2))
|
||||
records = self.read()
|
||||
self.assertEqual([
|
||||
False,
|
||||
True,
|
||||
], values(records))
|
||||
|
||||
def test_falses(self):
|
||||
self.assertEqual(
|
||||
self.import_(['value'], [
|
||||
[u'0'],
|
||||
[u'no'],
|
||||
[u'false'],
|
||||
[u'FALSE'],
|
||||
[u''],
|
||||
]),
|
||||
ok(5))
|
||||
self.assertEqual([
|
||||
False,
|
||||
False,
|
||||
False,
|
||||
False,
|
||||
False,
|
||||
],
|
||||
values(self.read()))
|
||||
|
||||
def test_trues(self):
|
||||
self.assertEqual(
|
||||
self.import_(['value'], [
|
||||
['off'],
|
||||
['None'],
|
||||
['nil'],
|
||||
['()'],
|
||||
['f'],
|
||||
['#f'],
|
||||
# Problem: OpenOffice (and probably excel) output localized booleans
|
||||
['VRAI'],
|
||||
[u'OFF'],
|
||||
[u'是的'],
|
||||
['!&%#${}'],
|
||||
['%(field)s'],
|
||||
]),
|
||||
ok(11))
|
||||
self.assertEqual(
|
||||
[True] * 11,
|
||||
values(self.read()))
|
||||
|
||||
class test_integer_field(ImporterCase):
|
||||
model_name = 'export.integer'
|
||||
|
||||
def test_none(self):
|
||||
self.assertEqual(
|
||||
self.import_(['value'], []),
|
||||
ok(0))
|
||||
|
||||
def test_empty(self):
|
||||
self.assertEqual(
|
||||
self.import_(['value'], [['']]),
|
||||
ok(1))
|
||||
self.assertEqual(
|
||||
[False],
|
||||
values(self.read()))
|
||||
|
||||
def test_zero(self):
|
||||
self.assertEqual(
|
||||
self.import_(['value'], [['0']]),
|
||||
ok(1))
|
||||
self.assertEqual(
|
||||
self.import_(['value'], [['-0']]),
|
||||
ok(1))
|
||||
self.assertEqual([False, False], values(self.read()))
|
||||
|
||||
def test_positives(self):
|
||||
self.assertEqual(
|
||||
self.import_(['value'], [
|
||||
['1'],
|
||||
['42'],
|
||||
[str(2**31-1)],
|
||||
['12345678']
|
||||
]),
|
||||
ok(4))
|
||||
self.assertEqual([
|
||||
1, 42, 2**31-1, 12345678
|
||||
], values(self.read()))
|
||||
|
||||
def test_negatives(self):
|
||||
self.assertEqual(
|
||||
self.import_(['value'], [
|
||||
['-1'],
|
||||
['-42'],
|
||||
[str(-(2**31 - 1))],
|
||||
[str(-(2**31))],
|
||||
['-12345678']
|
||||
]),
|
||||
ok(5))
|
||||
self.assertEqual([
|
||||
-1, -42, -(2**31 - 1), -(2**31), -12345678
|
||||
], values(self.read()))
|
||||
|
||||
@mute_logger('openerp.sql_db')
|
||||
def test_out_of_range(self):
|
||||
self.assertEqual(
|
||||
self.import_(['value'], [[str(2**31)]]),
|
||||
error(1, "integer out of range\n"))
|
||||
# auto-rollbacks if error is in process_liness, but not during
|
||||
# ir.model.data write. Can differentiate because former ends lines
|
||||
# error lines with "!"
|
||||
self.cr.rollback()
|
||||
self.assertEqual(
|
||||
self.import_(['value'], [[str(-2**32)]]),
|
||||
error(1, "integer out of range\n"))
|
||||
|
||||
|
||||
def test_nonsense(self):
|
||||
self.assertEqual(
|
||||
self.import_(['value'], [['zorglub']]),
|
||||
error(1, u"'zorglub' does not seem to be an integer for field 'unknown'"))
|
||||
|
||||
class test_float_field(ImporterCase):
|
||||
model_name = 'export.float'
|
||||
def test_none(self):
|
||||
self.assertEqual(
|
||||
self.import_(['value'], []),
|
||||
ok(0))
|
||||
|
||||
def test_empty(self):
|
||||
self.assertEqual(
|
||||
self.import_(['value'], [['']]),
|
||||
ok(1))
|
||||
self.assertEqual(
|
||||
[False],
|
||||
values(self.read()))
|
||||
|
||||
def test_zero(self):
|
||||
self.assertEqual(
|
||||
self.import_(['value'], [['0']]),
|
||||
ok(1))
|
||||
self.assertEqual(
|
||||
self.import_(['value'], [['-0']]),
|
||||
ok(1))
|
||||
self.assertEqual([False, False], values(self.read()))
|
||||
|
||||
def test_positives(self):
|
||||
self.assertEqual(
|
||||
self.import_(['value'], [
|
||||
['1'],
|
||||
['42'],
|
||||
[str(2**31-1)],
|
||||
['12345678'],
|
||||
[str(2**33)],
|
||||
['0.000001'],
|
||||
]),
|
||||
ok(6))
|
||||
self.assertEqual([
|
||||
1, 42, 2**31-1, 12345678, 2.0**33, .000001
|
||||
], values(self.read()))
|
||||
|
||||
def test_negatives(self):
|
||||
self.assertEqual(
|
||||
self.import_(['value'], [
|
||||
['-1'],
|
||||
['-42'],
|
||||
[str(-2**31 + 1)],
|
||||
[str(-2**31)],
|
||||
['-12345678'],
|
||||
[str(-2**33)],
|
||||
['-0.000001'],
|
||||
]),
|
||||
ok(7))
|
||||
self.assertEqual([
|
||||
-1, -42, -(2**31 - 1), -(2**31), -12345678, -2.0**33, -.000001
|
||||
], values(self.read()))
|
||||
|
||||
def test_nonsense(self):
|
||||
self.assertEqual(
|
||||
self.import_(['value'], [['foobar']]),
|
||||
error(1, u"'foobar' does not seem to be a number for field 'unknown'"))
|
||||
|
||||
class test_string_field(ImporterCase):
|
||||
model_name = 'export.string.bounded'
|
||||
|
||||
def test_empty(self):
|
||||
self.assertEqual(
|
||||
self.import_(['value'], [['']]),
|
||||
ok(1))
|
||||
self.assertEqual([False], values(self.read()))
|
||||
|
||||
def test_imported(self):
|
||||
self.assertEqual(
|
||||
self.import_(['value'], [
|
||||
[u'foobar'],
|
||||
[u'foobarbaz'],
|
||||
[u'Með suð í eyrum við spilum endalaust'],
|
||||
[u"People 'get' types. They use them all the time. Telling "
|
||||
u"someone he can't pound a nail with a banana doesn't much "
|
||||
u"surprise him."]
|
||||
]),
|
||||
ok(4))
|
||||
self.assertEqual([
|
||||
u"foobar",
|
||||
u"foobarbaz",
|
||||
u"Með suð í eyrum ",
|
||||
u"People 'get' typ",
|
||||
], values(self.read()))
|
||||
|
||||
class test_unbound_string_field(ImporterCase):
|
||||
model_name = 'export.string'
|
||||
|
||||
def test_imported(self):
|
||||
self.assertEqual(
|
||||
self.import_(['value'], [
|
||||
[u'í dag viðrar vel til loftárása'],
|
||||
# ackbar.jpg
|
||||
[u"If they ask you about fun, you tell them – fun is a filthy"
|
||||
u" parasite"]
|
||||
]),
|
||||
ok(2))
|
||||
self.assertEqual([
|
||||
u"í dag viðrar vel til loftárása",
|
||||
u"If they ask you about fun, you tell them – fun is a filthy parasite"
|
||||
], values(self.read()))
|
||||
|
||||
class test_text(ImporterCase):
|
||||
model_name = 'export.text'
|
||||
|
||||
def test_empty(self):
|
||||
self.assertEqual(
|
||||
self.import_(['value'], [['']]),
|
||||
ok(1))
|
||||
self.assertEqual([False], values(self.read()))
|
||||
|
||||
def test_imported(self):
|
||||
s = (u"Breiðskífa er notað um útgefna hljómplötu sem inniheldur "
|
||||
u"stúdíóupptökur frá einum flytjanda. Breiðskífur eru oftast "
|
||||
u"milli 25-80 mínútur og er lengd þeirra oft miðuð við 33⅓ "
|
||||
u"snúninga 12 tommu vínylplötur (sem geta verið allt að 30 mín "
|
||||
u"hvor hlið).\n\nBreiðskífur eru stundum tvöfaldar og eru þær þá"
|
||||
u" gefnar út á tveimur geisladiskum eða tveimur vínylplötum.")
|
||||
self.assertEqual(
|
||||
self.import_(['value'], [[s]]),
|
||||
ok(1))
|
||||
self.assertEqual([s], values(self.read()))
|
||||
|
||||
class test_selection(ImporterCase):
|
||||
model_name = 'export.selection'
|
||||
translations_fr = [
|
||||
("Qux", "toto"),
|
||||
("Bar", "titi"),
|
||||
("Foo", "tete"),
|
||||
]
|
||||
|
||||
def test_imported(self):
|
||||
self.assertEqual(
|
||||
self.import_(['value'], [
|
||||
['Qux'],
|
||||
['Bar'],
|
||||
['Foo'],
|
||||
['2'],
|
||||
]),
|
||||
ok(4))
|
||||
self.assertEqual([3, 2, 1, 2], values(self.read()))
|
||||
|
||||
def test_imported_translated(self):
|
||||
self.registry('res.lang').load_lang(self.cr, openerp.SUPERUSER_ID, 'fr_FR')
|
||||
Translations = self.registry('ir.translation')
|
||||
for source, value in self.translations_fr:
|
||||
Translations.create(self.cr, openerp.SUPERUSER_ID, {
|
||||
'name': 'export.selection,value',
|
||||
'lang': 'fr_FR',
|
||||
'type': 'selection',
|
||||
'src': source,
|
||||
'value': value
|
||||
})
|
||||
|
||||
self.assertEqual(
|
||||
self.import_(['value'], [
|
||||
['toto'],
|
||||
['tete'],
|
||||
['titi'],
|
||||
], context={'lang': 'fr_FR'}),
|
||||
ok(3))
|
||||
self.assertEqual([3, 1, 2], values(self.read()))
|
||||
self.assertEqual(
|
||||
self.import_(['value'], [['Foo']], context={'lang': 'fr_FR'}),
|
||||
ok(1))
|
||||
|
||||
def test_invalid(self):
|
||||
self.assertEqual(
|
||||
self.import_(['value'], [['Baz']]),
|
||||
error(1, u"Value 'Baz' not found in selection field 'unknown'"))
|
||||
self.cr.rollback()
|
||||
self.assertEqual(
|
||||
self.import_(['value'], [[42]]),
|
||||
error(1, u"Value '42' not found in selection field 'unknown'"))
|
||||
|
||||
class test_selection_function(ImporterCase):
|
||||
model_name = 'export.selection.function'
|
||||
translations_fr = [
|
||||
("Corge", "toto"),
|
||||
("Grault", "titi"),
|
||||
("Wheee", "tete"),
|
||||
("Moog", "tutu"),
|
||||
]
|
||||
|
||||
def test_imported(self):
|
||||
""" import uses fields_get, so translates import label (may or may not
|
||||
be good news) *and* serializes the selection function to reverse it:
|
||||
import does not actually know that the selection field uses a function
|
||||
"""
|
||||
# NOTE: conflict between a value and a label => ?
|
||||
self.assertEqual(
|
||||
self.import_(['value'], [
|
||||
['3'],
|
||||
["Grault"],
|
||||
]),
|
||||
ok(2))
|
||||
self.assertEqual(
|
||||
[3, 1],
|
||||
values(self.read()))
|
||||
|
||||
def test_translated(self):
|
||||
""" Expects output of selection function returns translated labels
|
||||
"""
|
||||
self.registry('res.lang').load_lang(self.cr, openerp.SUPERUSER_ID, 'fr_FR')
|
||||
Translations = self.registry('ir.translation')
|
||||
for source, value in self.translations_fr:
|
||||
Translations.create(self.cr, openerp.SUPERUSER_ID, {
|
||||
'name': 'export.selection,value',
|
||||
'lang': 'fr_FR',
|
||||
'type': 'selection',
|
||||
'src': source,
|
||||
'value': value
|
||||
})
|
||||
self.assertEqual(
|
||||
self.import_(['value'], [
|
||||
['toto'],
|
||||
['tete'],
|
||||
], context={'lang': 'fr_FR'}),
|
||||
ok(2))
|
||||
self.assertEqual(
|
||||
self.import_(['value'], [['Wheee']], context={'lang': 'fr_FR'}),
|
||||
ok(1))
|
||||
|
||||
class test_m2o(ImporterCase):
|
||||
model_name = 'export.many2one'
|
||||
|
||||
def test_by_name(self):
|
||||
# create integer objects
|
||||
integer_id1 = self.registry('export.integer').create(
|
||||
self.cr, openerp.SUPERUSER_ID, {'value': 42})
|
||||
integer_id2 = self.registry('export.integer').create(
|
||||
self.cr, openerp.SUPERUSER_ID, {'value': 36})
|
||||
# get its name
|
||||
name1 = dict(self.registry('export.integer').name_get(
|
||||
self.cr, openerp.SUPERUSER_ID,[integer_id1]))[integer_id1]
|
||||
name2 = dict(self.registry('export.integer').name_get(
|
||||
self.cr, openerp.SUPERUSER_ID,[integer_id2]))[integer_id2]
|
||||
|
||||
self.assertEqual(
|
||||
self.import_(['value'], [
|
||||
# import by name_get
|
||||
[name1],
|
||||
[name1],
|
||||
[name2],
|
||||
]),
|
||||
ok(3))
|
||||
# correct ids assigned to corresponding records
|
||||
self.assertEqual([
|
||||
(integer_id1, name1),
|
||||
(integer_id1, name1),
|
||||
(integer_id2, name2),],
|
||||
values(self.read()))
|
||||
|
||||
def test_by_xid(self):
|
||||
ExportInteger = self.registry('export.integer')
|
||||
integer_id = ExportInteger.create(
|
||||
self.cr, openerp.SUPERUSER_ID, {'value': 42})
|
||||
xid = self.xid(ExportInteger.browse(
|
||||
self.cr, openerp.SUPERUSER_ID, [integer_id])[0])
|
||||
|
||||
self.assertEqual(
|
||||
self.import_(['value/id'], [[xid]]),
|
||||
ok(1))
|
||||
b = self.browse()
|
||||
self.assertEqual(42, b[0].value.value)
|
||||
|
||||
def test_by_id(self):
|
||||
integer_id = self.registry('export.integer').create(
|
||||
self.cr, openerp.SUPERUSER_ID, {'value': 42})
|
||||
self.assertEqual(
|
||||
self.import_(['value/.id'], [[integer_id]]),
|
||||
ok(1))
|
||||
b = self.browse()
|
||||
self.assertEqual(42, b[0].value.value)
|
||||
|
||||
def test_by_names(self):
|
||||
integer_id1 = self.registry('export.integer').create(
|
||||
self.cr, openerp.SUPERUSER_ID, {'value': 42})
|
||||
integer_id2 = self.registry('export.integer').create(
|
||||
self.cr, openerp.SUPERUSER_ID, {'value': 42})
|
||||
name1 = dict(self.registry('export.integer').name_get(
|
||||
self.cr, openerp.SUPERUSER_ID,[integer_id1]))[integer_id1]
|
||||
name2 = dict(self.registry('export.integer').name_get(
|
||||
self.cr, openerp.SUPERUSER_ID,[integer_id2]))[integer_id2]
|
||||
# names should be the same
|
||||
self.assertEqual(name1, name2)
|
||||
|
||||
self.assertEqual(
|
||||
self.import_(['value'], [[name2]]),
|
||||
ok(1))
|
||||
self.assertEqual([
|
||||
(integer_id1, name1)
|
||||
], values(self.read()))
|
||||
|
||||
def test_fail_by_implicit_id(self):
|
||||
""" Can't implicitly import records by id
|
||||
"""
|
||||
# create integer objects
|
||||
integer_id1 = self.registry('export.integer').create(
|
||||
self.cr, openerp.SUPERUSER_ID, {'value': 42})
|
||||
integer_id2 = self.registry('export.integer').create(
|
||||
self.cr, openerp.SUPERUSER_ID, {'value': 36})
|
||||
|
||||
self.assertEqual(
|
||||
self.import_(['value'], [
|
||||
# import by id, without specifying it
|
||||
[integer_id1],
|
||||
[integer_id2],
|
||||
[integer_id1],
|
||||
]),
|
||||
error(1, u"No matching record found for name '%s' in field 'unknown'" % integer_id1))
|
||||
|
||||
def test_sub_field(self):
|
||||
""" Does not implicitly create the record, does not warn that you can't
|
||||
import m2o subfields (at all)...
|
||||
"""
|
||||
self.assertEqual(
|
||||
self.import_(['value/value'], [['42']]),
|
||||
error(1, u"Can not create Many-To-One records indirectly, import the field separately"))
|
||||
|
||||
def test_fail_noids(self):
|
||||
self.assertEqual(
|
||||
self.import_(['value'], [['nameisnoexist:3']]),
|
||||
error(1, u"No matching record found for name 'nameisnoexist:3' in field 'unknown'"))
|
||||
self.cr.rollback()
|
||||
self.assertEqual(
|
||||
self.import_(['value/id'], [['noxidhere']]),
|
||||
error(1, u"No matching record found for external id 'noxidhere' in field 'unknown'"))
|
||||
self.cr.rollback()
|
||||
self.assertEqual(
|
||||
self.import_(['value/.id'], [[66]]),
|
||||
error(1, u"No matching record found for database id '66' in field 'unknown'"))
|
||||
|
||||
class test_m2m(ImporterCase):
|
||||
model_name = 'export.many2many'
|
||||
|
||||
# apparently, one and only thing which works is a
|
||||
# csv_internal_sep-separated list of ids, xids, or names (depending if
|
||||
# m2m/.id, m2m/id or m2m[/anythingelse]
|
||||
def test_ids(self):
|
||||
id1 = self.registry('export.many2many.other').create(
|
||||
self.cr, openerp.SUPERUSER_ID, {'value': 3, 'str': 'record0'})
|
||||
id2 = self.registry('export.many2many.other').create(
|
||||
self.cr, openerp.SUPERUSER_ID, {'value': 44, 'str': 'record1'})
|
||||
id3 = self.registry('export.many2many.other').create(
|
||||
self.cr, openerp.SUPERUSER_ID, {'value': 84, 'str': 'record2'})
|
||||
id4 = self.registry('export.many2many.other').create(
|
||||
self.cr, openerp.SUPERUSER_ID, {'value': 9, 'str': 'record3'})
|
||||
id5 = self.registry('export.many2many.other').create(
|
||||
self.cr, openerp.SUPERUSER_ID, {'value': 99, 'str': 'record4'})
|
||||
|
||||
self.assertEqual(
|
||||
self.import_(['value/.id'], [
|
||||
['%d,%d' % (id1, id2)],
|
||||
['%d,%d,%d' % (id1, id3, id4)],
|
||||
['%d,%d,%d' % (id1, id2, id3)],
|
||||
['%d' % id5]
|
||||
]),
|
||||
ok(4))
|
||||
ids = lambda records: [record.id for record in records]
|
||||
|
||||
b = self.browse()
|
||||
self.assertEqual(ids(b[0].value), [id1, id2])
|
||||
self.assertEqual(values(b[0].value), [3, 44])
|
||||
|
||||
self.assertEqual(ids(b[2].value), [id1, id2, id3])
|
||||
self.assertEqual(values(b[2].value), [3, 44, 84])
|
||||
|
||||
def test_noids(self):
|
||||
self.assertEqual(
|
||||
self.import_(['value/.id'], [['42']]),
|
||||
error(1, u"No matching record found for database id '42' in field 'unknown'"))
|
||||
|
||||
def test_xids(self):
|
||||
M2O_o = self.registry('export.many2many.other')
|
||||
id1 = M2O_o.create(self.cr, openerp.SUPERUSER_ID, {'value': 3, 'str': 'record0'})
|
||||
id2 = M2O_o.create(self.cr, openerp.SUPERUSER_ID, {'value': 44, 'str': 'record1'})
|
||||
id3 = M2O_o.create(self.cr, openerp.SUPERUSER_ID, {'value': 84, 'str': 'record2'})
|
||||
id4 = M2O_o.create(self.cr, openerp.SUPERUSER_ID, {'value': 9, 'str': 'record3'})
|
||||
records = M2O_o.browse(self.cr, openerp.SUPERUSER_ID, [id1, id2, id3, id4])
|
||||
|
||||
self.assertEqual(
|
||||
self.import_(['value/id'], [
|
||||
['%s,%s' % (self.xid(records[0]), self.xid(records[1]))],
|
||||
['%s' % self.xid(records[3])],
|
||||
['%s,%s' % (self.xid(records[2]), self.xid(records[1]))],
|
||||
]),
|
||||
ok(3))
|
||||
|
||||
b = self.browse()
|
||||
self.assertEqual(values(b[0].value), [3, 44])
|
||||
self.assertEqual(values(b[2].value), [44, 84])
|
||||
def test_noxids(self):
|
||||
self.assertEqual(
|
||||
self.import_(['value/id'], [['noxidforthat']]),
|
||||
error(1, u"No matching record found for external id 'noxidforthat' in field 'unknown'"))
|
||||
|
||||
def test_names(self):
|
||||
M2O_o = self.registry('export.many2many.other')
|
||||
id1 = M2O_o.create(self.cr, openerp.SUPERUSER_ID, {'value': 3, 'str': 'record0'})
|
||||
id2 = M2O_o.create(self.cr, openerp.SUPERUSER_ID, {'value': 44, 'str': 'record1'})
|
||||
id3 = M2O_o.create(self.cr, openerp.SUPERUSER_ID, {'value': 84, 'str': 'record2'})
|
||||
id4 = M2O_o.create(self.cr, openerp.SUPERUSER_ID, {'value': 9, 'str': 'record3'})
|
||||
records = M2O_o.browse(self.cr, openerp.SUPERUSER_ID, [id1, id2, id3, id4])
|
||||
|
||||
name = lambda record: record.name_get()[0][1]
|
||||
|
||||
self.assertEqual(
|
||||
self.import_(['value'], [
|
||||
['%s,%s' % (name(records[1]), name(records[2]))],
|
||||
['%s,%s,%s' % (name(records[0]), name(records[1]), name(records[2]))],
|
||||
['%s,%s' % (name(records[0]), name(records[3]))],
|
||||
]),
|
||||
ok(3))
|
||||
|
||||
b = self.browse()
|
||||
self.assertEqual(values(b[1].value), [3, 44, 84])
|
||||
self.assertEqual(values(b[2].value), [3, 9])
|
||||
|
||||
def test_nonames(self):
|
||||
self.assertEqual(
|
||||
self.import_(['value'], [['wherethem2mhavenonames']]),
|
||||
error(1, u"No matching record found for name 'wherethem2mhavenonames' in field 'unknown'"))
|
||||
|
||||
def test_import_to_existing(self):
|
||||
M2O_o = self.registry('export.many2many.other')
|
||||
id1 = M2O_o.create(self.cr, openerp.SUPERUSER_ID, {'value': 3, 'str': 'record0'})
|
||||
id2 = M2O_o.create(self.cr, openerp.SUPERUSER_ID, {'value': 44, 'str': 'record1'})
|
||||
id3 = M2O_o.create(self.cr, openerp.SUPERUSER_ID, {'value': 84, 'str': 'record2'})
|
||||
id4 = M2O_o.create(self.cr, openerp.SUPERUSER_ID, {'value': 9, 'str': 'record3'})
|
||||
|
||||
xid = 'myxid'
|
||||
self.assertEqual(
|
||||
self.import_(['id', 'value/.id'], [[xid, '%d,%d' % (id1, id2)]]),
|
||||
ok(1))
|
||||
self.assertEqual(
|
||||
self.import_(['id', 'value/.id'], [[xid, '%d,%d' % (id3, id4)]]),
|
||||
ok(1))
|
||||
|
||||
b = self.browse()
|
||||
self.assertEqual(len(b), 1)
|
||||
# TODO: replacement of existing m2m values is correct?
|
||||
self.assertEqual(values(b[0].value), [84, 9])
|
||||
|
||||
class test_o2m(ImporterCase):
|
||||
model_name = 'export.one2many'
|
||||
|
||||
def test_name_get(self):
|
||||
s = u'Java is a DSL for taking large XML files and converting them to' \
|
||||
u' stack traces'
|
||||
self.assertEqual(
|
||||
self.import_(
|
||||
['const', 'value'],
|
||||
[['5', s]]),
|
||||
error(1, u"No matching record found for name '%s' in field 'unknown'" % s))
|
||||
|
||||
def test_single(self):
|
||||
self.assertEqual(
|
||||
self.import_(['const', 'value/value'], [
|
||||
['5', '63']
|
||||
]),
|
||||
ok(1))
|
||||
|
||||
(b,) = self.browse()
|
||||
self.assertEqual(b.const, 5)
|
||||
self.assertEqual(values(b.value), [63])
|
||||
|
||||
def test_multicore(self):
|
||||
self.assertEqual(
|
||||
self.import_(['const', 'value/value'], [
|
||||
['5', '63'],
|
||||
['6', '64'],
|
||||
]),
|
||||
ok(2))
|
||||
|
||||
b1, b2 = self.browse()
|
||||
self.assertEqual(b1.const, 5)
|
||||
self.assertEqual(values(b1.value), [63])
|
||||
self.assertEqual(b2.const, 6)
|
||||
self.assertEqual(values(b2.value), [64])
|
||||
|
||||
def test_multisub(self):
|
||||
self.assertEqual(
|
||||
self.import_(['const', 'value/value'], [
|
||||
['5', '63'],
|
||||
['', '64'],
|
||||
['', '65'],
|
||||
['', '66'],
|
||||
]),
|
||||
ok(4))
|
||||
|
||||
(b,) = self.browse()
|
||||
self.assertEqual(values(b.value), [63, 64, 65, 66])
|
||||
|
||||
def test_multi_subfields(self):
|
||||
self.assertEqual(
|
||||
self.import_(['value/str', 'const', 'value/value'], [
|
||||
['this', '5', '63'],
|
||||
['is', '', '64'],
|
||||
['the', '', '65'],
|
||||
['rhythm', '', '66'],
|
||||
]),
|
||||
ok(4))
|
||||
|
||||
(b,) = self.browse()
|
||||
self.assertEqual(values(b.value), [63, 64, 65, 66])
|
||||
self.assertEqual(
|
||||
values(b.value, 'str'),
|
||||
'this is the rhythm'.split())
|
||||
|
||||
def test_link_inline(self):
|
||||
id1 = self.registry('export.one2many.child').create(self.cr, openerp.SUPERUSER_ID, {
|
||||
'str': 'Bf', 'value': 109
|
||||
})
|
||||
id2 = self.registry('export.one2many.child').create(self.cr, openerp.SUPERUSER_ID, {
|
||||
'str': 'Me', 'value': 262
|
||||
})
|
||||
|
||||
try:
|
||||
self.import_(['const', 'value/.id'], [
|
||||
['42', '%d,%d' % (id1, id2)]
|
||||
])
|
||||
except ValueError, e:
|
||||
# should be Exception(Database ID doesn't exist: export.one2many.child : $id1,$id2)
|
||||
self.assertIs(type(e), ValueError)
|
||||
self.assertEqual(
|
||||
e.args[0],
|
||||
"invalid literal for int() with base 10: '%d,%d'" % (id1, id2))
|
||||
|
||||
def test_link(self):
|
||||
id1 = self.registry('export.one2many.child').create(self.cr, openerp.SUPERUSER_ID, {
|
||||
'str': 'Bf', 'value': 109
|
||||
})
|
||||
id2 = self.registry('export.one2many.child').create(self.cr, openerp.SUPERUSER_ID, {
|
||||
'str': 'Me', 'value': 262
|
||||
})
|
||||
|
||||
self.assertEqual(
|
||||
self.import_(['const', 'value/.id'], [
|
||||
['42', str(id1)],
|
||||
['', str(id2)],
|
||||
]),
|
||||
ok(2))
|
||||
|
||||
[b] = self.browse()
|
||||
self.assertEqual(b.const, 42)
|
||||
# automatically forces link between core record and o2ms
|
||||
self.assertEqual(values(b.value), [109, 262])
|
||||
self.assertEqual(values(b.value, field='parent_id'), [b, b])
|
||||
|
||||
def test_link_2(self):
|
||||
O2M_c = self.registry('export.one2many.child')
|
||||
id1 = O2M_c.create(self.cr, openerp.SUPERUSER_ID, {
|
||||
'str': 'Bf', 'value': 109
|
||||
})
|
||||
id2 = O2M_c.create(self.cr, openerp.SUPERUSER_ID, {
|
||||
'str': 'Me', 'value': 262
|
||||
})
|
||||
|
||||
self.assertEqual(
|
||||
self.import_(['const', 'value/.id', 'value/value'], [
|
||||
['42', str(id1), '1'],
|
||||
['', str(id2), '2'],
|
||||
]),
|
||||
ok(2))
|
||||
|
||||
[b] = self.browse()
|
||||
self.assertEqual(b.const, 42)
|
||||
self.assertEqual(values(b.value), [1, 2])
|
||||
self.assertEqual(values(b.value, field='parent_id'), [b, b])
|
||||
|
||||
class test_o2m_multiple(ImporterCase):
|
||||
model_name = 'export.one2many.multiple'
|
||||
|
||||
def test_multi_mixed(self):
|
||||
self.assertEqual(
|
||||
self.import_(['const', 'child1/value', 'child2/value'], [
|
||||
['5', '11', '21'],
|
||||
['', '12', '22'],
|
||||
['', '13', '23'],
|
||||
['', '14', ''],
|
||||
]),
|
||||
ok(4))
|
||||
|
||||
[b] = self.browse()
|
||||
self.assertEqual(values(b.child1), [11, 12, 13, 14])
|
||||
self.assertEqual(values(b.child2), [21, 22, 23])
|
||||
|
||||
def test_multi(self):
|
||||
self.assertEqual(
|
||||
self.import_(['const', 'child1/value', 'child2/value'], [
|
||||
['5', '11', '21'],
|
||||
['', '12', ''],
|
||||
['', '13', ''],
|
||||
['', '14', ''],
|
||||
['', '', '22'],
|
||||
['', '', '23'],
|
||||
]),
|
||||
ok(6))
|
||||
|
||||
[b] = self.browse()
|
||||
self.assertEqual(values(b.child1), [11, 12, 13, 14])
|
||||
self.assertEqual(values(b.child2), [21, 22, 23])
|
||||
|
||||
def test_multi_fullsplit(self):
|
||||
self.assertEqual(
|
||||
self.import_(['const', 'child1/value', 'child2/value'], [
|
||||
['5', '11', ''],
|
||||
['', '12', ''],
|
||||
['', '13', ''],
|
||||
['', '14', ''],
|
||||
['', '', '21'],
|
||||
['', '', '22'],
|
||||
['', '', '23'],
|
||||
]),
|
||||
ok(7))
|
||||
|
||||
[b] = self.browse()
|
||||
self.assertEqual(b.const, 5)
|
||||
self.assertEqual(values(b.child1), [11, 12, 13, 14])
|
||||
self.assertEqual(values(b.child2), [21, 22, 23])
|
||||
|
||||
# function, related, reference: written to db as-is...
|
||||
# => function uses @type for value coercion/conversion
|
||||
+8
-86
@@ -991,89 +991,6 @@ class BaseModel(object):
|
||||
self = self.with_context(export_raw_data=True)
|
||||
return {'datas': self.__export_rows(fields_to_export)}
|
||||
|
||||
def import_data(self, cr, uid, fields, datas, mode='init', current_module='', noupdate=False, context=None, filename=None):
|
||||
"""
|
||||
.. deprecated:: 7.0
|
||||
Use :meth:`~load` instead
|
||||
|
||||
Import given data in given module
|
||||
|
||||
This method is used when importing data via client menu.
|
||||
|
||||
Example of fields to import for a sale.order::
|
||||
|
||||
.id, (=database_id)
|
||||
partner_id, (=name_search)
|
||||
order_line/.id, (=database_id)
|
||||
order_line/name,
|
||||
order_line/product_id/id, (=xml id)
|
||||
order_line/price_unit,
|
||||
order_line/product_uom_qty,
|
||||
order_line/product_uom/id (=xml_id)
|
||||
|
||||
This method returns a 4-tuple with the following structure::
|
||||
|
||||
(return_code, errored_resource, error_message, unused)
|
||||
|
||||
* The first item is a return code, it is ``-1`` in case of
|
||||
import error, or the last imported row number in case of success
|
||||
* The second item contains the record data dict that failed to import
|
||||
in case of error, otherwise it's 0
|
||||
* The third item contains an error message string in case of error,
|
||||
otherwise it's 0
|
||||
* The last item is currently unused, with no specific semantics
|
||||
|
||||
:param fields: list of fields to import
|
||||
:param datas: data to import
|
||||
:param mode: 'init' or 'update' for record creation
|
||||
:param current_module: module name
|
||||
:param noupdate: flag for record creation
|
||||
:param filename: optional file to store partial import state for recovery
|
||||
:returns: 4-tuple in the form (return_code, errored_resource, error_message, unused)
|
||||
:rtype: (int, dict or 0, str or 0, str or 0)
|
||||
"""
|
||||
context = dict(context) if context is not None else {}
|
||||
context['_import_current_module'] = current_module
|
||||
|
||||
fields = map(fix_import_export_id_paths, fields)
|
||||
ir_model_data_obj = self.pool.get('ir.model.data')
|
||||
|
||||
def log(m):
|
||||
if m['type'] == 'error':
|
||||
raise Exception(m['message'])
|
||||
|
||||
if config.get('import_partial') and filename:
|
||||
with open(config.get('import_partial'), 'rb') as partial_import_file:
|
||||
data = pickle.load(partial_import_file)
|
||||
position = data.get(filename, 0)
|
||||
|
||||
position = 0
|
||||
try:
|
||||
for res_id, xml_id, res, info in self._convert_records(cr, uid,
|
||||
self._extract_records(cr, uid, fields, datas,
|
||||
context=context, log=log),
|
||||
context=context, log=log):
|
||||
ir_model_data_obj._update(cr, uid, self._name,
|
||||
current_module, res, mode=mode, xml_id=xml_id,
|
||||
noupdate=noupdate, res_id=res_id, context=context)
|
||||
position = info.get('rows', {}).get('to', 0) + 1
|
||||
if config.get('import_partial') and filename and (not (position%100)):
|
||||
with open(config.get('import_partial'), 'rb') as partial_import:
|
||||
data = pickle.load(partial_import)
|
||||
data[filename] = position
|
||||
with open(config.get('import_partial'), 'wb') as partial_import:
|
||||
pickle.dump(data, partial_import)
|
||||
if context.get('defer_parent_store_computation'):
|
||||
self._parent_store_compute(cr)
|
||||
cr.commit()
|
||||
except Exception, e:
|
||||
cr.rollback()
|
||||
return -1, {}, 'Line %d : %s' % (position + 1, tools.ustr(e)), ''
|
||||
|
||||
if context.get('defer_parent_store_computation'):
|
||||
self._parent_store_compute(cr)
|
||||
return position, 0, 0, 0
|
||||
|
||||
def load(self, cr, uid, fields, data, context=None):
|
||||
"""
|
||||
Attempts to load the data matrix, and returns a list of ids (or
|
||||
@@ -1099,9 +1016,14 @@ class BaseModel(object):
|
||||
|
||||
fg = self.fields_get(cr, uid, context=context)
|
||||
|
||||
mode = 'init'
|
||||
current_module = ''
|
||||
noupdate = False
|
||||
# determine values of mode, current_module and noupdate
|
||||
context = context or {}
|
||||
mode = context.get('mode', 'init')
|
||||
current_module = context.get('module', '')
|
||||
noupdate = context.get('noupdate', False)
|
||||
|
||||
# add current module in context for the conversion of xml ids
|
||||
context = dict(context, _import_current_module=current_module)
|
||||
|
||||
ids = []
|
||||
for id, xid, record, info in self._convert_records(cr, uid,
|
||||
|
||||
@@ -883,19 +883,6 @@ def convert_csv_import(cr, module, fname, csvcontent, idref=None, mode='init',
|
||||
input = cStringIO.StringIO(csvcontent) #FIXME
|
||||
reader = csv.reader(input, quotechar='"', delimiter=',')
|
||||
fields = reader.next()
|
||||
fname_partial = ""
|
||||
if config.get('import_partial'):
|
||||
fname_partial = module + '/'+ fname
|
||||
if not os.path.isfile(config.get('import_partial')):
|
||||
pickle.dump({}, file(config.get('import_partial'),'w+'))
|
||||
else:
|
||||
data = pickle.load(file(config.get('import_partial')))
|
||||
if fname_partial in data:
|
||||
if not data[fname_partial]:
|
||||
return
|
||||
else:
|
||||
for i in range(data[fname_partial]):
|
||||
reader.next()
|
||||
|
||||
if not (mode == 'init' or 'id' in fields):
|
||||
_logger.error("Import specification does not contain 'id' and we are in init mode, Cannot continue.")
|
||||
@@ -912,15 +899,16 @@ def convert_csv_import(cr, module, fname, csvcontent, idref=None, mode='init',
|
||||
_logger.error("Cannot import the line: %s", line)
|
||||
|
||||
registry = openerp.registry(cr.dbname)
|
||||
result, rows, warning_msg, dummy = registry[model].import_data(cr, uid, fields, datas,mode, module, noupdate, filename=fname_partial)
|
||||
if result < 0:
|
||||
context = {
|
||||
'mode': mode,
|
||||
'module': module,
|
||||
'noupdate': noupdate,
|
||||
}
|
||||
result = registry[model].load(cr, uid, fields, datas, context=context)
|
||||
if any(msg['type'] == 'error' for msg in result['messages']):
|
||||
# Report failed import and abort module install
|
||||
warning_msg = "\n".join(msg['message'] for msg in result['messages'])
|
||||
raise Exception(_('Module loading %s failed: file %s could not be processed:\n %s') % (module, fname, warning_msg))
|
||||
if config.get('import_partial'):
|
||||
data = pickle.load(file(config.get('import_partial')))
|
||||
data[fname_partial] = 0
|
||||
pickle.dump(data, file(config.get('import_partial'),'wb'))
|
||||
cr.commit()
|
||||
|
||||
def convert_xml_import(cr, module, xmlfile, idref=None, mode='init', noupdate=False, report=None):
|
||||
doc = etree.parse(xmlfile)
|
||||
|
||||
Reference in New Issue
Block a user