import bs4
import xml.etree.ElementTree as etree
import io
+import enum
from collections import namedtuple, OrderedDict
from markdown.extensions import Extension
from markdown.blockprocessors import BlockProcessor
class File:
+
+ class Op(enum.Enum):
+ R = enum.auto()
+ W = enum.auto()
+
def __init__(self, path=None, mode="r", content=None, **kwargs):
if path:
if content:
if content and not re.search(r"[r+]", mode):
raise RuntimeError("Cannot specify content without read mode")
- self._rcache = [content] if content else []
- self._wcache = []
+ self._cache = {
+ File.Op.R: [content] if content else [],
+ File.Op.W: []
+ }
+ self._lastop = None
self._mode = mode
self._kwargs = kwargs
self._file = None
def close(self):
self._file.close()
self._file = None
- self._rcache = self._wcache
-
- def _get_rcache(self):
- return (b"" if "b" in self._mode else "").join(self._rcache)
+ self._cache[File.Op.R] = self._cache[File.Op.W]
+ self._lastop = None
- def _get_wcache(self):
- return (b"" if "b" in self._mode else "").join(self._wcache)
+ def _get_cache(self, op):
+ return (b"" if "b" in self._mode else "").join(self._cache[op])
- def _add_to_rcache(self, s):
- self._rcache.append(s)
-
- def _add_to_wcache(self, s):
- self._wcache.append(s)
+ def _add_to_cache(self, op, s):
+ self._cache[op].append(s)
def read(self, *, cache=True):
- if cache and self._rcache:
- return self._get_rcache()
+ if cache and self._cache[File.Op.R]:
+ return self._get_cache(File.Op.R)
+
+ if self._lastop == File.Op.W:
+ try:
+ self._file.seek(0)
+ except io.UnsupportedOperation:
+ pass
- if not self._file:
- with self as f:
- return f.read(cache=cache)
+ self._lastop = File.Op.R
- self._file.seek(0)
if cache:
- self._add_to_rcache(self._file.read())
- return self._get_rcache()
+ self._add_to_cache(File.Op.R, self._file.read())
+ return self._get_cache(File.Op.R)
else:
return self._file.read()
def write(self, s, *, cache=True):
- if not self._file:
- with self as f:
- return f.write(s, cache=cache)
- self._file.seek(0)
- self._rcache = self._wcache
+ if self._lastop == File.Op.R:
+ try:
+ self._file.seek(0)
+ except io.UnsupportedOperation:
+ pass
if cache:
- self._add_to_wcache(s)
+ self._add_to_cache(File.Op.W, s)
+
+ self._cache[File.Op.R] = self._cache[File.Op.W]
written = self._file.write(s)
self._file.flush()
+ self._lastop = File.Op.W
return written
path = property(lambda s: s._path)
path = pathlib.Path(re.split(r" +", lines.pop(0), maxsplit=1)[1])
textsig = "\n".join(lines)
- sig_input = filefactory(path.expanduser()).read()
+ with filefactory(path.expanduser()) as sig_f:
+ sig_input = sig_f.read()
+
soup = bs4.BeautifulSoup(sig_input, "html.parser")
style = str(soup.style.extract()) if soup.style else ""
# variable used to identify the command file we're currently writing
# to.
cmds = MuttCommands(cmd_f, debug=debug_commands)
- cmds.cmd('set editor="$my_editor"')
- cmds.cmd('set edit_headers="$my_edit_headers"')
- cmds.cmd("unset my_editor")
- cmds.cmd("unset my_edit_headers")
-
- # let's flush those commands, as there'll be a lot of pushes from now
- # on, which need to be run in reverse order
- cmds.flush()
extensions = extensions.split(",") if extensions else []
tree = converter(
except AttributeError:
filename = "pytest_internal_file"
cmds.cmd(f"source 'rm -f {filename}|'")
+ cmds.cmd('set editor="$my_editor"')
+ cmds.cmd('set edit_headers="$my_edit_headers"')
+ cmds.cmd("unset my_editor")
+ cmds.cmd("unset my_edit_headers")
cmds.cmd("unset my_mdwn_postprocess_cmd_file")
cmds.flush()
)
lines = cmd_f.read().splitlines()
- assert '="$my_editor"' in lines.pop(0)
- assert '="$my_edit_headers"' in lines.pop(0)
- assert "unset my_editor" == lines.pop(0)
- assert "unset my_edit_headers" == lines.pop(0)
assert "send-message" in lines.pop(0)
assert "update-encoding" in lines.pop(0)
assert "first-entry" in lines.pop(0)
assert "source 'rm -f " in lines.pop(0)
+ assert '="$my_editor"' in lines.pop(0)
+ assert '="$my_edit_headers"' in lines.pop(0)
+ assert "unset my_editor" == lines.pop(0)
+ assert "unset my_edit_headers" == lines.pop(0)
assert "unset my_mdwn_postprocess_cmd_file" == lines.pop(0)
@pytest.mark.massage
max_other_attachments=max_attachments,
converter=converter,
)
- lines = cmd_f.read().splitlines()[4:-2]
+ lines = cmd_f.read().splitlines()[:-6]
assert "first-entry" in lines.pop()
assert "update-encoding" in lines.pop()
cmd_f=cmd_f,
converter=converter,
)
- lines = cmd_f.read().splitlines()[4:-2]
+ lines = cmd_f.read().splitlines()[:-6]
assert "first-entry" in lines.pop()
assert "update-encoding" in lines.pop()
@pytest.mark.converter
def test_converter_tree_basic(self, fakepath, const1, fakefilefactory):
- draft_f = fakefilefactory(fakepath, content=const1)
- tree = convert_markdown_to_html(
- draft_f, filefactory=fakefilefactory
- )
+ with fakefilefactory(fakepath, content=const1) as draft_f:
+ tree = convert_markdown_to_html(
+ draft_f, filefactory=fakefilefactory
+ )
assert tree.subtype == "alternative"
assert len(tree.children) == 2
def test_converter_writes(
self, fakepath, fakefilefactory, const1, monkeypatch
):
- draft_f = fakefilefactory(fakepath, content=const1)
- convert_markdown_to_html(draft_f, filefactory=fakefilefactory)
+ with fakefilefactory(fakepath, content=const1) as draft_f:
+ convert_markdown_to_html(draft_f, filefactory=fakefilefactory)
html = fakefilefactory.pop()
assert fakepath.with_suffix(".html") == html[0]
@pytest.mark.styling
def test_massage_styling_to_converter(self):
css = "p { color:red }"
- css_f = File(content=css)
css_applied = []
def converter(draft_f, css_f, **kwargs):
css_applied.append(css)
return Part("text", "plain", draft_f.path, orig=True)
- do_massage(
- draft_f=File(),
- cmd_f=File(),
- css_f=css_f,
- converter=converter,
- )
+ with (
+ File() as draft_f,
+ File(mode="w") as cmd_f,
+ File(content=css) as css_f
+ ):
+ do_massage(
+ draft_f=draft_f,
+ cmd_f=cmd_f,
+ css_f=css_f,
+ converter=converter,
+ )
assert css_applied[0] == css
@pytest.mark.converter
assert htmlsig == sigconst.format(path=fakepath)
@pytest.mark.sig
- def test_signature_extraction_file_not_found(self, const1):
- path = pathlib.Path("/does/not/exist")
+ def test_signature_extraction_file_not_found(self, fakepath, const1):
with pytest.raises(FileNotFoundError):
origtext, textsig, htmlsig = extract_signature(
- f"{const1}{EMAIL_SIG_SEP}{HTML_SIG_MARKER}{path}\n{const1}"
+ f"{const1}{EMAIL_SIG_SEP}{HTML_SIG_MARKER}{fakepath}\n{const1}"
)
@pytest.mark.imgproc
f.write(const1, cache=False)
assert f.read(cache=False) == const1
+ @pytest.mark.fileio
+ def test_file_class_path_no_exists(self, fakepath):
+ with pytest.raises(FileNotFoundError):
+ File(fakepath, mode="r").open()
+
@pytest.mark.fileio
def test_file_class_cache(self, tmp_path, const1, const2):
path = tmp_path / "file"
projects / etc / neomutt.git / blobdiff