test_swift_utils.py 文件源码

def test_cluster_rpc_notify_leader_changed(self, mock_uuid):
        mock_uuid.uuid4.return_value = 'e4b67426-6cc0-4aa3-829d-227999cd0a75'
        rpc = swift_utils.SwiftProxyClusterRPC()
        rq = rpc.notify_leader_changed('token1')
        self.assertEqual({'trigger': 'e4b67426-6cc0-4aa3-829d-227999cd0a75',
                          'broker-token': None,
                          'builder-broker': None,
                          'broker-timestamp': None,
                          'peers-only': None,
                          'leader-changed-notification': 'token1',
                          'stop-proxy-service': None,
                          'stop-proxy-service-ack': None,
                          'resync-request': None,
                          'sync-only-builders': None}, rq)

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