def test_does_not_update_package_with_multiline_spec(self):
requirements = 'tests/samples/requirements-multiline.txt'
tempdir = tempfile.mkdtemp()
tmpfile = os.path.join(tempdir, 'requirements.txt')
shutil.copy(requirements, tmpfile)
args = ['-r', tmpfile]
with utils.mock.patch('pip._internal.index.PackageFinder.find_all_candidates') as mock_find_all_candidates:
project = 'afakepackage'
version = '1.1'
link = Link('')
candidate = InstallationCandidate(project, version, link)
mock_find_all_candidates.return_value = [candidate]
result = self.runner.invoke(pur, args)
self.assertIsNone(result.exception)
expected_output = "All requirements up-to-date.\n"
self.assertEquals(u(result.output), u(expected_output))
self.assertEquals(result.exit_code, 0)
expected_requirements = open(requirements).read()
self.assertEquals(open(tmpfile).read(), expected_requirements)
python类copy()的实例源码
def test_updates_package_with_min_version_spec(self):
requirements = 'tests/samples/requirements-with-min-version-spec.txt'
tempdir = tempfile.mkdtemp()
tmpfile = os.path.join(tempdir, 'requirements.txt')
shutil.copy(requirements, tmpfile)
args = ['-r', tmpfile]
with utils.mock.patch('pip._internal.index.PackageFinder.find_all_candidates') as mock_find_all_candidates:
project = 'django'
version = '1.8.13'
link = Link('')
candidate = InstallationCandidate(project, version, link)
mock_find_all_candidates.return_value = [candidate]
result = self.runner.invoke(pur, args)
expected_output = "Updated django: 1.8.6 -> 1.8.13\nNew version for django found (1.8.13), but current spec prohibits updating: django > 1.8.6, < 1.9\nAll requirements up-to-date.\n"
self.assertEquals(u(result.output), u(expected_output))
self.assertIsNone(result.exception)
self.assertEquals(result.exit_code, 0)
expected_requirements = open('tests/samples/results/test_updates_package_with_min_version_spec').read()
self.assertEquals(open(tmpfile).read(), expected_requirements)
def test_dry_run(self):
requirements = 'tests/samples/requirements.txt'
tempdir = tempfile.mkdtemp()
tmpfile = os.path.join(tempdir, 'requirements.txt')
shutil.copy(requirements, tmpfile)
args = ['-r', tmpfile, '-d']
with utils.mock.patch('pip._internal.index.PackageFinder.find_all_candidates') as mock_find_all_candidates:
project = 'flask'
version = '0.10.1'
link = Link('')
candidate = InstallationCandidate(project, version, link)
mock_find_all_candidates.return_value = [candidate]
result = self.runner.invoke(pur, args)
self.assertIsNone(result.exception)
expected_output = '==> ' + tmpfile + ' <==\n' + \
open('tests/samples/results/test_updates_package').read() + "\n"
self.assertEquals(u(result.output), u(expected_output))
self.assertEquals(result.exit_code, 0)
self.assertEquals(open(tmpfile).read(), open(requirements).read())
def test_interactive_choice_default(self):
tempdir = tempfile.mkdtemp()
requirements = os.path.join(tempdir, 'requirements.txt')
shutil.copy('tests/samples/requirements.txt', requirements)
args = ['-r', requirements, '-i']
with utils.mock.patch('pip._internal.index.PackageFinder.find_all_candidates') as mock_find_all_candidates:
project = 'flask'
version = '0.10.1'
link = Link('')
candidate = InstallationCandidate(project, version, link)
mock_find_all_candidates.return_value = [candidate]
result = self.runner.invoke(pur, args)
self.assertIsNone(result.exception)
expected_output = "Update flask from 0.9 to 0.10.1? (y, n, q) [y]: \nUpdated flask: 0.9 -> 0.10.1\nAll requirements up-to-date.\n"
self.assertEquals(u(result.output), u(expected_output))
self.assertEquals(result.exit_code, 0)
expected_requirements = open('tests/samples/results/test_updates_package').read()
self.assertEquals(open(requirements).read(), expected_requirements)
def test_interactive_choice_yes(self):
tempdir = tempfile.mkdtemp()
requirements = os.path.join(tempdir, 'requirements.txt')
shutil.copy('tests/samples/requirements.txt', requirements)
args = ['-r', requirements, '-i']
with utils.mock.patch('pip._internal.index.PackageFinder.find_all_candidates') as mock_find_all_candidates:
project = 'flask'
version = '0.10.1'
link = Link('')
candidate = InstallationCandidate(project, version, link)
mock_find_all_candidates.return_value = [candidate]
result = self.runner.invoke(pur, args, input='y\n')
self.assertIsNone(result.exception)
expected_output = "Update flask from 0.9 to 0.10.1? (y, n, q) [y]: y\nUpdated flask: 0.9 -> 0.10.1\nAll requirements up-to-date.\n"
self.assertEquals(u(result.output), u(expected_output))
self.assertEquals(result.exit_code, 0)
expected_requirements = open('tests/samples/results/test_updates_package').read()
self.assertEquals(open(requirements).read(), expected_requirements)
def test_interactive_choice_quit(self):
tempdir = tempfile.mkdtemp()
requirements = os.path.join(tempdir, 'requirements.txt')
shutil.copy('tests/samples/requirements-multiple.txt', requirements)
args = ['-r', requirements, '-i']
with utils.mock.patch('pip._internal.index.PackageFinder.find_all_candidates') as mock_find_all_candidates:
project = 'flask'
version = '0.10.1'
link = Link('')
candidate = InstallationCandidate(project, version, link)
mock_find_all_candidates.return_value = [candidate]
result = self.runner.invoke(pur, args, input='y\nq\n')
self.assertIsNone(result.exception)
expected_output = "Update flask from 0.9 to 0.10.1? (y, n, q) [y]: y\nUpdated flask: 0.9 -> 0.10.1\nUpdate Alembic from 0.9 to 0.10.1? (y, n, q) [y]: q\nAll requirements up-to-date.\n"
self.assertEquals(u(result.output), u(expected_output))
self.assertEquals(result.exit_code, 0)
expected_requirements = open('tests/samples/results/test_interactive_choice_quit').read()
self.assertEquals(open(requirements).read(), expected_requirements)
def test_interactive_choice_invalid(self):
tempdir = tempfile.mkdtemp()
requirements = os.path.join(tempdir, 'requirements.txt')
shutil.copy('tests/samples/requirements-multiple.txt', requirements)
args = ['-r', requirements, '-i']
with utils.mock.patch('pip._internal.index.PackageFinder.find_all_candidates') as mock_find_all_candidates:
project = 'flask'
version = '0.10.1'
link = Link('')
candidate = InstallationCandidate(project, version, link)
mock_find_all_candidates.return_value = [candidate]
result = self.runner.invoke(pur, args, input='z\nn\ny\nq\n')
self.assertIsNone(result.exception)
expected_output = "Update flask from 0.9 to 0.10.1? (y, n, q) [y]: z\nPlease enter either y, n, q.\nUpdate flask from 0.9 to 0.10.1? (y, n, q) [y]: n\nUpdate Alembic from 0.9 to 0.10.1? (y, n, q) [y]: y\nUpdated Alembic: 0.9 -> 0.10.1\nUpdate sqlalchemy from 0.9 to 0.10.1? (y, n, q) [y]: q\nAll requirements up-to-date.\n"
self.assertEquals(u(result.output), u(expected_output))
self.assertEquals(result.exit_code, 0)
expected_requirements = open('tests/samples/results/test_interactive_choice_invalid').read()
self.assertEquals(open(requirements).read(), expected_requirements)
def _copy_file(filename, location, link):
copy = True
download_location = os.path.join(location, link.filename)
if os.path.exists(download_location):
response = ask_path_exists(
'The file %s exists. (i)gnore, (w)ipe, (b)ackup, (a)abort' %
display_path(download_location), ('i', 'w', 'b', 'a'))
if response == 'i':
copy = False
elif response == 'w':
logger.warning('Deleting %s', display_path(download_location))
os.remove(download_location)
elif response == 'b':
dest_file = backup_dir(download_location)
logger.warning(
'Backing up %s to %s',
display_path(download_location),
display_path(dest_file),
)
shutil.move(download_location, dest_file)
elif response == 'a':
sys.exit(-1)
if copy:
shutil.copy(filename, download_location)
logger.info('Saved %s', display_path(download_location))
def setup(): #Asks new user for information
print('Enter the directory for your iTunes library')
spacer_small()
print('This is typically stored in your \'user/Music/iTunes\' folder')
print('You can also make a copy wherever you want in iTunes')
print('Just go File>Library>Export Library...')
tunes_library_path = input() #Where is my itunes library?
spacer()
print('Now you gotta give me the root directory for your media player')
print('This is typically /Volumes/YOUR DEVICE NAME')
fiio_dir = input() #Device directory
if fiio_dir[len(fiio_dir) - 1] == '/':
fiio_dir = fiio_dir[0: len(fiio_dir) - 1] #User may add an extra slash, get rid of it
spacer()
print('Ima save those settings')
settings_body = 'iTunes Library Path:\n' + tunes_library_path + '\n' + 'Device Root Directory\n' + fiio_dir
newsettings = open(fiio_dir + '/X1_sync_settings.txt', 'w') #Write a settings file into the root directory of device
newsettings.write(settings_body)
newsettings.close()
return [tunes_library_path, fiio_dir]
def Dumpfile():
fv.write('##########')
fv.write('\n')
fv.write("local Administrators on machine ")
fv.write(host)
fv.write(" are : ....\n\n")
result,t,r= NetLocalGroupGetMembers(None,"Administrators",1)
for item in result:
fv.write(str(item))
fv.write('\n')
fv.write('\n\n')
fv.write ("##########\n")
fv.write('the following software is installed on ')
fv.write(host)
fv.write(': .... \n\n')
listofsoft=readSubKeys("HKLM", "SOFTWARE\Microsoft\Windows\CurrentVersion\Uninstall")
listofsoft.sort()
for software in listofsoft:
fv.write(str(software))
fv.write('\n')
fv.close()
shutil.copy(srcfile,r"\\10.1.1.12\gms")
win32api.DeleteFile(srcfile)
def cat_contig_files_from_subprocs( tup ):
contig = tup[0]
tmp = tup[1]
j = tup[2]
N_contigs = tup[3]
logging.info("...contig %s (%s/%s)" % (contig, j, N_contigs))
contig_fns = glob.glob( os.path.join(tmp, "chunk_*", "%s_*.tmp" % contig) )
ftypes = set( map(lambda x: os.path.basename(x).split("_")[-1], contig_fns) )
for ftype in ftypes:
if ftype in ["compkmers.tmp", "ipdskmers.tmp"]:
first_fn = glob.glob( os.path.join(tmp, "chunk_*", "%s_%s" % (contig, ftype)) )[0]
shutil.copy( first_fn, os.path.join(tmp, "%s_%s" % (contig,ftype)))
else:
to_cat = glob.glob( os.path.join(tmp, "chunk_*", "%s_%s" % (contig, ftype)) )
to_cat.sort()
outname = os.path.join(tmp, "%s_%s" % (contig,ftype) )
cat_list_of_files(to_cat, outname, del_ins=False)
def minimize(self):
self.tpr = "%s.tpr" % (self.basename)
self.mdout = "%s.out.mdp" % (self.basename)
self.pplog = "%s.pp.log" % (self.basename)
self.deffnm = "%s.md" % (self.basename)
self.mdgro = "%s.gro" % (self.deffnm)
self.mdtrr = "%s.trr" % (self.deffnm)
self.mdlog = "%s.log" % (self.deffnm)
self.mdedr = "%s.edr" % (self.deffnm)
ppargs = self.grompp + ["-f",self.mdp, "-p",self.top, "-c",self.gro, "-maxwarn","3", "-po",self.mdout, "-o",self.tpr]
mdargs = self.mdrun + ["-s",self.tpr, "-nt","1", "-deffnm",self.deffnm, "-cpt","0"]
with open(self.pplog, "w") as pplog:
md_env = os.environ.copy()
md_env["GMX_SUPPRESS_DUMP"] = '1'
md_env["GMX_MAXBACKUP"] = '-1'
pp = subprocess.call(ppargs, stdout=pplog, stderr=pplog, env=md_env)
if pp:
sys.stderr.write("grompp error: check %s\n" % (self.pplog))
sys.exit()
md = subprocess.call(mdargs, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=md_env)
if md:
sys.stderr.write("mdrun error: check %s\n" % (self.mdlog))
sys.exit()
def assess_status_func(configs):
"""Helper function to create the function that will assess_status() for
the unit.
Uses charmhelpers.contrib.openstack.utils.make_assess_status_func() to
create the appropriate status function and then returns it.
Used directly by assess_status() and also for pausing and resuming
the unit.
NOTE: REQUIRED_INTERFACES is augmented with the optional interfaces
depending on the current config before being passed to the
make_assess_status_func() function.
@param configs: a templating.OSConfigRenderer() object
@return f() -> None : a function that assesses the unit's workload status
"""
required_interfaces = REQUIRED_INTERFACES.copy()
required_interfaces.update(get_optional_interfaces())
return make_assess_status_func(
configs, required_interfaces,
charm_func=check_optional_relations,
services=services(),
ports=determine_ports())
def assess_status_func(configs):
"""Helper function to create the function that will assess_status() for
the unit.
Uses charmhelpers.contrib.openstack.utils.make_assess_status_func() to
create the appropriate status function and then returns it.
Used directly by assess_status() and also for pausing and resuming
the unit.
NOTE: REQUIRED_INTERFACES is augmented with the optional interfaces
depending on the current config before being passed to the
make_assess_status_func() function.
@param configs: a templating.OSConfigRenderer() object
@return f() -> None : a function that assesses the unit's workload status
"""
required_interfaces = REQUIRED_INTERFACES.copy()
required_interfaces.update(get_optional_interfaces())
return make_assess_status_func(
configs, required_interfaces,
charm_func=check_optional_relations,
services=services(),
ports=determine_ports())
def exit_cleanup():
try:
fileobj = sys.stdout.fileobj
except AttributeError:
pass
else:
sys.stdout.is_valid = False
sys.stderr.is_valid = False
fileobj.close()
filename = sys.stdout.filename
Logs.info('Output logged to %r' % filename)
# then copy the log file to "latest.log" if possible
up = os.path.dirname(os.path.abspath(filename))
try:
shutil.copy(filename, os.path.join(up, 'latest.log'))
except OSError:
# this may fail on windows due to processes spawned
#
pass
def exit_cleanup():
try:
fileobj = sys.stdout.fileobj
except AttributeError:
pass
else:
sys.stdout.is_valid = False
sys.stderr.is_valid = False
fileobj.close()
filename = sys.stdout.filename
Logs.info('Output logged to %r' % filename)
# then copy the log file to "latest.log" if possible
up = os.path.dirname(os.path.abspath(filename))
try:
shutil.copy(filename, os.path.join(up, 'latest.log'))
except OSError:
# this may fail on windows due to processes spawned
#
pass
def exit_cleanup():
try:
fileobj = sys.stdout.fileobj
except AttributeError:
pass
else:
sys.stdout.is_valid = False
sys.stderr.is_valid = False
fileobj.close()
filename = sys.stdout.filename
Logs.info('Output logged to %r' % filename)
# then copy the log file to "latest.log" if possible
up = os.path.dirname(os.path.abspath(filename))
try:
shutil.copy(filename, os.path.join(up, 'latest.log'))
except OSError:
# this may fail on windows due to processes spawned
#
pass
def _test_checkin_with_handoff(my):
search_type = "unittest/person"
code = "joe"
search_key = my.server.build_search_key(search_type, code)
# copy file to handoff dir
path = "%s/test/miso_ramen.jpg" % my.client_lib_dir
handoff_dir = my.server.get_handoff_dir()
shutil.copy(path, handoff_dir)
# check the file in
snapshot = my.server.simple_checkin(search_key, "publish", path, use_handoff_dir=True)
snapshot_code = snapshot.get("code")
# get the filename
file_type = "main"
path = my.server.get_path_from_snapshot(snapshot_code, file_type)
exists = os.path.exists(path)
my.assertEquals(True, exists)
# check that the file name is correct
filename = os.path.basename(path)
# changed naming.. adds "publish"
my.assertEquals("miso_ramen_publish_v001.jpg", filename)
def mean_variance_normalisation(h5f, mvn_h5f, vad=None):
"""Do mean variance normlization. Optionnaly use a vad.
Parameters:
----------
h5f: str. h5features file name
mvn_h5f: str, h5features output name
"""
dset = h5py.File(h5f).keys()[0]
if vad is not None:
raise NotImplementedError
else:
data = h5py.File(h5f)[dset]['features'][:]
features = data
epsilon = np.finfo(data.dtype).eps
mean = np.mean(data)
std = np.std(data)
mvn_features = (features - mean) / (std + epsilon)
shutil.copy(h5f, mvn_h5f)
h5py.File(mvn_h5f)[dset]['features'][:] = mvn_features
def _copy_file(filename, location, link):
copy = True
download_location = os.path.join(location, link.filename)
if os.path.exists(download_location):
response = ask_path_exists(
'The file %s exists. (i)gnore, (w)ipe, (b)ackup ' %
display_path(download_location), ('i', 'w', 'b'))
if response == 'i':
copy = False
elif response == 'w':
logger.warning('Deleting %s', display_path(download_location))
os.remove(download_location)
elif response == 'b':
dest_file = backup_dir(download_location)
logger.warning(
'Backing up %s to %s',
display_path(download_location),
display_path(dest_file),
)
shutil.move(download_location, dest_file)
if copy:
shutil.copy(filename, download_location)
logger.info('Saved %s', display_path(download_location))
