""" testing import """ from collections import namedtuple import pathlib from unittest.mock import patch import datetime import pytz from django.test import TestCase import responses from bookwyrm import models from bookwyrm.importers import Importer from bookwyrm.importers.importer import import_data, handle_imported_book def make_date(*args): """helper function to easily generate a date obj""" return datetime.datetime(*args, tzinfo=pytz.UTC) # pylint: disable=consider-using-with @patch("bookwyrm.suggested_users.rerank_suggestions_task.delay") @patch("bookwyrm.activitystreams.populate_stream_task.delay") @patch("bookwyrm.activitystreams.add_book_statuses_task.delay") class GenericImporter(TestCase): """importing from csv""" def setUp(self): """use a test csv""" self.importer = Importer() datafile = pathlib.Path(__file__).parent.joinpath("../data/generic.csv") self.csv = open(datafile, "r", encoding=self.importer.encoding) with patch("bookwyrm.suggested_users.rerank_suggestions_task.delay"), patch( "bookwyrm.activitystreams.populate_stream_task.delay" ): self.local_user = models.User.objects.create_user( "mouse", "mouse@mouse.mouse", "password", local=True ) work = models.Work.objects.create(title="Test Work") self.book = models.Edition.objects.create( title="Example Edition", remote_id="https://example.com/book/1", parent_work=work, ) def test_create_job(self, *_): """creates the import job entry and checks csv""" import_job = self.importer.create_job( self.local_user, self.csv, False, "public" ) self.assertEqual(import_job.user, self.local_user) self.assertEqual(import_job.include_reviews, False) self.assertEqual(import_job.privacy, "public") import_items = models.ImportItem.objects.filter(job=import_job).all() self.assertEqual(len(import_items), 4) self.assertEqual(import_items[0].index, 0) self.assertEqual(import_items[0].normalized_data["id"], "38") self.assertEqual(import_items[0].normalized_data["title"], "Gideon the Ninth") self.assertEqual(import_items[0].normalized_data["authors"], "Tamsyn Muir") self.assertEqual(import_items[0].normalized_data["isbn_13"], "9781250313195") self.assertIsNone(import_items[0].normalized_data["isbn_10"]) self.assertEqual(import_items[0].normalized_data["shelf"], "read") self.assertEqual(import_items[1].index, 1) self.assertEqual(import_items[1].normalized_data["id"], "48") self.assertEqual(import_items[1].normalized_data["title"], "Harrow the Ninth") self.assertEqual(import_items[2].index, 2) self.assertEqual(import_items[2].normalized_data["id"], "23") self.assertEqual(import_items[2].normalized_data["title"], "Subcutanean") self.assertEqual(import_items[3].index, 3) self.assertEqual(import_items[3].normalized_data["id"], "10") self.assertEqual(import_items[3].normalized_data["title"], "Patisserie at Home") def test_create_retry_job(self, *_): """trying again with items that didn't import""" import_job = self.importer.create_job( self.local_user, self.csv, False, "unlisted" ) import_items = models.ImportItem.objects.filter(job=import_job).all()[:2] retry = self.importer.create_retry_job( self.local_user, import_job, import_items ) self.assertNotEqual(import_job, retry) self.assertEqual(retry.user, self.local_user) self.assertEqual(retry.include_reviews, False) self.assertEqual(retry.privacy, "unlisted") retry_items = models.ImportItem.objects.filter(job=retry).all() self.assertEqual(len(retry_items), 2) self.assertEqual(retry_items[0].index, 0) self.assertEqual(retry_items[0].normalized_data["id"], "38") self.assertEqual(retry_items[1].index, 1) self.assertEqual(retry_items[1].normalized_data["id"], "48") def test_start_import(self, *_): """check that a task was created""" import_job = self.importer.create_job( self.local_user, self.csv, False, "unlisted" ) MockTask = namedtuple("Task", ("id")) mock_task = MockTask(7) with patch("bookwyrm.importers.importer.import_data.delay") as start: start.return_value = mock_task self.importer.start_import(import_job) import_job.refresh_from_db() self.assertEqual(import_job.task_id, "7") @responses.activate def test_import_data(self, *_): """resolve entry""" import_job = self.importer.create_job( self.local_user, self.csv, False, "unlisted" ) book = models.Edition.objects.create(title="Test Book") with patch( "bookwyrm.models.import_job.ImportItem.get_book_from_isbn" ) as resolve: resolve.return_value = book with patch("bookwyrm.importers.importer.handle_imported_book"): import_data(self.importer.service, import_job.id) import_item = models.ImportItem.objects.get(job=import_job, index=0) self.assertEqual(import_item.book.id, book.id) def test_handle_imported_book(self, *_): """import added a book, this adds related connections""" shelf = self.local_user.shelf_set.filter(identifier="read").first() self.assertIsNone(shelf.books.first()) import_job = self.importer.create_job( self.local_user, self.csv, False, "public" ) import_item = import_job.items.first() import_item.book = self.book import_item.save() with patch("bookwyrm.models.activitypub_mixin.broadcast_task.delay"): handle_imported_book( self.importer.service, self.local_user, import_item, False, "public" ) shelf.refresh_from_db() self.assertEqual(shelf.books.first(), self.book) def test_handle_imported_book_already_shelved(self, *_): """import added a book, this adds related connections""" with patch("bookwyrm.models.activitypub_mixin.broadcast_task.delay"): shelf = self.local_user.shelf_set.filter(identifier="to-read").first() models.ShelfBook.objects.create( shelf=shelf, user=self.local_user, book=self.book, shelved_date=make_date(2020, 2, 2), ) import_job = self.importer.create_job( self.local_user, self.csv, False, "unlisted" ) import_item = import_job.items.first() import_item.book = self.book import_item.save() with patch("bookwyrm.models.activitypub_mixin.broadcast_task.delay"): handle_imported_book( self.importer.service, self.local_user, import_item, False, "public" ) shelf.refresh_from_db() self.assertEqual(shelf.books.first(), self.book) self.assertEqual( shelf.shelfbook_set.first().shelved_date, make_date(2020, 2, 2) ) self.assertIsNone( self.local_user.shelf_set.get(identifier="read").books.first() ) def test_handle_import_twice(self, *_): """re-importing books""" shelf = self.local_user.shelf_set.filter(identifier="read").first() import_job = self.importer.create_job( self.local_user, self.csv, False, "public" ) import_item = import_job.items.first() import_item.book = self.book import_item.save() with patch("bookwyrm.models.activitypub_mixin.broadcast_task.delay"): handle_imported_book( self.importer.service, self.local_user, import_item, False, "public" ) handle_imported_book( self.importer.service, self.local_user, import_item, False, "public" ) shelf.refresh_from_db() self.assertEqual(shelf.books.first(), self.book) self.assertEqual(models.ReadThrough.objects.count(), 1) @patch("bookwyrm.activitystreams.add_status_task.delay") def test_handle_imported_book_review(self, *_): """review import""" import_job = self.importer.create_job(self.local_user, self.csv, True, "public") import_item = import_job.items.filter(index=3).first() import_item.book = self.book import_item.save() with patch("bookwyrm.models.activitypub_mixin.broadcast_task.delay"): with patch("bookwyrm.models.Status.broadcast") as broadcast_mock: handle_imported_book( self.importer.service, self.local_user, import_item, True, "unlisted", ) kwargs = broadcast_mock.call_args.kwargs self.assertEqual(kwargs["software"], "bookwyrm") review = models.Review.objects.get(book=self.book, user=self.local_user) self.assertEqual(review.content, "mixed feelings") self.assertEqual(review.rating, 2.0) self.assertEqual(review.privacy, "unlisted") @patch("bookwyrm.activitystreams.add_status_task.delay") def test_handle_imported_book_rating(self, *_): """rating import""" import_job = self.importer.create_job( self.local_user, self.csv, False, "public" ) import_item = import_job.items.filter(index=1).first() import_item.book = self.book import_item.save() with patch("bookwyrm.models.activitypub_mixin.broadcast_task.delay"): handle_imported_book( self.importer.service, self.local_user, import_item, True, "unlisted" ) review = models.ReviewRating.objects.get(book=self.book, user=self.local_user) self.assertIsInstance(review, models.ReviewRating) self.assertEqual(review.rating, 3.0) self.assertEqual(review.privacy, "unlisted") def test_handle_imported_book_reviews_disabled(self, *_): """review import""" import_job = self.importer.create_job( self.local_user, self.csv, False, "unlisted" ) import_item = import_job.items.filter(index=3).first() import_item.book = self.book import_item.save() with patch("bookwyrm.models.activitypub_mixin.broadcast_task.delay"): handle_imported_book( self.importer.service, self.local_user, import_item, False, "unlisted" ) self.assertFalse( models.Review.objects.filter(book=self.book, user=self.local_user).exists() )