Produits Entr'ouvert » Authentic 2

    python-tablib,

    python-chardet,

    python-attr

          'chardet',

          'attrs',

# authentic2 - versatile identity manager

# Copyright (C) 2010-2019 Entr'ouvert

#

# This program is free software: you can redistribute it and/or modify it

# under the terms of the GNU Affero General Public License as published

# by the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU Affero General Public License for more details.

#

# You should have received a copy of the GNU Affero General Public License

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

from __future__ import unicode_literals

import csv

import io

from chardet.universaldetector import UniversalDetector

import attr

from django.core.exceptions import FieldDoesNotExist

from django.db.transaction import atomic

from django.utils import six

from django.utils.translation import ugettext as _

from authentic2 import app_settings

from authentic2.a2_rbac.utils import get_default_ou

from authentic2.custom_user.models import User

from authentic2.forms.profile import modelform_factory, BaseUserForm

from authentic2.models import Attribute, AttributeValue

class UTF8Recoder(object):

    def __init__(self, fd):

        self.fd = fd

    def __iter__(self):

        return self

    def next(self):

        return self.fd.next().encode('utf-8')

class UnicodeReader(object):

    def __init__(self, fd, dialect='excel', **kwargs):

        self.reader = csv.reader(UTF8Recoder(fd), dialect=dialect, **kwargs)

    def next(self):

        row = self.reader.next()

        return [s.decode('utf-8') for s in row]

    def __iter__(self):

        return self

class CsvImporter(object):

    rows = None

    error = None

    error_description = None

    encoding = None

    def run(self, fd_or_str, encoding):

        if isinstance(fd_or_str, six.binary_type):

            input_fd = io.BytesIO(fd_or_str)

        elif isinstance(fd_or_str, six.text_type):

            input_fd = io.StringIO(fd_or_str)

        elif not hasattr(fd_or_str, 'read1'):

            try:

                input_fd = io.open(fd_or_str.fileno(), closefd=False)

            except Exception:

                try:

                    fd_or_str.seek(0)

                except Exception:

                    pass

                content = fd_or_str.read()

                if isinstance(content, six.text_type):

                    input_fd = io.StringIO(content)

                else:

                    input_fd = io.BytesIO(content)

        else:

            input_fd = fd_or_str

        assert hasattr(input_fd, 'read'), 'fd_or_str is not a string or a file object'

        def detect_encoding(input_fd, encoding):

            # detect StringIO

            if hasattr(input_fd, 'line_buffering'):

                return input_fd

            if encoding == 'detect':

                detector = UniversalDetector()

                try:

                    for line in input_fd:

                        detector.feed(line)

                        if detector.done:

                            break

                    else:

                        self.error = Error('cannot-detect-encoding', _('Cannot detect encoding'))

                        return None

                    detector.close()

                    encoding = detector.result['encoding']

                finally:

                    input_fd.seek(0)

            if not hasattr(input_fd, 'readable'):

                input_fd = io.open(input_fd.fileno(), 'rb', closefd=False)

            return io.TextIOWrapper(input_fd, encoding=encoding)

        def parse_csv():

            try:

                dialect = csv.Sniffer().sniff(input_fd.read().encode('utf-8'))

            except csv.Error as e:

                self.error = Error('unknown-csv-dialect', _('Unknown CSV dialect: %s') % e)

                return False

            finally:

                input_fd.seek(0)

            if not dialect:

                self.error = Error('unknown-csv-dialect', _('Unknown CSV dialect'))

                return False

            reader = UnicodeReader(input_fd, dialect)

            self.rows = list(reader)

            return True

        input_fd = detect_encoding(input_fd, encoding)

        if input_fd is None:

            return False

        return parse_csv()

@attr.s

class CsvHeader(object):

    column = attr.ib()

    name = attr.ib(default='')

    field = attr.ib(default=False, converter=bool)

    attribute = attr.ib(default=False, converter=bool)

    create = attr.ib(default=True, metadata={'flag': True})

    update = attr.ib(default=True, metadata={'flag': True})

    key = attr.ib(default=False, metadata={'flag': True})

    unique = attr.ib(default=False, metadata={'flag': True})

    globally_unique = attr.ib(default=False, metadata={'flag': True})

    verified = attr.ib(default=False, metadata={'flag': True})

    @property

    def flags(self):

        flags = []

        for attribute in attr.fields(self.__class__):

            if attribute.metadata.get('flag'):

                if getattr(self, attribute.name):

                    flags.append(attribute.name)

                else:

                    flags.append('no-' + attribute.name.replace('_', '-'))

        return flags

@attr.s

class Error(object):

    code = attr.ib()

    description = attr.ib(default='', cmp=False)

    def __eq__(self, other):

        if isinstance(other, Error):

            return self.as_error.__eq__(other)

        return super(LineError, self).__eq__(other)

@attr.s(cmp=False)

class LineError(Error):

    line = attr.ib(default=0)

    column = attr.ib(default=0)

    @classmethod

    def from_error(cls, error):

        return cls(**attr.asdict(error))

    def as_error(self):

        return Error(self.code, self.description)

    def __eq__(self, other):

        if isinstance(other, Error):

            return self.as_error() == other

        return (self.code, self.line, self.column) == (other.code, other.line, other.column)

class ImportUserForm(BaseUserForm):

    def clean(self):

        super(BaseUserForm, self).clean()

        self._validate_unique = False

@attr.s

class CsvRow(object):

    line = attr.ib()

    cells = attr.ib(default=[])

    errors = attr.ib(default=[])

    is_valid = attr.ib(default=True)

    action = attr.ib(default=None)

    def __getitem__(self, header):

        for cell in self.cells:

            if cell.header == header or cell.header.name == header:

                return cell

        raise KeyError(header.name)

    def __iter__(self):

        return iter(self.cells)

@attr.s

class CsvCell(object):

    line = attr.ib()

    header = attr.ib()

    value = attr.ib(default=None)

    missing = attr.ib(default=False)

    errors = attr.ib(default=[])

    action = attr.ib(default=None)

    @property

    def column(self):

        return self.header.column

class Simulate(Exception):

    pass

class UserCsvImporter(object):

    csv_importer = None

    errors = None

    headers = None

    headers_by_name = None

    rows = None

    has_errors = False

    ou = None

    updated = 0

    created = 0

    def add_error(self, line_error):

        if not hasattr(line_error, 'line'):

            line_error = LineError.from_error(line_error)

        self.errors.append(line_error)

    def run(self, fd_or_str, encoding, ou=None, simulate=False):

        self.ou = ou or get_default_ou()

        self.errors = []

        self.csv_importer = CsvImporter()

        def parse_csv():

            if not self.csv_importer.run(fd_or_str, encoding):

                self.add_error(self.csv_importer.error)

        def do_import():

            try:

                with atomic():

                    for row in self.rows:

                        self.do_import_row(row)

                    if simulate:

                        raise Simulate

            except Simulate:

                pass

        for action in [

                parse_csv,

                self.parse_header_row,

                self.parse_rows,

                do_import]:

            action()

            if self.errors:

                break

        self.has_errors = self.has_errors or bool(self.errors)

        return not bool(self.errors)

    def parse_header_row(self):

        self.headers = []

        self.headers_by_name = {}

        try:

            header_row = self.csv_importer.rows[0]

        except IndexError:

            self.add_error(Error('no-header-row', _('Missing header row')))

            return

        for i, head in enumerate(header_row):

            self.parse_header(head, column=i + 1)

        if not self.headers:

            self.add_error(Error('empty-header-row', _('Empty header row')))

            return

        key_counts = sum(1 for header in self.headers if header.key)

        if not key_counts:

            self.add_error(Error('missing-key-column', _('Missing key column')))

        if key_counts > 1:

            self.add_error(Error('too-many-key-columns', _('Too many key columns')))

    def parse_header(self, head, column):

        splitted = head.split()

        try:

            header = CsvHeader(column, splitted[0])

            self.headers_by_name[header.name] = header

        except IndexError:

            header = CsvHeader(column)

        else:

            try:

                if header.name in ['email', 'first_name', 'last_name', 'username']:

                    field = User._meta.get_field(header.name)  # NOQA: F841

                    header.field = True

                    if header.name == 'email' and self.email_is_unique:

                        header.unique = True

                        if app_settings.A2_EMAIL_IS_UNIQUE:

                            header.globally_unique = True

                    if header.name == 'username' and self.username_is_unique:

                        header.unique = True

                        if app_settings.A2_USERNAME_IS_UNIQUE:

                            header.globally_unique = True

            except FieldDoesNotExist:

                pass

            if not header.field:

                try:

                    attribute = Attribute.objects.get(name=header.name)  # NOQA: F841

                    header.attribute = True

                except Attribute.DoesNotExist:

                    pass

        self.headers.append(header)

        if not (header.field or header.attribute):

            self.add_error(LineError('unknown-or-missing-attribute',

                                     _('unknown or missing attribute "%s"') % head,

                                     line=1, column=column))

            return

        for flag in splitted[1:]:

            value = True

            if flag.startswith('no-'):

                value = False

                flag = flag[3:]

            flag = flag.replace('-', '_')

            try:

                if not getattr(attr.fields(CsvHeader), flag).metadata['flag']:

                    raise TypeError

                setattr(header, flag, value)

            except (AttributeError, TypeError, KeyError):

                self.add_error(LineError('unknown-flag', _('unknown flag "%s"'), line=1, column=column))

    def parse_rows(self):

        form_class = modelform_factory(User, fields=self.headers_by_name.keys(), form=ImportUserForm)

        rows = self.rows = []

        for i, row in enumerate(self.csv_importer.rows[1:]):

            csv_row = self.parse_row(form_class, row, line=i + 2)

            self.has_errors = self.has_errors or not(csv_row.is_valid)

            rows.append(csv_row)

    def parse_row(self, form_class, row, line):

        data = {}

        for header in self.headers:

            try:

                data[header.name] = row[header.column - 1]

            except IndexError:

                pass

        form = form_class(data=data)

        form.is_valid()

        def get_form_errors(form, name):

            return [Error('data-error', six.text_type(value)) for value in form.errors.get(name, [])]

        cells = [

            CsvCell(

                line=line,

                header=header,

                value=data.get(header.name),

                missing=header.name not in data,

                errors=get_form_errors(form, header.name))

            for header in self.headers]

        cell_errors = any(bool(cell.errors) for cell in cells)

        errors = get_form_errors(form, '__all__')

        return CsvRow(

            line=line,

            cells=cells,

            errors=errors,

            is_valid=not bool(cell_errors or errors))

    @property

    def email_is_unique(self):

        return app_settings.A2_EMAIL_IS_UNIQUE or self.ou.email_is_unique

    @property

    def username_is_unique(self):

        return app_settings.A2_USERNAME_IS_UNIQUE or self.ou.username_is_unique

    def check_unique_constraints(self, row, user=None):

        ou_users = User.objects.filter(ou=self.ou)

        users = User.objects.all()

        if user:

            users = users.exclude(pk=user.pk)

            ou_users = ou_users.exclude(pk=user.pk)

        errors = []

        for cell in row:

            if (not cell.header.globally_unique and not cell.header.unique) or (user and not cell.header.update):

                continue

            qs = ou_users

            if cell.header.globally_unique:

                qs = users

            if cell.header.field:

                unique = not qs.filter(**{cell.header.name: cell.value}).exists()

            elif cell.header.attribute:

                atvs = AttributeValue.objects.filter(attribute__name=cell.header.name, content=cell.value)

                unique = not qs.filter(attribute_values__in=atvs).exists()

            if not unique:

                errors.append(

                    Error('unique-constraint-failed', _('Unique constraint on column "%s" failed') % cell.header.name))

        row.errors.extend(errors)

        row.is_valid = row.is_valid and not bool(errors)

        return not bool(errors)

    def do_import_row(self, row):

        if not row.is_valid:

            return

        for header in self.headers:

            if header.key:

                header_key = header

                break

        else:

            assert False, 'no header key'

        user = None

        key_value = row[header_key].value

        if header_key.field:

            users = User.objects.filter(

                **{header_key.name: key_value})

        elif header_key.attribute:

            atvs = AttributeValue.objects.filter(attribute__name=header_key.name, content=key_value)

            users = User.objects.filter(attribute_values__in=atvs)

        users = users[:2]

        if users:

            row.action = 'update'

        else:

            row.action = 'create'

        if len(users) > 1:

            row.errors.append(

                Error('key-matches-too-many-users',

                      _('Key value "%s" matches too many users') % key_value))

            return

        user = None

        if users:

            user = users[0]

        if not self.check_unique_constraints(row, user=user):

            return

        if not user:

            user = User()

        for cell in row.cells:

            if not cell.header.field:

                continue

            if (row.action == 'create' and cell.header.create) or (row.action == 'update' and cell.header.update):

                if getattr(user, cell.header.name) != cell.value:

                    setattr(user, cell.header.name, cell.value)

                    if cell.header.name == 'email' and cell.header.verified:

                        user.email_verified = True

                    cell.action = 'updated'

                    continue

            cell.action = 'nothing'

        user.save()

        for cell in row.cells:

            if cell.header.field or not cell.header.attribute:

                continue

            if (row.action == 'create' and cell.header.create) or (row.action == 'update' and cell.header.update):

                attributes = user.attributes

                if cell.header.verified:

                    attributes = user.verified_attributes

                if getattr(attributes, cell.header.name) != cell.value:

                    setattr(attributes, cell.header.name, cell.value)

                    cell.action = 'updated'

                    continue

            cell.action = 'nothing'

        setattr(self, row.action + 'd', getattr(self, row.action + 'd') + 1)

    def __repr__(self):

        return '<%s %s>' % (self.__class__.__name__, repr(str(self)))

# -*- coding: utf-8 -*-

# authentic2 - versatile identity manager

# Copyright (C) 2010-2019 Entr'ouvert

#

# This program is free software: you can redistribute it and/or modify it

# under the terms of the GNU Affero General Public License as published

# by the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU Affero General Public License for more details.

#

# You should have received a copy of the GNU Affero General Public License

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

from __future__ import unicode_literals

import pytest

import io

from authentic2.custom_user.models import User

from authentic2.models import Attribute

from authentic2.a2_rbac.utils import get_default_ou

from authentic2.csv_import import CsvImporter, UserCsvImporter, CsvHeader, Error, LineError

ENCODINGS = [

    'iso-8859-1',

    'iso-8859-15',

    'utf-8',

    'cp1252',

]

def pytest_generate_tests(metafunc):

    if 'encoding' in metafunc.fixturenames:

        metafunc.parametrize('encoding', ENCODINGS)

    if 'style' in metafunc.fixturenames:

        metafunc.parametrize('style', ['str', 'file'])

@pytest.fixture

def profile(db):

    Attribute.objects.create(name='phone', kind='phone_number', label='Numéro de téléphone')

@pytest.fixture

def csv_importer_factory(encoding, style):

    def factory(content):

        content = content.encode(encoding)

        if style == 'file':

            content = io.BytesIO(content)

        importer = CsvImporter()

        run = importer.run

        importer.run = lambda *args, **kwargs: run(content, *args, encoding=encoding, **kwargs)

        return importer

    return factory

@pytest.fixture

def user_csv_importer_factory(encoding, style):

    def factory(content):

        content = content.encode(encoding)

        if style == 'file':

            content = io.BytesIO(content)

        importer = UserCsvImporter()

        run = importer.run

        importer.run = lambda *args, **kwargs: run(content, *args, encoding=encoding, **kwargs)

        return importer

    return factory

def test_unknown_csv_dialect_error(profile, user_csv_importer_factory):

    importer = user_csv_importer_factory('')

    assert not importer.run()

    assert importer.has_errors

    assert importer.errors == [Error('unknown-csv-dialect')]

def test_empty_header_row_error(profile, user_csv_importer_factory):

    importer = user_csv_importer_factory('\n1,2,3')

    assert not importer.run()

    assert importer.has_errors

    assert importer.errors == [Error('empty-header-row')]

def test_unknown_or_missing_attribute_error1(profile, user_csv_importer_factory):

    importer = user_csv_importer_factory('email key,first_name," "\n1,2,3')

    assert not importer.run()

    assert importer.has_errors

    assert importer.errors == [LineError('unknown-or-missing-attribute', line=1, column=2)]

def test_unknown_or_missing_attribute_error2(profile, user_csv_importer_factory):

    importer = user_csv_importer_factory('email key,first_name,x\n1,2,3')

    assert not importer.run()

    assert importer.has_errors

    assert importer.errors == [LineError('unknown-or-missing-attribute', line=1, column=3)]

def test_unknown_flag_error(profile, user_csv_importer_factory):

    importer = user_csv_importer_factory('email key,first_name xxx\n1,2')

    assert not importer.run()

    assert importer.has_errors

    assert importer.errors == [LineError('unknown-flag', line=1, column=2)]

def test_missing_key_column_error(profile, user_csv_importer_factory):

    importer = user_csv_importer_factory('email,first_name\n1,2')

    assert not importer.run()

    assert importer.has_errors

    assert importer.errors == [Error('missing-key-column')]

def test_too_many_key_columns_error(profile, user_csv_importer_factory):

    importer = user_csv_importer_factory('email key,first_name key\n1,2')

    assert not importer.run()

    assert importer.has_errors

    assert importer.errors == [Error('too-many-key-columns')]

def test_run(profile, user_csv_importer_factory):

    assert User.objects.count() == 0

    content = '''email key verified,first_name,last_name,phone update

tnoel@entrouvert.com,Thomas,Noël,1234

fpeters@entrouvert.com,Frédéric,Péters,5678

x,x,x,x'''

    importer = user_csv_importer_factory(content)

    assert importer.run(), importer.errors

    assert importer.headers == [

        CsvHeader(1, 'email', field=True, key=True, verified=True),

        CsvHeader(2, 'first_name', field=True),

        CsvHeader(3, 'last_name', field=True),

        CsvHeader(4, 'phone', attribute=True),

    ]

    assert importer.has_errors

    assert len(importer.rows) == 3

    assert all(row.is_valid for row in importer.rows[:2])

    assert not importer.rows[2].is_valid

    assert importer.rows[2].cells[0].errors

    assert all(error == Error('data-error') for error in importer.rows[2].cells[0].errors)

    assert not importer.rows[2].cells[1].errors

    assert not importer.rows[2].cells[2].errors

    assert importer.rows[2].cells[3].errors

    assert all(error == Error('data-error') for error in importer.rows[2].cells[3].errors)

    assert importer.updated == 0

    assert importer.created == 2

    assert User.objects.count() == 2

    thomas = User.objects.get(email='tnoel@entrouvert.com')

    assert thomas.email_verified is True

    assert thomas.first_name == 'Thomas'

    assert thomas.attributes.first_name == 'Thomas'

    assert thomas.last_name == 'Noël'

    assert thomas.attributes.last_name == 'Noël'

    assert thomas.attributes.phone == '1234'

    fpeters = User.objects.get(email='fpeters@entrouvert.com')

    assert fpeters.first_name == 'Frédéric'

    assert fpeters.email_verified is True

    assert fpeters.attributes.first_name == 'Frédéric'

    assert fpeters.last_name == 'Péters'

    assert fpeters.attributes.last_name == 'Péters'

    assert fpeters.attributes.phone == '5678'

def test_simulate(profile, user_csv_importer_factory):

    assert User.objects.count() == 0

    content = '''email key verified,first_name,last_name,phone update

tnoel@entrouvert.com,Thomas,Noël,1234

fpeters@entrouvert.com,Frédéric,Péters,5678

x,x,x,x'''

    importer = user_csv_importer_factory(content)

    assert importer.run(simulate=True), importer.errors

    assert importer.headers == [

        CsvHeader(1, 'email', field=True, key=True, verified=True),

        CsvHeader(2, 'first_name', field=True),

        CsvHeader(3, 'last_name', field=True),

        CsvHeader(4, 'phone', attribute=True),

    ]

    assert importer.has_errors

    assert len(importer.rows) == 3

    assert all(row.is_valid for row in importer.rows[:2])

    assert not importer.rows[2].is_valid

    assert importer.rows[2].cells[0].errors

    assert all(error == Error('data-error') for error in importer.rows[2].cells[0].errors)

    assert not importer.rows[2].cells[1].errors

    assert not importer.rows[2].cells[2].errors

    assert importer.rows[2].cells[3].errors

    assert all(error == Error('data-error') for error in importer.rows[2].cells[3].errors)

    assert importer.updated == 0

    assert importer.created == 2

    assert User.objects.count() == 0

def test_create_unique_error(profile, user_csv_importer_factory):

    content = '''email key verified,first_name,last_name,phone unique

tnoel@entrouvert.com,Thomas,Noël,1234'''

    importer = user_csv_importer_factory(content)

    user = User.objects.create(ou=get_default_ou())

    user.attributes.phone = '1234'

    assert importer.run()

    assert importer.created == 0

    assert importer.updated == 0

    assert len(importer.rows) == 1

    assert not importer.rows[0].is_valid

    assert importer.rows[0].action == 'create'

    assert all(not cell.errors for cell in importer.rows[0])

    assert all(not cell.action for cell in importer.rows[0])

    assert importer.rows[0].errors == [Error('unique-constraint-failed')]

def test_create_unique_in_ou(profile, user_csv_importer_factory):

    content = '''email key verified,first_name,last_name,phone unique

tnoel@entrouvert.com,Thomas,Noël,1234'''

    importer = user_csv_importer_factory(content)

    user = User.objects.create()

    user.attributes.phone = '1234'

    assert importer.run()

    assert len(importer.rows) == 1

    assert importer.rows[0].is_valid

    assert importer.rows[0].action == 'create'

    assert all(not cell.errors for cell in importer.rows[0])

    assert all(cell.action == 'updated' for cell in importer.rows[0])

    assert importer.created == 1

    assert importer.updated == 0

def test_create_unique_globally_error(profile, user_csv_importer_factory):

    content = '''email key verified,first_name,last_name,phone globally-unique

tnoel@entrouvert.com,Thomas,Noël,1234'''

    importer = user_csv_importer_factory(content)

    user = User.objects.create()

    user.attributes.phone = '1234'

    assert importer.run()

    assert importer.created == 0

    assert importer.updated == 0

    assert len(importer.rows) == 1

    assert not importer.rows[0].is_valid

    assert importer.rows[0].action == 'create'

    assert all(not cell.errors for cell in importer.rows[0])

    assert all(not cell.action for cell in importer.rows[0])

    assert importer.rows[0].errors == [Error('unique-constraint-failed')]

def test_update_unique_error(profile, user_csv_importer_factory):

    content = '''email key verified,first_name,last_name,phone unique update

tnoel@entrouvert.com,Thomas,Noël,1234'''

    importer = user_csv_importer_factory(content)

    user = User.objects.create(ou=get_default_ou())

    user.attributes.phone = '1234'

    user = User.objects.create(email='tnoel@entrouvert.com', ou=get_default_ou())

    assert importer.run()

    assert importer.created == 0

    assert importer.updated == 0

    assert len(importer.rows) == 1

    assert not importer.rows[0].is_valid

    assert importer.rows[0].action == 'update'

    assert all(not cell.errors for cell in importer.rows[0])

    assert all(not cell.action for cell in importer.rows[0])

    assert importer.rows[0].errors == [Error('unique-constraint-failed')]

def test_update_unique_globally_error(profile, user_csv_importer_factory):

    content = '''email key verified,first_name,last_name,phone globally-unique update

tnoel@entrouvert.com,Thomas,Noël,1234'''

    importer = user_csv_importer_factory(content)

    user = User.objects.create()

    user.attributes.phone = '1234'

    User.objects.create(email='tnoel@entrouvert.com', ou=get_default_ou())

    assert importer.run()

    assert importer.created == 0

    assert importer.updated == 0

    assert len(importer.rows) == 1

    assert not importer.rows[0].is_valid

    assert importer.rows[0].action == 'update'

    assert all(not cell.errors for cell in importer.rows[0])

    assert all(not cell.action for cell in importer.rows[0])

    assert importer.rows[0].errors == [Error('unique-constraint-failed')]

def test_update_unique_globally(profile, user_csv_importer_factory):

    content = '''email key verified no-update,first_name no-update,last_name no-update,phone unique update

tnoel@entrouvert.com,Thomas,Noël,1234'''

    importer = user_csv_importer_factory(content)

    user = User.objects.create()

    user.attributes.phone = '1234'

    thomas = User.objects.create(email='tnoel@entrouvert.com', ou=get_default_ou())

    assert importer.run()

    assert importer.created == 0

    assert importer.updated == 1

    assert len(importer.rows) == 1

    assert importer.rows[0].is_valid

    assert importer.rows[0].action == 'update'

    assert all(not cell.errors for cell in importer.rows[0])

    assert all(cell.action == 'nothing' for cell in importer.rows[0].cells[:3])

    assert importer.rows[0].cells[3].action == 'updated'

    thomas.refresh_from_db()

    assert not thomas.first_name

    assert not thomas.last_name

    assert thomas.attributes.phone == '1234'