Add tests for resumed import

beets/importer.py

@@ -1045,16 +1045,6 @@ def finalize(session):
 
         items = task.imported_items()
 
-        # Announce that we've added an album.
-        if task.is_album:
-            album = session.lib.get_album(task.album_id)
-            plugins.send('album_imported',
-                         lib=session.lib, album=album)
-        else:
-            for item in items:
-                plugins.send('item_imported',
-                             lib=session.lib, item=item)
-
         # When copying and deleting originals, delete old files.
         if config['import']['copy'] and config['import']['delete']:
             new_paths = [os.path.realpath(item.path) for item in items]
@@ -1077,6 +1067,16 @@ def finalize(session):
             task.save_history()
         task.cleanup()
 
+        # Announce that we've added an album.
+        if task.is_album:
+            album = session.lib.get_album(task.album_id)
+            plugins.send('album_imported',
+                         lib=session.lib, album=album)
+        else:
+            for item in items:
+                plugins.send('item_imported',
+                             lib=session.lib, item=item)
+
 
 # Singleton pipeline stages.
 

test/helper.py

@@ -45,6 +45,7 @@ import beets.plugins
 from beets.library import Library, Item
 from beets import importer
 from beets.autotag.hooks import AlbumInfo, TrackInfo
+from beets.mediafile import MediaFile
 
 # TODO Move AutotagMock here
 import _common
@@ -177,7 +178,7 @@ class TestHelper(object):
         beets.plugins._classes = set()
         beets.plugins._instances = {}
 
-    def create_importer(self, file_count=1):
+    def create_importer(self, item_count=1, album_count=1):
         """Returns import session with fixtures.
 
         Copies the specified number of files to a subdirectory of
@@ -188,11 +189,18 @@ class TestHelper(object):
         if not os.path.isdir(import_dir):
             os.mkdir(import_dir)
 
-        for i in range(file_count):
-            title = 'track {0}'.format(i)
-            src = os.path.join(_common.RSRC, 'full.mp3')
-            dest = os.path.join(import_dir, '{0}.mp3'.format(title))
-            shutil.copy(src, dest)
+        for i in range(album_count):
+            album = u'album {0}'.format(i)
+            album_dir = os.path.join(import_dir, album)
+            os.mkdir(album_dir)
+            for j in range(item_count):
+                title = 'track {0}'.format(j)
+                src = os.path.join(_common.RSRC, 'full.mp3')
+                dest = os.path.join(album_dir, '{0}.mp3'.format(title))
+                shutil.copy(src, dest)
+                mediafile = MediaFile(dest)
+                mediafile.update({'title': title, 'album': album})
+                mediafile.save()
 
         config['import']['quiet'] = True
         config['import']['autotag'] = False

test/test_importer.py

@@ -20,6 +20,7 @@ import StringIO
 from tempfile import mkstemp
 from zipfile import ZipFile
 from tarfile import TarFile
+from mock import patch
 
 import _common
 from _common import unittest
@@ -1079,6 +1080,53 @@ class TagLogTest(_common.TestCase):
         assert 'status caf' in sio.getvalue()
 
 
+class ResumeImportTest(unittest.TestCase, TestHelper):
+
+    def setUp(self):
+        self.setup_beets()
+
+    def tearDown(self):
+        self.teardown_beets()
+
+    @patch('beets.plugins.send')
+    def test_resume_album(self, plugins_send):
+        self.importer = self.create_importer(album_count=2)
+        self.config['import']['resume'] = True
+
+        def raise_exception(event, **kwargs):
+            if event == 'album_imported':
+                raise importer.ImportAbort
+        plugins_send.side_effect = raise_exception
+
+        self.importer.run()
+        self.assertEqual(len(self.lib.albums()), 1)
+        self.assertIsNotNone(self.lib.albums('album:album 0').get())
+
+        self.importer.run()
+        self.assertEqual(len(self.lib.albums()), 2)
+        self.assertIsNotNone(self.lib.albums('album:album 1').get())
+
+    @unittest.skip('not working yet')
+    @patch('beets.plugins.send')
+    def test_resume_singleton(self, plugins_send):
+        self.importer = self.create_importer(item_count=2)
+        self.config['import']['resume'] = True
+        self.config['import']['singletons'] = True
+
+        def raise_exception(event, **kwargs):
+            if event == 'item_imported':
+                raise importer.ImportAbort
+        plugins_send.side_effect = raise_exception
+
+        self.importer.run()
+        self.assertEqual(len(self.lib.items()), 1)
+        self.assertIsNotNone(self.lib.items('title:track 0').get())
+
+        self.importer.run()
+        self.assertEqual(len(self.lib.items()), 2)
+        self.assertIsNotNone(self.lib.items('title:track 1').get())
+
+
 def suite():
     return unittest.TestLoader().loadTestsFromName(__name__)