for arg, val in keep_args.items():
keep_cmd.append("{}={}".format(arg, val))
- logf = open(_logfilename('keep{}'.format(n)), 'a')
- kp0 = subprocess.Popen(
- keep_cmd, stdin=open('/dev/null'), stdout=logf, stderr=logf, close_fds=True)
+ with open(_logfilename('keep{}'.format(n)), 'a') as logf:
+ with open('/dev/null') as _stdin:
+ kp0 = subprocess.Popen(
+ keep_cmd, stdin=_stdin, stdout=logf, stderr=logf, close_fds=True)
with open(_pidfile('keep{}'.format(n)), 'w') as f:
f.write(str(kp0.pid))
pidfile = _pidfile('keepproxy')
if os.path.exists(pidfile):
try:
- os.kill(int(open(pidfile).read()), signal.SIGHUP)
+ with open(pidfile) as pid:
+ os.kill(int(pid.read()), signal.SIGHUP)
except OSError:
os.remove(pidfile)
# Returns 9 if program is not up.
def _getport(program):
try:
- return int(open(_portfile(program)).read())
+ with open(_portfile(program)) as prog:
+ return int(prog.read())
except IOError:
return 9
@contextlib.contextmanager
def nostderr():
- orig, sys.stderr = sys.stderr, open(os.devnull, 'w')
- try:
- yield
- finally:
- sys.stderr = orig
+ with open(os.devnull, 'w') as dn:
+ orig, sys.stderr = sys.stderr, dn
+ try:
+ yield
+ finally:
+ sys.stderr = orig
class MountArgsTest(unittest.TestCase):
quote(os.path.join(self.mnt, 'zzz', 'foo.txt')),
quote(os.path.join(self.mnt, 'zzz', '.arvados#collection')),
quote(os.path.join(self.okfile)))]))
- self.assertRegexpMatches(
- json.load(open(self.okfile))['manifest_text'],
- r' 0:3:foo.txt\n')
+ with open(self.okfile) as f:
+ self.assertRegexpMatches(
+ json.load(f)['manifest_text'],
+ r' 0:3:foo.txt\n')
def current_manifest(tmpdir):
- return json.load(open(
- os.path.join(tmpdir, '.arvados#collection')
- ))['manifest_text']
+ with open(os.path.join(tmpdir, '.arvados#collection')) as tmp:
+ return json.load(tmp)['manifest_text']
class TmpCollectionTest(IntegrationTest):