Merge "Add support for user correction"

etc/corrections.json

@@ -3212,5 +3212,12 @@
             "company_name": "ZTE Corporation",
             "correction_comment": "Related-Bug: #1634020"
         }
+    ],
+    "user_corrections": [
+        {
+            "correction_comment": "Reset emails (Related-Bug: #1634020)",
+            "user_id": "zhangyujun",
+            "emails": ["yujun.zhang@easystack.cn","284517620@qq.com"]
+        }
     ]
 }

etc/corrections.schema.json

@@ -1,7 +1,7 @@
 {
     "$schema": "http://json-schema.org/draft-04/schema#",
     "type": "object",
-    "required": ["corrections"],
+    "required": ["corrections", "user_corrections"],
     "properties": {
         "corrections": {
             "type": "array",
@@ -41,6 +41,28 @@
                 },
                 "required": ["primary_key", "correction_comment"]
             }
+        },
+        "user_corrections": {
+            "type": "array",
+            "items": {
+                "type": "object",
+                "properties": {
+                    "user_id": {
+                        "type": "string",
+                        "pattern": "^[\\S]+$"
+                    },
+                    "emails": {
+                        "type": "array",
+                        "items": {
+                            "type": "string",
+                            "pattern": "^[a-z\\d_\\.-]+@([a-z\\d\\.-]+\\.)+[a-z]+$"
+                        }
+                    },
+                    "correction_comment": {
+                        "type": "string"
+                    }
+                }
+            }
         }
     }
 }

stackalytics/processor/main.py

@@ -262,6 +262,14 @@ def apply_corrections(uri, runtime_storage_inst):
             LOG.warning('Correction misses primary key: %s', c)
     runtime_storage_inst.apply_corrections(valid_corrections)
 
+    valid_user_corrections = []
+    for u in corrections['user_corrections']:
+        if 'user_id' in u:
+            valid_user_corrections.append(c)
+        else:
+            LOG.warning('User correction misses user_id: %s', u)
+    runtime_storage_inst.apply_user_corrections(valid_user_corrections)
+
 
 def process_project_list(runtime_storage_inst):
     module_groups = runtime_storage_inst.get_by_key('module_groups') or {}

stackalytics/processor/runtime_storage.py

@@ -19,6 +19,8 @@ import memcache
 from oslo_log import log as logging
 import six
 
+
+from stackalytics.processor import user_processor
 from stackalytics.processor import utils
 
 
@@ -123,6 +125,15 @@ class MemcachedStorage(RuntimeStorage):
                 self.set_by_key(self._get_record_name(record_id), original)
                 self._commit_update(record_id)
 
+    def apply_user_corrections(self, user_corrections_iterator):
+        for user_correction in user_corrections_iterator:
+            stored_user = user_processor.load_user(self,
+                                                   user_id=user_correction[
+                                                       'user_id'])
+            updated_user = user_processor.update_user_profile(
+                stored_user, user_correction, is_correction=True)
+            user_processor.store_user(self, updated_user)
+
     def inc_user_count(self):
         return self.memcached.incr('user:count')
 

stackalytics/processor/user_processor.py

@@ -91,11 +91,17 @@ def delete_user(runtime_storage_inst, user):
     runtime_storage_inst.delete_by_key('user:%s' % user['seq'])
 
 
-def update_user_profile(stored_user, user):
+def update_user_profile(stored_user, user, is_correction=False):
     # update stored_user with user and return it
     if stored_user:
         updated_user = copy.deepcopy(stored_user)
         updated_user.update(user)
+        if is_correction:
+            updated_user['emails'] = user.get('emails',
+                                              stored_user.get('emails', []))
+            updated_user['corrections'] = stored_user.get('corrections', [])\
+                + [user.get('correction_comment', '')]
+        else:
             updated_user['emails'] = list(set(stored_user.get('emails', [])) |
                                           set(user.get('emails', [])))
     else:

stackalytics/tests/unit/test_user_processor.py

@@ -66,6 +66,44 @@ class TestUserProcessor(testtools.TestCase):
         # static flag must present
         self.assertTrue(updated_user.get('static'))
 
+    def test_update_user_with_correction(self):
+        user_correction = {
+            "user_id": "user",
+            "correction_comment": "Reset emails",
+            "emails": ["updated@smith.com"]
+        }
+
+        stored_user = {
+            "launchpad_id": "user",
+            "companies": [
+                {
+                    "company_name": "Rackspace",
+                    "end_date": "2011-Nov-20"
+                },
+                {
+                    "company_name": "IBM",
+                    "end_date": None
+                }
+            ],
+            "user_name": "Johnny",
+            "emails": ["obsoleted@smith.com"],
+            "corrections": ["Old correction"],
+            "static": True
+        }
+
+        updated_user = user_processor.update_user_profile(stored_user,
+                                                          user_correction,
+                                                          is_correction=True)
+
+        # reset emails from correction
+        self.assertEqual(set(user_correction['emails']),
+                         set(updated_user['emails']))
+        # save correction history
+        self.assertEqual(updated_user['corrections'],
+                         ["Old correction", "Reset emails"])
+        # static flag must present
+        self.assertTrue(updated_user.get('static'))
+
     def test_update_user_unknown_user(self):
         user = {
             "launchpad_id": "user",