GothenburgBitFactory/taskwarrior
1
0
Fork 0
mirror of https://github.com/GothenburgBitFactory/taskwarrior.git synced 2025-06-26 10:54:26 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

add initial bulk run from pre-commit over all files

Browse source
This commit is contained in:
Felix Schurk 2024-07-29 22:34:51 +02:00
parent 665aeeef61
commit 93356b39c3
418 changed files with 21354 additions and 23858 deletions
Download patch file Download diff file Expand all files Collapse all files

2
test/basetest/CMakeLists.txt
View file

@ -5,4 +5,4 @@ configure_file(hooks.py hooks.py COPYONLY)
configure_file(meta.py meta.py COPYONLY)
configure_file(task.py task.py COPYONLY)
configure_file(testing.py testing.py COPYONLY)
configure_file(utils.py utils.py)
configure_file(utils.py utils.py)

26
test/basetest/exceptions.py
View file

@ -1,25 +1,29 @@
import signal
sig_names = dict((k, v) for v, k in reversed(sorted(signal.__dict__.items()))
if v.startswith('SIG') and not v.startswith('SIG_'))
sig_names = dict(
(k, v)
for v, k in reversed(sorted(signal.__dict__.items()))
if v.startswith("SIG") and not v.startswith("SIG_")
)
class CommandError(Exception):
def __init__(self, cmd, code, out, err=None, msg=None):
DEFAULT = ("Command '{{0}}' was {signal}'ed. "
"SIGABRT usually means task timed out.\n")
DEFAULT = (
"Command '{{0}}' was {signal}'ed. "
"SIGABRT usually means task timed out.\n"
)
if msg is None:
msg_suffix = "\n*** Start STDOUT ***\n{2}\n*** End STDOUT ***\n"
if err is not None:
msg_suffix += (
"\n*** Start STDERR ***\n{3}\n*** End STDERR ***\n"
)
msg_suffix += "\n*** Start STDERR ***\n{3}\n*** End STDERR ***\n"
if code < 0:
self.msg = DEFAULT.format(signal=sig_names[abs(code)])
else:
self.msg = ("Command '{0}' finished with unexpected exit "
"code '{1}'.\n")
self.msg = (
"Command '{0}' finished with unexpected exit " "code '{1}'.\n"
)
self.msg += msg_suffix
else:
@ -43,12 +47,12 @@ class TimeoutWaitingFor(object):
self.name = name
def __repr__(self):
return "*** Timeout reached while waiting for {0} ***".format(
self.name)
return "*** Timeout reached while waiting for {0} ***".format(self.name)
class StreamsAreMerged(object):
def __repr__(self):
return "*** Streams are merged, STDERR is not available ***"
# vim: ai sts=4 et sw=4

176
test/basetest/hooks.py
View file

@ -4,6 +4,7 @@ import os
from sys import stderr
import shutil
import stat
try:
import simplejson as json
except ImportError:
@ -15,8 +16,8 @@ from .exceptions import HookError
class InvalidJSON(object):
"""Object representing the original unparsed JSON string and the JSON error
"""
"""Object representing the original unparsed JSON string and the JSON error"""
def __init__(self, original, error):
self.original = original
self.error = error
@ -38,6 +39,7 @@ class Hooks(object):
"""Abstraction to help interact with hooks (add, remove) during tests and
keep track of which are active.
"""
def __init__(self, datadir):
"""Initialize hooks container which keeps track of active hooks and
@ -63,8 +65,7 @@ class Hooks(object):
enabled = ", ".join(enabled) or None
disabled = ", ".join(disabled) or None
return "<Hooks: enabled: {0} | disabled: {1}>".format(enabled,
disabled)
return "<Hooks: enabled: {0} | disabled: {1}>".format(enabled, disabled)
def __getitem__(self, name):
return self._hooks[name]
@ -134,8 +135,7 @@ class Hooks(object):
hook._delete()
def clear(self):
"""Remove all existing hooks and empty the hook registry
"""
"""Remove all existing hooks and empty the hook registry"""
self._hooks = {}
# Remove any existing hooks
@ -150,10 +150,11 @@ class Hooks(object):
class Hook(object):
"""Represents a hook script and provides methods to enable/disable hooks
"""
def __init__(self, hookname, hookdir, content=None, default=False,
default_hookpath=None):
"""Represents a hook script and provides methods to enable/disable hooks"""
def __init__(
self, hookname, hookdir, content=None, default=False, default_hookpath=None
):
"""Initialize and create the hook
This class supports creating hooks in two ways:
@ -181,24 +182,25 @@ class Hook(object):
self._check_hook_not_exists(self.hookfile)
if not default and content is None:
raise HookError("Cannot create hookfile {0} without content. "
"If using a builtin hook pass default=True"
.format(self.hookname))
raise HookError(
"Cannot create hookfile {0} without content. "
"If using a builtin hook pass default=True".format(self.hookname)
)
if os.path.isfile(self.hookfile):
raise HookError("Hook with name {0} already exists. "
"Did you forget to remove() it before recreating?"
.format(self.hookname))
raise HookError(
"Hook with name {0} already exists. "
"Did you forget to remove() it before recreating?".format(self.hookname)
)
if default:
self.default_hookfile = os.path.join(self.default_hookpath,
self.hookname)
self.default_hookfile = os.path.join(self.default_hookpath, self.hookname)
self._check_hook_exists(self.default_hookfile)
# Symlinks change permission of source file, cannot use one here
shutil.copy(self.default_hookfile, self.hookfile)
else:
self.default_hookfile = None
with open(self.hookfile, 'w') as fh:
with open(self.hookfile, "w") as fh:
fh.write(content)
def __eq__(self, other):
@ -250,16 +252,19 @@ class Hook(object):
if self.hookname.startswith(hooktype):
break
else:
stderr.write("WARNING: {0} is not a valid hook type. "
"It will not be triggered\n".format(self.hookname))
stderr.write(
"WARNING: {0} is not a valid hook type. "
"It will not be triggered\n".format(self.hookname)
)
def _remove_file(self, file):
try:
os.remove(file)
except OSError as e:
if e.errno == errno.ENOENT:
raise HookError("Hook with name {0} was not found on "
"hooks/ folder".format(file))
raise HookError(
"Hook with name {0} was not found on " "hooks/ folder".format(file)
)
else:
raise
@ -271,18 +276,15 @@ class Hook(object):
self._remove_hookfile(self.hookfile)
def enable(self):
"""Make hookfile executable to allow triggering
"""
"""Make hookfile executable to allow triggering"""
os.chmod(self.hookfile, stat.S_IREAD | stat.S_IWRITE | stat.S_IEXEC)
def disable(self):
"""Remove hookfile executable bit to deny triggering
"""
"""Remove hookfile executable bit to deny triggering"""
os.chmod(self.hookfile, stat.S_IREAD | stat.S_IWRITE)
def is_active(self):
"""Check if hook is active by verifying the execute bit
"""
"""Check if hook is active by verifying the execute bit"""
return os.access(self.hookfile, os.X_OK)
@ -290,6 +292,7 @@ class LoggedHook(Hook):
"""A variant of a Hook that allows checking that the hook was called, what
was received via STDIN and what was answered to STDOUT
"""
def __init__(self, *args, **kwargs):
super(LoggedHook, self).__init__(*args, **kwargs)
@ -300,8 +303,7 @@ class LoggedHook(Hook):
self.wrappedname = "original_" + self.hookname
self.wrappedfile = os.path.join(self.hookdir, self.wrappedname)
self.original_wrapper = os.path.join(self.default_hookpath,
"wrapper.sh")
self.original_wrapper = os.path.join(self.default_hookpath, "wrapper.sh")
self.hooklog_in = self.wrappedfile + ".log.in"
self.hooklog_out = self.wrappedfile + ".log.out"
@ -326,11 +328,10 @@ class LoggedHook(Hook):
self._remove_file(self.hooklog_out)
def _setup_wrapper(self):
"""Setup wrapper shell script to allow capturing input/output of hook
"""
"""Setup wrapper shell script to allow capturing input/output of hook"""
# Create empty hooklog to allow checking that hook executed
open(self.hooklog_in, 'w').close()
open(self.hooklog_out, 'w').close()
open(self.hooklog_in, "w").close()
open(self.hooklog_out, "w").close()
# Rename the original hook to the name that will be used by wrapper
self._check_hook_not_exists(self.wrappedfile)
@ -340,8 +341,7 @@ class LoggedHook(Hook):
shutil.copy(self.original_wrapper, self.hookfile)
def _get_log_stat(self):
"""Return the most recent change timestamp and size of both logfiles
"""
"""Return the most recent change timestamp and size of both logfiles"""
stdin = os.stat(self.hooklog_in)
stdout = os.stat(self.hooklog_out)
@ -349,8 +349,7 @@ class LoggedHook(Hook):
return last_change, stdin.st_size, stdout.st_size
def _use_cache(self):
"""Check if log files were changed since last check
"""
"""Check if log files were changed since last check"""
try:
last_change = self._cache["last_change"]
except KeyError:
@ -367,20 +366,17 @@ class LoggedHook(Hook):
return True
def enable(self):
"""Make hookfile executable to allow triggering
"""
"""Make hookfile executable to allow triggering"""
super(LoggedHook, self).enable()
os.chmod(self.wrappedfile, stat.S_IREAD | stat.S_IWRITE | stat.S_IEXEC)
def disable(self):
"""Remove hookfile executable bit to deny triggering
"""
"""Remove hookfile executable bit to deny triggering"""
super(LoggedHook, self).disable()
os.chmod(self.wrappedfile, stat.S_IREAD | stat.S_IWRITE)
def is_active(self):
"""Check if hook is active by verifying the execute bit
"""
"""Check if hook is active by verifying the execute bit"""
parent_is_active = super(LoggedHook, self).disable()
return parent_is_active and os.access(self.wrappedfile, os.X_OK)
@ -407,16 +403,17 @@ class LoggedHook(Hook):
if self._use_cache():
return self._cache["log"]
log = {"calls": [],
"input": {
"json": [],
},
"output": {
"json": [],
"msgs": [],
},
"exitcode": None,
}
log = {
"calls": [],
"input": {
"json": [],
},
"output": {
"json": [],
"msgs": [],
},
"exitcode": None,
}
with open(self.hooklog_in) as fh:
for i, line in enumerate(fh):
@ -426,16 +423,19 @@ class LoggedHook(Hook):
# Timestamp includes nanosecond resolution
timestamp = tstamp.split(" ")[-1]
# convert timestamp to python datetime object
log["calls"].append({
"timestamp": datetime.fromtimestamp(float(timestamp)),
"args": args,
})
log["calls"].append(
{
"timestamp": datetime.fromtimestamp(float(timestamp)),
"args": args,
}
)
elif line.startswith("{"):
# Decode json input (to hook)
log["input"]["json"].append(json_decoder(line))
else:
raise IOError("Unexpected content on STDIN line {0}: {1}"
.format(i, line))
raise IOError(
"Unexpected content on STDIN line {0}: {1}".format(i, line)
)
with open(self.hooklog_out) as fh:
for line in fh:
@ -464,49 +464,43 @@ class LoggedHook(Hook):
"""
log = self.get_logs()
assert len(log["calls"]) == count, ("{0} calls expected for {1} but "
"found {2}".format(
count,
self.hookname,
log["calls"]
))
assert (
len(log["calls"]) == count
), "{0} calls expected for {1} but " "found {2}".format(
count, self.hookname, log["calls"]
)
def assertExitcode(self, exitcode):
"""Check if current hook finished with the expected exit code
"""
"""Check if current hook finished with the expected exit code"""
log = self.get_logs()
assert log["exitcode"] == exitcode, ("Expected exit code {0} for {1} "
"but found {2}".format(
exitcode,
self.hookname,
log["exitcode"]
))
assert (
log["exitcode"] == exitcode
), "Expected exit code {0} for {1} " "but found {2}".format(
exitcode, self.hookname, log["exitcode"]
)
def assertValidJSONOutput(self):
"""Check if current hook output is valid JSON in all expected replies
"""
"""Check if current hook output is valid JSON in all expected replies"""
log = self.get_logs()
for i, out in enumerate(log["output"]["json"]):
assert not isinstance(out, InvalidJSON), ("Invalid JSON found at "
"reply number {0} with "
"content {1}".format(
i + 1,
out.original
))
assert not isinstance(out, InvalidJSON), (
"Invalid JSON found at "
"reply number {0} with "
"content {1}".format(i + 1, out.original)
)
def assertInvalidJSONOutput(self):
"""Check if current hook output is invalid JSON in any expected reply
"""
"""Check if current hook output is invalid JSON in any expected reply"""
log = self.get_logs()
for i, out in enumerate(log["output"]["json"]):
assert isinstance(out, InvalidJSON), ("Valid JSON found at reply "
"number {0} with content "
"{1}".format(
i + 1,
out.original
))
assert isinstance(out, InvalidJSON), (
"Valid JSON found at reply "
"number {0} with content "
"{1}".format(i + 1, out.original)
)
# vim: ai sts=4 et sw=4

2
test/basetest/meta.py
View file

@ -7,6 +7,7 @@ class MetaTest(type):
Creates test_methods in the TestCase class dynamically named after the
arguments used.
"""
@staticmethod
def make_function(classname, *args, **kwargs):
def test(self):
@ -35,4 +36,5 @@ class MetaTest(type):
return super(MetaTest, meta).__new__(meta, classname, bases, dct)
# vim: ai sts=4 et sw=4

69
test/basetest/task.py
View file

@ -21,6 +21,7 @@ class Task(object):
A taskw client should not be used after being destroyed.
"""
DEFAULT_TASK = task_binary_location()
def __init__(self, taskw=DEFAULT_TASK):
@ -43,11 +44,13 @@ class Task(object):
self.reset_env()
with open(self.taskrc, 'w') as rc:
rc.write("data.location={0}\n"
"hooks=off\n"
"news.version=2.6.0\n"
"".format(self.datadir))
with open(self.taskrc, "w") as rc:
rc.write(
"data.location={0}\n"
"hooks=off\n"
"news.version=2.6.0\n"
"".format(self.datadir)
)
# Hooks disabled until requested
self.hooks = None
@ -61,14 +64,12 @@ class Task(object):
return self.runSuccess(*args, **kwargs)
def activate_hooks(self):
"""Enable self.hooks functionality and activate hooks on config
"""
"""Enable self.hooks functionality and activate hooks on config"""
self.config("hooks", "1")
self.hooks = Hooks(self.datadir)
def reset_env(self):
"""Set a new environment derived from the one used to launch the test
"""
"""Set a new environment derived from the one used to launch the test"""
# Copy all env variables to avoid clashing subprocess environments
self.env = os.environ.copy()
@ -78,15 +79,13 @@ class Task(object):
self.env["TASKRC"] = self.taskrc
def config(self, var, value):
"""Run setup `var` as `value` in taskd config
"""
"""Run setup `var` as `value` in taskd config"""
# Add -- to avoid misinterpretation of - in things like UUIDs
cmd = (self.taskw, "config", "--", var, value)
return run_cmd_wait(cmd, env=self.env, input="y\n")
def del_config(self, var):
"""Remove `var` from taskd config
"""
"""Remove `var` from taskd config"""
cmd = (self.taskw, "config", var)
return run_cmd_wait(cmd, env=self.env, input="y\n")
@ -104,8 +103,9 @@ class Task(object):
if export_filter is None:
export_filter = ""
code, out, err = self.runSuccess("rc.json.array=1 {0} export"
"".format(export_filter))
code, out, err = self.runSuccess(
"rc.json.array=1 {0} export" "".format(export_filter)
)
return json.loads(out)
@ -118,16 +118,16 @@ class Task(object):
result = self.export(export_filter=export_filter)
if len(result) != 1:
descriptions = [task.get('description') or '[description-missing]'
for task in result]
descriptions = [
task.get("description") or "[description-missing]" for task in result
]
raise ValueError(
"One task should match the '{0}' filter, '{1}' "
"matches:\n {2}".format(
export_filter or '',
len(result),
'\n '.join(descriptions)
))
export_filter or "", len(result), "\n ".join(descriptions)
)
)
return result[0]
@ -146,8 +146,7 @@ class Task(object):
return args
def runSuccess(self, args="", input=None, merge_streams=False,
timeout=5):
def runSuccess(self, args="", input=None, merge_streams=False, timeout=5):
"""Invoke task with given arguments and fail if exit code != 0
Use runError if you want exit_code to be tested automatically and
@ -171,10 +170,9 @@ class Task(object):
args = self._split_string_args_if_string(args)
command.extend(args)
output = run_cmd_wait_nofail(command, input,
merge_streams=merge_streams,
env=self.env,
timeout=timeout)
output = run_cmd_wait_nofail(
command, input, merge_streams=merge_streams, env=self.env, timeout=timeout
)
if output[0] != 0:
raise CommandError(command, *output)
@ -205,10 +203,9 @@ class Task(object):
args = self._split_string_args_if_string(args)
command.extend(args)
output = run_cmd_wait_nofail(command, input,
merge_streams=merge_streams,
env=self.env,
timeout=timeout)
output = run_cmd_wait_nofail(
command, input, merge_streams=merge_streams, env=self.env, timeout=timeout
)
# output[0] is the exit code
if output[0] == 0 or output[0] is None:
@ -217,8 +214,7 @@ class Task(object):
return output
def destroy(self):
"""Cleanup the data folder and release server port for other instances
"""
"""Cleanup the data folder and release server port for other instances"""
try:
shutil.rmtree(self.datadir)
except OSError as e:
@ -237,8 +233,10 @@ class Task(object):
self.destroy = lambda: None
def __destroyed(self, *args, **kwargs):
raise AttributeError("Task instance has been destroyed. "
"Create a new instance if you need a new client.")
raise AttributeError(
"Task instance has been destroyed. "
"Create a new instance if you need a new client."
)
def diag(self, merge_streams_with=None):
"""Run task diagnostics.
@ -302,4 +300,5 @@ class Task(object):
# Use advanced time format
self._command = [cmd, "-f", faketime] + self._command
# vim: ai sts=4 et sw=4

6
test/basetest/testing.py
View file

@ -7,14 +7,14 @@ class BaseTestCase(unittest.TestCase):
def tap(self, out):
sys.stderr.write("--- tap output start ---\n")
for line in out.splitlines():
sys.stderr.write(line + '\n')
sys.stderr.write(line + "\n")
sys.stderr.write("--- tap output end ---\n")
@unittest.skipIf(TASKW_SKIP, "TASKW_SKIP set, skipping task tests.")
class TestCase(BaseTestCase):
"""Automatically skips tests if TASKW_SKIP is present in the environment
"""
"""Automatically skips tests if TASKW_SKIP is present in the environment"""
pass

88
test/basetest/utils.py
View file

@ -9,11 +9,13 @@ import atexit
import tempfile
from subprocess import Popen, PIPE, STDOUT
from threading import Thread
try:
from Queue import Queue, Empty
except ImportError:
from queue import Queue, Empty
from time import sleep
try:
import simplejson as json
except ImportError:
@ -21,15 +23,13 @@ except ImportError:
from .exceptions import CommandError, TimeoutWaitingFor
USED_PORTS = set()
ON_POSIX = 'posix' in sys.builtin_module_names
ON_POSIX = "posix" in sys.builtin_module_names
# Directory relative to basetest module location
CURRENT_DIR = os.path.dirname(os.path.abspath(__file__))
# Location of binary files (usually the src/ folder)
BIN_PREFIX = os.path.abspath(
os.path.join("${CMAKE_BINARY_DIR}","src")
)
BIN_PREFIX = os.path.abspath(os.path.join("${CMAKE_BINARY_DIR}", "src"))
# Default location of test hooks
DEFAULT_HOOK_PATH = os.path.abspath(
@ -45,7 +45,7 @@ TASKW_SKIP = os.environ.get("TASKW_SKIP", False)
# Environment flags to control use of PATH or in-tree binaries
TASK_USE_PATH = os.environ.get("TASK_USE_PATH", False)
UUID_REGEXP = ("[0-9A-Fa-f]{8}-" + ("[0-9A-Fa-f]{4}-" * 3) + "[0-9A-Fa-f]{12}")
UUID_REGEXP = "[0-9A-Fa-f]{8}-" + ("[0-9A-Fa-f]{4}-" * 3) + "[0-9A-Fa-f]{12}"
def task_binary_location(cmd="task"):
@ -65,9 +65,8 @@ def binary_location(cmd, USE_PATH=False):
return os.path.join(BIN_PREFIX, cmd)
def wait_condition(cond, timeout=10, sleeptime=.01):
"""Wait for condition to return anything other than None
"""
def wait_condition(cond, timeout=10, sleeptime=0.01):
"""Wait for condition to return anything other than None"""
# NOTE Increasing sleeptime can dramatically increase testsuite runtime
# It also reduces CPU load significantly
if timeout is None:
@ -92,8 +91,8 @@ def wait_condition(cond, timeout=10, sleeptime=.01):
def wait_process(pid, timeout=None):
"""Wait for process to finish
"""
"""Wait for process to finish"""
def process():
try:
os.kill(pid, 0)
@ -120,13 +119,18 @@ def _queue_output(arguments, pidq, outputq):
# pid None is read by the main thread as a crash of the process
pidq.put(None)
outputq.put((
"",
("Unexpected exception caught during execution of taskw: '{0}' . "
"If you are running out-of-tree tests set TASK_USE_PATH=1 "
"in shell env before execution and add the "
"location of the task(d) binary to the PATH".format(e)),
255)) # false exitcode
outputq.put(
(
"",
(
"Unexpected exception caught during execution of taskw: '{0}' . "
"If you are running out-of-tree tests set TASK_USE_PATH=1 "
"in shell env before execution and add the "
"location of the task(d) binary to the PATH".format(e)
),
255,
)
) # false exitcode
return
@ -137,15 +141,14 @@ def _queue_output(arguments, pidq, outputq):
out, err = proc.communicate(input_data)
if sys.version_info > (3,):
out, err = out.decode('utf-8'), err.decode('utf-8')
out, err = out.decode("utf-8"), err.decode("utf-8")
# Give the output back to the caller
outputq.put((out, err, proc.returncode))
def _retrieve_output(thread, timeout, queue, thread_error):
"""Fetch output from taskw subprocess queues
"""
"""Fetch output from taskw subprocess queues"""
# Try to join the thread on failure abort
thread.join(timeout)
if thread.is_alive():
@ -184,16 +187,16 @@ def _get_output(arguments, timeout=None):
# Process crashed or timed out for some reason
if pid is None:
return _retrieve_output(t, output_timeout, outputq,
"TaskWarrior to start")
return _retrieve_output(t, output_timeout, outputq, "TaskWarrior to start")
# Wait for process to finish (normal execution)
state = wait_process(pid, timeout)
if state:
# Process finished
return _retrieve_output(t, output_timeout, outputq,
"TaskWarrior thread to join")
return _retrieve_output(
t, output_timeout, outputq, "TaskWarrior thread to join"
)
# If we reach this point we assume the process got stuck or timed out
for sig in (signal.SIGABRT, signal.SIGTERM, signal.SIGKILL):
@ -210,15 +213,21 @@ def _get_output(arguments, timeout=None):
if state:
# Process finished
return _retrieve_output(t, output_timeout, outputq,
"TaskWarrior to die")
return _retrieve_output(t, output_timeout, outputq, "TaskWarrior to die")
# This should never happen but in case something goes really bad
raise OSError("TaskWarrior stopped responding and couldn't be killed")
def run_cmd_wait(cmd, input=None, stdout=PIPE, stderr=PIPE,
merge_streams=False, env=os.environ, timeout=None):
def run_cmd_wait(
cmd,
input=None,
stdout=PIPE,
stderr=PIPE,
merge_streams=False,
env=os.environ,
timeout=None,
):
"Run a subprocess and wait for it to finish"
if input is None:
@ -265,8 +274,7 @@ def run_cmd_wait_nofail(*args, **kwargs):
def memoize(obj):
"""Keep an in-memory cache of function results given its inputs
"""
"""Keep an in-memory cache of function results given its inputs"""
cache = obj.cache = {}
@functools.wraps(obj)
@ -275,11 +283,13 @@ def memoize(obj):
if key not in cache:
cache[key] = obj(*args, **kwargs)
return cache[key]
return memoizer
try:
from shutil import which
which = memoize(which)
except ImportError:
# NOTE: This is shutil.which backported from python-3.3.3
@ -294,12 +304,12 @@ except ImportError:
path.
"""
# Check that a given file can be accessed with the correct mode.
# Additionally check that `file` is not a directory, as on Windows
# directories pass the os.access check.
def _access_check(fn, mode):
return (os.path.exists(fn) and os.access(fn, mode) and
not os.path.isdir(fn))
return os.path.exists(fn) and os.access(fn, mode) and not os.path.isdir(fn)
# If we're given a path with a directory part, look it up directly
# rather than referring to PATH directories. This includes checking
@ -348,16 +358,15 @@ except ImportError:
def parse_datafile(file):
"""Parse .data files on the client and server treating files as JSON
"""
"""Parse .data files on the client and server treating files as JSON"""
data = []
with open(file) as fh:
for line in fh:
line = line.rstrip("\n")
# Turn [] strings into {} to be treated properly as JSON hashes
if line.startswith('[') and line.endswith(']'):
line = '{' + line[1:-1] + '}'
if line.startswith("[") and line.endswith("]"):
line = "{" + line[1:-1] + "}"
if line.startswith("{"):
data.append(json.loads(line))
@ -370,6 +379,7 @@ def mkstemp(data):
"""
Create a temporary file that is removed at process exit
"""
def rmtemp(name):
try:
os.remove(name)
@ -377,7 +387,7 @@ def mkstemp(data):
pass
f = tempfile.NamedTemporaryFile(delete=False)
f.write(data.encode('utf-8') if not isinstance(data, bytes) else data)
f.write(data.encode("utf-8") if not isinstance(data, bytes) else data)
f.close()
# Ensure removal at end of python session
@ -387,11 +397,11 @@ def mkstemp(data):
def mkstemp_exec(data):
"""Create a temporary executable file that is removed at process exit
"""
"""Create a temporary executable file that is removed at process exit"""
name = mkstemp(data)
os.chmod(name, 0o755)
return name
# vim: ai sts=4 et sw=4