Fix broken notifications, add test.

filesorter/filesorter.py

@@ -387,6 +387,28 @@ def process_path(fspath):
     return tvepisodes, other_videos, failed
 
 
+def generic_main(torrent_path, name):
+    """Transmission specific code should become separate."""
+    tvepisodes, other_videos, failed = process_path(torrent_path)
+
+    if not tvepisodes and not other_videos:
+        if not failed:
+            # Extra message when we cannot even find a single video to process
+            log.info('No video files found')
+        send_notification('Download complete', name)
+    else:
+        if tvepisodes and other_videos:
+            subject = 'New Episodes/Videos'
+        elif tvepisodes:
+            subject = 'New Episodes' if len(tvepisodes)+len(failed)>1 else 'New Episode'
+        elif other_videos:
+            subject = 'New Videos' if len(other_videos)+len(failed)>1 else 'New Video'
+        message = ", ".join([e.title for e in tvepisodes]+other_videos)
+        if failed:
+            message += " and %d unknown" % len(failed)
+        send_notification(subject, message)
+
+
 def main():
     # XXX Need to support a non-move mode for torrents that are not compressed.
 
@@ -409,25 +431,7 @@ def main():
         log.error('Given torrent path "%s" does not exist' % torrent_path)
         return 1
 
-    tvepisodes, other_videos, failed = process_path(torrent_path)
-
-    if not tvepisodes and not other_videos:
-        if not failed:
-            # Extra message when we cannot even find a single video to process
-            log.info('No video files found')
-        send_notification('Download complete', torrent['name'])
-    else:
-        if tvepisodes and other_videos:
-            subject = 'New Episodes/Videos'
-        elif tvepisodes:
-            subject = 'New Episodes' if len(tvepisodes)+len(failed)>1 else 'New Episode'
-        elif other_videos:
-            subject = 'New Videos' if len(other_videos)+len(failed)>1 else 'New Video'
-        message = ", ".join(tvepisodes+other_videos)
-        if failed:
-            message += " and %d unknown" % len(failed)
-        send_notification(subject, message)
-
+    generic_main(torrent_path, torrent['name'])
 
 
 def main_wrapper(*a, **kw):

filesorter/tests.py

@@ -3,8 +3,10 @@
 from os import path
 import logging
 import pytest
+import mock
 from py.path import local
-from filesorter import process_path, config, log
+import filesorter
+from filesorter import process_path, config, log, generic_main
 
 
 log.addHandler(logging.StreamHandler(sys.stdout))
@@ -39,6 +41,17 @@ def close():
     return setup_maker
 
 
+@pytest.fixture()
+def notifications(request):
+    """Mock the notification sending."""
+    patcher = mock.patch('filesorter.filesorter.send_notification')
+    patcher.__enter__()
+    def close():
+        patcher.__exit__()
+    request.addfinalizer(close)
+    return filesorter.send_notification
+
+
 def test_agressive_sampler(setup):
     """If the input file contains both the proper video and a sample, we
     need to be sure we sort the real file.
@@ -48,11 +61,20 @@ def test_agressive_sampler(setup):
         u'sampside.er.s02e05.episode.title.72ts.720p.hds.720p.hdtv.x264-bwb-s.mkv',
         (u'show.title.S02E05.episode.title.720p.HDTV.x264-BWB-s.mkv', 2),
     ])
-
     tv_episodes, other_vids, failed = process_path(indir.strpath)
 
     assert other_vids == []
     assert len(failed) == 2
     assert len(tv_episodes) == 1
     assert local(tv_episodes[0].filename).readlines()[0].strip() == \
         'show.title.S02E05.episode.title.720p.HDTV.x264-BWB.mkv'
+
+
+class TestNotification(object):
+
+    def test_send(self, setup, notifications):
+        indir, outdir = setup('show title', [
+            u'show.title.S02E05.episode.title.720p.HDTV.x264-BWB.mkv',
+        ])
+        generic_main(indir.strpath, 'some title here')
+        assert notifications.mock_calls == [mock.call('New Episode', u'show title - 2x05')]