<reponame>ShaneKilkelly/bedquilt
import testutils
import json
import string
import psycopg2


class TestInsertDocument(testutils.BedquiltTestCase):

    def test_insert_into_non_existant_collection(self):
        doc = {
            "_id": "<EMAIL>",
            "name": "<NAME>",
            "age": 20
        }

        self.cur.execute("""
            select bq_insert('people', '{}');
        """.format(json.dumps(doc)))

        result = self.cur.fetchone()

        self.assertEqual(
            result, ('<EMAIL>',)
        )

        self.cur.execute("select bq_list_collections();")
        collections = self.cur.fetchall()
        self.assertIsNotNone(collections)

        self.assertEqual(collections, [("people",)])

    def test_with_non_string_id(self):
        docs = [
            {
                "_id": 42,
                "name": "Penguin",
                "age": "<EMAIL>"
            },
            {
                "_id": ['derp'],
                "name": "Penguin",
                "age": "<EMAIL>"
            },
            {
                "_id": {"name": "Penguin"},
                "age": "<EMAIL>"
            },
            {
                "_id": False,
                "name": "Penguin",
                "age": "<EMAIL>"
            },
            {
                "_id": None,
                "name": "Penguin",
                "age": "<EMAIL>"
            }
        ]
        for doc in docs:
            with self.assertRaises(psycopg2.InternalError):
                self.cur.execute("""
                select bq_insert('people', '{}');
                """.format(json.dumps(doc)))
            self.conn.rollback()

    def test_insert_without_id(self):
        doc = {
            "name": "<NAME>",
            "age": 20
        }
        self.cur.execute("""
            select bq_insert('people', '{}');
        """.format(json.dumps(doc)))

        result = self.cur.fetchone()

        self.assertIsNotNone(result)
        self.assertEqual(type(result), tuple)
        self.assertEqual(len(result), 1)

        _id = result[0]
        self.assertIn(type(_id), {str, unicode})
        self.assertEqual(len(_id), 24)
        for character in _id:
            self.assertIn(character, string.hexdigits)

    def test_with_single_quotes_in_field(self):
        doc = {
            "description": "Something I've eaten"
        }
        self.cur.execute("""
            select bq_insert('things', %s);
        """, (json.dumps(doc),))

        result = self.cur.fetchone()
        self.assertIsNotNone(result)

    def test_insert_with_repeat_id(self):
        doc = {
            "_id": "user_one",
            "name": "<NAME>",
            "age": 20
        }
        self.cur.execute("""
            select bq_insert('people', '{}');
        """.format(json.dumps(doc)))

        result = self.cur.fetchone()

        self.assertIsNotNone(result)
        self.assertEqual(type(result), tuple)
        self.assertEqual(len(result), 1)
        _id = result[0]
        self.assertEqual(_id, "user_one")

        self.conn.commit()

        with self.assertRaises(psycopg2.IntegrityError):
            self.cur.execute("""
            select bq_insert('people', '{}');
            """.format(json.dumps(doc)))
        self.conn.rollback()

        self.cur.execute("select count(*) from people;")
        result = self.cur.fetchone()
        self.assertEqual(result, (1,))

import testutils
import json
import string
import psycopg2


class TestInsertDocument(testutils.BedquiltTestCase):

    def test_insert_into_non_existant_collection(self):
        doc = {
            "_id": "<EMAIL>",
            "name": "<NAME>",
            "age": 20
        }

        self.cur.execute("""
            select bq_insert('people', '{}');
        """.format(json.dumps(doc)))

        result = self.cur.fetchone()

        self.assertEqual(
            result, ('<EMAIL>',)
        )

        self.cur.execute("select bq_list_collections();")
        collections = self.cur.fetchall()
        self.assertIsNotNone(collections)

        self.assertEqual(collections, [("people",)])

    def test_with_non_string_id(self):
        docs = [
            {
                "_id": 42,
                "name": "Penguin",
                "age": "<EMAIL>"
            },
            {
                "_id": ['derp'],
                "name": "Penguin",
                "age": "<EMAIL>"
            },
            {
                "_id": {"name": "Penguin"},
                "age": "<EMAIL>"
            },
            {
                "_id": False,
                "name": "Penguin",
                "age": "<EMAIL>"
            },
            {
                "_id": None,
                "name": "Penguin",
                "age": "<EMAIL>"
            }
        ]
        for doc in docs:
            with self.assertRaises(psycopg2.InternalError):
                self.cur.execute("""
                select bq_insert('people', '{}');
                """.format(json.dumps(doc)))
            self.conn.rollback()

    def test_insert_without_id(self):
        doc = {
            "name": "<NAME>",
            "age": 20
        }
        self.cur.execute("""
            select bq_insert('people', '{}');
        """.format(json.dumps(doc)))

        result = self.cur.fetchone()

        self.assertIsNotNone(result)
        self.assertEqual(type(result), tuple)
        self.assertEqual(len(result), 1)

        _id = result[0]
        self.assertIn(type(_id), {str, unicode})
        self.assertEqual(len(_id), 24)
        for character in _id:
            self.assertIn(character, string.hexdigits)

    def test_with_single_quotes_in_field(self):
        doc = {
            "description": "Something I've eaten"
        }
        self.cur.execute("""
            select bq_insert('things', %s);
        """, (json.dumps(doc),))

        result = self.cur.fetchone()
        self.assertIsNotNone(result)

    def test_insert_with_repeat_id(self):
        doc = {
            "_id": "user_one",
            "name": "<NAME>",
            "age": 20
        }
        self.cur.execute("""
            select bq_insert('people', '{}');
        """.format(json.dumps(doc)))

        result = self.cur.fetchone()

        self.assertIsNotNone(result)
        self.assertEqual(type(result), tuple)
        self.assertEqual(len(result), 1)
        _id = result[0]
        self.assertEqual(_id, "user_one")

        self.conn.commit()

        with self.assertRaises(psycopg2.IntegrityError):
            self.cur.execute("""
            select bq_insert('people', '{}');
            """.format(json.dumps(doc)))
        self.conn.rollback()

        self.cur.execute("select count(*) from people;")
        result = self.cur.fetchone()
        self.assertEqual(result, (1,))
select bq_insert('people', '{}'); select bq_insert('people', '{}'); select bq_insert('people', '{}'); select bq_insert('things', %s); select bq_insert('people', '{}'); select bq_insert('people', '{}');