made changes

tests/pulse_consumer/test_pulse_consumer.py

@@ -1,7 +1,6 @@
 from treeherder.pulse_consumer.consumer import PulseDataAdapter, TreeherderDataAdapter
 from ..sampledata import SampleData
 
-
 def test_process_data():
 
     sd = SampleData()
@@ -15,13 +14,15 @@ def test_process_data():
 
     msg = Message()
 
-    data = pda.process_data(sd.raw_pulse_data[0], msg)
+    for data in sd.raw_pulse_data:
 
-    missing_attributes = pda.required_attributes.difference(
-        set( data.keys() )
-        )
+        data = pda.process_data(data, msg)
 
-    assert set() == missing_attributes
+        missing_attributes = pda.required_attributes.difference(
+            set( data.keys() )
+            )
+
+        assert set() == missing_attributes
 
 class Message(object):
 

treeherder/model/sample_data/job_data.json.sample

@@ -8,20 +8,42 @@
      * The revision_hash is used for associating job data asynchronously 
      * with a single entry in project_jobs_1.result_set.
      **************/
-    sources: { 
-        gecko: "revision",
-        gaia: "revision",
-        mozharness: "revision,
-        ...
+    sources: [
+        {
+            repository:"gecko",
+            "revision":"",
+            "comments":"",
+            "push_timestamp":"",
+            "commit_timestamp":""
         },
-            
+        {
+            repository:"gaia",
+            "revision":"",
+            "comments":"",
+            "push_timestamp":"",
+            "commit_timestamp":""
+        },
+        {
+            repository:"mozharness",
+            "revision":"",
+            "comments":"",
+            "push_timestamp":"",
+            "commit_timestamp":""
+        },
+
+        ...
+
+        ],
+
+    revision_hash:"",
+
     jobs: [
         {
             #Stored in project_jobs_1.job.job_guid
             job_guid:"",
 
             #Stored in treeherder_reference_1.job_type.name
-            name:"build | mochitest_one | ...", 
+            name:"build | mochitest_one | ...",
 
             #Stored in treeherder_reference_1.product
             product_name:"Firefox | Firefox OS | Thunderbird | ...",
@@ -48,20 +70,20 @@
             end_timestamp:"",
 
             #Stored in treeherder_reference_1.machine.name
-            machine:"",    
+            machine:"",
 
             #Stored in:
             # treeherder_reference_1.build_platform.os_name,
-            # treeherder_reference_1.build_platform.platform, 
-            # treeherder_reference_1.build_platform.architecture, 
+            # treeherder_reference_1.build_platform.platform,
+            # treeherder_reference_1.build_platform.architecture,
             build_platform:{ os_name:"", platform:"", architecture:"" },
 
             #Stored in:
             # treeherder_reference_1.machine_platform.os_name,
-            # treeherder_reference_1.machine_platform.platform, 
-            # treeherder_reference_1.machine_platform.architecture, 
+            # treeherder_reference_1.machine_platform.platform,
+            # treeherder_reference_1.machine_platform.architecture,
             machine_platform:{ os_name:"", platform:"", architecture:"" },
-    
+
             #Stored in treeherder_reference_1.option_collection and
             #treeherder_reference_1.option
             option_collection: {
@@ -77,7 +99,7 @@
                   name:"" },
                   ...
                   ],
-                  
+
             #Stored project_jobs_1.job_artifact
             artifact:"{
                 type:" json | img | ...",

treeherder/pulse_consumer/consumer.py

@@ -410,6 +410,7 @@ class PulseDataAdapter(object):
                     'attr_table':[
                         { 'attr':'who' },
                         { 'attr':'when' },
+                        { 'attr':'comments' },
                         ]
                     },
                 'payload.build.properties': {
@@ -468,7 +469,15 @@ class PulseDataAdapter(object):
 
         #configure consumers
         self.pulse.configure(
-            topic=['#.finished', '#.log_uploaded'],
+            #####
+            #TODO: Register a specialized adapter for #.finished
+            #      to record the heartbeat of the push. This will
+            #      require adding the request_ids and request_times
+            #      to the .finished data structure.
+            #
+            #topic=['#.finished', '#.log_uploaded'],
+            #####
+            topic=['#.log_uploaded'],
             callback=self.process_data,
             durable=self.durable
             )
@@ -516,6 +525,7 @@ class PulseDataAdapter(object):
             #    missing_attributes, data, raw_data
             #    )
             pass
+
         else:
             #Carry out data processing that requires all of the
             #attributes being populated
@@ -559,7 +569,8 @@ class PulseDataAdapter(object):
                 attr = attr_data.get('attr', None)
                 attr_test = attr_data.get('attr_test', None)
 
-                if ( attr_test and (datum[0] in attr_test) ) or ( attr and (attr in datum[0]) ):
+                if ( attr_test and (datum[0] in attr_test) ) or \
+                    ( attr and (attr in datum[0]) ):
 
                     cb = attr_data.get('cb', None)
 
@@ -690,7 +701,22 @@ class TreeherderDataAdapter(PulseDataAdapter):
             'jobs': []
             }
 
-        treeherder_data['sources'][ data['branch'] ] = data['revision']
+        treeherder_data['sources'] = []
+
+        ####
+        #TODO: This is a temporary fix, this data will not be located
+        #      in the sourceStamp in the pulse stream. It will likely
+        #      be in other build properties but for now this will work.
+        #      Once the new properties are added they need to be incorporated
+        #      here.
+        ####
+        treeherder_data['sources'].append(
+            { 'repository':data['branch'],
+              'revision':data['revision'],
+              'push_timestamp':data['when'],
+              'commit_timestamp':data['when'],
+              'comments':data['comments'] }
+            )
 
         job = {
             'job_guid': self.get_job_guid(