Switched to pytest

Author: catherinedevlin

run_tests.sh

@@ -1,3 +1,3 @@
 #!/bin/bash
-ipython -c "import nose; nose.run()"
-# Insert breakpoints with `from nose.tools import set_trace; set_trace()`
+python -c "import pytest; pytest.main(['.', '-x', '--pdb'])"
+# Insert breakpoints with `import pytest; pytest.set_trace()`

src/sql/run.py

@@ -1,17 +1,20 @@
-import operator
-import csv
-import six
 import codecs
+import csv
+import operator
 import os.path
 import re
+
+import prettytable
+import six
 import sqlalchemy
 import sqlparse
-import prettytable
+
+from .column_guesser import ColumnGuesserMixin
+
 try:
     from pgspecial.main import PGSpecial
 except ImportError:
     PGSpecial = None
-from .column_guesser import ColumnGuesserMixin
 
 
 def unduplicate_field_names(field_names):
@@ -26,6 +29,7 @@ def unduplicate_field_names(field_names):
         res.append(k)
     return res
 
+
 class UnicodeWriter(object):
     """
     A CSV writer which will write rows to CSV file "f",
@@ -41,19 +45,17 @@ def __init__(self, f, dialect=csv.excel, encoding="utf-8", **kwds):
 
     def writerow(self, row):
         if six.PY2:
-            _row = [s.encode("utf-8")
-                    if hasattr(s, "encode")
-                    else s
+            _row = [s.encode("utf-8") if hasattr(s, "encode") else s
                     for s in row]
         else:
             _row = row
         self.writer.writerow(_row)
         # Fetch UTF-8 output from the queue ...
         data = self.queue.getvalue()
         if six.PY2:
-           data = data.decode("utf-8")
-           # ... and reencode it into the target encoding
-           data = self.encoder.encode(data)
+            data = data.decode("utf-8")
+            # ... and reencode it into the target encoding
+            data = self.encoder.encode(data)
         # write to the target stream
         self.stream.write(data)
         # empty queue
@@ -64,14 +66,20 @@ def writerows(self, rows):
         for row in rows:
             self.writerow(row)
 
+
 class CsvResultDescriptor(object):
     """Provides IPython Notebook-friendly output for the feedback after a ``.csv`` called."""
+
     def __init__(self, file_path):
         self.file_path = file_path
+
     def __repr__(self):
-        return 'CSV results at %s' % os.path.join(os.path.abspath('.'), self.file_path)
+        return 'CSV results at %s' % os.path.join(
+            os.path.abspath('.'), self.file_path)
+
     def _repr_html_(self):
-        return '<a href="%s">CSV results</a>' % os.path.join('.', 'files', self.file_path)
+        return '<a href="%s">CSV results</a>' % os.path.join('.', 'files',
+                                                             self.file_path)
 
 
 def _nonbreaking_spaces(match_obj):
@@ -84,6 +92,7 @@ def _nonbreaking_spaces(match_obj):
     spaces = '&nbsp;' * len(match_obj.group(2))
     return '%s%s' % (match_obj.group(1), spaces)
 
+
 _cell_with_spaces_pattern = re.compile(r'(<td>)( {2,})')
 
 
@@ -93,6 +102,7 @@ class ResultSet(list, ColumnGuesserMixin):
 
     Can access rows listwise, or by string value of leftmost column.
     """
+
     def __init__(self, sqlaproxy, sql, config):
         self.keys = sqlaproxy.keys()
         self.sql = sql
@@ -118,7 +128,8 @@ def _repr_html_(self):
             self.pretty.add_rows(self)
             result = self.pretty.get_html_string()
             result = _cell_with_spaces_pattern.sub(_nonbreaking_spaces, result)
-            if self.config.displaylimit and len(self) > self.config.displaylimit:
+            if self.config.displaylimit and len(
+                    self) > self.config.displaylimit:
                 result = '%s\n<span style="font-style:italic;text-align:center;">%d rows, truncated to displaylimit of %d</span>' % (
                     result, len(self), self.config.displaylimit)
             return result
@@ -143,6 +154,7 @@ def __getitem__(self, key):
             if len(result) > 1:
                 raise KeyError('%d results for "%s"' % (len(result), key))
             return result[0]
+
     def dict(self):
         """Returns a single dict built from the result set
 
@@ -217,7 +229,7 @@ def plot(self, title=None, **kwargs):
         plt.ylabel(ylabel)
         return plot
 
-    def bar(self, key_word_sep = " ", title=None, **kwargs):
+    def bar(self, key_word_sep=" ", title=None, **kwargs):
         """Generates a pylab bar plot from the result set.
 
         ``matplotlib`` must be installed, and in an
@@ -241,8 +253,7 @@ def bar(self, key_word_sep = " ", title=None, **kwargs):
         self.guess_pie_columns(xlabel_sep=key_word_sep)
         plot = plt.bar(range(len(self.ys[0])), self.ys[0], **kwargs)
         if self.xlabels:
-            plt.xticks(range(len(self.xlabels)), self.xlabels,
-                       rotation=45)
+            plt.xticks(range(len(self.xlabels)), self.xlabels, rotation=45)
         plt.xlabel(self.xlabel)
         plt.ylabel(self.ys[0].name)
         return plot
@@ -251,7 +262,7 @@ def csv(self, filename=None, **format_params):
         """Generate results in comma-separated form.  Write to ``filename`` if given.
            Any other parameters will be passed on to csv.writer."""
         if not self.pretty:
-            return None # no results
+            return None  # no results
         self.pretty.add_rows(self)
         if filename:
             encoding = format_params.get('encoding', 'utf-8')
@@ -279,17 +290,37 @@ def interpret_rowcount(rowcount):
         result = '%d rows affected.' % rowcount
     return result
 
+
 class FakeResultProxy(object):
     """A fake class that pretends to behave like the ResultProxy from
     SqlAlchemy.
     """
+
     def __init__(self, cursor, headers):
         self.fetchall = cursor.fetchall
         self.fetchmany = cursor.fetchmany
         self.rowcount = cursor.rowcount
         self.keys = lambda: headers
         self.returns_rows = True
 
+# some dialects have autocommit
+# specific dialects break when commit is used:
+_COMMIT_BLACKLIST_DIALECTS = ('mssql', 'clickhouse')
+
+
+def _commit(conn, config):
+    """Issues a commit, if appropriate for current config and dialect"""
+
+    _should_commit = config.autocommit and all(
+        dialect not in str(conn.dialect)
+        for dialect in _COMMIT_BLACKLIST_DIALECTS)
+
+    if _should_commit:
+        try:
+            conn.session.execute('commit')
+        except sqlalchemy.exc.OperationalError:
+            pass  # not all engines can commit
+
 
 def run(conn, sql, config, user_namespace):
     if sql.strip():
@@ -302,21 +333,12 @@ def run(conn, sql, config, user_namespace):
                     raise ImportError('pgspecial not installed')
                 pgspecial = PGSpecial()
                 _, cur, headers, _ = pgspecial.execute(
-                                              conn.session.connection.cursor(),
-                                              statement)[0]
+                    conn.session.connection.cursor(), statement)[0]
                 result = FakeResultProxy(cur, headers)
             else:
                 txt = sqlalchemy.sql.text(statement)
                 result = conn.session.execute(txt, user_namespace)
-            try:
-                # some dialects have autocommit
-                # specific dialects break when commit is used:
-                dialects_blacklist = ('mssql', 'clickhouse')
-                if config.autocommit \
-                and all(dialect not in str(conn.dialect) for dialect in dialects_blacklist):
-                    conn.session.execute('commit')
-            except sqlalchemy.exc.OperationalError:
-                pass # not all engines can commit
+            _commit(conn=conn, config=config)
             if result and config.feedback:
                 print(interpret_rowcount(result.rowcount))
         resultset = ResultSet(result, statement, config)
@@ -330,11 +352,10 @@ def run(conn, sql, config, user_namespace):
 
 
 class PrettyTable(prettytable.PrettyTable):
-
     def __init__(self, *args, **kwargs):
         self.row_count = 0
         self.displaylimit = None
-        return super(PrettyTable, self).__init__(*args,  **kwargs)
+        return super(PrettyTable, self).__init__(*args, **kwargs)
 
     def add_rows(self, data):
         if self.row_count and (data.config.displaylimit == self.displaylimit):

src/tests/test_column_guesser.py

@@ -1,46 +1,53 @@
 import re
 import sys
-from nose.tools import with_setup, raises
+
+import pytest
+
 from sql.magic import SqlMagic
 
 ip = get_ipython()
 
+
 class SqlEnv(object):
     def __init__(self, connectstr):
         self.connectstr = connectstr
+
     def query(self, txt):
         return ip.run_line_magic('sql', "%s %s" % (self.connectstr, txt))
 
+
 sql_env = SqlEnv('sqlite://')
 
-def setup():
+
+@pytest.fixture
+def tbl():
     sqlmagic = SqlMagic(shell=ip)
     ip.register_magics(sqlmagic)
     creator = """
         DROP TABLE IF EXISTS manycoltbl;
-        CREATE TABLE manycoltbl 
+        CREATE TABLE manycoltbl
                      (name TEXT, y1 REAL, y2 REAL, name2 TEXT, y3 INT);
-        INSERT INTO manycoltbl VALUES 
+        INSERT INTO manycoltbl VALUES
                      ('r1-txt1', 1.01, 1.02, 'r1-txt2', 1.04);
-        INSERT INTO manycoltbl VALUES 
+        INSERT INTO manycoltbl VALUES
                      ('r2-txt1', 2.01, 2.02, 'r2-txt2', 2.04);
         INSERT INTO manycoltbl VALUES ('r3-txt1', 3.01, 3.02, 'r3-txt2', 3.04);
         """
     for qry in creator.split(";"):
         sql_env.query(qry)
-
-def teardown():
+    yield
     sql_env.query("DROP TABLE manycoltbl")
 
+
 class Harness(object):
     def run_query(self):
         return sql_env.query(self.query)
 
+
 class TestOneNum(Harness):
     query = "SELECT y1 FROM manycoltbl"
-    
-    @with_setup(setup, teardown)
-    def test_pie(self):
+
+    def test_pie(self, tbl):
         results = self.run_query()
         results.guess_pie_columns(xlabel_sep="//")
         assert results.ys[0].is_quantity
@@ -49,28 +56,26 @@ def test_pie(self):
         assert results.xlabels == []
         assert results.xlabel == ''
 
-    @with_setup(setup, teardown)
-    def test_plot(self):
+    def test_plot(self, tbl):
         results = self.run_query()
         results.guess_plot_columns()
         assert results.ys == [[1.01, 2.01, 3.01]]
         assert results.x == []
         assert results.x.name == ''
 
+
 class TestOneStrOneNum(Harness):
     query = "SELECT name, y1 FROM manycoltbl"
-    
-    @with_setup(setup, teardown)
-    def test_pie(self):
+
+    def test_pie(self, tbl):
         results = self.run_query()
         results.guess_pie_columns(xlabel_sep="//")
         assert results.ys[0].is_quantity
         assert results.ys == [[1.01, 2.01, 3.01]]
         assert results.xlabels == ['r1-txt1', 'r2-txt1', 'r3-txt1']
         assert results.xlabel == 'name'
 
-    @with_setup(setup, teardown)
-    def test_plot(self):
+    def test_plot(self, tbl):
         results = self.run_query()
         results.guess_plot_columns()
         assert results.ys == [[1.01, 2.01, 3.01]]
@@ -79,20 +84,19 @@ def test_plot(self):
 
 class TestTwoStrTwoNum(Harness):
     query = "SELECT name2, y3, name, y1 FROM manycoltbl"
-    
-    @with_setup(setup, teardown)
-    def test_pie(self):
+
+    def test_pie(self, tbl):
         results = self.run_query()
         results.guess_pie_columns(xlabel_sep="//")
         assert results.ys[0].is_quantity
         assert results.ys == [[1.01, 2.01, 3.01]]
-        assert results.xlabels == ['r1-txt2//1.04//r1-txt1',
-                                  'r2-txt2//2.04//r2-txt1',
-                                  'r3-txt2//3.04//r3-txt1']
+        assert results.xlabels == [
+            'r1-txt2//1.04//r1-txt1', 'r2-txt2//2.04//r2-txt1',
+            'r3-txt2//3.04//r3-txt1'
+        ]
         assert results.xlabel == 'name2, y3, name'
 
-    @with_setup(setup, teardown)
-    def test_plot(self):
+    def test_plot(self, tbl):
         results = self.run_query()
         results.guess_plot_columns()
         assert results.ys == [[1.01, 2.01, 3.01]]
@@ -101,21 +105,19 @@ def test_plot(self):
 
 class TestTwoStrThreeNum(Harness):
     query = "SELECT name, y1, name2, y2, y3 FROM manycoltbl"
-    
-    @with_setup(setup, teardown)
-    def test_pie(self):
+
+    def test_pie(self, tbl):
         results = self.run_query()
         results.guess_pie_columns(xlabel_sep="//")
         assert results.ys[0].is_quantity
         assert results.ys == [[1.04, 2.04, 3.04]]
-        assert results.xlabels == ['r1-txt1//1.01//r1-txt2//1.02',
-                                  'r2-txt1//2.01//r2-txt2//2.02',
-                                  'r3-txt1//3.01//r3-txt2//3.02']
+        assert results.xlabels == [
+            'r1-txt1//1.01//r1-txt2//1.02', 'r2-txt1//2.01//r2-txt2//2.02',
+            'r3-txt1//3.01//r3-txt2//3.02'
+        ]
 
-    @with_setup(setup, teardown)
-    def test_plot(self):
+    def test_plot(self, tbl):
         results = self.run_query()
         results.guess_plot_columns()
         assert results.ys == [[1.02, 2.02, 3.02], [1.04, 2.04, 3.04]]
         assert results.x == [1.01, 2.01, 3.01]
-