Python utils.wait_task函数代码示例

OGeek|极客世界-中国程序员成长平台 › 门户 › 编程› Python›Python编程经验

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

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

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

示例1: test_tasks

    def test_tasks(self):
        id1 = AsyncTask('/plugins/ginger/tasks/1', self._async_op).id
        id2 = AsyncTask('/plugins/ginger/tasks/2', self._except_op).id
        id3 = AsyncTask('/plugins/ginger/tasks/3', self._intermid_op).id

        target_uri = urllib2.quote('^/plugins/ginger/tasks/*', safe="")
        filter_data = 'status=running&target_uri=%s' % target_uri
        tasks = json.loads(
            self.request('/plugins/ginger/tasks?%s' % filter_data).read()
        )
        self.assertEquals(3, len(tasks))

        tasks = json.loads(self.request('/plugins/ginger/tasks').read())
        tasks_ids = [t['id'] for t in tasks]
        self.assertEquals(set([id1, id2, id3]) - set(tasks_ids), set([]))
        wait_task(self._task_lookup, id2)
        foo2 = json.loads(
            self.request('/plugins/ginger/tasks/%s' % id2).read()
        )
        keys = ['id', 'status', 'message', 'target_uri']
        self.assertEquals(sorted(keys), sorted(foo2.keys()))
        self.assertEquals('failed', foo2['status'])
        wait_task(self._task_lookup, id3)
        foo3 = json.loads(
            self.request('/plugins/ginger/tasks/%s' % id3).read()
        )
        self.assertEquals('in progress', foo3['message'])
        self.assertEquals('running', foo3['status'])

开发者ID:kimchi-project，项目名称:ginger，代码行数:28，代码来源:test_tasks.py

示例2: add_vm

 def add_vm(name):
     # Create a VM
     req = json.dumps({'name': name,
                       'template': '/plugins/kimchi/templates/test'})
     task = json.loads(request(host, ssl_port, '/plugins/kimchi/vms',
                       req, 'POST').read())
     wait_task(model.task_lookup, task['id'])

开发者ID:Finn10111，项目名称:kimchi，代码行数:7，代码来源:test_mockmodel.py

示例3: test_vm_info

    def test_vm_info(self):
        model.templates_create({'name': u'test',
                                'cdrom': fake_iso})
        task = model.vms_create({'name': u'test-vm',
                                 'template': '/plugins/kimchi/templates/test'})
        wait_task(model.task_lookup, task['id'])
        vms = model.vms_get_list()
        self.assertEquals(2, len(vms))
        self.assertIn(u'test-vm', vms)

        keys = set(('name', 'state', 'stats', 'uuid', 'memory', 'cpu_info',
                    'screenshot', 'icon', 'graphics', 'users', 'groups',
                    'access', 'persistent'))

        stats_keys = set(('cpu_utilization', 'mem_utilization',
                          'net_throughput', 'net_throughput_peak',
                          'io_throughput', 'io_throughput_peak'))

        info = model.vm_lookup(u'test-vm')
        self.assertEquals(keys, set(info.keys()))
        self.assertEquals('shutoff', info['state'])
        self.assertEquals('test-vm', info['name'])
        self.assertEquals(get_template_default('old', 'memory'),
                          info['memory'])
        self.assertEquals(1, info['cpu_info']['vcpus'])
        self.assertEquals(1, info['cpu_info']['maxvcpus'])
        self.assertEquals('plugins/kimchi/images/icon-vm.png', info['icon'])
        self.assertEquals(stats_keys, set(info['stats'].keys()))
        self.assertEquals('vnc', info['graphics']['type'])
        self.assertEquals('127.0.0.1', info['graphics']['listen'])

开发者ID:Finn10111，项目名称:kimchi，代码行数:30，代码来源:test_mockmodel.py

示例4: test_screenshot_refresh

    def test_screenshot_refresh(self):
        # Create a VM
        req = json.dumps({'name': 'test', 'cdrom': fake_iso})
        request(host, ssl_port, '/plugins/kimchi/templates', req, 'POST')
        req = json.dumps({'name': 'test-vm',
                          'template': '/plugins/kimchi/templates/test'})
        resp = request(host, ssl_port, '/plugins/kimchi/vms', req, 'POST')
        task = json.loads(resp.read())
        wait_task(model.task_lookup, task['id'])

        # Test screenshot refresh for running vm
        request(host, ssl_port, '/plugins/kimchi/vms/test-vm/start', '{}',
                'POST')
        resp = request(host, ssl_port,
                       '/plugins/kimchi/vms/test-vm/screenshot')
        self.assertEquals(200, resp.status)
        self.assertEquals('image/png', resp.getheader('content-type'))
        resp1 = request(host, ssl_port, '/plugins/kimchi/vms/test-vm')
        rspBody = resp1.read()
        testvm_Data = json.loads(rspBody)
        screenshotURL = '/' + testvm_Data['screenshot']
        time.sleep(5)
        resp2 = request(host, ssl_port, screenshotURL)
        self.assertEquals(200, resp2.status)
        self.assertEquals(resp2.getheader('content-type'),
                          resp.getheader('content-type'))
        self.assertEquals(resp2.getheader('content-length'),
                          resp.getheader('content-length'))
        self.assertEquals(resp2.getheader('last-modified'),
                          resp.getheader('last-modified'))

开发者ID:Finn10111，项目名称:kimchi，代码行数:30，代码来源:test_mockmodel.py

示例5: test_debugreport_download

 def test_debugreport_download(self):
     req = json.dumps({'name': 'test_rest_report1'})
     with RollbackContext() as rollback:
         resp = request('/plugins/gingerbase/debugreports', req, 'POST')
         self.assertEquals(202, resp.status)
         task = json.loads(resp.read())
         # make sure the debugreport doesn't exist until the
         # the task is finished
         wait_task(self._task_lookup, task['id'], 20)
         rollback.prependDefer(self._report_delete, 'test_rest_report1')
         resp = request(
             '/plugins/gingerbase/debugreports/test_rest_report1'
         )
         debugreport = json.loads(resp.read())
         self.assertEquals("test_rest_report1", debugreport['name'])
         self.assertEquals(200, resp.status)
         resp = request(
             '/plugins/gingerbase/debugreports/test_rest_report1/content'
         )
         self.assertEquals(200, resp.status)
         resp = request(
             '/plugins/gingerbase/debugreports/test_rest_report1'
         )
         debugre = json.loads(resp.read())
         resp = request('/' + debugre['uri'])
         self.assertEquals(200, resp.status)

开发者ID:kimchi-project，项目名称:gingerbase，代码行数:26，代码来源:test_rest.py

示例6: test_vm_livemigrate_persistent_API

    def test_vm_livemigrate_persistent_API(self):
        patch_auth()

        inst = model.Model(libvirt_uri='qemu:///system',
                           objstore_loc=self.tmp_store)

        host = '127.0.0.1'
        port = get_free_port('http')
        ssl_port = get_free_port('https')
        cherrypy_port = get_free_port('cherrypy_port')

        with RollbackContext() as rollback:
            test_server = run_server(host, port, ssl_port, test_mode=True,
                                     cherrypy_port=cherrypy_port, model=inst)
            rollback.prependDefer(test_server.stop)

            self.request = partial(request, host, ssl_port)

            self.create_vm_test()
            rollback.prependDefer(rollback_wrapper, self.inst.vm_delete,
                                  u'test_vm_migrate')

            # removing cdrom because it is not shared storage and will make
            # the migration fail
            dev_list = self.inst.vmstorages_get_list('test_vm_migrate')
            self.inst.vmstorage_delete('test_vm_migrate',  dev_list[0])

            try:
                self.inst.vm_start('test_vm_migrate')
            except Exception, e:
                self.fail('Failed to start the vm, reason: %s' % e.message)

            migrate_url = "/plugins/kimchi/vms/%s/migrate" % 'test_vm_migrate'

            req = json.dumps({'remote_host': KIMCHI_LIVE_MIGRATION_TEST,
                             'user': 'root'})
            resp = self.request(migrate_url, req, 'POST')
            self.assertEquals(202, resp.status)
            task = json.loads(resp.read())
            wait_task(self._task_lookup, task['id'])
            task = json.loads(
                self.request(
                    '/plugins/kimchi/tasks/%s' % task['id'],
                    '{}'
                ).read()
            )
            self.assertEquals('finished', task['status'])

            try:
                remote_conn = self.get_remote_conn()
                rollback.prependDefer(remote_conn.close)
                remote_vm = remote_conn.lookupByName('test_vm_migrate')
                self.assertTrue(remote_vm.isPersistent())
                remote_vm.destroy()
                remote_vm.undefine()
            except Exception, e:
                self.fail('Migration test failed: %s' % e.message)

开发者ID:Finn10111，项目名称:kimchi，代码行数:57，代码来源:test_livemigration.py

示例7: test_packages_update

    def test_packages_update(self):
        def _task_lookup(taskid):
            return json.loads(self.request('/plugins/gingerbase/tasks/%s' %
                                           taskid).read())

        resp = self.request('/plugins/gingerbase/host/packagesupdate',
                            None, 'GET')
        pkgs = json.loads(resp.read())
        self.assertEquals(5, len(pkgs))

        pkg_keys = ['package_name', 'repository', 'arch', 'version', 'depends']
        for p in pkgs:
            name = p['package_name']
            resp = self.request('/plugins/gingerbase/host/packagesupdate/' +
                                name, None, 'GET')
            info = json.loads(resp.read())
            self.assertEquals(sorted(pkg_keys), sorted(info.keys()))

        # Test system update of specific package. Since package 'ginger' has
        # 'wok' as dependency, both packages must be selected to be updated
        # and, in the end of the process, we have only 3 packages to update.
        uri = '/plugins/gingerbase/host/packagesupdate/ginger/upgrade'
        resp = self.request(uri, '{}', 'POST')
        task = json.loads(resp.read())
        task_params = [u'id', u'message', u'status', u'target_uri']
        self.assertEquals(sorted(task_params), sorted(task.keys()))

        resp = self.request('/plugins/gingerbase/tasks/' + task[u'id'],
                            None, 'GET')
        task_info = json.loads(resp.read())
        self.assertEquals(task_info['status'], 'running')
        wait_task(_task_lookup, task_info['id'])
        resp = self.request('/plugins/gingerbase/tasks/' + task[u'id'],
                            None, 'GET')
        task_info = json.loads(resp.read())
        self.assertEquals(task_info['status'], 'finished')
        self.assertIn(u'All packages updated', task_info['message'])
        pkgs = model.packagesupdate_get_list()
        self.assertEquals(3, len(pkgs))

        # test system update of the rest of packages
        resp = self.request('/plugins/gingerbase/host/swupdate', '{}', 'POST')
        task = json.loads(resp.read())
        task_params = [u'id', u'message', u'status', u'target_uri']
        self.assertEquals(sorted(task_params), sorted(task.keys()))

        resp = self.request('/tasks/' + task[u'id'], None, 'GET')
        task_info = json.loads(resp.read())
        self.assertEquals(task_info['status'], 'running')
        wait_task(_task_lookup, task_info['id'])
        resp = self.request('/tasks/' + task[u'id'], None, 'GET')
        task_info = json.loads(resp.read())
        self.assertEquals(task_info['status'], 'finished')
        self.assertIn(u'All packages updated', task_info['message'])
        pkgs = model.packagesupdate_get_list()
        self.assertEquals(0, len(pkgs))

开发者ID:pvital，项目名称:gingerbase，代码行数:56，代码来源:test_host.py

示例8: test_memory_window_changes

    def test_memory_window_changes(self):
        model.templates_create({'name': u'test',
                                'source_media': fake_iso})
        task = model.vms_create({'name': u'test-vm',
                                 'template': '/plugins/kimchi/templates/test'})
        wait_task(model.task_lookup, task['id'])

        info = model.device_lookup('pci_0000_1a_00_0')
        model.vmhostdevs_update_mmio_guest(u'test-vm', True)
        model._attach_device(u'test-vm',
                             model._get_pci_device_xml(info, 0, False))

开发者ID:cclauss，项目名称:kimchi，代码行数:11，代码来源:test_mockmodel.py

示例9: test_vm_livemigrate_persistent_API

    def test_vm_livemigrate_persistent_API(self):
        patch_auth()
        with RollbackContext() as rollback:
            test_server = run_server()
            rollback.prependDefer(test_server.stop)

            self.request = partial(request)

            self.create_vm_test()
            rollback.prependDefer(
                rollback_wrapper, self.inst.vm_delete, 'test_vm_migrate'
            )

            # removing cdrom because it is not shared storage and will make
            # the migration fail
            dev_list = self.inst.vmstorages_get_list('test_vm_migrate')
            self.inst.vmstorage_delete('test_vm_migrate', dev_list[0])

            try:
                self.inst.vm_start('test_vm_migrate')
            except Exception as e:
                self.fail('Failed to start the vm, reason: %s' % e.message)

            migrate_url = '/plugins/kimchi/vms/%s/migrate' % 'test_vm_migrate'

            req = json.dumps(
                {'remote_host': KIMCHI_LIVE_MIGRATION_TEST, 'user': 'root'}
            )
            resp = self.request(migrate_url, req, 'POST')
            self.assertEqual(202, resp.status)
            task = json.loads(resp.read())
            wait_task(self._task_lookup, task['id'])
            task = json.loads(
                self.request('/plugins/kimchi/tasks/%s' %
                             task['id'], '{}').read()
            )
            self.assertEqual('finished', task['status'])

            try:
                remote_conn = self.get_remote_conn()
                rollback.prependDefer(remote_conn.close)
                remote_vm = remote_conn.lookupByName('test_vm_migrate')
                self.assertTrue(remote_vm.isPersistent())
                remote_vm.destroy()
                remote_vm.undefine()
            except Exception as e:
                self.fail('Migration test failed: %s' % e.message)

开发者ID:alinefm，项目名称:kimchi，代码行数:47，代码来源:test_livemigration.py

示例10: test_hotplug_3D_card

    def test_hotplug_3D_card(self):
        model.templates_create({'name': u'test',
                                'source_media': fake_iso})
        task = model.vms_create({'name': u'test-vm',
                                 'template': '/plugins/kimchi/templates/test'})
        wait_task(model.task_lookup, task['id'])
        model.vm_start(u'test-vm')

        # attach the 3D cards found to a running guest
        all_devices = model.devices_get_list()
        for device in all_devices:
            device_info = model.device_lookup(device)
            if model.device_is_device_3D_controller(device_info):
                try:
                    model.vmhostdevs_create(u'test-vm', {'name': device})

                # expect the error: KCHVMHDEV0006E
                except InvalidOperation as e:
                    self.assertEqual(e.message[:14], u'KCHVMHDEV0006E')

开发者ID:cclauss，项目名称:kimchi，代码行数:19，代码来源:test_mockmodel.py

示例11: test_create_debugreport

 def test_create_debugreport(self):
     req = json.dumps({'name': 'report1'})
     with RollbackContext() as rollback:
         resp = request(host, ssl_port, '/plugins/gingerbase/debugreports',
                        req, 'POST')
         self.assertEquals(202, resp.status)
         task = json.loads(resp.read())
         # make sure the debugreport doesn't exist until the
         # the task is finished
         wait_task(self._task_lookup, task['id'])
         rollback.prependDefer(self._report_delete, 'report2')
         resp = request(host, ssl_port,
                        '/plugins/gingerbase/debugreports/report1')
         debugreport = json.loads(resp.read())
         self.assertEquals("report1", debugreport['name'])
         self.assertEquals(200, resp.status)
         req = json.dumps({'name': 'report2'})
         resp = request(host, ssl_port,
                        '/plugins/gingerbase/debugreports/report1',
                        req, 'PUT')
         self.assertEquals(303, resp.status)

开发者ID:Clevero，项目名称:gingerbase，代码行数:21，代码来源:test_rest.py

示例12: test_host_actions

    def test_host_actions(self):
        def _task_lookup(taskid):
            return json.loads(self.request('/plugins/gingerbase/tasks/%s' %
                                           taskid).read())

        resp = self.request('/plugins/gingerbase/host/shutdown', '{}', 'POST')
        self.assertEquals(200, resp.status)
        resp = self.request('/plugins/gingerbase/host/reboot', '{}', 'POST')
        self.assertEquals(200, resp.status)

        # Test system update
        resp = self.request('/plugins/gingerbase/host/packagesupdate',
                            None, 'GET')
        pkgs = json.loads(resp.read())
        self.assertEquals(3, len(pkgs))

        pkg_keys = ['package_name', 'repository', 'arch', 'version']
        for p in pkgs:
            name = p['package_name']
            resp = self.request('/plugins/gingerbase/host/packagesupdate/' +
                                name, None, 'GET')
            info = json.loads(resp.read())
            self.assertEquals(sorted(pkg_keys), sorted(info.keys()))

        resp = self.request('/plugins/gingerbase/host/swupdate', '{}', 'POST')
        task = json.loads(resp.read())
        task_params = [u'id', u'message', u'status', u'target_uri']
        self.assertEquals(sorted(task_params), sorted(task.keys()))

        resp = self.request('/tasks/' + task[u'id'], None, 'GET')
        task_info = json.loads(resp.read())
        self.assertEquals(task_info['status'], 'running')
        wait_task(_task_lookup, task_info['id'])
        resp = self.request('/tasks/' + task[u'id'], None, 'GET')
        task_info = json.loads(resp.read())
        self.assertEquals(task_info['status'], 'finished')
        self.assertIn(u'All packages updated', task_info['message'])
        pkgs = model.packagesupdate_get_list()
        self.assertEquals(0, len(pkgs))

开发者ID:frediz，项目名称:gingerbase，代码行数:39，代码来源:test_host.py

示例13: _do_volume_test

def _do_volume_test(self, model, host, ssl_port, pool_name):
    def _task_lookup(taskid):
        return json.loads(
            self.request('/plugins/kimchi/tasks/%s' % taskid).read()
        )

    uri = '/plugins/kimchi/storagepools/%s/storagevolumes' \
          % pool_name.encode('utf-8')
    resp = self.request(uri)
    self.assertEquals(200, resp.status)

    resp = self.request('/plugins/kimchi/storagepools/%s' %
                        pool_name.encode('utf-8'))
    pool_info = json.loads(resp.read())
    with RollbackContext() as rollback:
        # Create storage volume with 'capacity'
        vol = 'test-volume'
        vol_uri = uri + '/' + vol
        req = json.dumps({'name': vol, 'format': 'raw',
                          'capacity': 1073741824})  # 1 GiB
        resp = self.request(uri, req, 'POST')
        if pool_info['type'] in READONLY_POOL_TYPE:
            self.assertEquals(400, resp.status)
        else:
            rollback.prependDefer(rollback_wrapper, model.storagevolume_delete,
                                  pool_name, vol)
            self.assertEquals(202, resp.status)
            task_id = json.loads(resp.read())['id']
            wait_task(_task_lookup, task_id)
            status = json.loads(
                self.request('/plugins/kimchi/tasks/%s' % task_id).read()
            )
            self.assertEquals('finished', status['status'])
            vol_info = json.loads(self.request(vol_uri).read())
            vol_info['name'] = vol
            vol_info['format'] = 'raw'
            vol_info['capacity'] = 1073741824

            # Resize the storage volume: increase its capacity to 2 GiB
            req = json.dumps({'size': 2147483648})  # 2 GiB
            resp = self.request(vol_uri + '/resize', req, 'POST')
            self.assertEquals(200, resp.status)
            storagevolume = json.loads(self.request(vol_uri).read())
            self.assertEquals(2147483648, storagevolume['capacity'])

            # Resize the storage volume: decrease its capacity to 512 MiB
            # This test case may fail if libvirt does not include the fix for
            # https://bugzilla.redhat.com/show_bug.cgi?id=1021802
            req = json.dumps({'size': 536870912})  # 512 MiB
            resp = self.request(vol_uri + '/resize', req, 'POST')
            self.assertEquals(200, resp.status)
            storagevolume = json.loads(self.request(vol_uri).read())
            self.assertEquals(536870912, storagevolume['capacity'])

            # Wipe the storage volume
            resp = self.request(vol_uri + '/wipe', '{}', 'POST')
            self.assertEquals(200, resp.status)
            storagevolume = json.loads(self.request(vol_uri).read())
            self.assertEquals(0, storagevolume['allocation'])

            # Clone the storage volume
            vol_info = json.loads(self.request(vol_uri).read())
            resp = self.request(vol_uri + '/clone', '{}', 'POST')
            self.assertEquals(202, resp.status)
            task = json.loads(resp.read())
            cloned_vol_name = task['target_uri'].split('/')[-2]
            rollback.prependDefer(model.storagevolume_delete, pool_name,
                                  cloned_vol_name)
            wait_task(_task_lookup, task['id'])
            task = json.loads(
                self.request('/plugins/kimchi/tasks/%s' % task['id']).read()
            )
            self.assertEquals('finished', task['status'])
            resp = self.request(uri + '/' + cloned_vol_name.encode('utf-8'))

            self.assertEquals(200, resp.status)
            cloned_vol = json.loads(resp.read())

            self.assertNotEquals(vol_info['name'], cloned_vol['name'])
            self.assertNotEquals(vol_info['path'], cloned_vol['path'])
            for key in ['name', 'path', 'allocation']:
                del vol_info[key]
                del cloned_vol[key]

            self.assertEquals(vol_info, cloned_vol)

            # Delete the storage volume
            resp = self.request(vol_uri, '{}', 'DELETE')
            self.assertEquals(204, resp.status)
            resp = self.request(vol_uri)
            self.assertEquals(404, resp.status)

        # Storage volume upload
        # It is done through a sequence of POST and several PUT requests
        filename = 'COPYING.LGPL'
        filepath = os.path.join(paths.get_prefix(), filename)
        filesize = os.stat(filepath).st_size

        # Create storage volume for upload
        req = json.dumps({'name': filename, 'format': 'raw',
#.........这里部分代码省略.........

开发者ID:Pojen-Huang，项目名称:kimchi，代码行数:101，代码来源:test_model_storagevolume.py

示例14: test_nonroot_access

    def test_nonroot_access(self, validate_users):
        validate_users.return_value = True

        # Non-root users can not create or delete network (only get)
        resp = self.request('/plugins/kimchi/networks', '{}', 'GET')
        self.assertEqual(200, resp.status)
        resp = self.request('/plugins/kimchi/networks', '{}', 'POST')
        self.assertEqual(403, resp.status)
        resp = self.request(
            '/plugins/kimchi/networks/default/activate', '{}', 'POST')
        self.assertEqual(403, resp.status)
        resp = self.request('/plugins/kimchi/networks/default', '{}', 'DELETE')
        self.assertEqual(403, resp.status)

        # Non-root users can not create or delete storage pool (only get)
        resp = self.request('/plugins/kimchi/storagepools', '{}', 'GET')
        self.assertEqual(200, resp.status)
        resp = self.request('/plugins/kimchi/storagepools', '{}', 'POST')
        self.assertEqual(403, resp.status)
        resp = self.request(
            '/plugins/kimchi/storagepools/default/activate', '{}', 'POST'
        )
        self.assertEqual(403, resp.status)
        resp = self.request(
            '/plugins/kimchi/storagepools/default', '{}', 'DELETE')
        self.assertEqual(403, resp.status)

        # Non-root users can not update or delete a template
        # but he can get and create a new one
        resp = self.request('/plugins/kimchi/templates', '{}', 'GET')
        self.assertEqual(403, resp.status)
        req = json.dumps({'name': 'test', 'source_media': fake_iso})
        resp = self.request('/plugins/kimchi/templates', req, 'POST')
        self.assertEqual(403, resp.status)
        resp = self.request('/plugins/kimchi/templates/test', '{}', 'PUT')
        self.assertEqual(403, resp.status)
        resp = self.request('/plugins/kimchi/templates/test', '{}', 'DELETE')
        self.assertEqual(403, resp.status)

        # Non-root users can only get vms authorized to them
        model.templates_create(
            {'name': u'test', 'source_media': {'type': 'disk', 'path': fake_iso}}
        )

        task_info = model.vms_create(
            {'name': u'test-me', 'template': '/plugins/kimchi/templates/test'}
        )
        wait_task(model.task_lookup, task_info['id'])

        model.vm_update(u'test-me', {'users': ['user'], 'groups': []})

        task_info = model.vms_create(
            {'name': u'test-usera', 'template': '/plugins/kimchi/templates/test'}
        )
        wait_task(model.task_lookup, task_info['id'])

        non_root = list(set(model.users_get_list()) - set(['admin']))[0]
        model.vm_update(u'test-usera', {'users': [non_root], 'groups': []})

        task_info = model.vms_create(
            {'name': u'test-groupa', 'template': '/plugins/kimchi/templates/test'}
        )
        wait_task(model.task_lookup, task_info['id'])
        a_group = model.groups_get_list()[0]
        model.vm_update(u'test-groupa', {'groups': [a_group]})

        resp = self.request('/plugins/kimchi/vms', '{}', 'GET')
        self.assertEqual(200, resp.status)
        vms_data = json.loads(resp.read())
        self.assertEqual(
            [u'test-groupa', u'test-me'], sorted([v['name'] for v in vms_data])
        )
        resp = self.request('/plugins/kimchi/vms', req, 'POST')
        self.assertEqual(403, resp.status)

        # Create a vm using mockmodel directly to test Resource access
        task_info = model.vms_create(
            {'name': 'kimchi-test', 'template': '/plugins/kimchi/templates/test'}
        )
        wait_task(model.task_lookup, task_info['id'])
        resp = self.request('/plugins/kimchi/vms/kimchi-test', '{}', 'PUT')
        self.assertEqual(403, resp.status)
        resp = self.request('/plugins/kimchi/vms/kimchi-test', '{}', 'DELETE')
        self.assertEqual(403, resp.status)

        # Non-root users can only update VMs authorized by them
        resp = self.request('/plugins/kimchi/vms/test-me/start', '{}', 'POST')
        self.assertEqual(200, resp.status)
        resp = self.request(
            '/plugins/kimchi/vms/test-usera/start', '{}', 'POST')
        self.assertEqual(403, resp.status)

        model.template_delete('test')
        model.vm_delete('test-me')

开发者ID:alinefm，项目名称:kimchi，代码行数:94，代码来源:test_authorization.py

示例15: test_nonroot_access

    def test_nonroot_access(self):
        # Non-root users can not create or delete network (only get)
        resp = self.request("/plugins/kimchi/networks", "{}", "GET")
        self.assertEquals(200, resp.status)
        resp = self.request("/plugins/kimchi/networks", "{}", "POST")
        self.assertEquals(403, resp.status)
        resp = self.request("/plugins/kimchi/networks/default/activate", "{}", "POST")
        self.assertEquals(403, resp.status)
        resp = self.request("/plugins/kimchi/networks/default", "{}", "DELETE")
        self.assertEquals(403, resp.status)

        # Non-root users can not create or delete storage pool (only get)
        resp = self.request("/plugins/kimchi/storagepools", "{}", "GET")
        self.assertEquals(200, resp.status)
        resp = self.request("/plugins/kimchi/storagepools", "{}", "POST")
        self.assertEquals(403, resp.status)
        resp = self.request("/plugins/kimchi/storagepools/default/activate", "{}", "POST")
        self.assertEquals(403, resp.status)
        resp = self.request("/plugins/kimchi/storagepools/default", "{}", "DELETE")
        self.assertEquals(403, resp.status)

        # Non-root users can not update or delete a template
        # but he can get and create a new one
        resp = self.request("/plugins/kimchi/templates", "{}", "GET")
        self.assertEquals(403, resp.status)
        req = json.dumps({"name": "test", "source_media": fake_iso})
        resp = self.request("/plugins/kimchi/templates", req, "POST")
        self.assertEquals(403, resp.status)
        resp = self.request("/plugins/kimchi/templates/test", "{}", "PUT")
        self.assertEquals(403, resp.status)
        resp = self.request("/plugins/kimchi/templates/test", "{}", "DELETE")
        self.assertEquals(403, resp.status)

        # Non-root users can only get vms authorized to them
        model.templates_create({"name": u"test", "source_media": {"type": "disk", "path": fake_iso}})

        task_info = model.vms_create({"name": u"test-me", "template": "/plugins/kimchi/templates/test"})
        wait_task(model.task_lookup, task_info["id"])

        fake_user = get_fake_user()

        model.vm_update(u"test-me", {"users": [fake_user.keys()[0]], "groups": []})

        task_info = model.vms_create({"name": u"test-usera", "template": "/plugins/kimchi/templates/test"})
        wait_task(model.task_lookup, task_info["id"])

        non_root = list(set(model.users_get_list()) - set(["root"]))[0]
        model.vm_update(u"test-usera", {"users": [non_root], "groups": []})

        task_info = model.vms_create({"name": u"test-groupa", "template": "/plugins/kimchi/templates/test"})
        wait_task(model.task_lookup, task_info["id"])
        a_group = model.groups_get_list()[0]
        model.vm_update(u"test-groupa", {"groups": [a_group]})

        resp = self.request("/plugins/kimchi/vms", "{}", "GET")
        self.assertEquals(200, resp.status)
        vms_data = json.loads(resp.read())
        self.assertEquals([u"test-groupa", u"test-me"], sorted([v["name"] for v in vms_data]))
        resp = self.request("/plugins/kimchi/vms", req, "POST")
        self.assertEquals(403, resp.status)

        # Create a vm using mockmodel directly to test Resource access
        task_info = model.vms_create({"name": "kimchi-test", "template": "/plugins/kimchi/templates/test"})
        wait_task(model.task_lookup, task_info["id"])
        resp = self.request("/plugins/kimchi/vms/kimchi-test", "{}", "PUT")
        self.assertEquals(403, resp.status)
        resp = self.request("/plugins/kimchi/vms/kimchi-test", "{}", "DELETE")
        self.assertEquals(403, resp.status)

        # Non-root users can only update VMs authorized by them
        resp = self.request("/plugins/kimchi/vms/test-me/start", "{}", "POST")
        self.assertEquals(200, resp.status)
        resp = self.request("/plugins/kimchi/vms/test-usera/start", "{}", "POST")
        self.assertEquals(403, resp.status)

        model.template_delete("test")
        model.vm_delete("test-me")

开发者ID:kimchi-project，项目名称:kimchi，代码行数:77，代码来源:test_authorization.py

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

鲜花

握手

雷人

路过

鸡蛋

该文章已有0人参与评论

请发表评论

全部评论

专题导读

More+

10-27 六六分期app的软件客服如何联系？(六六分期

11-06 可心卡盟:win10系统火狐flash插件崩溃怎么

11-06 亲亲特价:怎么删除回收站图标

11-06 济南大学虚拟社区:鲁大师节能降温的具体办

11-06 xlueops.exe:无线网络安装向导

11-06 女斗合众国:win7系统cf与主机连接不稳定怎

11-06 0xc000022-[cf烟雾头]cf怎么调烟雾头

11-06 qizideyouhuo:应用程序无法正常启动0xc0000

11-06 ipz-185:win7系统vcf文件怎么打开

11-06 傻哥蹦迪:win10系统s4怎么打开usb调试

11-06 八神浩树gtaste:回收站清空了怎么恢复

11-06 妖尾之黑色守护:win10系统电脑没有1440x900

11-06 校园至尊魔王小说:win7系统浏览网页时字体

11-06 女斗合众国:win10系统访问共享文件夹提示请

11-06 tokyo hot n0654:恢复win7系统默认字体一招

11-06 雨酷仙境:设置win7系统转移临时文件夹腾出

11-06 阿穆纳伊之杖:win7系统开始菜单在右边还原

11-06 tunespotting:win10系统火狐flash插件总是

11-06 甘尔葛分析师：计谋网站seo关键词暴涨有什

11-06 蔡贵霖: 计谋网站seo关键词暴涨有什么秘密

11-06 博益网首页:ao3网页版进入不了解决方法

11-06 漏斗子专栏: 网站数据分析小白易懂精华篇

11-06 见证双虹怎么做:win7系统开启telnet命令的

11-06 颾狐蝶蜋:系统资源不足无法完成请求的服务

11-06 国光中学校歌:提交网站到alexa查询详细步骤

11-06 西安有情天:静态网页和动态网页的区别

11-06 红木雅尚斋:外部链接构造对网站的好处

11-06 前官礼遇：防止域名劫持–增强域安全性的10

11-06 密传二转答案: 中文分词算法有哪些

11-06 金泉家园邮编:百度快照劫持的表现及应对方

Python utils.wrap函数代码示例发布时间：2022-05-27

Python utils.test_create函数代码示例发布时间：2022-05-27

Python util.grid_equal函数代码示例

1 Python 入门教程

Python入门教程 Python 是一种解释型、面向对象、动态数据类型的高级程序设计语言。 P

阅读：13804|2022-01-22

2 Python wikiutil.getFrontPage函数代码示例

Python wikiutil.getFrontPage函数代码示例

阅读：10190|2022-05-24

3 Python 简介

Python 简介 Python 是一个高层次的结合了解释性、编译性、互动性和面向对象的脚本

阅读：4086|2022-01-22

4 Python tests.group函数代码示例

Python tests.group函数代码示例

阅读：4042|2022-05-27

5 Python util.check_if_user_has_permission

Python util.check_if_user_has_permission函数代码示例

阅读：3842|2022-05-27

6 Python 操练实例98

Python 练习实例98 Python 100例题目：从键盘输入一个字符串，将小写字母全部转换成大

阅读：3510|2022-01-22

7 Python 环境搭建

Python 环境搭建本章节我们将向大家介绍如何在本地搭建 Python 开发环境。 Py

阅读：3030|2022-01-22

8 Python output.darkgreen函数代码示例

Python output.darkgreen函数代码示例

阅读：2653|2022-05-25

9 Python 基础语法

Python 基础语法 Python 语言与 Perl，C 和 Java 等语言有许多相似之处。但是，也

阅读：2646|2022-01-22

10 Python 中文编码

Python 中文编码前面章节中我们已经学会了如何用 Python 输出 Hello, World!，英文没

阅读：2302|2022-01-22

客服电话

电子邮件

Python utils.wait_task函数代码示例

示例1: test_tasks

示例2: add_vm

示例3: test_vm_info

示例4: test_screenshot_refresh

示例5: test_debugreport_download

示例6: test_vm_livemigrate_persistent_API

示例7: test_packages_update

示例8: test_memory_window_changes

示例9: test_vm_livemigrate_persistent_API

示例10: test_hotplug_3D_card

示例11: test_create_debugreport

示例12: test_host_actions

示例13: _do_volume_test

示例14: test_nonroot_access

示例15: test_nonroot_access

请发表评论

全部评论

上一篇：

下一篇：

Python util.grid_equal函数代码示例

Python util.get_worker_name函数代码示例

Python util.get_webmention_target函数代

Python util.get_uuid函数代码示例

Python util.get_type_by_name函数代码示例

Python util.grid_equal函数代码示例

Python util.get_worker_name函数代码示例

Python util.get_webmention_target函数代

Python util.get_uuid函数代码示例

Python util.get_type_by_name函数代码示例

Python util.get_stdout函数代码示例

关于我们

产品与服务

解决方案

139-2527-9053