test_coordinator.py 文件源码

def test_save_state(self, leader_set, relation_set):
        c = coordinator.BaseCoordinator()
        unit = hookenv.local_unit()
        c.grants = {'directdump': True}
        c.requests = {unit: 'data1', 'foo/2': 'data2'}

        # grants is dumped to leadership settings, if the unit is leader.
        with patch.object(c, '_save_local_state') as save_loc:
            c._save_state()
            self.assertFalse(leader_set.called)
            hookenv.is_leader.return_value = True
            c._save_state()
            leader_set.assert_called_once_with({c.key: '{"directdump": true}'})

        # If there is no relation id, the local units requests is dumped
        # to a local stash.
        with patch.object(c, '_save_local_state') as save_loc:
            c._save_state()
            save_loc.assert_called_once_with('data1')

        # If there is a relation id, the local units requests is dumped
        # to the peer relation.
        with patch.object(c, '_save_local_state') as save_loc:
            c.relid = 'cluster:1'
            c._save_state()
            self.assertFalse(save_loc.called)
            relation_set.assert_called_once_with(
                c.relid, relation_settings={c.key: '"data1"'})  # JSON encoded