csv: set verification source for user attributes (#66053)

src/authentic2/csv_import.py

@@ -774,12 +774,32 @@ class UserCsvImporter:
                 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
+                # lock attribute values objects
+                list(AttributeValue.objects.with_owner(user).select_for_update().values_list('id', flat=True))
+                updated = False
+
+                with atomic():
+                    if cell.header.verified:
+                        attributes = user.verified_attributes
+                        is_known_unverified_value = False
+                        unverified_value = cell.header.attribute.get_value(user, verified=False)
+                        if unverified_value:
+                            is_known_unverified_value = (
+                                cell.value in unverified_value
+                                if cell.header.attribute.multiple
+                                else cell.value == unverified_value
+                            )
+                        if getattr(attributes, cell.header.name) != cell.value or is_known_unverified_value:
+                            attributes._set_sourced_attr(cell.header.name, cell.value, 'csv')
+                        updated = True
+                    else:
+                        # TODO remove only 'csv' among verification sources
+                        setattr(attributes, cell.header.name, cell.value)
+                        updated = True
+
+                    if updated:
+                        cell.action = 'updated'
+                        continue
             cell.action = 'nothing'
 
         for cell in row.cells:

tests/test_csv_import.py

@@ -27,7 +27,7 @@ from authentic2.a2_rbac.models import Role
 from authentic2.a2_rbac.utils import get_default_ou
 from authentic2.csv_import import CsvHeader, CsvImporter, Error, LineError, UserCsvImporter
 from authentic2.custom_user.models import User
-from authentic2.models import Attribute, PasswordReset
+from authentic2.models import Attribute, AttributeValue, PasswordReset
 
 ENCODINGS = [
     'iso-8859-1',
@@ -728,3 +728,34 @@ john.doe@example.com,John,Doe2'''
     user2.refresh_from_db()
     assert user1.last_name == 'Doe'
     assert user2.last_name == 'Doe2'
+
+
+def test_user_attributes_verified(db, profile, user_csv_importer_factory):
+    content = '''email key,first_name,last_name,phone verified
+jdoe@nowhere.null,John,Doe,0101010101
+jsmith@nowhere.null,Jimmy,Smith,0202020202'''
+    importer = user_csv_importer_factory(content)
+
+    assert importer.run(), importer.all_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', field=True, verified=True),
+    ]
+
+    jdoe = User.objects.get(email='jdoe@nowhere.null')
+    assert jdoe.verified_attributes.phone == '0101010101'
+
+    jsmith = User.objects.get(email='jsmith@nowhere.null')
+    assert jsmith.verified_attributes.phone == '0202020202'
+
+    for user in (
+        jdoe,
+        jsmith,
+    ):
+        attribute = Attribute.objects.get(name='phone')
+        av = AttributeValue.objects.with_owner(user).get(attribute=attribute)
+        assert av.verified
+        assert av.verification_sources == ['csv']
+        assert av.last_verified_on