<commit_before>#!/usr/bin/env python

import datetime
import os
import re
import shutil
import sqlite3
import urllib2

from subprocess import Popen, PIPE
from textwrap import dedent

from unittest import TestCase, main

import scraperwiki

# scraperwiki.sql._State.echo = True

DB_NAME = 'scraperwiki.sqlite'

class Setup(TestCase):
    def test_setup(self):
        try:
            os.remove('scraperwiki.sqlite')
        except OSError:
            pass


class TestException(TestCase):
    def testExceptionSaved(self):
        script = dedent("""
            import scraperwiki.runlog
            print scraperwiki.runlog.setup()
            raise ValueError
        """)
        process = Popen(["python", "-c", script],
                        stdout=PIPE, stderr=PIPE, stdin=open("/dev/null"))
        stdout, stderr = process.communicate()

        assert 'Traceback' in stderr, "stderr should contain the original Python traceback"
        match = re.match(r'^\w{8}-\w{4}-\w{4}-\w{4}-\w{12}', stdout)
        assert match, "runlog.setup() should return a run_id"

        l = scraperwiki.sql.select(
            "exception_type, run_id, time from _sw_runlog order by time desc limit 1")

        # Check that some record is stored.
        assert l
        # Check that the exception name appears.
        assert 'ValueError' in l[0][
            'exception_type'], "runlog should save exception types to the database"
        # Check that the run_id from earlier has been saved.
        assert match.group() == l[0].get(
            'run_id'), "runlog should save a run_id to the database"
        # Check that the time recorded is relatively recent.
        time_str = l[0]['time']
        then = datetime.datetime.strptime(time_str, '%Y-%m-%d %H:%M:%S.%f')
        assert (datetime.datetime.now() - then).total_seconds() < 5 * \
            60, "run log should save a time to the database"

    def testRunlogSuccess(self):
        script = dedent("""
            import scraperwiki.runlog
            print scraperwiki.runlog.setup()
        """)
        process = Popen(["python", "-c", script],
                        stdout=PIPE, stderr=PIPE, stdin=open("/dev/null"))
        stdout, stderr = process.communicate()

        l = scraperwiki.sql.select(
            "time, run_id, success from _sw_runlog order by time desc limit 1")

        # Check that some record is stored.
        assert l
        # Check that it has saved a success column.
        assert l[0]['success']
        # Check that a run_id has been saved.
        match = re.match(r'^\w{8}-\w{4}-\w{4}-\w{4}-\w{12}', stdout)
        assert match.group() == l[0].get(
            'run_id'), "runlog should save a run_id to the database"
        # Check that the time is relatively recent.
        then = datetime.datetime.strptime(l[0]['time'], '%Y-%m-%d %H:%M:%S.%f')
        assert (datetime.datetime.now() - then).total_seconds() < 5 * 60

class TestSaveGetVar(TestCase):
    def savegetvar(self, var):
        scraperwiki.sql.save_var("weird", var)
        self.assertEqual(scraperwiki.sql.get_var("weird"), var)

    def test_string(self):
        self.savegetvar("asdio")

    def test_int(self):
        self.savegetvar(1)

    def test_date(self):
        date1 = datetime.datetime.now()
        date2 = datetime.date.today()
        scraperwiki.sql.save_var("weird", date1)
        self.assertEqual(scraperwiki.sql.get_var("weird"), unicode(date1))
        scraperwiki.sql.save_var("weird", date2)
        self.assertEqual(scraperwiki.sql.get_var("weird"), unicode(date2))

    def test_save_multiple_values(self):
        scraperwiki.sql.save_var('foo', 'hello')
        scraperwiki.sql.save_var('bar', 'goodbye')

        self.assertEqual('hello', scraperwiki.sql.get_var('foo'))
        self.assertEqual('goodbye', scraperwiki.sql.get_var('bar'))

class TestGetNonexistantVar(TestCase):
    def test_get(self):
        self.assertIsNone(scraperwiki.sql.get_var('meatball'))

class TestSaveVar(TestCase):
    def setUp(self):
        super(TestSaveVar, self).setUp()
        scraperwiki.sql.save_var("birthday", "November 30, 1888")
        connection = sqlite3.connect(DB_NAME)
        self.cursor = connection.cursor()

    def test_insert(self):
        self.cursor.execute("""
          SELECT name, value_blob, type
          FROM `swvariables`
          WHERE name == "birthday"
          """)
        observed = self.cursor.fetchall()
        expected = [("birthday", "November 30, 1888", "text",)]
        ((a, b, c),) = observed
        observed = [(a, str(b), c)]
        self.assertEqual(observed, expected)

class SaveAndCheck(TestCase):
    def save_and_check(self, dataIn, tableIn, dataOut, tableOut=None, twice=True):
        if tableOut == None:
            tableOut = '[' + tableIn + ']'

        # Insert
        with scraperwiki.sql.Transaction():
            scraperwiki.sql.save([], dataIn, tableIn)

        # Observe with pysqlite
        connection = sqlite3.connect(DB_NAME)
        cursor = connection.cursor()
        cursor.execute("SELECT * FROM %s" % tableOut)
        observed1 = cursor.fetchall()
        connection.close()

        if twice:
            # Observe with DumpTruck
            observed2 = scraperwiki.sql.select('* FROM %s' % tableOut)

            # Check
            expected1 = dataOut
            expected2 = [dataIn] if type(dataIn) == dict else dataIn

            self.assertListEqual(observed1, expected1)
            self.assertListEqual(observed2, expected2)

class SaveAndSelect(TestCase):
    def save_and_select(self, d):
        scraperwiki.sql.save([], {"foo": d})
        observed = scraperwiki.sql.select('* from swdata')[0]['foo']
        self.assertEqual(d, observed)
<commit_msg>Remove reference to Dumptruck in comments<commit_after>#!/usr/bin/env python

import datetime
import os
import re
import shutil
import sqlite3
import urllib2

from subprocess import Popen, PIPE
from textwrap import dedent

from unittest import TestCase, main

import scraperwiki

# scraperwiki.sql._State.echo = True

DB_NAME = 'scraperwiki.sqlite'

class Setup(TestCase):
    def test_setup(self):
        try:
            os.remove('scraperwiki.sqlite')
        except OSError:
            pass


class TestException(TestCase):
    def testExceptionSaved(self):
        script = dedent("""
            import scraperwiki.runlog
            print scraperwiki.runlog.setup()
            raise ValueError
        """)
        process = Popen(["python", "-c", script],
                        stdout=PIPE, stderr=PIPE, stdin=open("/dev/null"))
        stdout, stderr = process.communicate()

        assert 'Traceback' in stderr, "stderr should contain the original Python traceback"
        match = re.match(r'^\w{8}-\w{4}-\w{4}-\w{4}-\w{12}', stdout)
        assert match, "runlog.setup() should return a run_id"

        l = scraperwiki.sql.select(
            "exception_type, run_id, time from _sw_runlog order by time desc limit 1")

        # Check that some record is stored.
        assert l
        # Check that the exception name appears.
        assert 'ValueError' in l[0][
            'exception_type'], "runlog should save exception types to the database"
        # Check that the run_id from earlier has been saved.
        assert match.group() == l[0].get(
            'run_id'), "runlog should save a run_id to the database"
        # Check that the time recorded is relatively recent.
        time_str = l[0]['time']
        then = datetime.datetime.strptime(time_str, '%Y-%m-%d %H:%M:%S.%f')
        assert (datetime.datetime.now() - then).total_seconds() < 5 * \
            60, "run log should save a time to the database"

    def testRunlogSuccess(self):
        script = dedent("""
            import scraperwiki.runlog
            print scraperwiki.runlog.setup()
        """)
        process = Popen(["python", "-c", script],
                        stdout=PIPE, stderr=PIPE, stdin=open("/dev/null"))
        stdout, stderr = process.communicate()

        l = scraperwiki.sql.select(
            "time, run_id, success from _sw_runlog order by time desc limit 1")

        # Check that some record is stored.
        assert l
        # Check that it has saved a success column.
        assert l[0]['success']
        # Check that a run_id has been saved.
        match = re.match(r'^\w{8}-\w{4}-\w{4}-\w{4}-\w{12}', stdout)
        assert match.group() == l[0].get(
            'run_id'), "runlog should save a run_id to the database"
        # Check that the time is relatively recent.
        then = datetime.datetime.strptime(l[0]['time'], '%Y-%m-%d %H:%M:%S.%f')
        assert (datetime.datetime.now() - then).total_seconds() < 5 * 60

class TestSaveGetVar(TestCase):
    def savegetvar(self, var):
        scraperwiki.sql.save_var("weird", var)
        self.assertEqual(scraperwiki.sql.get_var("weird"), var)

    def test_string(self):
        self.savegetvar("asdio")

    def test_int(self):
        self.savegetvar(1)

    def test_date(self):
        date1 = datetime.datetime.now()
        date2 = datetime.date.today()
        scraperwiki.sql.save_var("weird", date1)
        self.assertEqual(scraperwiki.sql.get_var("weird"), unicode(date1))
        scraperwiki.sql.save_var("weird", date2)
        self.assertEqual(scraperwiki.sql.get_var("weird"), unicode(date2))

    def test_save_multiple_values(self):
        scraperwiki.sql.save_var('foo', 'hello')
        scraperwiki.sql.save_var('bar', 'goodbye')

        self.assertEqual('hello', scraperwiki.sql.get_var('foo'))
        self.assertEqual('goodbye', scraperwiki.sql.get_var('bar'))

class TestGetNonexistantVar(TestCase):
    def test_get(self):
        self.assertIsNone(scraperwiki.sql.get_var('meatball'))

class TestSaveVar(TestCase):
    def setUp(self):
        super(TestSaveVar, self).setUp()
        scraperwiki.sql.save_var("birthday", "November 30, 1888")
        connection = sqlite3.connect(DB_NAME)
        self.cursor = connection.cursor()

    def test_insert(self):
        self.cursor.execute("""
          SELECT name, value_blob, type
          FROM `swvariables`
          WHERE name == "birthday"
          """)
        observed = self.cursor.fetchall()
        expected = [("birthday", "November 30, 1888", "text",)]
        ((a, b, c),) = observed
        observed = [(a, str(b), c)]
        self.assertEqual(observed, expected)

class SaveAndCheck(TestCase):
    def save_and_check(self, dataIn, tableIn, dataOut, tableOut=None, twice=True):
        if tableOut == None:
            tableOut = '[' + tableIn + ']'

        # Insert
        with scraperwiki.sql.Transaction():
            scraperwiki.sql.save([], dataIn, tableIn)

        # Observe with pysqlite
        connection = sqlite3.connect(DB_NAME)
        cursor = connection.cursor()
        cursor.execute("SELECT * FROM %s" % tableOut)
        observed1 = cursor.fetchall()
        connection.close()

        if twice:
            # Observe using this module
            observed2 = scraperwiki.sql.select('* FROM %s' % tableOut)

            # Check
            expected1 = dataOut
            expected2 = [dataIn] if type(dataIn) == dict else dataIn

            self.assertListEqual(observed1, expected1)
            self.assertListEqual(observed2, expected2)

class SaveAndSelect(TestCase):
    def save_and_select(self, d):
        scraperwiki.sql.save([], {"foo": d})
        observed = scraperwiki.sql.select('* from swdata')[0]['foo']
        self.assertEqual(d, observed)

{'fraction_non_alphanumeric': 0.12611830105098584, 'fraction_numerical': 0.010075566750629723, 'mean_word_length': 2.4838124054462933, 'pattern_counts': {'":': 2, '<': 7, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '387857', 'n_tokens_mistral': 3582, 'n_tokens_neox': 3465, 'n_words': 850}