Merge branch 'issue_2581' of github.com:nathan-rice/sqlalchemy into pg_json

1 files changed, 190 insertions, 1 deletions

diff --git a/test/dialect/postgresql/test_types.py b/test/dialect/postgresql/test_types.py
index 6e8609448..19df131fd 100644
--- a/test/dialect/postgresql/test_types.py
+++ b/test/dialect/postgresql/test_types.py
@@ -15,7 +15,8 @@ from sqlalchemy.orm import Session, mapper, aliased
 from sqlalchemy import exc, schema, types
 from sqlalchemy.dialects.postgresql import base as postgresql
 from sqlalchemy.dialects.postgresql import HSTORE, hstore, array, \
-            INT4RANGE, INT8RANGE, NUMRANGE, DATERANGE, TSRANGE, TSTZRANGE
+            INT4RANGE, INT8RANGE, NUMRANGE, DATERANGE, TSRANGE, TSTZRANGE, \
+            JSON
 import decimal
 from sqlalchemy import util
 from sqlalchemy.testing.util import round_decimal
@@ -1663,3 +1664,191 @@ class DateTimeTZRangeTests(_RangeTypeMixin, fixtures.TablesTest):
 
     def _data_obj(self):
         return self.extras.DateTimeTZRange(*self.tstzs())
+
+
+class JSONTest(fixtures.TestBase):
+    def _assert_sql(self, construct, expected):
+        dialect = postgresql.dialect()
+        compiled = str(construct.compile(dialect=dialect))
+        compiled = re.sub(r'\s+', ' ', compiled)
+        expected = re.sub(r'\s+', ' ', expected)
+        eq_(compiled, expected)
+
+    def setup(self):
+        metadata = MetaData()
+        self.test_table = Table('test_table', metadata,
+            Column('id', Integer, primary_key=True),
+            Column('test_column', JSON)
+        )
+        self.jsoncol = self.test_table.c.test_column
+
+    def _test_where(self, whereclause, expected):
+        stmt = select([self.test_table]).where(whereclause)
+        self._assert_sql(
+            stmt,
+            "SELECT test_table.id, test_table.test_column FROM test_table "
+            "WHERE %s" % expected
+        )
+
+    def _test_cols(self, colclause, expected, from_=True):
+        stmt = select([colclause])
+        self._assert_sql(
+            stmt,
+            (
+                "SELECT %s" +
+                (" FROM test_table" if from_ else "")
+            ) % expected
+        )
+
+    def test_bind_serialize_default(self):
+        from sqlalchemy.engine import default
+
+        dialect = default.DefaultDialect()
+        proc = self.test_table.c.test_column.type._cached_bind_processor(dialect)
+        eq_(
+            proc({"A": [1, 2, 3, True, False]}),
+            '{"A": [1, 2, 3, true, false]}'
+        )
+
+    def test_result_deserialize_default(self):
+        from sqlalchemy.engine import default
+
+        dialect = default.DefaultDialect()
+        proc = self.test_table.c.test_column.type._cached_result_processor(
+                    dialect, None)
+        eq_(
+            proc('{"A": [1, 2, 3, true, false]}'),
+            {"A": [1, 2, 3, True, False]}
+        )
+
+    # This test is a bit misleading -- in real life you will need to cast to do anything
+    def test_where_getitem(self):
+        self._test_where(
+            self.jsoncol['bar'] == None,
+            "(test_table.test_column -> %(test_column_1)s) IS NULL"
+        )
+
+    def test_where_path(self):
+        self._test_where(
+            self.jsoncol.get_path('{"foo", 1}') == None,
+            "(test_table.test_column #> %(test_column_1)s) IS NULL"
+        )
+
+    def test_where_getitem_as_text(self):
+        self._test_where(
+            self.jsoncol.get_item_as_text('bar') == None,
+            "(test_table.test_column ->> %(test_column_1)s) IS NULL"
+        )
+
+    def test_where_path_as_text(self):
+        self._test_where(
+            self.jsoncol.get_path_as_text('{"foo", 1}') == None,
+            "(test_table.test_column #>> %(test_column_1)s) IS NULL"
+        )
+
+    def test_cols_get(self):
+        self._test_cols(
+            self.jsoncol['foo'],
+            "test_table.test_column -> %(test_column_1)s AS anon_1",
+            True
+        )
+
+
+class JSONRoundTripTest(fixtures.TablesTest):
+    __only_on__ = 'postgresql'
+
+    @classmethod
+    def define_tables(cls, metadata):
+        Table('data_table', metadata,
+            Column('id', Integer, primary_key=True),
+            Column('name', String(30), nullable=False),
+            Column('data', JSON)
+        )
+
+    def _fixture_data(self, engine):
+        data_table = self.tables.data_table
+        engine.execute(
+                data_table.insert(),
+                {'name': 'r1', 'data': {"k1": "r1v1", "k2": "r1v2"}},
+                {'name': 'r2', 'data': {"k1": "r2v1", "k2": "r2v2"}},
+                {'name': 'r3', 'data': {"k1": "r3v1", "k2": "r3v2"}},
+                {'name': 'r4', 'data': {"k1": "r4v1", "k2": "r4v2"}},
+                {'name': 'r5', 'data': {"k1": "r5v1", "k2": "r5v2"}},
+        )
+
+    def _assert_data(self, compare):
+        data = testing.db.execute(
+            select([self.tables.data_table.c.data]).
+                order_by(self.tables.data_table.c.name)
+        ).fetchall()
+        eq_([d for d, in data], compare)
+
+    def _test_insert(self, engine):
+        engine.execute(
+            self.tables.data_table.insert(),
+            {'name': 'r1', 'data': {"k1": "r1v1", "k2": "r1v2"}}
+        )
+        self._assert_data([{"k1": "r1v1", "k2": "r1v2"}])
+
+    def _non_native_engine(self):
+        if testing.against("postgresql+psycopg2"):
+            engine = engines.testing_engine()
+        else:
+            engine = testing.db
+        engine.connect()
+        return engine
+
+    def test_reflect(self):
+        from sqlalchemy import inspect
+        insp = inspect(testing.db)
+        cols = insp.get_columns('data_table')
+        assert isinstance(cols[2]['type'], JSON)
+
+    @testing.only_on("postgresql+psycopg2")
+    def test_insert_native(self):
+        engine = testing.db
+        self._test_insert(engine)
+
+    def test_insert_python(self):
+        engine = self._non_native_engine()
+        self._test_insert(engine)
+
+    @testing.only_on("postgresql+psycopg2")
+    def test_criterion_native(self):
+        engine = testing.db
+        self._fixture_data(engine)
+        self._test_criterion(engine)
+
+    def test_criterion_python(self):
+        engine = self._non_native_engine()
+        self._fixture_data(engine)
+        self._test_criterion(engine)
+
+    def test_path_query(self):
+        engine = testing.db
+        self._fixture_data(engine)
+        data_table = self.tables.data_table
+        result = engine.execute(
+            select([data_table.c.data]).where(
+                data_table.c.data.get_path_as_text('{k1}') == 'r3v1'
+            )
+        ).first()
+        eq_(result, ({'k1': 'r3v1', 'k2': 'r3v2'},))
+
+    def test_query_returned_as_text(self):
+        engine = testing.db
+        self._fixture_data(engine)
+        data_table = self.tables.data_table
+        result = engine.execute(
+            select([data_table.c.data.get_item_as_text('k1')])
+        ).first()
+        assert isinstance(result[0], basestring)
+
+    def _test_criterion(self, engine):
+        data_table = self.tables.data_table
+        result = engine.execute(
+            select([data_table.c.data]).where(
+                data_table.c.data.get_item_as_text('k1') == 'r3v1'
+            )
+        ).first()
+        eq_(result, ({'k1': 'r3v1', 'k2': 'r3v2'},))