Add context to load_csv function

anthem/lyrics/loaders.py

@@ -1,28 +1,28 @@
 # -*- coding: utf-8 -*-
-# Copyright 2016 Camptocamp SA
-# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html)
+# Copyright 2016-2017 Camptocamp SA
+# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl)
 
 import codecs
 import csv
 
 from ..exceptions import AnthemError
 
 
-def load_csv(ctx, model_name, path, dialect='excel', **fmtparams):
+def load_csv(ctx, model, path, dialect='excel', **fmtparams):
     """ Load a CSV from a filename
 
     Usage example::
 
       from pkg_resources import Requirement, resource_string
 
       req = Requirement.parse('my-project')
-      load_csv(ctx, 'res.users',
+      load_csv(ctx, ctx.env['res.users'],
                resource_string(req, 'data/users.csv'),
                delimiter=',')
 
     """
     with open(path, 'rb') as data:
-        load_csv_stream(ctx, model_name, data, dialect=dialect, **fmtparams)
+        load_csv_stream(ctx, model, data, dialect=dialect, **fmtparams)
 
 
 def csv_unireader(f, encoding="utf-8", **fmtparams):
@@ -33,7 +33,7 @@ def csv_unireader(f, encoding="utf-8", **fmtparams):
         yield [e.decode("utf-8") for e in row]
 
 
-def load_csv_stream(ctx, model_name, data, dialect='excel', encoding='utf-8',
+def load_csv_stream(ctx, model, data, dialect='excel', encoding='utf-8',
                     **fmtparams):
     """ Load a CSV from a stream
 
@@ -42,7 +42,7 @@ def load_csv_stream(ctx, model_name, data, dialect='excel', encoding='utf-8',
       from pkg_resources import Requirement, resource_stream
 
       req = Requirement.parse('my-project')
-      load_csv_stream(ctx, 'res.users',
+      load_csv_stream(ctx, ctx.env['res.users'],
                       resource_stream(req, 'data/users.csv'),
                       delimiter=',')
 
@@ -51,16 +51,18 @@ def load_csv_stream(ctx, model_name, data, dialect='excel', encoding='utf-8',
     head = data.next()
     values = list(data)
     if values:
-        result = ctx.env[model_name].load(head, values)
+        if isinstance(model, basestring):
+            model = ctx.env[model]
+        result = model.load(head, values)
         ids = result['ids']
         if not ids:
             messages = u'\n'.join(
                 u'- %s' % msg for msg in result['messages']
             )
             ctx.log_line(u"Failed to load CSV "
                          u"in '%s'. Details:\n%s" %
-                         (model_name, messages))
+                         (model._name, messages))
             raise AnthemError(u'Could not import CSV. See the logs')
         else:
             ctx.log_line(u"Imported %d records in '%s'" %
-                         (len(ids), model_name))
+                         (len(ids), model._name))

tasks.py

@@ -38,6 +38,7 @@ def tests_prepare(ctx, version):
         ctx.run('tar xfz odoo.tar.gz')
         ctx.run('rm -f odoo.tar.gz')
     print('Installing odoo')
+    ctx.run('pip install -r odoo-{}/requirements.txt -q'.format(version))
     ctx.run('pip install -e odoo-{} -q'.format(version))
 
 

tests/test_lyrics_loaders.py

@@ -19,7 +19,35 @@
                )
 
 
-def test_load_csv_stream():
+def test_load_csv_stream_model():
+    csv_stream = StringIO()
+    csv_stream.write(csv_partner)
+    csv_stream.seek(0)
+    with anthem.cli.Context(None, anthem.cli.Options(test_mode=True)) as ctx:
+        load_csv_stream(ctx, ctx.env['res.partner'], csv_stream, delimiter=',')
+        partner1 = ctx.env.ref('__test__.partner1', raise_if_not_found=False)
+        assert partner1
+        assert partner1.name == 'Partner 1'
+        partner2 = ctx.env.ref('__test__.partner2', raise_if_not_found=False)
+        assert partner2
+        assert partner2.name == 'Partner 2'
+
+
+def test_load_csv_file_model(tmpdir):
+    csvfile = tmpdir.mkdir("files").join("res.partner.csv")
+    csvfile.write(csv_partner)
+    with anthem.cli.Context(None, anthem.cli.Options(test_mode=True)) as ctx:
+        load_csv(ctx, ctx.env['res.partner'], csvfile.strpath, delimiter=',')
+        partner1 = ctx.env.ref('__test__.partner1', raise_if_not_found=False)
+        assert partner1
+        assert partner1.name == 'Partner 1'
+        partner2 = ctx.env.ref('__test__.partner2', raise_if_not_found=False)
+        assert partner2
+        assert partner2.name == 'Partner 2'
+
+
+def test_load_csv_stream_model_string():
+    """ Pass string instead of model to load_csv_stream """
     csv_stream = StringIO()
     csv_stream.write(csv_partner)
     csv_stream.seek(0)
@@ -33,7 +61,7 @@ def test_load_csv_stream():
         assert partner2.name == 'Partner 2'
 
 
-def test_load_csv_file(tmpdir):
+def test_load_csv_file_model_string(tmpdir):
     csvfile = tmpdir.mkdir("files").join("res.partner.csv")
     csvfile.write(csv_partner)
     with anthem.cli.Context(None, anthem.cli.Options(test_mode=True)) as ctx: