Merge pull request #258 from Codeacious/jrm_parallel_store_trond

Implemented #221, called tronstore

docs/config.rst

@@ -215,6 +215,18 @@ State Persistence
         The number of save calls to buffer before writing the state.  Defaults to 1,
         which is no buffering.
 
+    **db_store_method**
+        The method to use for saving state information to a SQL database. Only used if store_type is sql.
+
+        Valid options are:
+            **json** - uses the `simplejson` module.
+
+            **msgpack** - uses the `msgpack` module, from the msgpack-python package (tested with version 0.3.0).
+
+            **pickle** - uses the `cPickle` module. be careful with this one, as pickle is Turing complete.
+
+            **yaml** - uses the `yaml` module, from the PyYaml package (tested with version 3.10).
+
 
 Example::
 
@@ -223,6 +235,7 @@ Example::
         name: local_sqlite
         connection_details: "sqlite:///dest_state.db"
         buffer_size: 1 # No buffer
+        db_store_method: json
 
 
 .. _action_runners:

docs/man_tronview.rst

@@ -69,6 +69,9 @@ Options
 ``-s, --save``
     Save server and color options to client config file (~/.tron)
 
+``--namespace``
+    Only show jobs and services from the specified namespace
+
 
 States
 ----------

tests/core/job_test.py

@@ -103,12 +103,17 @@ def test_restore_state(self):
         job_runs = [mock.Mock(), mock.Mock()]
         state_data = ({'enabled': False, 'run_ids': [1, 2]}, run_data)
 
-        with mock.patch.object(self.job.job_runs, 'restore_state', return_value=job_runs):
+        with contextlib.nested(
+            mock.patch.object(self.job.job_runs, 'restore_state', return_value=job_runs),
+            mock.patch.object(self.job.job_runs, 'get_run_numbers', return_value=state_data[0]['run_ids'])
+        ):
             self.job.restore_state(state_data)
 
             assert not self.job.enabled
             calls = [mock.call(job_runs[i]) for i in xrange(len(job_runs))]
             self.job.watcher.watch.assert_has_calls(calls)
+            calls = [mock.call(job_runs[i], jobrun.JobRun.NOTIFY_DONE) for i in xrange(len(job_runs))]
+            self.job_scheduler.watch.assert_has_calls(calls)
             assert_equal(self.job.job_state.state_data, state_data[0])
             self.job.job_runs.restore_state.assert_called_once_with(
                 sorted(run_data, key=lambda data: data['run_num'], reverse=True),
@@ -117,6 +122,7 @@ def test_restore_state(self):
                 self.job.context,
                 self.job.node_pool
             )
+            self.job.job_runs.get_run_numbers.assert_called_once_with()
             self.job.job_scheduler.restore_state.assert_called_once_with()
             self.job.event.ok.assert_called_with('restored')
 
@@ -500,6 +506,35 @@ def mock_eventloop(self):
     def teardown_job(self):
         event.EventManager.reset()
 
+    def test_restore_state_scheduled(self):
+        mock_scheduled = [mock.Mock(), mock.Mock()]
+        with contextlib.nested(
+            mock.patch.object(self.job_scheduler.job_runs, 'get_scheduled',
+                return_value=iter(mock_scheduled)),
+            mock.patch.object(self.job_scheduler, 'schedule'),
+            mock.patch.object(self.job_scheduler, '_set_callback')
+        ) as (get_patch, sched_patch, back_patch):
+            self.job_scheduler.restore_state()
+            get_patch.assert_called_once_with()
+            calls = [mock.call(m) for m in mock_scheduled]
+            back_patch.assert_has_calls(calls)
+            sched_patch.assert_called_once_with()
+
+    def test_restore_state_queued(self):
+        queued = mock.Mock()
+        with contextlib.nested(
+            mock.patch.object(self.job_scheduler.job_runs, 'get_scheduled',
+                return_value=iter([])),
+            mock.patch.object(self.job_scheduler.job_runs, 'get_first_queued',
+                return_value=queued),
+            mock.patch.object(self.job_scheduler, 'schedule'),
+            mock.patch.object(job.eventloop, 'call_later')
+        ) as (get_patch, queue_patch, sched_patch, later_patch):
+            self.job_scheduler.restore_state()
+            get_patch.assert_called_once_with()
+            later_patch.assert_called_once_with(0, self.job_scheduler.run_job, queued, run_queued=True)
+            sched_patch.assert_called_once_with()
+
     def test_schedule(self):
         with mock.patch.object(self.job_scheduler.job_state, 'is_enabled',
         new=True):

tests/mcp_reconfigure_test.py

@@ -143,6 +143,7 @@ def teardown_mcp(self):
         event.EventManager.reset()
         filehandler.OutputPath(self.test_dir).delete()
         filehandler.FileHandleManager.reset()
+        self.mcp.state_watcher.shutdown()
 
     def reconfigure(self):
         config = {schema.MASTER_NAMESPACE: self._get_config(1, self.test_dir)}

tests/mcp_test.py

@@ -45,10 +45,9 @@ class MasterControlProgramTestCase(TestCase):
     def setup_mcp(self):
         self.working_dir    = tempfile.mkdtemp()
         self.config_path    = tempfile.mkdtemp()
-        self.mcp            = mcp.MasterControlProgram(
+        with mock.patch('tron.serialize.runstate.statemanager.StateChangeWatcher', autospec=True):
+            self.mcp        = mcp.MasterControlProgram(
                                 self.working_dir, self.config_path)
-        self.mcp.state_watcher = mock.create_autospec(
-                                statemanager.StateChangeWatcher)
 
     @teardown
     def teardown_mcp(self):
@@ -134,11 +133,12 @@ class MasterControlProgramRestoreStateTestCase(TestCase):
     def setup_mcp(self):
         self.working_dir        = tempfile.mkdtemp()
         self.config_path        = tempfile.mkdtemp()
-        self.mcp                = mcp.MasterControlProgram(
-                                    self.working_dir, self.config_path)
-        self.mcp.jobs           = mock.create_autospec(job.JobCollection)
-        self.mcp.services       = mock.create_autospec(service.ServiceCollection)
-        self.mcp.state_watcher  = mock.create_autospec(statemanager.StateChangeWatcher)
+        with mock.patch('tron.serialize.runstate.statemanager.StateChangeWatcher', autospec=True):
+            self.mcp               = mcp.MasterControlProgram(
+                                        self.working_dir, self.config_path)
+            self.mcp.jobs          = mock.create_autospec(job.JobCollection)
+            self.mcp.services      = mock.create_autospec(service.ServiceCollection)
+            self.mcp.state_watcher = mock.create_autospec(statemanager.StateChangeWatcher)
 
     @teardown
     def teardown_mcp(self):

tests/serialize/runstate/statemanager_test.py

@@ -1,32 +1,15 @@
-import os
 import mock
+import contextlib
 from testify import TestCase, assert_equal, setup, run
 
 from tests.assertions import assert_raises
 from tests.testingutils import autospec_method
-from tron.config import schema
 from tron.serialize import runstate
-from tron.serialize.runstate.shelvestore import ShelveStateStore
 from tron.serialize.runstate.statemanager import PersistentStateManager, StateChangeWatcher
 from tron.serialize.runstate.statemanager import StateSaveBuffer
 from tron.serialize.runstate.statemanager import StateMetadata
 from tron.serialize.runstate.statemanager import PersistenceStoreError
 from tron.serialize.runstate.statemanager import VersionMismatchError
-from tron.serialize.runstate.statemanager import PersistenceManagerFactory
-
-
-class PersistenceManagerFactoryTestCase(TestCase):
-
-    def test_from_config_shelve(self):
-        thefilename = 'thefilename'
-        config = schema.ConfigState(
-            store_type='shelve', name=thefilename, buffer_size=0,
-            connection_details=None)
-        manager = PersistenceManagerFactory.from_config(config)
-        store = manager._impl
-        assert_equal(store.filename, config.name)
-        assert isinstance(store, ShelveStateStore)
-        os.unlink(thefilename)
 
 
 class StateMetadataTestCase(TestCase):
@@ -73,13 +56,19 @@ class PersistentStateManagerTestCase(TestCase):
 
     @setup
     def setup_manager(self):
-        self.store = mock.Mock()
-        self.store.build_key.side_effect = lambda t, i: '%s%s' % (t, i)
-        self.buffer = StateSaveBuffer(1)
-        self.manager = PersistentStateManager(self.store, self.buffer)
+        with mock.patch('tron.serialize.runstate.statemanager.ParallelStore', autospec=True) \
+        as self.store_patch:
+            self.store = self.store_patch.return_value
+            self.build_patch = mock.Mock(side_effect=lambda t, i: '%s%s' % (t, i))
+            self.store_patch.return_value.configure_mock(build_key=self.build_patch)
+            self.buffer = StateSaveBuffer(1)
+            self.manager = PersistentStateManager()
+            self.manager._buffer = self.buffer
 
     def test__init__(self):
-        assert_equal(self.manager._impl, self.store)
+        self.store_patch.assert_called_once_with()
+        self.build_patch.assert_called_once_with(runstate.MCP_STATE, StateMetadata.name)
+        assert_equal(self.manager.metadata_key, self.manager._impl.build_key(runstate.MCP_STATE, StateMetadata.name))
 
     def test_keys_for_items(self):
         names = ['namea', 'nameb']
@@ -137,14 +126,43 @@ def test_disabled_nested(self):
             pass
         assert not self.manager.enabled
 
+    def test_update_config_success(self):
+        new_config = mock.Mock(buffer_size=5)
+        self.store.load_config.configure_mock(return_value=True)
+        with contextlib.nested(
+            mock.patch.object(self.manager, '_save_from_buffer'),
+            mock.patch('tron.serialize.runstate.statemanager.StateSaveBuffer', autospec=True)
+        ) as (save_patch, buffer_patch):
+            assert_equal(self.manager.update_from_config(new_config), True)
+            save_patch.assert_called_once_with()
+            self.store.load_config.assert_called_once_with(new_config)
+            buffer_patch.assert_called_once_with(new_config.buffer_size)
+
+    def test_update_config_failure(self):
+        new_config = mock.Mock(buffer_size=5)
+        self.store.load_config.configure_mock(return_value=False)
+        with contextlib.nested(
+            mock.patch.object(self.manager, '_save_from_buffer'),
+            mock.patch('tron.serialize.runstate.statemanager.StateSaveBuffer', autospec=True)
+        ) as (save_patch, buffer_patch):
+            assert_equal(self.manager.update_from_config(new_config), False)
+            save_patch.assert_called_once_with()
+            self.store.load_config.assert_called_once_with(new_config)
+            assert not buffer_patch.called
+
 
 class StateChangeWatcherTestCase(TestCase):
 
     @setup
     def setup_watcher(self):
-        self.watcher = StateChangeWatcher()
-        self.state_manager = mock.create_autospec(PersistentStateManager)
-        self.watcher.state_manager = self.state_manager
+        with mock.patch('tron.serialize.runstate.statemanager.PersistentStateManager', autospec=True) \
+        as self.persistence_patch:
+            self.watcher = StateChangeWatcher()
+            self.state_manager = mock.create_autospec(PersistentStateManager)
+            self.watcher.state_manager = self.state_manager
+
+    def test__init__(self):
+        self.persistence_patch.assert_called_once_with()
 
     def test_update_from_config_no_change(self):
         self.watcher.config = state_config = mock.Mock()
@@ -153,17 +171,39 @@ def test_update_from_config_no_change(self):
         assert_equal(self.watcher.state_manager, self.state_manager)
         assert not self.watcher.shutdown.mock_calls
 
-    @mock.patch('tron.serialize.runstate.statemanager.PersistenceManagerFactory',
-    autospec=True)
-    def test_update_from_config_changed(self, mock_factory):
-        state_config = mock.Mock()
-        autospec_method(self.watcher.shutdown)
+    def test_update_from_config_success(self):
+        state_config = mock.Mock(store_type="shelve")
         assert self.watcher.update_from_config(state_config)
         assert_equal(self.watcher.config, state_config)
-        self.watcher.shutdown.assert_called_with()
-        assert_equal(self.watcher.state_manager,
-            mock_factory.from_config.return_value)
-        mock_factory.from_config.assert_called_with(state_config)
+        self.state_manager.update_from_config.assert_called_once_with(state_config)
+
+    def test_update_from_config_failure_same_config(self):
+        state_config = self.watcher.config
+        assert not self.watcher.update_from_config(state_config)
+        assert_equal(self.watcher.config, state_config)
+        assert not self.state_manager.update_from_config.called
+
+    def test_update_from_config_failure_from_state_manager(self):
+        self.state_manager.update_from_config.configure_mock(return_value=False)
+        state_config = self.watcher.config
+        fake_config = mock.Mock(store_type="shelve")
+        assert not self.watcher.update_from_config(fake_config)
+        assert_equal(self.watcher.config, state_config)
+        self.state_manager.update_from_config.assert_called_once_with(fake_config)
+
+    def test_update_from_config_failure_bad_store_type(self):
+        state_config = self.watcher.config
+        fake_config = mock.Mock(store_type="hue_hue_hue")
+        assert_raises(PersistenceStoreError, self.watcher.update_from_config, fake_config)
+        assert_equal(self.watcher.config, state_config)
+        assert not self.state_manager.update_from_config.called
+
+    def test_update_from_config_failure_bad_db_type(self):
+        state_config = self.watcher.config
+        fake_config = mock.Mock(store_type="sql", store_method="make_it_rain")
+        assert_raises(PersistenceStoreError, self.watcher.update_from_config, fake_config)
+        assert_equal(self.watcher.config, state_config)
+        assert not self.state_manager.update_from_config.called
 
     def test_save_job(self):
         mock_job = mock.Mock()