test_swift_utils.py 文件源码

def test_cluster_rpc_sync_request(self, mock_uuid, mock_time):
        mock_time.time = mock.Mock(return_value=float(1.234))
        mock_uuid.uuid4.return_value = 'token2'
        rpc = swift_utils.SwiftProxyClusterRPC()
        rq = rpc.sync_rings_request('token1')
        self.assertEqual({'trigger': 'token2',
                          'broker-token': 'token1',
                          'broker-timestamp': '1.234000',
                          'builder-broker': '1.2.3.4',
                          'peers-only': None,
                          'leader-changed-notification': None,
                          'resync-request': None,
                          'stop-proxy-service': None,
                          'stop-proxy-service-ack': None,
                          'sync-only-builders': None}, rq)

        template_keys = set(rpc.template())
        self.assertTrue(set(rq.keys()).issubset(template_keys))