Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_suffix(self):
p1 = self.longpath
p2 = local.path("file.tar.gz")
assert p1.suffix == ".txt"
assert p1.suffixes == [".txt"]
assert p2.suffix == ".gz"
assert p2.suffixes == [".tar",".gz"]
assert p1.with_suffix(".tar.gz") == local.path("/some/long/path/to/file.tar.gz")
assert p2.with_suffix(".other") == local.path("file.tar.other")
assert p2.with_suffix(".other", 2) == local.path("file.other")
assert p2.with_suffix(".other", 0) == local.path("file.tar.gz.other")
assert p2.with_suffix(".other", None) == local.path("file.other")
with pytest.raises(ValueError):
p1.with_suffix('nodot')
af = AtomicFile("tmp.txt")
code = """from __future__ import with_statement
from plumbum.fs.atomic import AtomicFile
af = AtomicFile("tmp.txt")
try:
with af.locked(blocking = False):
raise ValueError("this should have failed")
except (OSError, IOError):
print("already locked")
"""
with af.locked():
output = local.python("-c", code)
assert output.strip() == "already locked"
local.path("tmp.txt").delete()
def __init__(self, path=None):
if path is None:
# autogenerate a tmpdir of the form '/tmp/pytest-'
self.path = plumbum.local.path(str(
py.path.local.make_numbered_dir(
prefix="pytest-",
rootdir=py.path.local("/tmp/"),
keep=5,
)
))
else:
self.path = plumbum.local.path(path)
self.path.mkdir()
for entity in self.path.list():
self.path.join(entity).delete()
def test_pickle(self):
path1 = local.path('.')
path2 = local.path('~')
assert pickle.loads(pickle.dumps(self.longpath)) == self.longpath
assert pickle.loads(pickle.dumps(path1)) == path1
assert pickle.loads(pickle.dumps(path2)) == path2
def test_iterdir(self):
cwd = local.path('.')
files = list(cwd.iterdir())
assert cwd / 'test_local.py' in files
assert cwd / 'test_remote.py' in files
assert cwd['test_local.py'] in files
assert cwd['test_remote.py'] in files
def parse_config_file(self, config_file):
try:
logging.debug('Parsing config %s', config_file)
with local.path(config_file).open('r') as file:
for line in file:
line = line.strip()
if len(line) == 0 or line[0] == '#':
continue
parts = line.split()
if len(parts) > 2:
raise OcrmypdfConfigParsingError('More than one option and one value in line: {}'.format(line))
opt = parts[0]
val = parts[1] if len(parts) > 1 else None
if opt[0] != '-':
raise OcrmypdfConfigParsingError('Line must start with an option (-f, --foo): {}'.format(line))
self.options[opt] = val
except IOError as io:
str = 'IOError parsing {}: {}'.format(config_file, io)
logger.debug(str)
raise OcrmypdfConfigParsingError(str)
def on_modified(self, event):
self.logger.debug("File modified: %s", event.src_path)
self.touch_file(local.path(event.src_path))
def ExistingDirectory(val):
"""A switch-type validator that ensures that the given argument is an existing directory"""
p = local.path(val)
if not p.is_dir():
raise ValueError(_("{0} is not a directory").format(val))
return p
def main(self, src, dst):
if local.path(dst).exists():
if not self.overwrite:
logger.debug("Oh no! That's terrible")
raise ValueError("Destination already exists")
else:
delete(dst)
logger.debug("I'm going to copy %s to %s", src, dst)
copy(src, dst)
logger.debug("Great success")
def on_moved(self, event):
self.logger.debug("File moved: %s -> %s", event.src_path, event.dest_path)
self.delete_file(local.path(event.src_path))
self.touch_file(local.path(event.dest_path))