Started in on unittests for command line utilities

.travis.yml

@@ -5,6 +5,7 @@ python:
   - "3.4"
 install:
   - pip install -r requirements.txt
+  - export PYTHONPATH="$PYTHONPATH:`pwd`"
 script:
   - pep8 storytracker
   - pyflakes storytracker

command.py

@@ -0,0 +1,48 @@
+#! /usr/bin/env python
+import six
+import threading
+import subprocess
+import traceback
+import shlex
+
+
+class Command(object):
+    """
+    Enables to run subprocess commands in a different thread with TIMEOUT option.
+
+    Based on jcollado's solution:
+    http://stackoverflow.com/questions/1191374/subprocess-with-timeout/4825933#4825933
+    """
+    command = None
+    process = None
+    status = None
+    output, error = '', ''
+
+    def __init__(self, command):
+        if isinstance(command, six.string_types):
+            command = shlex.split(command)
+        self.command = command
+
+    def run(self, timeout=None, **kwargs):
+        """ Run a command then return: (status, output, error). """
+        def target(**kwargs):
+            try:
+                self.process = subprocess.Popen(self.command, **kwargs)
+                self.output, self.error = self.process.communicate()
+                self.status = self.process.returncode
+            except:
+                self.error = traceback.format_exc()
+                self.status = -1
+        # default stdout and stderr
+        if 'stdout' not in kwargs:
+            kwargs['stdout'] = subprocess.PIPE
+        if 'stderr' not in kwargs:
+            kwargs['stderr'] = subprocess.PIPE
+        # thread
+        thread = threading.Thread(target=target, kwargs=kwargs)
+        thread.start()
+        thread.join(timeout)
+        if thread.is_alive():
+            self.process.terminate()
+            thread.join()
+        return self.status, self.output, self.error

test.py

@@ -1,16 +1,22 @@
 import os
 import sys
 import six
+import site
 import glob
 import tempfile
 import unittest
+import subprocess
 import storytracker
+from command import Command
 from datetime import datetime
 from bs4 import BeautifulSoup
 from storytracker.analysis import ArchivedURL
 from storytracker.analysis import ArchivedURLSet
 from storytracker.analysis import Hyperlink, Image
 
+#
+# Base tests
+#
 
 class NullDevice():
     """
@@ -33,12 +39,22 @@ def setUp(self):
         self.img = "http://www.trbimg.com/img-5359922b/turbine/\
 la-me-lafd-budget-20140415-001/750/16x9"
         self.tmpdir = tempfile.mkdtemp()
+
+
+class MutedTest(BaseTest):
+
+    def setUp(self):
+        super(MutedTest, self).setUp()
         # Turning off stdout temporarily
         original_stdout = sys.stdout
         sys.stdout = NullDevice()
 
 
-class ArchiveTest(BaseTest):
+#
+# Python tests
+#
+
+class ArchiveTest(MutedTest):
 
     def test_get(self):
         storytracker.get(self.url)
@@ -81,7 +97,7 @@ def test_long_url(self):
         os.remove(obj.archive_path)
 
 
-class AnalysisTest(BaseTest):
+class AnalysisTest(MutedTest):
 
     def test_open_archive_gzip(self):
         obj1 = storytracker.archive(self.url, output_dir=self.tmpdir)
@@ -150,13 +166,13 @@ def test_urlset_creation(self):
         self.assertEqual(len(urlset), 2)
         with self.assertRaises(TypeError):
             urlset.append(1)
-            urlset.append([1,2,3])
+            urlset.append([1, 2, 3])
         with self.assertRaises(ValueError):
             urlset.append(obj)
         urlset.append(obj3)
         self.assertEqual(len(urlset), 3)
         with self.assertRaises(TypeError):
-            ArchivedURLSet([1,2, obj])
+            ArchivedURLSet([1, 2, obj])
         with self.assertRaises(ValueError):
             ArchivedURLSet([obj, obj])
 
@@ -177,6 +193,45 @@ def test_open_archive_directory(self):
         self.assertTrue(len(urlset), 2)
         [self.assertTrue(isinstance(o, ArchivedURL)) for o in urlset]
 
+#
+# CLI tests
+#
+
+class CLITest(BaseTest):
+
+    def setUp(self):
+        super(CLITest, self).setUp()
+        self.this_dir = os.path.dirname(os.path.abspath(__file__))
+        sys.path.append(self.this_dir)
+        self.simple_url = "http://www.example.com"
+
+    def test_get(self):
+        path = os.path.join(self.this_dir, "bin/storytracker-get")
+        cmd = '%s %s' % (path, self.simple_url)
+        process = Command(cmd)
+        code, out, err = process.run(timeout=3)
+        python = storytracker.get(self.simple_url).encode("utf-8")
+        self.assertEqual(type(out), type(python))
+        self.assertEqual(out, python)
+
+    def test_get_no_verify(self):
+        path = os.path.join(self.this_dir, "bin/storytracker-get")
+        cmd = '%s %s --do-not-verify' % (path, self.simple_url)
+        process = Command(cmd)
+        code, out, err = process.run(timeout=3)
+        python = storytracker.get(self.simple_url).encode("utf-8")
+        self.assertEqual(type(out), type(python))
+        self.assertEqual(out, python)
+
+    def test_archive(self):
+        path = os.path.join(self.this_dir, "bin/storytracker-archive")
+        cmd = '%s %s --do-not-compress' % (path, self.simple_url)
+        process = Command(cmd)
+        code, out, err = process.run(timeout=3)
+        python = storytracker.archive(self.simple_url).html.encode("utf-8")
+        self.assertEqual(type(out), type(python))
+        self.assertEqual(out, python)
+
 
 if __name__ == '__main__':
     if six.PY3:

tox.ini

@@ -2,6 +2,8 @@
 envlist=py27,py34
 
 [testenv]
+setenv=
+    PYTHONPATH="$PYTHONPATH:{toxinidir}"
 deps=
   requests
   BeautifulSoup4