test_instance.py 文件源码

def test_remove_success(self):
        create_instance(self.ins)
        req = dict(
            container_serial=self.ins.get('container_serial'),
            user_name=test_cfg.USER_NAME
        )
        with mock.patch('pika.BlockingConnection') as mock_block,\
            mock.patch.object(pika.BlockingConnection,'channel') as mock_block_channel,\
            mock.patch.object(UiQueue, 'send') as mock_queue_send:
            worker_res = dict(
                code='0x1',
                message='pass',
                container_serial=self.ins.get('container_serial')
            )
            mock_block.return_value = mock.Mock()
            mock_block_channel.return_value = mock.Mock()
            mock_queue_send.return_value = json.dumps(worker_res)
            response = self.app.post('/remove_ins', data=json.dumps(req), follow_redirects=True)
            remove_instance_by_serial(self.ins.get('container_serial'))
            mock_queue_send.assert_called_once()
            res_dict = json.loads(response.data)
            self.assertEqual(res_dict.get('code'), '0x1')