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_lazy_list_str():
x = ['I', 'have', 'seen', 'the', 'wind', 'blow']
f = StringIO()
dump(x, f)
f.seek(0)
lj = LazyJSON(f)
assert_equal('the', lj[3])
assert_equal(x[:2:-2], lj[:2:-2])
assert_equal(x, [_ for _ in lj])
assert_equal(x, lj.load())
def test_lazy_dict():
f = StringIO()
ljdump({"wakka": 42}, f)
f.seek(0)
lj = LazyJSON(f)
assert ["wakka"] == list(lj.keys())
assert 42 == lj["wakka"]
assert 1 == len(lj)
assert {"wakka": 42} == lj.load()
def test_lazy_int():
f = StringIO()
dump(42, f)
f.seek(0)
lj = LazyJSON(f)
assert_equal(42, lj.load())
def test_lazy_list_list_ints():
x = [[0, 1], [6, 28], [496, 8128]]
f = StringIO()
ljdump(x, f)
f.seek(0)
lj = LazyJSON(f)
assert isinstance(lj[1], LJNode)
assert 28 == lj[1][1]
assert [6 == 28], lj[1].load()
assert x == lj.load()
def dump(self):
with open(self.filename, 'r', newline='\n') as f:
hist = lazyjson.LazyJSON(f).load()
hist['cmds'].extend(self.buffer)
if self.at_exit:
hist['ts'][1] = time.time() # apply end time
hist['locked'] = False
with open(self.filename, 'w', newline='\n') as f:
lazyjson.dump(hist, f, sort_keys=True)
def __init__(self, f, reopen=True):
"""
Parameters
----------
f : file handle or str
Path to xonsh history file.
reopen : bool, optional
Whether new file handle should be opened for each load, passed directly into
LazyJSON class.
"""
self._lj = LazyJSON(f, reopen=reopen)
(timestamp, number of cmds, file name) tuples.
"""
# pylint: disable=no-member
env = getattr(builtins, "__xonsh_env__", None)
if env is None:
return []
boot = uptime.boottime()
fs = _xhj_get_history_files(sort=False)
files = []
for f in fs:
try:
if os.path.getsize(f) == 0:
# collect empty files (for gc)
files.append((time.time(), 0, f))
continue
lj = xlj.LazyJSON(f, reopen=False)
if lj["locked"] and lj["ts"][0] < boot:
# computer was rebooted between when this history was created
# and now and so this history should be unlocked.
hist = lj.load()
lj.close()
hist["locked"] = False
with open(f, "w", newline="\n") as fp:
xlj.ljdump(hist, fp, sort_keys=True)
lj = xlj.LazyJSON(f, reopen=False)
if only_unlocked and lj["locked"]:
continue
# info: closing timestamp, number of commands, filename
files.append((lj["ts"][1] or lj["ts"][0], len(lj.sizes["cmds"]) - 1, f))
lj.close()
except (IOError, OSError, ValueError):
continue
def unlocked_files(self):
"""Finds the history files and returns the ones that are unlocked, this is
sorted by the last closed time. Returns a list of (timestamp, file) tuples.
"""
xdd = os.path.abspath(builtins.__xonsh_env__.get('XONSH_DATA_DIR'))
fs = [f for f in iglob(os.path.join(xdd, 'xonsh-*.json'))]
files = []
for f in fs:
try:
lj = lazyjson.LazyJSON(f, reopen=False)
if lj['locked']:
continue
# info: closing timestamp, number of commands, filename
files.append((lj['ts'][1], len(lj.sizes['cmds']) - 1, f))
lj.close()
except (IOError, OSError, ValueError):
continue
files.sort()
return files
if isinstance(key, slice):
return [self[i] for i in range(*key.indices(size))]
elif not isinstance(key, int):
raise IndexError('CommandField may only be indexed by int or slice.')
# now we know we have an int
key = size + key if key < 0 else key # ensure key is non-negative
bufsize = len(self.hist.buffer)
if size - bufsize <= key: # key is in buffer
return self.hist.buffer[key + bufsize - size].get(self.field, self.default)
# now we know we have to go into the file
queue = self.hist._queue
queue.append(self)
with self.hist._cond:
self.hist._cond.wait_for(self.i_am_at_the_front)
with open(self.hist.filename, 'r', newline='\n') as f:
lj = lazyjson.LazyJSON(f, reopen=False)
rtn = lj['cmds'][key].get(self.field, self.default)
if isinstance(rtn, lazyjson.Node):
rtn = rtn.load()
queue.popleft()
return rtn
def dump(self):
"""Write the cached history to external storage."""
opts = builtins.__xonsh_env__.get("HISTCONTROL")
last_inp = None
cmds = []
for cmd in self.buffer:
if "ignoredups" in opts and cmd["inp"] == last_inp:
# Skipping dup cmd
continue
if "ignoreerr" in opts and cmd["rtn"] != 0:
# Skipping failed cmd
continue
cmds.append(cmd)
last_inp = cmd["inp"]
with open(self.filename, "r", newline="\n") as f:
hist = xlj.LazyJSON(f).load()
load_hist_len = len(hist["cmds"])
hist["cmds"].extend(cmds)
if self.at_exit:
hist["ts"][1] = time.time() # apply end time
hist["locked"] = False
if not builtins.__xonsh_env__.get("XONSH_STORE_STDOUT", False):
[cmd.pop("out") for cmd in hist["cmds"][load_hist_len:] if "out" in cmd]
with open(self.filename, "w", newline="\n") as f:
xlj.ljdump(hist, f, sort_keys=True)