From 3ef4a7aecc0f15f11691931ad8c71ef00bbf2f0e Mon Sep 17 00:00:00 2001 From: Joakim Soderlund Date: Sun, 8 Jan 2017 00:38:38 +0100 Subject: [PATCH] Add writers module --- fimfarchive/writers.py | 181 +++++++++++++++++++++++++++++++++++++++++ tests/test_writers.py | 153 ++++++++++++++++++++++++++++++++++ 2 files changed, 334 insertions(+) create mode 100644 fimfarchive/writers.py create mode 100644 tests/test_writers.py diff --git a/fimfarchive/writers.py b/fimfarchive/writers.py new file mode 100644 index 0000000..0147647 --- /dev/null +++ b/fimfarchive/writers.py @@ -0,0 +1,181 @@ +""" +Writers for Fimfarchive. +""" + + +# +# Fimfarchive, preserves stories from Fimfiction. +# Copyright (C) 2015 Joakim Soderlund +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . +# + + +import json +import os + +from fimfarchive.mappers import StaticMapper, StoryPathMapper + + +__all__ = ( + 'Writer', + 'DirectoryWriter', +) + + +class Writer(): + """ + Abstract base class for story writers. + """ + + def write(self, story): + """ + Saves the story to somewhere. + + Args: + story: Intance of the `Story` class. + + throws: + IOError: If writing the story failed. + """ + raise NotImplementedError() + + +class DirectoryWriter(Writer): + """ + Writes story meta and data to file system. + """ + + def __init__( + self, meta_path=None, data_path=None, + overwrite=False, make_dirs=True): + """ + Constructor. + + Writing of meta and data can be enabled by setting either of + the path parameters. If both path parameters are set to None, + then this writer instance will essentially be doing nothing. + + Args: + meta_path: Directory path, or callable returning a path. + data_path: Directory path, or callable returning a path. + overwrite: Enable to overwrite already existing files. + make_dirs: Enable to create parent directories. + """ + self.meta_path = self.get_mapper(meta_path) + self.data_path = self.get_mapper(data_path) + self.overwrite = overwrite + self.make_dirs = make_dirs + + def get_mapper(self, obj): + """ + Returns a callable for mapping story to file path. + """ + if callable(obj): + return obj + elif isinstance(obj, str): + return StoryPathMapper(obj) + elif obj is None: + return StaticMapper(obj) + else: + raise TypeError("Path must be callable or string.") + + def check_overwrite(self, path): + """ + Checks that a file is not overwritten unless requested. + + Args: + path: File path which is to be written to. + + Raises: + FileExistsError: If overwrite is disabled and path exists. + """ + if not self.overwrite and os.path.exists(path): + raise FileExistsError("Would overwrite: '{}'." .format(path)) + + def check_directory(self, path): + """ + Checks that the path's parent directory exists. + + The parent directory can optionally be created if not. + + Args: + path: File path which is to be written to. + + Raises: + IOError: If the parent directory is a file. + FileNotFoundError: If the parent directory does not exist, + and if directory creation has been disabled. + """ + parent = os.path.dirname(path) + + if os.path.isdir(parent): + return + elif self.make_dirs: + os.makedirs(parent) + else: + raise FileNotFoundError(parent) + + def perform_write(self, contents, path): + """ + Performs the actual file write. + + Args: + contents: Bytes to write. + path: File path to write to. + """ + self.check_overwrite(path) + self.check_directory(path) + + with open(path, 'wb') as fobj: + fobj.write(contents) + + def write_meta(self, story, path): + """ + Prepares the story meta for writing. + + Args: + story: Story containing the meta. + path: File path to write to. + """ + text = json.dumps( + story.meta, + indent=4, + sort_keys=True, + ensure_ascii=False, + ) + + contents = text.encode('utf-8') + self.perform_write(contents, path) + + def write_data(self, story, path): + """ + Prepares the story data for writing. + + Args: + story: Story containing the data. + path: File path to write to. + """ + contents = story.data + self.perform_write(contents, path) + + def write(self, story): + meta_path = self.meta_path(story) + data_path = self.data_path(story) + + if meta_path: + self.write_meta(story, meta_path) + + if data_path: + self.write_data(story, data_path) diff --git a/tests/test_writers.py b/tests/test_writers.py new file mode 100644 index 0000000..2ff0e79 --- /dev/null +++ b/tests/test_writers.py @@ -0,0 +1,153 @@ +""" +Writer tests. +""" + + +# +# Fimfarchive, preserves stories from Fimfiction. +# Copyright (C) 2015 Joakim Soderlund +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . +# + + +import json +import os + +import pytest + +from fimfarchive.mappers import StoryPathMapper +from fimfarchive.writers import DirectoryWriter + + +class TestDirectoryWriter: + """ + DirectoryWriter tests. + """ + + @pytest.fixture + def mapper(self, tmpdir): + """ + Returns a story path mapper for a temporary directory. + """ + return StoryPathMapper(tmpdir) + + def test_story_meta_is_written(self, story, mapper): + """ + Tests story meta is written in its entirety. + """ + writer = DirectoryWriter(meta_path=mapper) + writer.write(story) + + with open(mapper(story), 'rt') as fobj: + assert story.meta == json.load(fobj) + + def test_story_data_is_written(self, story, mapper): + """ + Tests story data is written in its entirety. + """ + writer = DirectoryWriter(data_path=mapper) + writer.write(story) + + with open(mapper(story), 'rb') as fobj: + assert story.data == fobj.read() + + def test_string_paths_become_mappers(self, tmpdir): + """ + Tests `StoryPathMapper` instances are created for string paths. + """ + writer = DirectoryWriter(meta_path='meta', data_path='data') + + assert isinstance(writer.meta_path, StoryPathMapper) + assert isinstance(writer.data_path, StoryPathMapper) + + def test_rejects_integer_path(self): + """ + Tests `TypeError` is raised for invalid path types. + """ + with pytest.raises(TypeError): + DirectoryWriter(meta_path=1) + + def test_parent_directory_creation(self, story, tmpdir): + """ + Tests parent directory is created by default. + """ + directory = str(tmpdir.join('meta')) + + writer = DirectoryWriter(meta_path=directory) + writer.write(story) + + assert os.path.isdir(directory) + + def test_disable_directory_creation(self, story, tmpdir): + """ + Tests `FileNotFoundError` is raised if directory creation is disabled. + """ + directory = str(tmpdir.join('meta')) + + writer = DirectoryWriter(meta_path=directory, make_dirs=False) + + with pytest.raises(FileNotFoundError): + writer.write(story) + + def test_refuse_meta_overwrite(self, story, mapper): + """ + Tests raises `FileExistsError` on meta unless overwrite is enabled. + """ + writer = DirectoryWriter(meta_path=mapper) + + with open(mapper(story), 'at'): + pass + + with pytest.raises(FileExistsError): + writer.write(story) + + def test_refuse_data_overwrite(self, story, mapper): + """ + Tests raises `FileExistsError` on data unless overwrite is enabled. + """ + writer = DirectoryWriter(data_path=mapper) + + with open(mapper(story), 'ab'): + pass + + with pytest.raises(FileExistsError): + writer.write(story) + + def test_overwrites_when_enabled(self, story, tmpdir): + """ + Tests overwrites meta and data when requested. + """ + meta_path = StoryPathMapper(tmpdir.mkdir('meta')) + data_path = StoryPathMapper(tmpdir.mkdir('data')) + + writer = DirectoryWriter( + meta_path=meta_path, + data_path=data_path, + overwrite=True, + ) + + with open(meta_path(story), 'wt'): + pass + + with open(data_path(story), 'wb'): + pass + + writer.write(story) + + with open(meta_path(story), 'rt') as fobj: + assert story.meta == json.load(fobj) + + with open(data_path(story), 'rb') as fobj: + assert story.data == fobj.read()