def test_upgrade_volatile_running_config(self):
        with namedTemporaryDir() as pdir, namedTemporaryDir() as vdir:
            with mock.patch.object(netconf, 'CONF_RUN_DIR', pdir),\
                    mock.patch.object(netconf, 'CONF_VOLATILE_RUN_DIR', vdir):

                vol_rconfig = netconf.RunningConfig(volatile=True)
                vol_rconfig.save()

                netupgrade.upgrade()

                pers_rconfig = netconf.RunningConfig()
                self.assertFalse(vol_rconfig.config_exists())
                self.assertTrue(pers_rconfig.config_exists())

    def _commonConvertExternalVM(self, url):
        with namedTemporaryDir() as v2v._V2V_DIR, \
                namedTemporaryDir() as v2v._LOG_DIR:
            v2v.convert_external_vm(url,
                                    'root',
                                    ProtectedPassword('mypassword'),
                                    self.vminfo,
                                    self.job_id,
                                    FakeIRS())
            job = v2v._jobs[self.job_id]
            job.wait()

            self.assertEqual(job.status, v2v.STATUS.DONE)

 def test_getblocksize(self):
     with namedTemporaryDir():
         metadata = {blockSD.DMDK_LOGBLKSIZE: 2048,
                     blockSD.DMDK_PHYBLKSIZE: 1024}
         manifest = self.make_manifest(metadata)
         self.assertEquals(2048, manifest.logBlkSize)
         self.assertEquals(1024, manifest.phyBlkSize)

 def test_getmetaparam(self):
     with namedTemporaryDir():
         metadata = {}
         manifest = self.make_manifest(metadata)
         metadata[sd.DMDK_SDUUID] = manifest.sdUUID
         self.assertEquals(manifest.sdUUID,
                           manifest.getMetaParam(sd.DMDK_SDUUID))

def VM(params=None, devices=None, runCpu=False,
       arch=cpuarch.X86_64, status=None,
       cif=None, create_device_objects=False,
       post_copy=None, recover=False):
    with namedTemporaryDir() as tmpDir:
        with MonkeyPatchScope([(constants, 'P_VDSM_RUN', tmpDir),
                               (libvirtconnection, 'get', Connection),
                               (containersconnection, 'get', Connection),
                               (vm.Vm, '_updateDomainDescriptor',
                                   _updateDomainDescriptor),
                               (vm.Vm, 'send_status_event',
                                   lambda _, **kwargs: None)]):
            vmParams = {'vmId': 'TESTING', 'vmName': 'nTESTING'}
            vmParams.update({} if params is None else params)
            cif = ClientIF() if cif is None else cif
            fake = vm.Vm(cif, vmParams, recover=recover)
            cif.vmContainer[fake.id] = fake
            fake.arch = arch
            fake.guestAgent = GuestAgent()
            fake.conf['devices'] = [] if devices is None else devices
            if create_device_objects:
                fake._devices = common.dev_map_from_dev_spec_map(
                    fake._devSpecMapFromConf(), fake.log
                )
            fake._guestCpuRunning = runCpu
            if status is not None:
                fake._lastStatus = status
            if post_copy is not None:
                fake._post_copy = post_copy
            sampling.stats_cache.add(fake.id)
            yield fake

 def test_full(self):
     with namedTemporaryDir() as tmpdir:
         filename = os.path.join(tmpdir, 'test')
         with io.open(filename, "wb") as f:
             f.write(b"x" * qcow2.CLUSTER_SIZE * 3)
         runs = qemuimg.map(filename)
         self.assertEqual(qcow2.count_clusters(runs), 3)

 def test_empty_sparse(self):
     with namedTemporaryDir() as tmpdir:
         filename = os.path.join(tmpdir, 'test')
         with io.open(filename, "wb") as f:
             f.truncate(MB)
         runs = qemuimg.map(filename)
         self.assertEqual(qcow2.count_clusters(runs), 0)

 def check_best_small(self, compat, size):
     with namedTemporaryDir() as tmpdir:
         filename = os.path.join(tmpdir, 'test')
         with io.open(filename, "wb") as f:
             f.truncate(size)
             f.write("x" * MB)
         self.check_estimate(filename, compat)

 def test_check(self):
     with namedTemporaryDir() as tmpdir:
         path = os.path.join(tmpdir, 'test.qcow2')
         qemuimg.create(path, size=1048576, format=qemuimg.FORMAT.QCOW2)
         info = qemuimg.check(path)
         # The exact value depends on qcow2 internals
         self.assertEqual(int, type(info['offset']))

 def testLoopMount(self):
     with namedTemporaryDir() as mpath:
         # two nested with blocks to be python 2.6 friendly
         with createFloppyImage(FLOPPY_SIZE) as path:
             m = mount.Mount(path, mpath)
             with loop_mount(m):
                 self.assertTrue(m.isMounted())

 def test_getmetaparam(self):
     with namedTemporaryDir() as tmpdir:
         metadata = {sd.DMDK_VERSION: 3}
         manifest = make_filesd_manifest(tmpdir, metadata)
         metadata[sd.DMDK_SDUUID] = manifest.sdUUID
         self.assertEquals(manifest.sdUUID,
                           manifest.getMetaParam(sd.DMDK_SDUUID))

 def test_getblocksize_defaults(self):
     with namedTemporaryDir() as tmpdir:
         lvm = FakeLVM(tmpdir)
         with MonkeyPatchScope([(blockSD, 'lvm', lvm)]):
             manifest = make_blocksd(tmpdir, lvm)
             self.assertEquals(512, manifest.logBlkSize)
             self.assertEquals(512, manifest.phyBlkSize)

    def testFullDir(self, persist=False):
        """
        Test that rotator does it's basic functionality.
        """
        # Prepare
        prefix = "prefix"
        stubContent = ('"Multiple exclamation marks", '
                       'he went on, shaking his head, '
                       '"are a sure sign of a diseased mind."')
        # (C) Terry Pratchet - Small Gods
        with namedTemporaryDir() as dir:
            gen = 10

            expectedDirContent = []
            for i in range(gen):
                fname = "%s.txt.%d" % (prefix, i + 1)
                expectedDirContent.append("%s.txt.%d" % (prefix, i + 1))
                f = open(os.path.join(dir, fname), "wb")
                f.write(stubContent)
                f.flush()
                f.close()

            # Rotate
            misc.rotateFiles(dir, prefix, gen, persist=persist)

            # Test result
            currentDirContent = os.listdir(dir)
            expectedDirContent.sort()
            currentDirContent.sort()
            self.assertEquals(currentDirContent, expectedDirContent)

 def test_lvpath(self):
     with namedTemporaryDir() as tmpdir:
         lvm = FakeLVM(tmpdir)
         vg_name = 'foo'
         lv_name = 'bar'
         expected = os.path.join(tmpdir, 'dev', vg_name, lv_name)
         self.assertEqual(expected, lvm.lvPath(vg_name, lv_name))

 def test_unsafe_create_volume(self):
     with namedTemporaryDir() as tmpdir:
         path = os.path.join(tmpdir, 'test.qcow2')
         # Using unsafe=True to verify that it is possible to create an
         # image based on a non-existing backing file, like an inactive LV.
         qemuimg.create(path, size=1048576, format=qemuimg.FORMAT.QCOW2,
                        backing='no-such-file', unsafe=True)

 def testEmptyDir(self, persist=False):
     """
     Test that when given an empty dir the rotator works correctly.
     """
     prefix = "prefix"
     with namedTemporaryDir() as dir:
         misc.rotateFiles(dir, prefix, 0, persist=persist)

 def test_create_new(self):
     with namedTemporaryDir() as tmpdir:
         target = os.path.join(tmpdir, "target")
         link = os.path.join(tmpdir, "link")
         fileUtils.atomic_symlink(target, link)
         self.assertEqual(os.readlink(link), target)
         self.assertFalse(os.path.exists(link + ".tmp"))

 def test_error_isfile(self):
     with namedTemporaryDir() as tmpdir:
         target = os.path.join(tmpdir, "target")
         link = os.path.join(tmpdir, "link")
         with open(link, 'w') as f:
             f.write('data')
         self.assertRaises(OSError, fileUtils.atomic_symlink, target, link)

 def test_getallimages(self):
     with namedTemporaryDir() as tmpdir:
         manifest = make_filesd_manifest(tmpdir)
         self.assertEqual(set(), manifest.getAllImages())
         img_uuid = str(uuid.uuid4())
         make_file_volume(manifest.domaindir, VOLSIZE, img_uuid)
         self.assertIn(img_uuid, manifest.getAllImages())

    def test_teardown_failure(self):
        job_id = make_uuid()
        sp_id = make_uuid()
        sd_id = make_uuid()
        img0_id = make_uuid()
        img1_id = TEARDOWN_ERROR_IMAGE_ID
        vol0_id = make_uuid()
        vol1_id = make_uuid()
        images = [
            {'sd_id': sd_id, 'img_id': img0_id, 'vol_id': vol0_id},
            {'sd_id': sd_id, 'img_id': img1_id, 'vol_id': vol1_id},
        ]

        expected = [
            ('prepareImage', (sd_id, sp_id, img0_id, vol0_id),
             {'allowIllegal': True}),
            ('prepareImage', (sd_id, sp_id, img1_id, vol1_id),
             {'allowIllegal': True}),
            ('teardownImage', (sd_id, sp_id, img1_id), {}),
            ('teardownImage', (sd_id, sp_id, img0_id), {}),
        ]

        with namedTemporaryDir() as base:
            irs = FakeIRS(base)

            job = seal.Job(job_id, sp_id, images, irs)
            job.autodelete = False
            job.run()
            wait_for_job(job)

            self.assertEqual(jobs.STATUS.FAILED, job.status)
            self.assertEqual(expected, irs.__calls__)