Actions Taken Data Unit test.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "The sql extension is already loaded. To reload it, use:\n",
      "  %reload_ext sql\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "u'Connected: buffer@buffermetrics'"
      ]
     },
     "execution_count": 23,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# %load http://bit.do/redshift\n",
    "import os\n",
    "\n",
    "rs_dbname = 'buffermetrics'\n",
    "rs_user = os.environ['BUFFER_REDSHIFT_USER']\n",
    "rs_pwd = os.environ['BUFFER_REDSHIFT_PWD']\n",
    "rs_host = 'buffer-metrics.cgbexruym8z7.us-east-1.redshift.amazonaws.com'\n",
    "rs_port = '5439'\n",
    "\n",
    "%load_ext sql\n",
    "%config SqlMagic.autopandas = True\n",
    "\n",
    "%sql redshift+psycopg2://$rs_user:$rs_pwd@$rs_host:$rs_port/$rs_dbname"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create an IPython magic that will run unit tests"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 73,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "from IPython.core.magic import register_line_magic\n",
    "import pandas as pd\n",
    "\n",
    "@register_line_magic\n",
    "def runtests(line):\n",
    "    \"\"\"\n",
    "    The %runtests magic searches your IPython namespace for functions\n",
    "    with names that begin with 'test'. It will attempt to run these\n",
    "    functions (calling them with no arguments), and report whether they\n",
    "    pass, fail (raise an AssertionError), or error (raise any other\n",
    "    kind of error).\n",
    "    For tests that fail or error %runtests will show the exception raised\n",
    "    but not the traceback, so write informative messages!\n",
    "    \"\"\"\n",
    "    import collections\n",
    "    import time\n",
    "\n",
    "    #setup tables for result\n",
    "    results = []\n",
    "    \n",
    "    ip = get_ipython()\n",
    "\n",
    "    tests = {}\n",
    "\n",
    "    # collect tests\n",
    "    # search will only find functions that start with 'test'\n",
    "    for k, v in ip.user_ns.iteritems():\n",
    "        if k.startswith('test') and isinstance(v, collections.Callable):\n",
    "            tests[k] = v\n",
    "    print 'Collected {} tests.\\n'.format(len(tests))\n",
    "    # run tests\n",
    "    ok = []\n",
    "    fail = {}\n",
    "    error = {}\n",
    "\n",
    "    t1 = time.time()\n",
    "\n",
    "    for name, func in tests.iteritems():\n",
    "        print '{} ... '.format(name),\n",
    "\n",
    "        try:\n",
    "            func()\n",
    "        except AssertionError as e:\n",
    "            results.append({\n",
    "                'name' : name,\n",
    "                'status' : 'fail',\n",
    "                'message' : '{}: {}'.format(name, repr(e))\n",
    "            })\n",
    "        except Exception as e:\n",
    "            results.append({\n",
    "                'name' : name,\n",
    "                'status' : 'error',\n",
    "                'message' : '{}: {}'.format(name, repr(e))\n",
    "            })\n",
    "        else:\n",
    "            results.append({\n",
    "                'name' : name,\n",
    "                'status' : 'success',\n",
    "                'message' : ''\n",
    "            })\n",
    "            \n",
    "    t2 = time.time()\n",
    "\n",
    "    results = pd.DataFrame(results, columns=['name', 'status', 'message'])\n",
    "    def highlight_fail_success(status):\n",
    "        return ['background-color: green;color: white' if v == 'success' else 'background-color: red' for v in status]\n",
    "\n",
    "    results.style.apply(highlight_fail_success)\n",
    "    print ''\n",
    "    print 'Ran {} tests in {:.3g} seconds.'.format(len(tests), t2 - t1)\n",
    "    print 'ok = {}, fail = {}, error = {}'.format(\n",
    "        len(results[results.status == 'ok']), \n",
    "        len(results[results.status == 'fail']), \n",
    "        len(results[results.status == 'error'])\n",
    "    )\n",
    "    \n",
    "    return results.style.apply(highlight_fail_success, subset=['status'])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Define test functions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 77,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def test_pk_is_distinct():\n",
    "    result = %sql select id, count(*) from actions_taken group by id having count(*) > 1\n",
    "    self.assertEqual(len(result), 0)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Run all the test functions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 78,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Collected 1 tests.\n",
      "\n",
      "test_pk_is_distinct ...  0 rows affected.\n",
      "\n",
      "Ran 1 tests in 461 seconds.\n",
      "ok = 0, fail = 0, error = 0\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "\n",
       "        <style  type=\"text/css\" >\n",
       "        \n",
       "        \n",
       "            #T_82b12f70_0146_11e6_acb3_7831c1d32516row0_col1 {\n",
       "            \n",
       "                background-color:  green;\n",
       "            \n",
       "                color:  white;\n",
       "            \n",
       "            }\n",
       "        \n",
       "        </style>\n",
       "\n",
       "        <table id=\"T_82b12f70_0146_11e6_acb3_7831c1d32516\" None>\n",
       "        \n",
       "\n",
       "        <thead>\n",
       "            \n",
       "            <tr>\n",
       "                \n",
       "                <th class=\"blank\">\n",
       "                \n",
       "                <th class=\"col_heading level0 col0\">name\n",
       "                \n",
       "                <th class=\"col_heading level0 col1\">status\n",
       "                \n",
       "                <th class=\"col_heading level0 col2\">message\n",
       "                \n",
       "            </tr>\n",
       "            \n",
       "        </thead>\n",
       "        <tbody>\n",
       "            \n",
       "            <tr>\n",
       "                \n",
       "                <th id=\"T_82b12f70_0146_11e6_acb3_7831c1d32516\" class=\"row_heading level2 row0\">\n",
       "                    0\n",
       "                \n",
       "                <td id=\"T_82b12f70_0146_11e6_acb3_7831c1d32516row0_col0\" class=\"data row0 col0\">\n",
       "                    test_pk_is_distinct\n",
       "                \n",
       "                <td id=\"T_82b12f70_0146_11e6_acb3_7831c1d32516row0_col1\" class=\"data row0 col1\">\n",
       "                    success\n",
       "                \n",
       "                <td id=\"T_82b12f70_0146_11e6_acb3_7831c1d32516row0_col2\" class=\"data row0 col2\">\n",
       "                    \n",
       "                \n",
       "            </tr>\n",
       "            \n",
       "        </tbody>\n",
       "        </table>\n",
       "        "
      ],
      "text/plain": [
       "<pandas.core.style.Styler at 0x11123aa90>"
      ]
     },
     "execution_count": 78,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%runtests"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.11"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}