diff --git a/.travis.yml b/.travis.yml index 9dd59d0..088402d 100644 --- a/.travis.yml +++ b/.travis.yml @@ -5,6 +5,7 @@ python: - "3.4" install: - pip install -r requirements.txt + - export PYTHONPATH="$PYTHONPATH:`pwd`" script: - pep8 storytracker - pyflakes storytracker diff --git a/command.py b/command.py new file mode 100644 index 0000000..71a678a --- /dev/null +++ b/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 diff --git a/test.py b/test.py index e9b6811..4b72a9c 100644 --- a/test.py +++ b/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: diff --git a/tox.ini b/tox.ini index c107f47..fca33bf 100644 --- a/tox.ini +++ b/tox.ini @@ -2,6 +2,8 @@ envlist=py27,py34 [testenv] +setenv= + PYTHONPATH="$PYTHONPATH:{toxinidir}" deps= requests BeautifulSoup4