• 设为首页
  • 点击收藏
  • 手机版
    手机扫一扫访问
    迪恩网络手机版
  • 关注官方公众号
    微信扫一扫关注
    迪恩网络公众号
登陆 注册
描述文字

Python callbacks.update_db函数代码示例

原作者: [db:作者] 来自: [db:来源] 收藏 邀请

本文整理汇总了Python中synnefo.logic.callbacks.update_db函数的典型用法代码示例。如果您正苦于以下问题：Python update_db函数的具体用法？Python update_db怎么用？Python update_db使用的例子？那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。



在下文中一共展示了update_db函数的20个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。

示例1: test_changed_nic

    def test_changed_nic(self, client):
        ip = mfactory.IPv4AddressFactory(subnet__cidr="10.0.0.0/24",
                                         address="10.0.0.2")
        network = ip.network
        subnet = ip.subnet
        vm = ip.nic.machine
        pool = subnet.get_ip_pools()[0]
        pool.reserve("10.0.0.2")
        pool.save()

        msg = self.create_msg(instance_nics=[{'network': network.backend_id,
                                              'ip': '10.0.0.3',
                                              'mac': 'aa:bb:cc:00:11:22',
                                              'name': ip.nic.backend_uuid}],
                              instance=vm.backend_vm_id)
        update_db(client, msg)
        self.assertTrue(client.basic_ack.called)
        db_vm = VirtualMachine.objects.get(id=vm.id)
        nics = db_vm.nics.all()
        self.assertEqual(len(nics), 1)
        self.assertEqual(nics[0].index, 0)
        self.assertEqual(nics[0].ipv4_address, '10.0.0.3')
        self.assertEqual(nics[0].mac, 'aa:bb:cc:00:11:22')
        pool = subnet.get_ip_pools()[0]
        self.assertTrue(pool.is_available('10.0.0.2'))
        self.assertFalse(pool.is_available('10.0.0.3'))
        pool.save()
开发者ID:antonis-m，项目名称:synnefo，代码行数:27，代码来源:callbacks.py


示例2: test_reboot

 def test_reboot(self, client):
     vm = mfactory.VirtualMachineFactory()
     msg = self.create_msg(operation='OP_INSTANCE_REBOOT',
                           instance=vm.backend_vm_id)
     update_db(client, msg)
     client.basic_ack.assert_called_once()
     db_vm = VirtualMachine.objects.get(id=vm.id)
     self.assertEqual(db_vm.operstate, 'STARTED')
开发者ID:cstavr，项目名称:synnefo，代码行数:8，代码来源:tests.py


示例3: test_create

 def test_create(self, client):
     vm = mfactory.VirtualMachineFactory()
     msg = self.create_msg(operation='OP_INSTANCE_CREATE',
                           instance=vm.backend_vm_id)
     update_db(client, msg)
     self.assertTrue(client.basic_ack.called)
     db_vm = VirtualMachine.objects.get(id=vm.id)
     self.assertEqual(db_vm.operstate, 'STARTED')
开发者ID:antonis-m，项目名称:synnefo，代码行数:8，代码来源:callbacks.py


示例4: test_stop

 def test_stop(self, client):
     vm = mfactory.VirtualMachineFactory()
     msg = self.create_msg(operation='OP_INSTANCE_SHUTDOWN',
                           instance=vm.backend_vm_id)
     with mocked_quotaholder():
         update_db(client, msg)
     self.assertTrue(client.basic_ack.called)
     db_vm = VirtualMachine.objects.get(id=vm.id)
     self.assertEqual(db_vm.operstate, 'STOPPED')
开发者ID:antonis-m，项目名称:synnefo，代码行数:9，代码来源:callbacks.py


示例5: test_create_error

 def test_create_error(self, client):
     """Test that error create sets vm to ERROR state"""
     vm = mfactory.VirtualMachineFactory()
     msg = self.create_msg(operation='OP_INSTANCE_CREATE',
                           instance=vm.backend_vm_id,
                           status='error')
     update_db(client, msg)
     self.assertTrue(client.basic_ack.called)
     db_vm = VirtualMachine.objects.get(id=vm.id)
     self.assertEqual(db_vm.operstate, 'ERROR')
开发者ID:antonis-m，项目名称:synnefo，代码行数:10，代码来源:callbacks.py


示例6: test_other_error

 def test_other_error(self, client):
     """Test that other error messages do no affect the VM"""
     vm = mfactory.VirtualMachineFactory()
     msg = self.create_msg(operation='OP_INSTANCE_STARTUP',
                           instance=vm.backend_vm_id,
                           status='error')
     update_db(client, msg)
     self.assertTrue(client.basic_ack.called)
     db_vm = VirtualMachine.objects.get(id=vm.id)
     self.assertEqual(db_vm.operstate, vm.operstate)
     self.assertEqual(db_vm.backendtime, vm.backendtime)
开发者ID:antonis-m，项目名称:synnefo，代码行数:11，代码来源:callbacks.py


示例7: test_no_nics

 def test_no_nics(self, client):
     vm = mfactory.VirtualMachineFactory(operstate='ERROR')
     mfactory.NetworkInterfaceFactory(machine=vm, state="ACTIVE")
     mfactory.NetworkInterfaceFactory(machine=vm, state="ACTIVE")
     mfactory.NetworkInterfaceFactory(machine=vm, state="ACTIVE")
     self.assertEqual(len(vm.nics.all()), 3)
     msg = self.create_msg(instance_nics=[],
                           instance=vm.backend_vm_id)
     update_db(client, msg)
     self.assertTrue(client.basic_ack.called)
     db_vm = VirtualMachine.objects.get(id=vm.id)
     self.assertEqual(len(db_vm.nics.all()), 0)
开发者ID:antonis-m，项目名称:synnefo，代码行数:12，代码来源:callbacks.py


示例8: test_remove

 def test_remove(self, client):
     vm = mfactory.VirtualMachineFactory(flavor__cpu=1, flavor__ram=128)
     mfactory.VolumeFactory(userid=vm.userid, machine=vm, size=1)
     mfactory.VolumeFactory(userid=vm.userid, machine=vm, size=3)
     # Also create a NIC
     ip = mfactory.IPv4AddressFactory(nic__machine=vm)
     nic = ip.nic
     nic.network.get_ip_pools()[0].reserve(nic.ipv4_address)
     msg = self.create_msg(operation='OP_INSTANCE_REMOVE',
                           instance=vm.backend_vm_id)
     with mocked_quotaholder() as m:
         update_db(client, msg)
     self.assertTrue(client.basic_ack.called)
     db_vm = VirtualMachine.objects.get(id=vm.id)
     self.assertEqual(db_vm.operstate, 'DESTROYED')
     self.assertTrue(db_vm.deleted)
     # Check that nics are deleted
     self.assertFalse(db_vm.nics.all())
     self.assertTrue(nic.network.get_ip_pools()[0].is_available(ip.address))
     # Check that volumes are deleted
     self.assertFalse(db_vm.volumes.filter(deleted=False))
     # Check quotas
     name, args, kwargs = m.mock_calls[0]
     for (userid, res), value in args[1].items():
         if res == 'cyclades.disk':
             self.assertEqual(value, -4 << 30)
         elif res == 'cyclades.cpu':
             self.assertEqual(value, -1)
         elif res == 'cyclades.ram':
             self.assertEqual(value, -128 << 20)
     vm2 = mfactory.VirtualMachineFactory()
     fp1 = mfactory.IPv4AddressFactory(nic__machine=vm2, floating_ip=True,
                                       network__floating_ip_pool=True)
     network = fp1.network
     nic1 = mfactory.NetworkInterfaceFactory(machine=vm2)
     fp1.nic = nic1
     fp1.save()
     pool = network.get_ip_pools()[0]
     pool.reserve(fp1.address)
     pool.save()
     msg = self.create_msg(operation='OP_INSTANCE_REMOVE',
                           instance=vm2.backend_vm_id)
     with mocked_quotaholder():
         update_db(client, msg)
     client.basic_ack.assert_called_once()
     db_vm = VirtualMachine.objects.get(id=vm.id)
     self.assertEqual(db_vm.operstate, 'DESTROYED')
     self.assertTrue(db_vm.deleted)
     self.assertEqual(IPAddress.objects.get(id=fp1.id).nic, None)
     pool = network.get_ip_pools()[0]
     # Test that floating ips are not released
     self.assertFalse(pool.is_available(fp1.address))
开发者ID:Erethon，项目名称:synnefo，代码行数:52，代码来源:callbacks.py


示例9: test_remove

 def test_remove(self, client):
     vm = mfactory.VirtualMachineFactory()
     # Also create a NIC
     mfactory.NetworkInterfaceFactory(machine=vm)
     msg = self.create_msg(operation='OP_INSTANCE_REMOVE',
                           instance=vm.backend_vm_id)
     with mocked_quotaholder():
         update_db(client, msg)
     client.basic_ack.assert_called_once()
     db_vm = VirtualMachine.objects.get(id=vm.id)
     self.assertEqual(db_vm.operstate, 'DESTROYED')
     self.assertTrue(db_vm.deleted)
     # Check that nics are deleted
     self.assertFalse(db_vm.nics.all())
开发者ID:cstavr，项目名称:synnefo，代码行数:14，代码来源:tests.py


示例10: test_old_msg

 def test_old_msg(self, client):
     from time import sleep
     from datetime import datetime
     old_time = time()
     sleep(0.01)
     new_time = datetime.fromtimestamp(time())
     vm = mfactory.VirtualMachineFactory(backendtime=new_time)
     vm.operstate = 'STOPPED'
     vm.save()
     msg = self.create_msg(operation='OP_INSTANCE_STARTUP',
                           event_time=split_time(old_time),
                           instance=vm.backend_vm_id)
     update_db(client, msg)
     self.assertTrue(client.basic_ack.called)
     db_vm = VirtualMachine.objects.get(id=vm.id)
     self.assertEquals(db_vm.operstate, "STOPPED")
     self.assertEquals(db_vm.backendtime, new_time)
开发者ID:antonis-m，项目名称:synnefo，代码行数:17，代码来源:callbacks.py


示例11: test_empty_nic

 def test_empty_nic(self, client):
     vm = mfactory.VirtualMachineFactory(operstate='ERROR')
     for public in [True, False]:
         net = mfactory.NetworkFactory(public=public)
         msg = self.create_msg(nics=[{'network': net.backend_id}],
                               instance=vm.backend_vm_id)
         update_db(client, msg)
         client.basic_ack.assert_called_once()
         db_vm = VirtualMachine.objects.get(id=vm.id)
         nics = db_vm.nics.all()
         self.assertEqual(len(nics), 1)
         self.assertEqual(nics[0].index, 0)
         self.assertEqual(nics[0].ipv4, '')
         self.assertEqual(nics[0].ipv6, '')
         self.assertEqual(nics[0].mac, '')
         if public:
             self.assertEqual(nics[0].firewall_profile,
                              settings.DEFAULT_FIREWALL_PROFILE)
         else:
             self.assertEqual(nics[0].firewall_profile, '')
开发者ID:cstavr，项目名称:synnefo，代码行数:20，代码来源:tests.py


示例12: test_error_snapshot

    def test_error_snapshot(self, pithos_backend, client):
        vm = mfactory.VirtualMachineFactory()
        disks = [
            (0, {"snapshot_info": json.dumps({"snapshot_id":
                                              "test_snapshot_id"})})
        ]
        msg = self.create_msg(operation='OP_INSTANCE_SNAPSHOT',
                              instance=vm.backend_vm_id,
                              job_fields={'disks': disks},
                              status="running")
        update_db(client, msg)
        self.assertEqual(pithos_backend().update_object_status.mock_calls, [])

        msg = self.create_msg(operation='OP_INSTANCE_SNAPSHOT',
                              instance=vm.backend_vm_id,
                              job_fields={'disks': disks},
                              event_time=split_time(time()),
                              status="error")
        update_db(client, msg)

        pithos_backend().update_object_status\
                        .assert_called_once_with("test_snapshot_id", state=-1)

        pithos_backend.reset_mock()
        msg = self.create_msg(operation='OP_INSTANCE_SNAPSHOT',
                              instance=vm.backend_vm_id,
                              job_fields={'disks': disks},
                              event_time=split_time(time()),
                              status="success")
        update_db(client, msg)
        pithos_backend().update_object_status\
                        .assert_called_once_with("test_snapshot_id", state=1)
开发者ID:Erethon，项目名称:synnefo，代码行数:32，代码来源:callbacks.py


示例13: test_full_nic

 def test_full_nic(self, client):
     vm = mfactory.VirtualMachineFactory(operstate='ERROR')
     net = mfactory.NetworkFactory(subnet='10.0.0.0/24')
     pool = net.get_pool()
     self.assertTrue(pool.is_available('10.0.0.22'))
     pool.save()
     msg = self.create_msg(nics=[{'network': net.backend_id,
                                  'ip': '10.0.0.22',
                                  'mac': 'aa:bb:cc:00:11:22'}],
                           instance=vm.backend_vm_id)
     update_db(client, msg)
     client.basic_ack.assert_called_once()
     db_vm = VirtualMachine.objects.get(id=vm.id)
     nics = db_vm.nics.all()
     self.assertEqual(len(nics), 1)
     self.assertEqual(nics[0].index, 0)
     self.assertEqual(nics[0].ipv4, '10.0.0.22')
     self.assertEqual(nics[0].ipv6, '')
     self.assertEqual(nics[0].mac, 'aa:bb:cc:00:11:22')
     pool = net.get_pool()
     self.assertFalse(pool.is_available('10.0.0.22'))
     pool.save()
开发者ID:cstavr，项目名称:synnefo，代码行数:22，代码来源:tests.py


示例14: test_remove

 def test_remove(self, client):
     vm = mfactory.VirtualMachineFactory()
     # Also create a NIC
     ip = mfactory.IPv4AddressFactory(nic__machine=vm)
     nic = ip.nic
     nic.network.get_ip_pools()[0].reserve(nic.ipv4_address)
     msg = self.create_msg(operation='OP_INSTANCE_REMOVE',
                           instance=vm.backend_vm_id)
     with mocked_quotaholder():
         update_db(client, msg)
     self.assertTrue(client.basic_ack.called)
     db_vm = VirtualMachine.objects.get(id=vm.id)
     self.assertEqual(db_vm.operstate, 'DESTROYED')
     self.assertTrue(db_vm.deleted)
     # Check that nics are deleted
     self.assertFalse(db_vm.nics.all())
     self.assertTrue(nic.network.get_ip_pools()[0].is_available(ip.address))
     vm2 = mfactory.VirtualMachineFactory()
     fp1 = mfactory.IPv4AddressFactory(nic__machine=vm2, floating_ip=True,
                                       network__floating_ip_pool=True)
     network = fp1.network
     nic1 = mfactory.NetworkInterfaceFactory(machine=vm2)
     fp1.nic = nic1
     fp1.save()
     pool = network.get_ip_pools()[0]
     pool.reserve(fp1.address)
     pool.save()
     msg = self.create_msg(operation='OP_INSTANCE_REMOVE',
                           instance=vm2.backend_vm_id)
     with mocked_quotaholder():
         update_db(client, msg)
     client.basic_ack.assert_called_once()
     db_vm = VirtualMachine.objects.get(id=vm.id)
     self.assertEqual(db_vm.operstate, 'DESTROYED')
     self.assertTrue(db_vm.deleted)
     self.assertEqual(IPAddress.objects.get(id=fp1.id).nic, None)
     pool = network.get_ip_pools()[0]
     # Test that floating ips are not released
     self.assertFalse(pool.is_available(fp1.address))
开发者ID:antonis-m，项目名称:synnefo，代码行数:39，代码来源:callbacks.py


示例15: test_remove_error

    def test_remove_error(self, rapi, client):
        vm = mfactory.VirtualMachineFactory()
        # Also create a NIC
        msg = self.create_msg(operation='OP_INSTANCE_REMOVE',
                              status="error",
                              instance=vm.backend_vm_id)
        rapi().GetInstance.return_value = {}
        update_db(client, msg)
        db_vm = VirtualMachine.objects.get(id=vm.id)
        self.assertFalse(db_vm.deleted)

        rapi().GetInstance.side_effect = GanetiApiError(msg="msg",
                                                        code=503)
        update_db(client, msg)
        db_vm = VirtualMachine.objects.get(id=vm.id)
        self.assertFalse(db_vm.deleted)

        rapi().GetInstance.side_effect = GanetiApiError(msg="msg",
                                                        code=404)
        with mocked_quotaholder():
            update_db(client, msg)
        db_vm = VirtualMachine.objects.get(id=vm.id)
        self.assertTrue(db_vm.deleted)
开发者ID:antonis-m，项目名称:synnefo，代码行数:23，代码来源:callbacks.py


示例16: test_unhandled_exception

 def test_unhandled_exception(self, client):
     update_db(client, None)
     self.assertEqual(1, client.basic_reject.call_count)
开发者ID:AthinaB，项目名称:synnefo，代码行数:3，代码来源:callbacks.py


示例17: test_wrong_type

 def test_wrong_type(self, client):
     msg = self.create_msg(type="WRONG_TYPE")
     update_db(client, msg)
     self.assertTrue(client.basic_nack.called)
开发者ID:antonis-m，项目名称:synnefo，代码行数:4，代码来源:callbacks.py


示例18: test_missing_instance

 def test_missing_instance(self, client):
     msg = self.create_msg(operation='OP_INSTANCE_STARTUP',
                           instance='foo')
     update_db(client, msg)
     self.assertTrue(client.basic_ack.called)
开发者ID:antonis-m，项目名称:synnefo，代码行数:5，代码来源:callbacks.py


示例19: test_unhandled_exception

 def test_unhandled_exception(self, client):
     update_db(client, {})
     client.basic_reject.assert_called_once()
开发者ID:antonis-m，项目名称:synnefo，代码行数:3，代码来源:callbacks.py


示例20: test_missing_attribute

 def test_missing_attribute(self, client):
     update_db(client, json.dumps({'body': {}}))
     self.assertTrue(client.basic_reject.called)
开发者ID:antonis-m，项目名称:synnefo，代码行数:3，代码来源:callbacks.py



注：本文中的synnefo.logic.callbacks.update_db函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。

鲜花
握手
雷人
路过
鸡蛋
该文章已有0人参与评论

请发表评论

全部评论

专题导读
More+
上一篇：
Python utils.render_to_string函数代码示例发布时间：2022-05-27
下一篇：
Python services.get_service_path函数代码示例发布时间：2022-05-27
热门推荐
More+
热门话题
More+
阅读排行榜

扫描微信二维码

查看手机版网站

随时了解更新最新资讯

139-2527-9053

在线客服（服务时间 9:00～18:00）

在线QQ客服
地址：深圳市南山区西丽大学城创智工业园
电邮：jeky_zhao#qq.com
移动电话：139-2527-9053

Powered by 互联科技 X3.4© 2001-2213 极客世界.|Sitemap