def test_copy_tree_verbosity(self):
mkpath(self.target, verbose=0)
copy_tree(self.target, self.target2, verbose=0)
self.assertEqual(self._logs, [])
remove_tree(self.root_target, verbose=0)
mkpath(self.target, verbose=0)
a_file = os.path.join(self.target, 'ok.txt')
with open(a_file, 'w') as f:
f.write('some content')
wanted = ['copying %s -> %s' % (a_file, self.target2)]
copy_tree(self.target, self.target2, verbose=1)
self.assertEqual(self._logs, wanted)
remove_tree(self.root_target, verbose=0)
remove_tree(self.target2, verbose=0)
python类copy_tree()的实例源码
def datadir(tmpdir, request):
'''
Fixture responsible for locating the test data directory and copying it
into a temporary directory.
'''
filename = request.module.__file__
test_dir = os.path.dirname(filename)
data_dir = os.path.join(test_dir, 'data')
dir_util.copy_tree(data_dir, str(tmpdir))
def getter(filename, as_str=True):
filepath = tmpdir.join(filename)
if as_str:
return str(filepath)
return filepath
return getter
def test_copy_tree_verbosity(self):
mkpath(self.target, verbose=0)
copy_tree(self.target, self.target2, verbose=0)
self.assertEqual(self._logs, [])
remove_tree(self.root_target, verbose=0)
mkpath(self.target, verbose=0)
a_file = os.path.join(self.target, 'ok.txt')
with open(a_file, 'w') as f:
f.write('some content')
wanted = ['copying %s -> %s' % (a_file, self.target2)]
copy_tree(self.target, self.target2, verbose=1)
self.assertEqual(self._logs, wanted)
remove_tree(self.root_target, verbose=0)
remove_tree(self.target2, verbose=0)
def copy_tree(self, infile, outfile,
preserve_mode=1, preserve_times=1, preserve_symlinks=0,
level=1):
"""Copy an entire directory tree respecting verbose, dry-run,
and force flags.
"""
return dir_util.copy_tree(
infile, outfile,
preserve_mode,preserve_times,preserve_symlinks,
not self.force,
dry_run=self.dry_run)
def do_step(context):
settings = context.meta['settings']
index_file = context.meta['index-file']
username = settings["username"]
home_dir = os.path.join("/home", username)
# Copy all the files in ./bosh into the home directory
dir_util.copy_tree("./bosh/", home_dir)
copy("./manifests/{0}".format(index_file), "{0}/manifests/".format(home_dir))
call("chown -R {0} {1}".format(username, home_dir), shell=True)
call("chmod 400 {0}/bosh".format(home_dir), shell=True)
return context
def install_overrides(self, destination):
# FIXME: distutils.dir_util.copy_tree seems to keep some internal state when it does its copy.
#
# If the destination directory disappears, copy_tree will not recreate missing path
# components.
#
# In practice this should not be an issue since our script will execute once to install a
# modpack, then exit.
copy_tree(os.path.join(self.unpack_directory, self.manifest["overrides"]), destination)
def copy_tree(self, infile, outfile,
preserve_mode=1, preserve_times=1, preserve_symlinks=0,
level=1):
"""Copy an entire directory tree respecting verbose, dry-run,
and force flags.
"""
return dir_util.copy_tree(
infile, outfile,
preserve_mode,preserve_times,preserve_symlinks,
not self.force,
dry_run=self.dry_run)
def lp_init(self, pf_w_path, providers=['libvirt']):
"""
Initializes a linchpin project. Creates the necessary directory
structure, includes PinFile, topologies and layouts for the given
provider. (Default: Libvirt. Other providers not yet implemented.)
:param pf_w_path: Path to where the PinFile might exist. Gets created
if it doesn't exist.
:param providers: A list of providers for which templates
(and a target) will be provided into the workspace.
NOT YET IMPLEMENTED
"""
src = self.get_cfg('init', 'source', 'templates/')
src_w_path = os.path.realpath('{0}/{1}'.format(self.ctx.lib_path, src))
src_pf = os.path.realpath('{0}.lp_example'.format(pf_w_path))
try:
if os.path.exists(pf_w_path):
if not click.confirm('{0} already exists,'
'overwrite it?'.format(pf_w_path),
default=False):
sys.exit(0)
dir_util.copy_tree(src_w_path, self.workspace, verbose=1)
os.rename(src_pf, pf_w_path)
self.ctx.log_state('{0} and file structure created at {1}'.format(
self.pinfile, self.workspace))
except Exception as e:
self.ctx.log_state('Error: {0}'.format(e))
sys.exit(1)
def do_activate(self, *args):
print('LinterClang activated')
self.enabled = True
dir_util.copy_tree(self.parent.projectPath, '/tmp/pyidetmp')
if self.live:
self.connection = self.parent.sbuff.connect('changed', self.set_file_changed)
self.do_live_linting()
def replace_tools_from_product_repo(node_list, **kwargs):
""" This will clone the tools from product repo and then replaces
tools directory in warrior main with this tools repo.
"""
logfile = kwargs.get("logfile")
config_file_name = kwargs.get("config_file_name")
console_log_name = kwargs.get("console_log_name")
print_log_name = kwargs.get("print_log_name")
if "tools" in node_list:
tools_node = get_node(config_file_name, "tools")
tools_url = get_attribute_value(tools_node, "url")
tools_root = get_repository_name(tools_url)
tools_clone = get_attribute_value(tools_node, "clone")
tools_base_path = ""
warrior_node = get_node(config_file_name, "warriorframework")
warrior_base_path = get_attribute_value(warrior_node, "destination")
if tools_url and tools_clone == "yes":
tools_base_path = validate_base_path(
tools_base_path, logfile=logfile,
config_file_name=config_file_name,
console_log_name=console_log_name,
print_log_name=print_log_name)
warrior_base_path = validate_base_path(
warrior_base_path, logfile=logfile,
config_file_name=config_file_name,
console_log_name=console_log_name, print_log_name=print_log_name)
warrior_tools_path = os.path.join(warrior_base_path,
"warrior", "Tools")
product_tools_path = os.path.join(tools_base_path, tools_root, "Tools")
dir_util.copy_tree(product_tools_path, warrior_tools_path, update=1)
delete_directory(os.path.join(tools_base_path, tools_root), logfile, print_log_name)
def do_merge(backups, precision, dry_run, include_latest, skip_errors,
**kwargs):
data = sorted(os.listdir(backups))
if not data:
raise MergeError("Nothing found.")
groups, helper = tee(group_by_timestamp(data, precision * 60 * 60,
skip_errors))
next(helper, None)
for group in groups:
try:
next(helper)
except StopIteration:
if not include_latest:
logger.info("Folders `{0}` were excluded from merge because "
"they can be not "
"complete.".format(', '.join(group)))
continue
logger.info("GROUP: {0}".format(group))
if will_override(backups, group):
raise MergeError("Group `{0}` contains overlapping files. May be "
"precision was bigger then backup "
"periodicity.".format(group))
dst = os.path.join(backups, group.pop(0))
for item in group:
src = os.path.join(backups, item)
dir_util.copy_tree(src, dst, verbose=True, dry_run=dry_run)
dir_util.remove_tree(src, verbose=True, dry_run=dry_run)
def _commit(self):
"""Sync the temporary directory to the final path.
"""
dir_util.copy_tree(self.path, self._final_path)
def merge_tree(self, dst, symlinks=False, *args, **kwargs):
"""
Copy entire contents of self to dst, overwriting existing
contents in dst with those in self.
If the additional keyword `update` is True, each
`src` will only be copied if `dst` does not exist,
or `src` is newer than `dst`.
Note that the technique employed stages the files in a temporary
directory first, so this function is not suitable for merging
trees with large files, especially if the temporary directory
is not capable of storing a copy of the entire source tree.
"""
update = kwargs.pop('update', False)
with tempdir() as _temp_dir:
# first copy the tree to a stage directory to support
# the parameters and behavior of copytree.
stage = _temp_dir / str(hash(self))
self.copytree(stage, symlinks, *args, **kwargs)
# now copy everything from the stage directory using
# the semantics of dir_util.copy_tree
dir_util.copy_tree(stage, dst, preserve_symlinks=symlinks,
update=update)
#
# --- Special stuff from os
def copy_tree(self, infile, outfile,
preserve_mode=1, preserve_times=1, preserve_symlinks=0,
level=1):
"""Copy an entire directory tree respecting verbose, dry-run,
and force flags.
"""
return dir_util.copy_tree(
infile, outfile,
preserve_mode,preserve_times,preserve_symlinks,
not self.force,
dry_run=self.dry_run)
def copy_tree(self, infile, outfile,
preserve_mode=1, preserve_times=1, preserve_symlinks=0,
level=1):
"""Copy an entire directory tree respecting verbose, dry-run,
and force flags.
"""
return dir_util.copy_tree(
infile, outfile,
preserve_mode,preserve_times,preserve_symlinks,
not self.force,
dry_run=self.dry_run)
def copy_tree(self, infile, outfile,
preserve_mode=1, preserve_times=1, preserve_symlinks=0,
level=1):
"""Copy an entire directory tree respecting verbose, dry-run,
and force flags.
"""
return dir_util.copy_tree(
infile, outfile,
preserve_mode,preserve_times,preserve_symlinks,
not self.force,
dry_run=self.dry_run)
def merge_tree(self, dst, symlinks=False, *args, **kwargs):
"""
Copy entire contents of self to dst, overwriting existing
contents in dst with those in self.
If the additional keyword `update` is True, each
`src` will only be copied if `dst` does not exist,
or `src` is newer than `dst`.
Note that the technique employed stages the files in a temporary
directory first, so this function is not suitable for merging
trees with large files, especially if the temporary directory
is not capable of storing a copy of the entire source tree.
"""
update = kwargs.pop('update', False)
with tempdir() as _temp_dir:
# first copy the tree to a stage directory to support
# the parameters and behavior of copytree.
stage = _temp_dir / str(hash(self))
self.copytree(stage, symlinks, *args, **kwargs)
# now copy everything from the stage directory using
# the semantics of dir_util.copy_tree
dir_util.copy_tree(stage, dst, preserve_symlinks=symlinks,
update=update)
#
# --- Special stuff from os
def copy_tree(self, infile, outfile,
preserve_mode=1, preserve_times=1, preserve_symlinks=0,
level=1):
"""Copy an entire directory tree respecting verbose, dry-run,
and force flags.
"""
return dir_util.copy_tree(
infile, outfile,
preserve_mode,preserve_times,preserve_symlinks,
not self.force,
dry_run=self.dry_run)
def copy_tree(self, infile, outfile, preserve_mode=1, preserve_times=1,
preserve_symlinks=0, level=1):
"""Copy an entire directory tree respecting verbose, dry-run,
and force flags.
"""
return dir_util.copy_tree(infile, outfile, preserve_mode,
preserve_times, preserve_symlinks,
not self.force, dry_run=self.dry_run)
def test_copy_tree_skips_nfs_temp_files(self):
mkpath(self.target, verbose=0)
a_file = os.path.join(self.target, 'ok.txt')
nfs_file = os.path.join(self.target, '.nfs123abc')
for f in a_file, nfs_file:
with open(f, 'w') as fh:
fh.write('some content')
copy_tree(self.target, self.target2)
self.assertEqual(os.listdir(self.target2), ['ok.txt'])
remove_tree(self.root_target, verbose=0)
remove_tree(self.target2, verbose=0)
def test_copy_tree_exception_in_listdir(self):
"""
An exception in listdir should raise a DistutilsFileError
"""
with patch("os.listdir", side_effect=OSError()), \
self.assertRaises(errors.DistutilsFileError):
src = self.tempdirs[-1]
dir_util.copy_tree(src, None)
def copy_tree(self, infile, outfile,
preserve_mode=1, preserve_times=1, preserve_symlinks=0,
level=1):
"""Copy an entire directory tree respecting verbose, dry-run,
and force flags.
"""
return dir_util.copy_tree(
infile, outfile,
preserve_mode,preserve_times,preserve_symlinks,
not self.force,
dry_run=self.dry_run)
def download_and_extract_dataset(self):
if not os.path.exists(self.get_dataset_filename()):
print("Downloading OpenOMR dataset...")
self.download_file(self.get_dataset_download_url(), self.get_dataset_filename())
print("Extracting OpenOMR dataset...")
absolute_path_to_temp_folder = os.path.abspath('OpenOmrDataset')
self.extract_dataset(absolute_path_to_temp_folder)
os.makedirs(self.destination_directory, exist_ok=True)
dir_util.copy_tree(os.path.join(absolute_path_to_temp_folder, "OpenOMR-Dataset"), self.destination_directory)
self.clean_up_temp_directory(absolute_path_to_temp_folder)
FornesMusicSymbolsDatasetDownloader.py 文件源码
项目:OMR-Datasets
作者: apacha
项目源码
文件源码
阅读 18
收藏 0
点赞 0
评论 0
def download_and_extract_dataset(self):
if not os.path.exists(self.get_dataset_filename()):
print("Downloading Fornes Music Symbol dataset...")
self.download_file(self.get_dataset_download_url(), self.get_dataset_filename())
print("Extracting Fornes Music Symbol dataset...")
absolute_path_to_temp_folder = os.path.abspath('Fornes-Music-Symbols')
self.extract_dataset(absolute_path_to_temp_folder)
self.__fix_capital_file_endings(absolute_path_to_temp_folder)
os.makedirs(self.destination_directory, exist_ok=True)
dir_util.copy_tree(os.path.join(absolute_path_to_temp_folder, "Music_Symbols"),
self.destination_directory)
self.clean_up_temp_directory(absolute_path_to_temp_folder)
RebeloMusicSymbolDataset2Downloader.py 文件源码
项目:OMR-Datasets
作者: apacha
项目源码
文件源码
阅读 23
收藏 0
点赞 0
评论 0
def download_and_extract_dataset(self):
if not os.path.exists(self.get_dataset_filename()):
print("Downloading Rebelo Symbol Dataset 2...")
self.download_file(self.get_dataset_download_url(), self.get_dataset_filename())
print("Extracting Rebelo Symbol Dataset 2...")
absolute_path_to_temp_folder = os.path.abspath('Rebelo-Music-Symbol-Dataset2')
self.extract_dataset(absolute_path_to_temp_folder)
os.makedirs(self.destination_directory, exist_ok=True)
dir_util.copy_tree(os.path.join(absolute_path_to_temp_folder, "database2"), self.destination_directory)
self.clean_up_temp_directory(absolute_path_to_temp_folder)
PrintedMusicSymbolsDatasetDownloader.py 文件源码
项目:OMR-Datasets
作者: apacha
项目源码
文件源码
阅读 17
收藏 0
点赞 0
评论 0
def download_and_extract_dataset(self):
if not os.path.exists(self.get_dataset_filename()):
print("Downloading Printed Music Symbol dataset...")
self.download_file(self.get_dataset_download_url(), self.get_dataset_filename())
print("Extracting Printed Music Symbol dataset...")
absolute_path_to_temp_folder = os.path.abspath('PrintedMusicSymbolsDataset')
self.extract_dataset(absolute_path_to_temp_folder)
os.makedirs(self.destination_directory, exist_ok=True)
dir_util.copy_tree(os.path.join(absolute_path_to_temp_folder, "PrintedMusicSymbolsDataset"),
self.destination_directory)
self.clean_up_temp_directory(absolute_path_to_temp_folder)
RebeloMusicSymbolDataset1Downloader.py 文件源码
项目:OMR-Datasets
作者: apacha
项目源码
文件源码
阅读 18
收藏 0
点赞 0
评论 0
def download_and_extract_dataset(self):
if not os.path.exists(self.get_dataset_filename()):
print("Downloading Rebelo Symbol Dataset 1...")
self.download_file(self.get_dataset_download_url(), self.get_dataset_filename())
print("Extracting Rebelo Symbol Dataset 1...")
absolute_path_to_temp_folder = os.path.abspath('Rebelo-Music-Symbol-Dataset1')
self.extract_dataset(absolute_path_to_temp_folder)
os.makedirs(self.destination_directory, exist_ok=True)
dir_util.copy_tree(os.path.join(absolute_path_to_temp_folder, "database1"), self.destination_directory)
self.clean_up_temp_directory(absolute_path_to_temp_folder)
def copy_tree(self, infile, outfile,
preserve_mode=1, preserve_times=1, preserve_symlinks=0,
level=1):
"""Copy an entire directory tree respecting verbose, dry-run,
and force flags.
"""
return dir_util.copy_tree(
infile, outfile,
preserve_mode,preserve_times,preserve_symlinks,
not self.force,
dry_run=self.dry_run)
def copy_tree(self, infile, outfile, preserve_mode=1, preserve_times=1,
preserve_symlinks=0, level=1):
"""Copy an entire directory tree respecting verbose, dry-run,
and force flags.
"""
return dir_util.copy_tree(infile, outfile, preserve_mode,
preserve_times, preserve_symlinks,
not self.force, dry_run=self.dry_run)
def test_copy_tree_skips_nfs_temp_files(self):
mkpath(self.target, verbose=0)
a_file = os.path.join(self.target, 'ok.txt')
nfs_file = os.path.join(self.target, '.nfs123abc')
for f in a_file, nfs_file:
with open(f, 'w') as fh:
fh.write('some content')
copy_tree(self.target, self.target2)
self.assertEqual(os.listdir(self.target2), ['ok.txt'])
remove_tree(self.root_target, verbose=0)
remove_tree(self.target2, verbose=0)