Repository: openaddresses/machine
Pull request #576 (2017-03-09 23:18:59)

Complete RunState usage
Switched to using RunState instances everywhere and converting to/from dictionaries on every database boundary. Finished work started in #572, prepares for #567.


Patch:
diff --git a/openaddr/ci/__init__.py b/openaddr/ci/__init__.py
index 356b879..8f0c135 100644
--- a/openaddr/ci/__init__.py
+++ b/openaddr/ci/__init__.py
@@ -1007,11 +1007,15 @@ def pop_task_from_donequeue(queue, github_auth):
         if task is None:
             return
     
+        # Convert dictionary into RunState
+        if 'result' in task.data:
+            task.data['result'] = objects.result_dictionary2runstate(task.data['result'])
+        
         donedata = queuedata.Done(**task.data)
         _L.info(u'Got file {} from done queue'.format(donedata.name))
         results = donedata.result
         message = results['message']
-        run_state = RunState(results.get('state', None) or results.get('output', None)) # TODO: stop looking for old "output" here.
+        run_state = results['state']
         content_b64 = donedata.content_b64
         commit_sha = donedata.commit_sha
         worker_id = donedata.worker_id
diff --git a/openaddr/ci/objects.py b/openaddr/ci/objects.py
index 3e2a0c3..c1410ef 100644
--- a/openaddr/ci/objects.py
+++ b/openaddr/ci/objects.py
@@ -1,6 +1,6 @@
 import logging; _L = logging.getLogger('openaddr.ci.objects')
 
-import json, pickle
+import json, pickle, copy
 
 # Todo: make this a Python 3 enum
 FAIL_REASONS = {
@@ -128,9 +128,12 @@ class RunState:
     
     def get(self, json_key):
         return getattr(self, RunState.key_attrs[json_key])
-        
+    
+    def to_dict(self):
+        return {k: self.get(k) for k in self.keys}
+    
     def to_json(self):
-        return json.dumps({k: self.get(k) for k in self.keys})
+        return json.dumps(self.to_dict())
 
 class Zip:
     '''
@@ -139,23 +142,59 @@ class Zip:
         self.url = url
         self.content_length = content_length
 
+def _result_runstate2dictionary(result):
+    '''
+    '''
+    actual_result = copy.copy(result)
+
+    if result and 'state' in result:
+        actual_result['state'] = result['state'].to_dict()
+    elif result and 'output' in result:
+        # old-style
+        actual_result['state'] = result.pop('output').to_dict()
+
+    return actual_result
+
+def result_dictionary2runstate(result):
+    '''
+    '''
+    actual_result = copy.copy(result)
+
+    if result and 'state' in result:
+        actual_result['state'] = RunState(result['state'])
+    elif result and 'output' in result:
+        # old-style
+        actual_result['state'] = RunState(result.pop('output'))
+    elif result:
+        actual_result['state'] = RunState(None)
+
+    return actual_result
+
 def add_job(db, job_id, status, task_files, file_states, file_results, owner, repo, status_url, comments_url):
     ''' Save information about a job to the database.
     
         Throws an IntegrityError exception if the job ID exists.
     '''
+    # Find RunState instances in file_results and turn them into dictionaries.
+    actual_results = {path: _result_runstate2dictionary(result)
+                      for (path, result) in file_results.items()}
+    
     db.execute('''INSERT INTO jobs
                   (task_files, file_states, file_results, github_owner,
                    github_repository, github_status_url, github_comments_url,
                    status, id, datetime_start)
                   VALUES (%s::json, %s::json, %s::json, %s, %s, %s, %s, %s, %s, NOW())''',
                (json.dumps(task_files), json.dumps(file_states),
-                json.dumps(file_results), owner, repo, status_url,
+                json.dumps(actual_results), owner, repo, status_url,
                 comments_url, status, job_id))
 
 def write_job(db, job_id, status, task_files, file_states, file_results, owner, repo, status_url, comments_url):
     ''' Save information about a job to the database.
     '''
+    # Find RunState instances in file_results and turn them into dictionaries.
+    actual_results = {path: _result_runstate2dictionary(result)
+                      for (path, result) in file_results.items()}
+    
     is_complete = bool(status is not None)
     
     db.execute('''UPDATE jobs
@@ -165,7 +204,7 @@ def write_job(db, job_id, status, task_files, file_states, file_results, owner,
                       datetime_end=CASE WHEN %s THEN NOW() ELSE null END
                   WHERE id = %s''',
                (json.dumps(task_files), json.dumps(file_states),
-                json.dumps(file_results), owner, repo, status_url, comments_url,
+                json.dumps(actual_results), owner, repo, status_url, comments_url,
                 status, is_complete, job_id))
 
 def read_job(db, job_id):
@@ -185,7 +224,11 @@ def read_job(db, job_id):
     except TypeError:
         return None
     else:
-        return Job(job_id, status, task_files, states, file_results,
+        # Find dictionaries in file_results and turn them into RunState instances.
+        actual_results = {path: result_dictionary2runstate(result)
+                          for (path, result) in file_results.items()}
+    
+        return Job(job_id, status, task_files, states, actual_results,
                    github_owner, github_repository, github_status_url,
                    github_comments_url, datetime_start, datetime_end)
     
@@ -205,7 +248,18 @@ def read_jobs(db, past_id):
                   ORDER BY sequence DESC LIMIT 25''',
                (past_id, ))
     
-    return [Job(*row) for row in db.fetchall()]
+    jobs = []
+    
+    for row in db.fetchall():
+        # Find dictionaries in file_results and turn them into RunState instances.
+        job_args = list(row)
+        file_results = job_args.pop(4)
+        actual_results = {path: result_dictionary2runstate(result)
+                          for (path, result) in file_results.items()}
+        job_args.insert(4, actual_results)
+        jobs.append(Job(*job_args))
+    
+    return jobs
 
 def add_set(db, owner, repository):
     '''
diff --git a/openaddr/ci/queuedata.py b/openaddr/ci/queuedata.py
index a9f6136..78f3f9a 100644
--- a/openaddr/ci/queuedata.py
+++ b/openaddr/ci/queuedata.py
@@ -56,8 +56,7 @@ class Done:
 
         # Convert RunState to a plain dictionary
         if data['result'] and 'state' in data['result']:
-            state = data['result']['state']
-            data['result']['state'] = {k: state.get(k) for k in state.keys}
+            data['result']['state'] = data['result']['state'].to_dict()
     
         return data
 


Test patch:
diff --git a/openaddr/tests/ci.py b/openaddr/tests/ci.py
index 7f77ed7..3f882db 100644
--- a/openaddr/tests/ci.py
+++ b/openaddr/tests/ci.py
@@ -40,7 +40,8 @@ from ..ci.objects import (
     add_run, set_run, copy_run, get_completed_file_run, get_completed_run,
     old_read_completed_set_runs, read_completed_set_runs, read_latest_set,
     read_run, read_completed_runs_to_date, read_latest_run, Run, RunState,
-    read_completed_source_runs, read_completed_set_runs_count
+    read_completed_source_runs, read_completed_set_runs_count,
+    result_dictionary2runstate
     )
 
 from ..ci.collect import (
@@ -121,7 +122,8 @@ class TestObjects (unittest.TestCase):
     def test_add_job(self):
         ''' Check behavior of objects.add_job()
         '''
-        add_job(self.db, 'xyz', True, {}, {}, {}, 'o', 'a', 'http://', 'https://')
+        results = {'sources/file.json': {'message': 'Yo', 'state': RunState({'source': 'file.txt'})}}
+        add_job(self.db, 'xyz', True, {}, {}, results, 'o', 'a', 'http://', 'https://')
 
         self.db.execute.assert_called_once_with(
                '''INSERT INTO jobs
@@ -129,12 +131,36 @@ class TestObjects (unittest.TestCase):
                    github_repository, github_status_url, github_comments_url,
                    status, id, datetime_start)
                   VALUES (%s::json, %s::json, %s::json, %s, %s, %s, %s, %s, %s, NOW())''',
-                  ('{}', '{}', '{}', 'o', 'a', 'http://', 'https://', True, 'xyz'))
+                  ('{}', '{}', '{"sources/file.json": {"message": "Yo", "state": {"source": "file.txt"}}}',
+                   'o', 'a', 'http://', 'https://', True, 'xyz'))
 
+    def test_result_dictionary2runstate(self):
+        '''
+        '''
+        result1 = {'message': 'Yo', 'state': {'source': 'Hello'}}
+        result1b = result_dictionary2runstate(result1)
+        self.assertEqual(result1b['message'], 'Yo')
+        self.assertEqual(result1b['state'].source, 'Hello')
+
+        result2 = {'message': 'Yo', 'output': {'source': 'Hello'}}
+        result2b = result_dictionary2runstate(result2)
+        self.assertEqual(result2b['message'], 'Yo')
+        self.assertEqual(result2b['state'].source, 'Hello')
+
+        result3 = {'message': 'Yo'}
+        result3b = result_dictionary2runstate(result3)
+        self.assertEqual(result3b['message'], 'Yo')
+        self.assertIsNone(result3b['state'].source)
+
+        result4 = None
+        result4b = result_dictionary2runstate(result4)
+        self.assertIsNone(result4b)
+    
     def test_write_job_success(self):
         ''' Check behavior of objects.write_job()
         '''
-        write_job(self.db, 'xyz', True, {}, {}, {}, 'o', 'a', 'http://', 'https://')
+        results = {'sources/file.json': {'message': 'Yo', 'state': RunState({'source': 'file.txt'})}}
+        write_job(self.db, 'xyz', True, {}, {}, results, 'o', 'a', 'http://', 'https://')
 
         self.db.execute.assert_called_once_with(
                '''UPDATE jobs
@@ -143,12 +169,14 @@ class TestObjects (unittest.TestCase):
                       github_status_url=%s, github_comments_url=%s, status=%s,
                       datetime_end=CASE WHEN %s THEN NOW() ELSE null END
                   WHERE id = %s''',
-                  ('{}', '{}', '{}', 'o', 'a', 'http://', 'https://', True, True, 'xyz'))
+                  ('{}', '{}', '{"sources/file.json": {"message": "Yo", "state": {"source": "file.txt"}}}',
+                   'o', 'a', 'http://', 'https://', True, True, 'xyz'))
 
     def test_write_job_failure(self):
         ''' Check behavior of objects.write_job()
         '''
-        write_job(self.db, 'xyz', False, {}, {}, {}, 'o', 'a', 'http://', 'https://')
+        results = {'sources/file.json': {'message': 'Yo', 'state': RunState({'source': 'file.txt'})}}
+        write_job(self.db, 'xyz', False, {}, {}, results, 'o', 'a', 'http://', 'https://')
 
         self.db.execute.assert_called_once_with(
                '''UPDATE jobs
@@ -157,12 +185,14 @@ class TestObjects (unittest.TestCase):
                       github_status_url=%s, github_comments_url=%s, status=%s,
                       datetime_end=CASE WHEN %s THEN NOW() ELSE null END
                   WHERE id = %s''',
-                  ('{}', '{}', '{}', 'o', 'a', 'http://', 'https://', False, True, 'xyz'))
+                  ('{}', '{}', '{"sources/file.json": {"message": "Yo", "state": {"source": "file.txt"}}}',
+                   'o', 'a', 'http://', 'https://', False, True, 'xyz'))
 
     def test_write_job_ongoing(self):
         ''' Check behavior of objects.write_job()
         '''
-        write_job(self.db, 'xyz', None, {}, {}, {}, 'o', 'a', 'http://', 'https://')
+        results = {'sources/file.json': {'message': 'Yo', 'state': RunState({'source': 'file.txt'})}}
+        write_job(self.db, 'xyz', None, {}, {}, results, 'o', 'a', 'http://', 'https://')
 
         self.db.execute.assert_called_once_with(
                '''UPDATE jobs
@@ -171,12 +201,13 @@ class TestObjects (unittest.TestCase):
                       github_status_url=%s, github_comments_url=%s, status=%s,
                       datetime_end=CASE WHEN %s THEN NOW() ELSE null END
                   WHERE id = %s''',
-                  ('{}', '{}', '{}', 'o', 'a', 'http://', 'https://', None, False, 'xyz'))
+                  ('{}', '{}', '{"sources/file.json": {"message": "Yo", "state": {"source": "file.txt"}}}',
+                   'o', 'a', 'http://', 'https://', None, False, 'xyz'))
 
     def test_read_job_yes(self):
         ''' Check behavior of objects.read_job()
         '''
-        self.db.fetchone.return_value = True, {}, {}, {}, 'o', 'a', 'http://', 'https://', None, None
+        self.db.fetchone.return_value = True, {}, {}, {"sources/file.json": {"message": "Yo", "state": {"source": "file.txt"}}}, 'o', 'a', 'http://', 'https://', None, None
         
         job = read_job(self.db, 'xyz')
         self.assertEqual(job.id, 'xyz')
@@ -185,6 +216,9 @@ class TestObjects (unittest.TestCase):
         self.assertEqual(job.github_repository, 'a')
         self.assertEqual(job.github_status_url, 'http://')
         self.assertEqual(job.github_comments_url, 'https://')
+        self.assertIn('sources/file.json', job.file_results)
+        self.assertEqual(job.file_results['sources/file.json']['message'], 'Yo')
+        self.assertEqual(job.file_results['sources/file.json']['state'].source, 'file.txt')
         self.assertIsNone(job.datetime_start)
         self.assertIsNone(job.datetime_end)
 
@@ -196,6 +230,16 @@ class TestObjects (unittest.TestCase):
                   LIMIT 1''',
                   ('xyz', ))
 
+    def test_read_job_old(self):
+        ''' Check old-dict behavior of objects.read_job()
+        '''
+        self.db.fetchone.return_value = True, {}, {}, {"sources/file.json": {"message": "Yo", "output": {"source": "file.txt"}}}, 'o', 'a', 'http://', 'https://', None, None
+        
+        job = read_job(self.db, 'xyz')
+        self.assertIn('sources/file.json', job.file_results)
+        self.assertEqual(job.file_results['sources/file.json']['message'], 'Yo')
+        self.assertEqual(job.file_results['sources/file.json']['state'].source, 'file.txt')
+
     def test_read_job_no(self):
         ''' Check behavior of objects.read_job()
         '''
@@ -215,11 +259,14 @@ class TestObjects (unittest.TestCase):
     def test_read_jobs(self):
         ''' Check behavior of objects.read_jobs()
         '''
-        self.db.fetchall.return_value = (('xyz', True, {}, {}, {}, 'o', 'a', 'http://', 'https://', None, None), )
+        self.db.fetchall.return_value = (('xyz', True, {}, {}, {"sources/file.json": {"message": "Yo", "state": {"source": "file.txt"}}}, 'o', 'a', 'http://', 'https://', None, None), )
         
         (job, ) = read_jobs(self.db, None)
         self.assertEqual(job.id, 'xyz')
         self.assertEqual(job.status, True)
+        self.assertIn('sources/file.json', job.file_results)
+        self.assertEqual(job.file_results['sources/file.json']['message'], 'Yo')
+        self.assertEqual(job.file_results['sources/file.json']['state'].source, 'file.txt')
 
         self.db.execute.assert_called_once_with(
                '''SELECT id, status, task_files, file_states, file_results,
@@ -3525,7 +3572,10 @@ class TestQueue (unittest.TestCase):
         self.assertEqual(run_data[6:], (False, 'j', None, 'sss', is_merged_to_master.return_value, None))
         
         job_data = update_job_status.mock_calls[0][1]
-        self.assertEqual(job_data[1:6], ('j', 'u', 'sources/xx/f.json', False, {'message': 'Yo', 'output': output_data}))
+        self.assertEqual(job_data[1:5], ('j', 'u', 'sources/xx/f.json', False))
+        self.assertEqual(job_data[5]['message'], 'Yo')
+        self.assertEqual(job_data[5]['state'].source, 'sources/xx/f.json')
+        self.assertEqual(job_data[5]['state'].process_hash, 'f00')
     
     @patch('openaddr.ci.WORKER_COOLDOWN', new=timedelta(seconds=0))
     @patch('openaddr.ci.work.do_work')