mirror of
https://github.com/jrnl-org/jrnl.git
synced 2025-07-07 00:36:13 +02:00
Merge branch 'develop' into v2.5
This commit is contained in:
Jonathan Wren
GitHub
commit
97cf65e516
75 changed files with 2578 additions and 1204 deletions
|
|
@ -1,19 +1,30 @@
|
|||
from __future__ import unicode_literals
|
||||
from __future__ import absolute_import
|
||||
from unittest.mock import patch
|
||||
|
||||
from behave import given, when, then
|
||||
from jrnl import cli, install, Journal, util, plugins
|
||||
from jrnl import __version__
|
||||
from dateutil import parser as date_parser
|
||||
from collections import defaultdict
|
||||
|
||||
try:
|
||||
import parsedatetime.parsedatetime_consts as pdt
|
||||
except ImportError:
|
||||
import parsedatetime as pdt
|
||||
import time
|
||||
import os
|
||||
import json
|
||||
import yaml
|
||||
import keyring
|
||||
import tzlocal
|
||||
import shlex
|
||||
import sys
|
||||
|
||||
consts = pdt.Constants(usePyICU=False)
|
||||
consts.DOWParseStyle = -1 # Prefers past weekdays
|
||||
CALENDAR = pdt.Calendar(consts)
|
||||
|
||||
|
||||
class TestKeyring(keyring.backend.KeyringBackend):
|
||||
"""A test keyring that just stores its valies in a hash"""
|
||||
"""A test keyring that just stores its values in a hash"""
|
||||
|
||||
priority = 1
|
||||
keys = defaultdict(dict)
|
||||
|
|
@ -24,42 +35,38 @@ class TestKeyring(keyring.backend.KeyringBackend):
|
|||
def get_password(self, servicename, username):
|
||||
return self.keys[servicename].get(username)
|
||||
|
||||
def delete_password(self, servicename, username, password):
|
||||
def delete_password(self, servicename, username):
|
||||
self.keys[servicename][username] = None
|
||||
|
||||
|
||||
# set the keyring for keyring lib
|
||||
keyring.set_keyring(TestKeyring())
|
||||
|
||||
|
||||
try:
|
||||
from io import StringIO
|
||||
except ImportError:
|
||||
from cStringIO import StringIO
|
||||
import tzlocal
|
||||
import shlex
|
||||
import sys
|
||||
|
||||
|
||||
def ushlex(command):
|
||||
if sys.version_info[0] == 3:
|
||||
return shlex.split(command)
|
||||
return map(lambda s: s.decode('UTF8'), shlex.split(command.encode('utf8')))
|
||||
return map(lambda s: s.decode("UTF8"), shlex.split(command.encode("utf8")))
|
||||
|
||||
|
||||
def read_journal(journal_name="default"):
|
||||
config = util.load_config(install.CONFIG_FILE_PATH)
|
||||
with open(config['journals'][journal_name]) as journal_file:
|
||||
with open(config["journals"][journal_name]) as journal_file:
|
||||
journal = journal_file.read()
|
||||
return journal
|
||||
|
||||
|
||||
def open_journal(journal_name="default"):
|
||||
config = util.load_config(install.CONFIG_FILE_PATH)
|
||||
journal_conf = config['journals'][journal_name]
|
||||
if type(journal_conf) is dict: # We can override the default config on a by-journal basis
|
||||
journal_conf = config["journals"][journal_name]
|
||||
|
||||
if type(journal_conf) is dict:
|
||||
# We can override the default config on a by-journal basis
|
||||
config.update(journal_conf)
|
||||
else: # But also just give them a string to point to the journal file
|
||||
config['journal'] = journal_conf
|
||||
else:
|
||||
# But also just give them a string to point to the journal file
|
||||
config["journal"] = journal_conf
|
||||
|
||||
return Journal.open_journal(journal_name, config)
|
||||
|
||||
|
||||
|
|
@ -69,22 +76,80 @@ def set_config(context, config_file):
|
|||
install.CONFIG_FILE_PATH = os.path.abspath(full_path)
|
||||
if config_file.endswith("yaml"):
|
||||
# Add jrnl version to file for 2.x journals
|
||||
with open(install.CONFIG_FILE_PATH, 'a') as cf:
|
||||
with open(install.CONFIG_FILE_PATH, "a") as cf:
|
||||
cf.write("version: {}".format(__version__))
|
||||
|
||||
|
||||
@when('we open the editor and enter ""')
|
||||
@when('we open the editor and enter "{text}"')
|
||||
def open_editor_and_enter(context, text=""):
|
||||
text = text or context.text
|
||||
|
||||
def _mock_editor_function(command):
|
||||
tmpfile = command[-1]
|
||||
with open(tmpfile, "w+") as f:
|
||||
if text is not None:
|
||||
f.write(text)
|
||||
else:
|
||||
f.write("")
|
||||
|
||||
return tmpfile
|
||||
|
||||
with patch("subprocess.call", side_effect=_mock_editor_function):
|
||||
run(context, "jrnl")
|
||||
|
||||
|
||||
def _mock_getpass(inputs):
|
||||
def prompt_return(prompt="Password: "):
|
||||
print(prompt)
|
||||
return next(inputs)
|
||||
|
||||
return prompt_return
|
||||
|
||||
|
||||
def _mock_input(inputs):
|
||||
def prompt_return(prompt=""):
|
||||
val = next(inputs)
|
||||
print(prompt, val)
|
||||
return val
|
||||
|
||||
return prompt_return
|
||||
|
||||
|
||||
@when('we run "{command}" and enter')
|
||||
@when('we run "{command}" and enter ""')
|
||||
@when('we run "{command}" and enter "{inputs}"')
|
||||
def run_with_input(context, command, inputs=None):
|
||||
text = inputs or context.text
|
||||
def run_with_input(context, command, inputs=""):
|
||||
# create an iterator through all inputs. These inputs will be fed one by one
|
||||
# to the mocked calls for 'input()', 'util.getpass()' and 'sys.stdin.read()'
|
||||
if context.text:
|
||||
text = iter(context.text.split("\n"))
|
||||
else:
|
||||
text = iter([inputs])
|
||||
|
||||
args = ushlex(command)[1:]
|
||||
buffer = StringIO(text.strip())
|
||||
util.STDIN = buffer
|
||||
try:
|
||||
cli.run(args or [])
|
||||
context.exit_status = 0
|
||||
except SystemExit as e:
|
||||
context.exit_status = e.code
|
||||
|
||||
# fmt: off
|
||||
# see: https://github.com/psf/black/issues/557
|
||||
with patch("builtins.input", side_effect=_mock_input(text)) as mock_input, \
|
||||
patch("getpass.getpass", side_effect=_mock_getpass(text)) as mock_getpass, \
|
||||
patch("sys.stdin.read", side_effect=text) as mock_read:
|
||||
|
||||
try:
|
||||
cli.run(args or [])
|
||||
context.exit_status = 0
|
||||
except SystemExit as e:
|
||||
context.exit_status = e.code
|
||||
|
||||
# at least one of the mocked input methods got called
|
||||
assert mock_input.called or mock_getpass.called or mock_read.called
|
||||
# all inputs were used
|
||||
try:
|
||||
next(text)
|
||||
assert False, "Not all inputs were consumed"
|
||||
except StopIteration:
|
||||
pass
|
||||
# fmt: on
|
||||
|
||||
|
||||
@when('we run "{command}"')
|
||||
|
|
@ -106,112 +171,66 @@ def load_template(context, filename):
|
|||
|
||||
@when('we set the keychain password of "{journal}" to "{password}"')
|
||||
def set_keychain(context, journal, password):
|
||||
keyring.set_password('jrnl', journal, password)
|
||||
keyring.set_password("jrnl", journal, password)
|
||||
|
||||
|
||||
@then('we should get an error')
|
||||
@then("we should get an error")
|
||||
def has_error(context):
|
||||
assert context.exit_status != 0, context.exit_status
|
||||
|
||||
|
||||
@then('we should get no error')
|
||||
@then("we should get no error")
|
||||
def no_error(context):
|
||||
assert context.exit_status is 0, context.exit_status
|
||||
assert context.exit_status == 0, context.exit_status
|
||||
|
||||
|
||||
@then('the output should be parsable as json')
|
||||
def check_output_json(context):
|
||||
out = context.stdout_capture.getvalue()
|
||||
assert json.loads(out), out
|
||||
|
||||
|
||||
@then('"{field}" in the json output should have {number:d} elements')
|
||||
@then('"{field}" in the json output should have 1 element')
|
||||
def check_output_field(context, field, number=1):
|
||||
out = context.stdout_capture.getvalue()
|
||||
out_json = json.loads(out)
|
||||
assert field in out_json, [field, out_json]
|
||||
assert len(out_json[field]) == number, len(out_json[field])
|
||||
|
||||
|
||||
@then('"{field}" in the json output should not contain "{key}"')
|
||||
def check_output_field_not_key(context, field, key):
|
||||
out = context.stdout_capture.getvalue()
|
||||
out_json = json.loads(out)
|
||||
assert field in out_json
|
||||
assert key not in out_json[field]
|
||||
|
||||
|
||||
@then('"{field}" in the json output should contain "{key}"')
|
||||
def check_output_field_key(context, field, key):
|
||||
out = context.stdout_capture.getvalue()
|
||||
out_json = json.loads(out)
|
||||
assert field in out_json
|
||||
assert key in out_json[field]
|
||||
|
||||
|
||||
@then('the json output should contain {path} = "{value}"')
|
||||
def check_json_output_path(context, path, value):
|
||||
""" E.g.
|
||||
the json output should contain entries.0.title = "hello"
|
||||
"""
|
||||
out = context.stdout_capture.getvalue()
|
||||
struct = json.loads(out)
|
||||
|
||||
for node in path.split('.'):
|
||||
try:
|
||||
struct = struct[int(node)]
|
||||
except ValueError:
|
||||
struct = struct[node]
|
||||
assert struct == value, struct
|
||||
|
||||
|
||||
@then('the output should be')
|
||||
@then("the output should be")
|
||||
@then('the output should be "{text}"')
|
||||
def check_output(context, text=None):
|
||||
text = (text or context.text).strip().splitlines()
|
||||
out = context.stdout_capture.getvalue().strip().splitlines()
|
||||
assert len(text) == len(out), "Output has {} lines (expected: {})".format(len(out), len(text))
|
||||
assert len(text) == len(out), "Output has {} lines (expected: {})".format(
|
||||
len(out), len(text)
|
||||
)
|
||||
for line_text, line_out in zip(text, out):
|
||||
assert line_text.strip() == line_out.strip(), [line_text.strip(), line_out.strip()]
|
||||
assert line_text.strip() == line_out.strip(), [
|
||||
line_text.strip(),
|
||||
line_out.strip(),
|
||||
]
|
||||
|
||||
|
||||
@then('the output should contain "{text}" in the local time')
|
||||
def check_output_time_inline(context, text):
|
||||
out = context.stdout_capture.getvalue()
|
||||
local_tz = tzlocal.get_localzone()
|
||||
utc_time = date_parser.parse(text)
|
||||
local_date = utc_time.astimezone(local_tz).strftime("%Y-%m-%d %H:%M")
|
||||
assert local_date in out, local_date
|
||||
date, flag = CALENDAR.parse(text)
|
||||
output_date = time.strftime("%Y-%m-%d %H:%M", date)
|
||||
assert output_date in out, output_date
|
||||
|
||||
|
||||
@then('the output should contain')
|
||||
@then("the output should contain")
|
||||
@then('the output should contain "{text}"')
|
||||
def check_output_inline(context, text=None):
|
||||
text = text or context.text
|
||||
out = context.stdout_capture.getvalue()
|
||||
if isinstance(out, bytes):
|
||||
out = out.decode('utf-8')
|
||||
assert text in out, text
|
||||
|
||||
|
||||
@then('the output should not contain "{text}"')
|
||||
def check_output_not_inline(context, text):
|
||||
out = context.stdout_capture.getvalue()
|
||||
if isinstance(out, bytes):
|
||||
out = out.decode('utf-8')
|
||||
assert text not in out
|
||||
|
||||
|
||||
@then('we should see the message "{text}"')
|
||||
def check_message(context, text):
|
||||
out = context.messages.getvalue()
|
||||
out = context.stderr_capture.getvalue()
|
||||
assert text in out, [text, out]
|
||||
|
||||
|
||||
@then('we should not see the message "{text}"')
|
||||
def check_not_message(context, text):
|
||||
out = context.messages.getvalue()
|
||||
out = context.stderr_capture.getvalue()
|
||||
assert text not in out, [text, out]
|
||||
|
||||
|
||||
|
|
@ -226,7 +245,7 @@ def check_journal_content(context, text, journal_name="default"):
|
|||
def journal_doesnt_exist(context, journal_name="default"):
|
||||
with open(install.CONFIG_FILE_PATH) as config_file:
|
||||
config = yaml.load(config_file, Loader=yaml.FullLoader)
|
||||
journal_path = config['journals'][journal_name]
|
||||
journal_path = config["journals"][journal_name]
|
||||
assert not os.path.exists(journal_path)
|
||||
|
||||
|
||||
|
|
@ -234,11 +253,7 @@ def journal_doesnt_exist(context, journal_name="default"):
|
|||
@then('the config for journal "{journal}" should have "{key}" set to "{value}"')
|
||||
def config_var(context, key, value, journal=None):
|
||||
t, value = value.split(":")
|
||||
value = {
|
||||
"bool": lambda v: v.lower() == "true",
|
||||
"int": int,
|
||||
"str": str
|
||||
}[t](value)
|
||||
value = {"bool": lambda v: v.lower() == "true", "int": int, "str": str}[t](value)
|
||||
config = util.load_config(install.CONFIG_FILE_PATH)
|
||||
if journal:
|
||||
config = config["journals"][journal]
|
||||
|
|
@ -246,8 +261,8 @@ def config_var(context, key, value, journal=None):
|
|||
assert config[key] == value
|
||||
|
||||
|
||||
@then('the journal should have {number:d} entries')
|
||||
@then('the journal should have {number:d} entry')
|
||||
@then("the journal should have {number:d} entries")
|
||||
@then("the journal should have {number:d} entry")
|
||||
@then('journal "{journal_name}" should have {number:d} entries')
|
||||
@then('journal "{journal_name}" should have {number:d} entry')
|
||||
def check_journal_entries(context, number, journal_name="default"):
|
||||
|
|
@ -264,6 +279,6 @@ def list_journal_directory(context, journal="default"):
|
|||
for file in f:
|
||||
print(os.path.join(root,file))
|
||||
|
||||
@then('fail')
|
||||
@then("fail")
|
||||
def debug_fail(context):
|
||||
assert False
|
||||
|
|
|
|||
124
features/steps/export_steps.py
Normal file
124
features/steps/export_steps.py
Normal file
|
|
@ -0,0 +1,124 @@
|
|||
import json
|
||||
import os
|
||||
import shutil
|
||||
from xml.etree import ElementTree
|
||||
|
||||
from behave import then, given
|
||||
|
||||
|
||||
@then("the output should be parsable as json")
|
||||
def check_output_json(context):
|
||||
out = context.stdout_capture.getvalue()
|
||||
assert json.loads(out), out
|
||||
|
||||
|
||||
@then('"{field}" in the json output should have {number:d} elements')
|
||||
@then('"{field}" in the json output should have 1 element')
|
||||
def check_output_field(context, field, number=1):
|
||||
out = context.stdout_capture.getvalue()
|
||||
out_json = json.loads(out)
|
||||
assert field in out_json, [field, out_json]
|
||||
assert len(out_json[field]) == number, len(out_json[field])
|
||||
|
||||
|
||||
@then('"{field}" in the json output should not contain "{key}"')
|
||||
def check_output_field_not_key(context, field, key):
|
||||
out = context.stdout_capture.getvalue()
|
||||
out_json = json.loads(out)
|
||||
assert field in out_json
|
||||
assert key not in out_json[field]
|
||||
|
||||
|
||||
@then('"{field}" in the json output should contain "{key}"')
|
||||
def check_output_field_key(context, field, key):
|
||||
out = context.stdout_capture.getvalue()
|
||||
out_json = json.loads(out)
|
||||
assert field in out_json
|
||||
assert key in out_json[field]
|
||||
|
||||
|
||||
@then('the json output should contain {path} = "{value}"')
|
||||
def check_json_output_path(context, path, value):
|
||||
""" E.g.
|
||||
the json output should contain entries.0.title = "hello"
|
||||
"""
|
||||
out = context.stdout_capture.getvalue()
|
||||
struct = json.loads(out)
|
||||
|
||||
for node in path.split("."):
|
||||
try:
|
||||
struct = struct[int(node)]
|
||||
except ValueError:
|
||||
struct = struct[node]
|
||||
assert struct == value, struct
|
||||
|
||||
|
||||
@then("the output should be a valid XML string")
|
||||
def assert_valid_xml_string(context):
|
||||
output = context.stdout_capture.getvalue()
|
||||
xml_tree = ElementTree.fromstring(output)
|
||||
assert xml_tree, output
|
||||
|
||||
|
||||
@then('"entries" node in the xml output should have {number:d} elements')
|
||||
def assert_xml_output_entries_count(context, number):
|
||||
output = context.stdout_capture.getvalue()
|
||||
xml_tree = ElementTree.fromstring(output)
|
||||
|
||||
xml_tags = (node.tag for node in xml_tree)
|
||||
assert "entries" in xml_tags, str(list(xml_tags))
|
||||
|
||||
actual_entry_count = len(xml_tree.find("entries"))
|
||||
assert actual_entry_count == number, actual_entry_count
|
||||
|
||||
|
||||
@then('"tags" in the xml output should contain {expected_tags_json_list}')
|
||||
def assert_xml_output_tags(context, expected_tags_json_list):
|
||||
output = context.stdout_capture.getvalue()
|
||||
xml_tree = ElementTree.fromstring(output)
|
||||
|
||||
xml_tags = (node.tag for node in xml_tree)
|
||||
assert "tags" in xml_tags, str(list(xml_tags))
|
||||
|
||||
expected_tags = json.loads(expected_tags_json_list)
|
||||
actual_tags = set(t.attrib["name"] for t in xml_tree.find("tags"))
|
||||
assert actual_tags == set(expected_tags), [actual_tags, set(expected_tags)]
|
||||
|
||||
|
||||
@given('we created a directory named "{dir_name}"')
|
||||
def create_directory(context, dir_name):
|
||||
if os.path.exists(dir_name):
|
||||
shutil.rmtree(dir_name)
|
||||
os.mkdir(dir_name)
|
||||
|
||||
|
||||
@then('"{dir_name}" should contain the files {expected_files_json_list}')
|
||||
def assert_dir_contains_files(context, dir_name, expected_files_json_list):
|
||||
actual_files = os.listdir(dir_name)
|
||||
expected_files = json.loads(expected_files_json_list)
|
||||
assert actual_files == expected_files, [actual_files, expected_files]
|
||||
|
||||
|
||||
@then('the content of exported yaml "{file_path}" should be')
|
||||
def assert_exported_yaml_file_content(context, file_path):
|
||||
expected_content = context.text.strip().splitlines()
|
||||
|
||||
with open(file_path, "r") as f:
|
||||
actual_content = f.read().strip().splitlines()
|
||||
|
||||
for actual_line, expected_line in zip(actual_content, expected_content):
|
||||
if actual_line.startswith("tags: ") and expected_line.startswith("tags: "):
|
||||
assert_equal_tags_ignoring_order(actual_line, expected_line)
|
||||
else:
|
||||
assert actual_line.strip() == expected_line.strip(), [
|
||||
actual_line.strip(),
|
||||
expected_line.strip(),
|
||||
]
|
||||
|
||||
|
||||
def assert_equal_tags_ignoring_order(actual_line, expected_line):
|
||||
actual_tags = set(tag.strip() for tag in actual_line[len("tags: ") :].split(","))
|
||||
expected_tags = set(
|
||||
tag.strip() for tag in expected_line[len("tags: ") :].split(",")
|
||||
)
|
||||
assert actual_tags == expected_tags, [actual_tags, expected_tags]
|
||||
Loading…
Add table
Add a link
Reference in a new issue