forked from openlp/openlp
Refactor csv importer
This commit is contained in:
Philip Ridout
parent
a78c3d67e7
commit
a0882bd523
@ -24,9 +24,15 @@ The :mod:`~openlp.core.lib.exceptions` module contains custom exceptions
|
||||
"""
|
||||
|
||||
|
||||
# TODO: Test __init__ & __str__
|
||||
class ValidationError(Exception):
|
||||
"""
|
||||
The :class:`~openlp.core.lib.exceptions.ValidationError` exception provides a custom exception for validating
|
||||
import files.
|
||||
"""
|
||||
pass
|
||||
|
||||
def __init__(self, msg="Validation Error"):
|
||||
self.msg = msg
|
||||
|
||||
def __str__(self):
|
||||
return '{error_message}'.format(error_message=self.msg)
|
||||
|
||||
@ -25,7 +25,8 @@ import logging
|
||||
from lxml import etree, objectify
|
||||
|
||||
from openlp.core.common import languages
|
||||
from openlp.plugins.bibles.lib.db import BibleDB
|
||||
from openlp.core.lib import ValidationError
|
||||
from openlp.plugins.bibles.lib.db import BibleDB, BiblesResourcesDB
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
@ -64,6 +65,29 @@ class BibleImport(BibleDB):
|
||||
self.save_meta('language_id', language_id)
|
||||
return language_id
|
||||
|
||||
def find_and_create_book(self, name, no_of_books, language_id, guess_id=None):
|
||||
"""
|
||||
Find the OpenLP book id and then create the book in this bible db
|
||||
|
||||
:param name: Name of the book. If None, then fall back to the guess_id Str
|
||||
:param no_of_books: The total number of books contained in this bible Int
|
||||
:param language_id: The OpenLP id of the language of this bible Int
|
||||
:param guess_id: The guessed id of the book, used if name is None Int
|
||||
:return:
|
||||
"""
|
||||
if name:
|
||||
book_ref_id = self.get_book_ref_id_by_name(name, no_of_books, language_id)
|
||||
else:
|
||||
log.debug('No book name supplied. Falling back to guess_id')
|
||||
book_ref_id = guess_id
|
||||
if not book_ref_id:
|
||||
raise ValidationError(msg='Could not resolve book_ref_id in "{}"'.format(self.filename))
|
||||
book_details = BiblesResourcesDB.get_book_by_id(book_ref_id)
|
||||
if book_details is None:
|
||||
raise ValidationError(msg='book_ref_id: {book_ref} Could not be found in the BibleResourcesDB while '
|
||||
'importing {file}'.format(book_ref=book_ref_id, file=self.filename))
|
||||
return self.create_book(name, book_ref_id, book_details['testament_id'])
|
||||
|
||||
@staticmethod
|
||||
def parse_xml(filename, use_objectify=False, elements=None, tags=None):
|
||||
"""
|
||||
|
||||
@ -49,17 +49,22 @@ There are two acceptable formats of the verses file. They are:
|
||||
|
||||
All CSV files are expected to use a comma (',') as the delimiter and double quotes ('"') as the quote symbol.
|
||||
"""
|
||||
import logging
|
||||
import csv
|
||||
import logging
|
||||
from collections import namedtuple
|
||||
|
||||
from openlp.core.common import translate
|
||||
from openlp.core.lib import get_file_encoding
|
||||
from openlp.core.lib.exceptions import ValidationError
|
||||
from openlp.plugins.bibles.lib.bibleimport import BibleImport
|
||||
from openlp.plugins.bibles.lib.db import BiblesResourcesDB
|
||||
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
Book = namedtuple('Book', 'id, testament_id, name, abbreviation')
|
||||
Verse = namedtuple('Verse', 'book_id_name, chapter_number, number, text')
|
||||
|
||||
|
||||
class CSVBible(BibleImport):
|
||||
"""
|
||||
@ -77,81 +82,107 @@ class CSVBible(BibleImport):
|
||||
self.books_file = kwargs['booksfile']
|
||||
self.verses_file = kwargs['versefile']
|
||||
|
||||
@staticmethod
|
||||
def get_book_name(name, books):
|
||||
"""
|
||||
Normalize a book name or id.
|
||||
|
||||
:param name: The name, or id of a book. Str
|
||||
:param books: A dict of books parsed from the books file.
|
||||
:return: The normalized name. Str
|
||||
"""
|
||||
try:
|
||||
book_name = books[int(name)]
|
||||
except ValueError:
|
||||
book_name = name
|
||||
return book_name
|
||||
|
||||
@staticmethod
|
||||
def parse_csv_file(filename, results_tuple):
|
||||
"""
|
||||
Parse the supplied CSV file.
|
||||
|
||||
:param filename: The name of the file to parse. Str
|
||||
:param results_tuple: The namedtuple to use to store the results. namedtuple
|
||||
:return: An iterable yielding namedtuples of type results_tuple
|
||||
"""
|
||||
try:
|
||||
encoding = get_file_encoding(filename)['encoding']
|
||||
with open(filename, 'r', encoding=encoding, newline='') as csv_file:
|
||||
csv_reader = csv.reader(csv_file, delimiter=',', quotechar='"')
|
||||
return [results_tuple(*line) for line in csv_reader]
|
||||
except (OSError, csv.Error):
|
||||
raise ValidationError(msg='Parsing "{file}" failed'.format(file=filename))
|
||||
|
||||
def process_books(self, books):
|
||||
"""
|
||||
Process the books parsed from the books file.
|
||||
|
||||
:param books: An a list Book namedtuples
|
||||
:return: A dict of books or None
|
||||
"""
|
||||
book_list = {}
|
||||
number_of_books = len(books)
|
||||
for book in books:
|
||||
if self.stop_import_flag:
|
||||
return None
|
||||
self.wizard.increment_progress_bar(
|
||||
translate('BiblesPlugin.CSVBible', 'Importing books... {book}').format(book=book.name))
|
||||
self.find_and_create_book(book.name, number_of_books, self.language_id)
|
||||
book_list.update({int(book.id): book.name})
|
||||
self.application.process_events()
|
||||
return book_list
|
||||
|
||||
def process_verses(self, verses, books):
|
||||
"""
|
||||
Process the verses parsed from the verses file.
|
||||
|
||||
:param verses: A list of Verse namedtuples
|
||||
:param books: A dict of books
|
||||
:return: None
|
||||
"""
|
||||
book_ptr = None
|
||||
for verse in verses:
|
||||
if self.stop_import_flag:
|
||||
return None
|
||||
verse_book = self.get_book_name(verse.book_id_name, books)
|
||||
if book_ptr != verse_book:
|
||||
book = self.get_book(verse_book)
|
||||
book_ptr = book.name
|
||||
self.wizard.increment_progress_bar(
|
||||
translate('BiblesPlugin.CSVBible', 'Importing verses from {book}...',
|
||||
'Importing verses from <book name>...').format(book=book.name))
|
||||
self.session.commit()
|
||||
self.create_verse(book.id, verse.chapter_number, verse.number, verse.text)
|
||||
self.wizard.increment_progress_bar(translate('BiblesPlugin.CSVBible', 'Importing verses... done.'))
|
||||
self.application.process_events()
|
||||
self.session.commit()
|
||||
|
||||
def do_import(self, bible_name=None):
|
||||
"""
|
||||
Import the bible books and verses.
|
||||
Import a bible from the CSV files.
|
||||
|
||||
:param bible_name: Optional name of the bible being imported. Str or None
|
||||
:return: True if the import was successful, False if it failed or was cancelled
|
||||
"""
|
||||
self.wizard.progress_bar.setValue(0)
|
||||
self.wizard.progress_bar.setMinimum(0)
|
||||
self.wizard.progress_bar.setMaximum(66)
|
||||
success = True
|
||||
language_id = self.get_language_id(bible_name=self.books_file)
|
||||
if not language_id:
|
||||
return False
|
||||
books_file = None
|
||||
book_list = {}
|
||||
# Populate the Tables
|
||||
try:
|
||||
details = get_file_encoding(self.books_file)
|
||||
books_file = open(self.books_file, 'r', encoding=details['encoding'])
|
||||
books_reader = csv.reader(books_file, delimiter=',', quotechar='"')
|
||||
for line in books_reader:
|
||||
if self.stop_import_flag:
|
||||
break
|
||||
self.wizard.increment_progress_bar(translate('BiblesPlugin.CSVBible',
|
||||
'Importing books... {text}').format(text=line[2]))
|
||||
book_ref_id = self.get_book_ref_id_by_name(line[2], 67, language_id)
|
||||
if not book_ref_id:
|
||||
log.error('Importing books from "{name}" failed'.format(name=self.books_file))
|
||||
return False
|
||||
book_details = BiblesResourcesDB.get_book_by_id(book_ref_id)
|
||||
self.create_book(line[2], book_ref_id, book_details['testament_id'])
|
||||
book_list.update({int(line[0]): line[2]})
|
||||
self.application.process_events()
|
||||
except (IOError, IndexError):
|
||||
log.exception('Loading books from file failed')
|
||||
success = False
|
||||
finally:
|
||||
if books_file:
|
||||
books_file.close()
|
||||
if self.stop_import_flag or not success:
|
||||
self.language_id = self.get_language(bible_name)
|
||||
if not self.language_id:
|
||||
raise ValidationError(msg='Invalid language selected')
|
||||
books = self.parse_csv_file(self.books_file, Book)
|
||||
self.wizard.progress_bar.setValue(0)
|
||||
self.wizard.progress_bar.setMinimum(0)
|
||||
self.wizard.progress_bar.setMaximum(len(books))
|
||||
book_list = self.process_books(books)
|
||||
if self.stop_import_flag:
|
||||
return False
|
||||
verses = self.parse_csv_file(self.verses_file, Verse)
|
||||
self.wizard.progress_bar.setValue(0)
|
||||
self.wizard.progress_bar.setMaximum(len(books) + 1)
|
||||
self.process_verses(verses, book_list)
|
||||
if self.stop_import_flag:
|
||||
return False
|
||||
except ValidationError:
|
||||
log.exception('Could not import CSV bible')
|
||||
return False
|
||||
self.wizard.progress_bar.setValue(0)
|
||||
self.wizard.progress_bar.setMaximum(67)
|
||||
verse_file = None
|
||||
try:
|
||||
book_ptr = None
|
||||
details = get_file_encoding(self.verses_file)
|
||||
verse_file = open(self.verses_file, 'r', encoding=details['encoding'])
|
||||
verse_reader = csv.reader(verse_file, delimiter=',', quotechar='"')
|
||||
for line in verse_reader:
|
||||
if self.stop_import_flag:
|
||||
break
|
||||
try:
|
||||
line_book = book_list[int(line[0])]
|
||||
except ValueError:
|
||||
line_book = line[0]
|
||||
if book_ptr != line_book:
|
||||
book = self.get_book(line_book)
|
||||
book_ptr = book.name
|
||||
# TODO: Check out this conversion in translations
|
||||
self.wizard.increment_progress_bar(
|
||||
translate('BiblesPlugin.CSVBible',
|
||||
'Importing verses from {name}...'.format(name=book.name),
|
||||
'Importing verses from <book name>...'))
|
||||
self.session.commit()
|
||||
verse_text = line[3]
|
||||
self.create_verse(book.id, line[1], line[2], verse_text)
|
||||
self.wizard.increment_progress_bar(translate('BiblesPlugin.CSVBible', 'Importing verses... done.'))
|
||||
self.application.process_events()
|
||||
self.session.commit()
|
||||
except IOError:
|
||||
log.exception('Loading verses from file failed')
|
||||
success = False
|
||||
finally:
|
||||
if verse_file:
|
||||
verse_file.close()
|
||||
if self.stop_import_flag:
|
||||
return False
|
||||
else:
|
||||
return success
|
||||
return True
|
||||
|
||||
@ -30,6 +30,7 @@ from openlp.plugins.bibles.lib.db import BiblesResourcesDB
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
NS = {'ns': 'http://www.bibletechnologies.net/2003/OSIS/namespace'}
|
||||
# Tags we don't use and can remove the content
|
||||
REMOVABLE_ELEMENTS = ('{http://www.bibletechnologies.net/2003/OSIS/namespace}note',
|
||||
'{http://www.bibletechnologies.net/2003/OSIS/namespace}milestone',
|
||||
@ -88,18 +89,17 @@ class OSISBible(BibleImport):
|
||||
self.wizard.increment_progress_bar(translate('BiblesPlugin.OsisImport',
|
||||
'Removing unused tags (this may take a few minutes)...'))
|
||||
osis_bible_tree = self.parse_xml(self.filename, elements=REMOVABLE_ELEMENTS, tags=REMOVABLE_TAGS)
|
||||
namespace = {'ns': 'http://www.bibletechnologies.net/2003/OSIS/namespace'}
|
||||
# Find bible language]
|
||||
language = osis_bible_tree.xpath("//ns:osisText/@xml:lang", namespaces=namespace)
|
||||
language = osis_bible_tree.xpath("//ns:osisText/@xml:lang", namespaces=NS)
|
||||
language_id = self.get_language_id(language[0] if language else None, bible_name=self.filename)
|
||||
if not language_id:
|
||||
return False
|
||||
num_books = int(osis_bible_tree.xpath("count(//ns:div[@type='book'])", namespaces=namespace))
|
||||
num_books = int(osis_bible_tree.xpath("count(//ns:div[@type='book'])", namespaces=NS))
|
||||
# Precompile a few xpath-querys
|
||||
verse_in_chapter = etree.XPath('count(//ns:chapter[1]/ns:verse)', namespaces=namespace)
|
||||
text_in_verse = etree.XPath('count(//ns:verse[1]/text())', namespaces=namespace)
|
||||
verse_in_chapter = etree.XPath('count(//ns:chapter[1]/ns:verse)', namespaces=NS)
|
||||
text_in_verse = etree.XPath('count(//ns:verse[1]/text())', namespaces=NS)
|
||||
# Find books in the bible
|
||||
bible_books = osis_bible_tree.xpath("//ns:div[@type='book']", namespaces=namespace)
|
||||
bible_books = osis_bible_tree.xpath("//ns:div[@type='book']", namespaces=NS)
|
||||
for book in bible_books:
|
||||
if self.stop_import_flag:
|
||||
break
|
||||
|
||||
@ -23,13 +23,17 @@
|
||||
This module contains tests for the CSV Bible importer.
|
||||
"""
|
||||
|
||||
import os
|
||||
import csv
|
||||
import json
|
||||
import os
|
||||
from collections import namedtuple
|
||||
from unittest import TestCase
|
||||
|
||||
from tests.functional import MagicMock, patch
|
||||
from openlp.plugins.bibles.lib.csvbible import CSVBible
|
||||
from openlp.plugins.bibles.lib.db import BibleDB
|
||||
from tests.functional import ANY, MagicMock, PropertyMock, call, patch
|
||||
from openlp.core.lib.exceptions import ValidationError
|
||||
from openlp.plugins.bibles.lib.bibleimport import BibleImport
|
||||
from openlp.plugins.bibles.lib.csvbible import Book, CSVBible, Verse
|
||||
|
||||
|
||||
TEST_PATH = os.path.abspath(os.path.join(os.path.dirname(__file__),
|
||||
'..', '..', '..', 'resources', 'bibles'))
|
||||
@ -41,14 +45,12 @@ class TestCSVImport(TestCase):
|
||||
"""
|
||||
|
||||
def setUp(self):
|
||||
self.registry_patcher = patch('openlp.plugins.bibles.lib.db.Registry')
|
||||
self.registry_patcher.start()
|
||||
self.manager_patcher = patch('openlp.plugins.bibles.lib.db.Manager')
|
||||
self.registry_patcher = patch('openlp.plugins.bibles.lib.db.Registry')
|
||||
self.addCleanup(self.manager_patcher.stop)
|
||||
self.addCleanup(self.registry_patcher.stop)
|
||||
self.manager_patcher.start()
|
||||
|
||||
def tearDown(self):
|
||||
self.registry_patcher.stop()
|
||||
self.manager_patcher.stop()
|
||||
self.registry_patcher.start()
|
||||
|
||||
def test_create_importer(self):
|
||||
"""
|
||||
@ -58,12 +60,330 @@ class TestCSVImport(TestCase):
|
||||
mocked_manager = MagicMock()
|
||||
|
||||
# WHEN: An importer object is created
|
||||
importer = CSVBible(mocked_manager, path='.', name='.', booksfile='.', versefile='.')
|
||||
importer = CSVBible(mocked_manager, path='.', name='.', booksfile='books.csv', versefile='verse.csv')
|
||||
|
||||
# THEN: The importer should be an instance of BibleDB
|
||||
self.assertIsInstance(importer, BibleDB)
|
||||
# THEN: The importer should be an instance of BibleImport
|
||||
self.assertIsInstance(importer, BibleImport)
|
||||
self.assertEqual(importer.books_file, 'books.csv')
|
||||
self.assertEqual(importer.verses_file, 'verse.csv')
|
||||
|
||||
def test_file_import(self):
|
||||
def book_namedtuple_test(self):
|
||||
"""
|
||||
Test that the Book namedtuple is created as expected
|
||||
"""
|
||||
# GIVEN: The Book namedtuple
|
||||
# WHEN: Creating an instance of Book
|
||||
result = Book('id', 'testament_id', 'name', 'abbreviation')
|
||||
|
||||
# THEN: The attributes should match up with the data we used
|
||||
self.assertEqual(result.id, 'id')
|
||||
self.assertEqual(result.testament_id, 'testament_id')
|
||||
self.assertEqual(result.name, 'name')
|
||||
self.assertEqual(result.abbreviation, 'abbreviation')
|
||||
|
||||
def verse_namedtuple_test(self):
|
||||
"""
|
||||
Test that the Verse namedtuple is created as expected
|
||||
"""
|
||||
# GIVEN: The Verse namedtuple
|
||||
# WHEN: Creating an instance of Verse
|
||||
result = Verse('book_id_name', 'chapter_number', 'number', 'text')
|
||||
|
||||
# THEN: The attributes should match up with the data we used
|
||||
self.assertEqual(result.book_id_name, 'book_id_name')
|
||||
self.assertEqual(result.chapter_number, 'chapter_number')
|
||||
self.assertEqual(result.number, 'number')
|
||||
self.assertEqual(result.text, 'text')
|
||||
|
||||
def get_book_name_id_test(self):
|
||||
"""
|
||||
Test that get_book_name() returns the correct book when called with an id
|
||||
"""
|
||||
# GIVEN: A dictionary of books with their id as the keys
|
||||
books = {1: 'Book 1', 2: 'Book 2', 3: 'Book 3'}
|
||||
|
||||
# WHEN: Calling get_book_name() and the name is an integer represented as a string
|
||||
test_data = [['1', 'Book 1'], ['2', 'Book 2'], ['3', 'Book 3']]
|
||||
for name, expected_result in test_data:
|
||||
actual_result = CSVBible.get_book_name(name, books)
|
||||
|
||||
# THEN: get_book_name() should return the book name associated with that id from the books dictionary
|
||||
self.assertEqual(actual_result, expected_result)
|
||||
|
||||
def get_book_name_test(self):
|
||||
"""
|
||||
Test that get_book_name() returns the name when called with a non integer value
|
||||
"""
|
||||
# GIVEN: A dictionary of books with their id as the keys
|
||||
books = {1: 'Book 1', 2: 'Book 2', 3: 'Book 3'}
|
||||
|
||||
# WHEN: Calling get_book_name() and the name is not an integer represented as a string
|
||||
test_data = [['Book 4', 'Book 4'], ['Book 5', 'Book 5'], ['Book 6', 'Book 6']]
|
||||
for name, expected_result in test_data:
|
||||
actual_result = CSVBible.get_book_name(name, books)
|
||||
|
||||
# THEN: get_book_name() should return the input
|
||||
self.assertEqual(actual_result, expected_result)
|
||||
|
||||
def parse_csv_file_test(self):
|
||||
"""
|
||||
Test the parse_csv_file() with sample data
|
||||
"""
|
||||
# GIVEN: A mocked csv.reader which returns an iterator with test data
|
||||
test_data = [['1', 'Line 1', 'Data 1'], ['2', 'Line 2', 'Data 2'], ['3', 'Line 3', 'Data 3']]
|
||||
TestTuple = namedtuple('TestTuple', 'line_no line_description line_data')
|
||||
|
||||
with patch('openlp.plugins.bibles.lib.csvbible.get_file_encoding',
|
||||
return_value={'encoding': 'utf-8', 'confidence': 0.99}),\
|
||||
patch('openlp.plugins.bibles.lib.csvbible.open', create=True) as mocked_open,\
|
||||
patch('openlp.plugins.bibles.lib.csvbible.csv.reader', return_value=iter(test_data)) as mocked_reader:
|
||||
|
||||
# WHEN: Calling the CSVBible parse_csv_file method with a file name and TestTuple
|
||||
result = CSVBible.parse_csv_file('file.csv', TestTuple)
|
||||
|
||||
# THEN: A list of TestTuple instances with the parsed data should be returned
|
||||
self.assertEqual(result, [TestTuple('1', 'Line 1', 'Data 1'), TestTuple('2', 'Line 2', 'Data 2'),
|
||||
TestTuple('3', 'Line 3', 'Data 3')])
|
||||
mocked_open.assert_called_once_with('file.csv', 'r', encoding='utf-8', newline='')
|
||||
mocked_reader.assert_called_once_with(ANY, delimiter=',', quotechar='"')
|
||||
|
||||
def parse_csv_file_oserror_test(self):
|
||||
"""
|
||||
Test the parse_csv_file() handles an OSError correctly
|
||||
"""
|
||||
# GIVEN: Mocked a mocked open object which raises an OSError
|
||||
with patch('openlp.plugins.bibles.lib.csvbible.get_file_encoding',
|
||||
return_value={'encoding': 'utf-8', 'confidence': 0.99}),\
|
||||
patch('openlp.plugins.bibles.lib.csvbible.open', side_effect=OSError, create=True):
|
||||
|
||||
# WHEN: Calling CSVBible.parse_csv_file
|
||||
# THEN: A ValidationError should be raised
|
||||
with self.assertRaises(ValidationError) as context:
|
||||
CSVBible.parse_csv_file('file.csv', None)
|
||||
self.assertEqual(context.exception.msg, 'Parsing "file.csv" failed')
|
||||
|
||||
def parse_csv_file_csverror_test(self):
|
||||
"""
|
||||
Test the parse_csv_file() handles an csv.Error correctly
|
||||
"""
|
||||
# GIVEN: Mocked a csv.reader which raises an csv.Error
|
||||
with patch('openlp.plugins.bibles.lib.csvbible.get_file_encoding',
|
||||
return_value={'encoding': 'utf-8', 'confidence': 0.99}),\
|
||||
patch('openlp.plugins.bibles.lib.csvbible.open', create=True),\
|
||||
patch('openlp.plugins.bibles.lib.csvbible.csv.reader', side_effect=csv.Error):
|
||||
|
||||
# WHEN: Calling CSVBible.parse_csv_file
|
||||
# THEN: A ValidationError should be raised
|
||||
with self.assertRaises(ValidationError) as context:
|
||||
CSVBible.parse_csv_file('file.csv', None)
|
||||
self.assertEqual(context.exception.msg, 'Parsing "file.csv" failed')
|
||||
|
||||
def process_books_stopped_import_test(self):
|
||||
"""
|
||||
Test process books when the import is stopped
|
||||
"""
|
||||
# GIVEN: An instance of CSVBible with the stop_import_flag set to True
|
||||
mocked_manager = MagicMock()
|
||||
with patch('openlp.plugins.bibles.lib.db.BibleDB._setup'):
|
||||
importer = CSVBible(mocked_manager, path='.', name='.', booksfile='books.csv', versefile='verse.csv')
|
||||
type(importer).application = PropertyMock()
|
||||
importer.stop_import_flag = True
|
||||
importer.wizard = MagicMock()
|
||||
|
||||
# WHEN: Calling process_books
|
||||
result = importer.process_books(['Book 1'])
|
||||
|
||||
# THEN: increment_progress_bar should not be called and the return value should be None
|
||||
self.assertFalse(importer.wizard.increment_progress_bar.called)
|
||||
self.assertIsNone(result)
|
||||
|
||||
def process_books_test(self):
|
||||
"""
|
||||
Test process books when it completes successfully
|
||||
"""
|
||||
# GIVEN: An instance of CSVBible with the stop_import_flag set to False, and some sample data
|
||||
mocked_manager = MagicMock()
|
||||
with patch('openlp.plugins.bibles.lib.db.BibleDB._setup'),\
|
||||
patch('openlp.plugins.bibles.lib.csvbible.translate'):
|
||||
importer = CSVBible(mocked_manager, path='.', name='.', booksfile='books.csv', versefile='verse.csv')
|
||||
type(importer).application = PropertyMock()
|
||||
importer.find_and_create_book = MagicMock()
|
||||
importer.language_id = 10
|
||||
importer.stop_import_flag = False
|
||||
importer.wizard = MagicMock()
|
||||
|
||||
books = [Book('1', '1', '1. Mosebog', '1Mos'), Book('2', '1', '2. Mosebog', '2Mos')]
|
||||
|
||||
# WHEN: Calling process_books
|
||||
result = importer.process_books(books)
|
||||
|
||||
# THEN: translate and find_and_create_book should have been called with both book names.
|
||||
# The returned data should be a dictionary with both song's id and names.
|
||||
self.assertEqual(importer.find_and_create_book.mock_calls,
|
||||
[call('1. Mosebog', 2, 10), call('2. Mosebog', 2, 10)])
|
||||
importer.application.process_events.assert_called_once_with()
|
||||
self.assertDictEqual(result, {1: '1. Mosebog', 2: '2. Mosebog'})
|
||||
|
||||
def process_verses_stopped_import_test(self):
|
||||
"""
|
||||
Test process_verses when the import is stopped
|
||||
"""
|
||||
# GIVEN: An instance of CSVBible with the stop_import_flag set to True
|
||||
mocked_manager = MagicMock()
|
||||
with patch('openlp.plugins.bibles.lib.db.BibleDB._setup'):
|
||||
importer = CSVBible(mocked_manager, path='.', name='.', booksfile='books.csv', versefile='verse.csv')
|
||||
type(importer).application = PropertyMock()
|
||||
importer.get_book_name = MagicMock()
|
||||
importer.session = MagicMock()
|
||||
importer.stop_import_flag = True
|
||||
importer.wizard = MagicMock()
|
||||
|
||||
# WHEN: Calling process_verses
|
||||
result = importer.process_verses([], [])
|
||||
|
||||
# THEN: get_book_name should not be called and the return value should be None
|
||||
self.assertFalse(importer.get_book_name.called)
|
||||
importer.wizard.increment_progress_bar.assert_called_once_with('Importing verses... done.')
|
||||
importer.application.process_events.assert_called_once_with()
|
||||
self.assertIsNone(result)
|
||||
|
||||
def process_verses_successful_test(self):
|
||||
"""
|
||||
Test process_verses when the import is successful
|
||||
"""
|
||||
# GIVEN: An instance of CSVBible with the application and wizard attributes mocked out, and some test data.
|
||||
mocked_manager = MagicMock()
|
||||
with patch('openlp.plugins.bibles.lib.db.BibleDB._setup'),\
|
||||
patch('openlp.plugins.bibles.lib.csvbible.translate'):
|
||||
importer = CSVBible(mocked_manager, path='.', name='.', booksfile='books.csv', versefile='verse.csv')
|
||||
type(importer).application = PropertyMock()
|
||||
importer.create_verse = MagicMock()
|
||||
importer.get_book = MagicMock(return_value=Book('1', '1', '1. Mosebog', '1Mos'))
|
||||
importer.get_book_name = MagicMock(return_value='1. Mosebog')
|
||||
importer.session = MagicMock()
|
||||
importer.stop_import_flag = False
|
||||
importer.wizard = MagicMock()
|
||||
verses = [Verse(1, 1, 1, 'I Begyndelsen skabte Gud Himmelen og Jorden.'),
|
||||
Verse(1, 1, 2, 'Og Jorden var øde og tom, og der var Mørke over Verdensdybet. '
|
||||
'Men Guds Ånd svævede over Vandene.')]
|
||||
books = {1: '1. Mosebog'}
|
||||
|
||||
# WHEN: Calling process_verses
|
||||
importer.process_verses(verses, books)
|
||||
|
||||
# THEN: create_verse is called with the test data
|
||||
self.assertEqual(importer.get_book_name.mock_calls, [call(1, books), call(1, books)])
|
||||
importer.get_book.assert_called_once_with('1. Mosebog')
|
||||
self.assertEqual(importer.session.commit.call_count, 2)
|
||||
self.assertEqual(importer.create_verse.mock_calls,
|
||||
[call('1', 1, 1, 'I Begyndelsen skabte Gud Himmelen og Jorden.'),
|
||||
call('1', 1, 2, 'Og Jorden var øde og tom, og der var Mørke over Verdensdybet. '
|
||||
'Men Guds Ånd svævede over Vandene.')])
|
||||
importer.application.process_events.assert_called_once_with()
|
||||
|
||||
def do_import_invalid_language_id_test(self):
|
||||
"""
|
||||
Test do_import when the user cancels the language selection dialog box
|
||||
"""
|
||||
# GIVEN: An instance of CSVBible and a mocked get_language which simulates the user cancelling the language box
|
||||
mocked_manager = MagicMock()
|
||||
with patch('openlp.plugins.bibles.lib.db.BibleDB._setup'),\
|
||||
patch('openlp.plugins.bibles.lib.csvbible.log') as mocked_log:
|
||||
importer = CSVBible(mocked_manager, path='.', name='.', booksfile='books.csv', versefile='verse.csv')
|
||||
importer.get_language = MagicMock(return_value=None)
|
||||
|
||||
# WHEN: Calling do_import
|
||||
result = importer.do_import('Bible Name')
|
||||
|
||||
# THEN: The log.exception method should have been called to show that it reached the except clause.
|
||||
# False should be returned.
|
||||
importer.get_language.assert_called_once_with('Bible Name')
|
||||
mocked_log.exception.assert_called_once_with('Could not import CSV bible')
|
||||
self.assertFalse(result)
|
||||
|
||||
def do_import_stop_import_test(self):
|
||||
"""
|
||||
Test do_import when the import is stopped
|
||||
"""
|
||||
# GIVEN: An instance of CSVBible with stop_import set to True
|
||||
mocked_manager = MagicMock()
|
||||
with patch('openlp.plugins.bibles.lib.db.BibleDB._setup'),\
|
||||
patch('openlp.plugins.bibles.lib.csvbible.log') as mocked_log:
|
||||
importer = CSVBible(mocked_manager, path='.', name='.', booksfile='books.csv', versefile='verse.csv')
|
||||
importer.get_language = MagicMock(return_value=10)
|
||||
importer.parse_csv_file = MagicMock(return_value=['Book 1', 'Book 2', 'Book 3'])
|
||||
importer.process_books = MagicMock()
|
||||
importer.stop_import_flag = True
|
||||
importer.wizard = MagicMock()
|
||||
|
||||
# WHEN: Calling do_import
|
||||
result = importer.do_import('Bible Name')
|
||||
|
||||
# THEN: log.exception should not be called, parse_csv_file should only be called once,
|
||||
# and False should be returned.
|
||||
self.assertFalse(mocked_log.exception.called)
|
||||
importer.parse_csv_file.assert_called_once_with('books.csv', Book)
|
||||
importer.process_books.assert_called_once_with(['Book 1', 'Book 2', 'Book 3'])
|
||||
self.assertFalse(result)
|
||||
|
||||
def do_import_stop_import_2_test(self):
|
||||
"""
|
||||
Test do_import when the import is stopped
|
||||
"""
|
||||
# GIVEN: An instance of CSVBible with stop_import which is True the second time of calling
|
||||
mocked_manager = MagicMock()
|
||||
with patch('openlp.plugins.bibles.lib.db.BibleDB._setup'),\
|
||||
patch('openlp.plugins.bibles.lib.csvbible.log') as mocked_log:
|
||||
CSVBible.stop_import_flag = PropertyMock(side_effect=[False, True])
|
||||
importer = CSVBible(mocked_manager, path='.', name='.', booksfile='books.csv', versefile='verses.csv')
|
||||
importer.get_language = MagicMock(return_value=10)
|
||||
importer.parse_csv_file = MagicMock(side_effect=[['Book 1'], ['Verse 1']])
|
||||
importer.process_books = MagicMock(return_value=['Book 1'])
|
||||
importer.process_verses = MagicMock(return_value=['Verse 1'])
|
||||
importer.wizard = MagicMock()
|
||||
|
||||
# WHEN: Calling do_import
|
||||
result = importer.do_import('Bible Name')
|
||||
|
||||
# THEN: log.exception should not be called, parse_csv_file should be called twice,
|
||||
# and False should be returned.
|
||||
self.assertFalse(mocked_log.exception.called)
|
||||
self.assertEqual(importer.parse_csv_file.mock_calls, [call('books.csv', Book), call('verses.csv', Verse)])
|
||||
importer.process_verses.assert_called_once_with(['Verse 1'], ['Book 1'])
|
||||
self.assertFalse(result)
|
||||
|
||||
# Cleanup
|
||||
del CSVBible.stop_import_flag
|
||||
|
||||
def do_import_success_test(self):
|
||||
"""
|
||||
Test do_import when the import succeeds
|
||||
"""
|
||||
# GIVEN: An instance of CSVBible
|
||||
mocked_manager = MagicMock()
|
||||
with patch('openlp.plugins.bibles.lib.db.BibleDB._setup'),\
|
||||
patch('openlp.plugins.bibles.lib.csvbible.log') as mocked_log:
|
||||
importer = CSVBible(mocked_manager, path='.', name='.', booksfile='books.csv', versefile='verses.csv')
|
||||
importer.get_language = MagicMock(return_value=10)
|
||||
importer.parse_csv_file = MagicMock(side_effect=[['Book 1'], ['Verse 1']])
|
||||
importer.process_books = MagicMock(return_value=['Book 1'])
|
||||
importer.process_verses = MagicMock(return_value=['Verse 1'])
|
||||
importer.session = MagicMock()
|
||||
importer.stop_import_flag = False
|
||||
importer.wizard = MagicMock()
|
||||
|
||||
# WHEN: Calling do_import
|
||||
result = importer.do_import('Bible Name')
|
||||
|
||||
# THEN: log.exception should not be called, parse_csv_file should be called twice,
|
||||
# and True should be returned.
|
||||
self.assertFalse(mocked_log.exception.called)
|
||||
self.assertEqual(importer.parse_csv_file.mock_calls, [call('books.csv', Book), call('verses.csv', Verse)])
|
||||
importer.process_books.assert_called_once_with(['Book 1'])
|
||||
importer.process_verses.assert_called_once_with(['Verse 1'], ['Book 1'])
|
||||
self.assertTrue(result)
|
||||
|
||||
def file_import_test(self):
|
||||
"""
|
||||
Test the actual import of CSV Bible file
|
||||
"""
|
||||
|
||||
Loading…
Reference in New Issue
Block a user