DML tests

Author: mxmzdlv

scripts/load_test_data.sh

@@ -1,6 +1,7 @@
 bq mk test_pybigquery
 bq rm -f -t test_pybigquery.sample
 bq rm -f -t test_pybigquery.sample_one_row
+bq rm -f -t test_pybigquery.sample_dml
 bq load --source_format=NEWLINE_DELIMITED_JSON --schema=$(dirname $0)/schema.json test_pybigquery.sample $(dirname $0)/sample.json
 bq load --source_format=NEWLINE_DELIMITED_JSON --schema=$(dirname $0)/schema.json test_pybigquery.sample_one_row $(dirname $0)/sample_one_row.json
-
+bq mk --schema=$(dirname $0)/schema.json -t test_pybigquery.sample_dml

test/test_sqlalchemy_bigquery.py

@@ -26,6 +26,19 @@
 ]
 
 
+ONE_ROW_CONTENTS_DML = [
+    588,
+    datetime.datetime(2013, 10, 10, 11, 27, 16, tzinfo=timezone('UTC')),
+    'test',
+    40.76727216,
+    False,
+    datetime.date(2013, 10, 10),
+    datetime.datetime(2013, 10, 10, 11, 27, 16),
+    datetime.time(11, 27, 16),
+    'test_bytes'
+]
+
+
 @pytest.fixture(scope='session')
 def engine():
     engine = create_engine('bigquery://', echo=True)
@@ -42,6 +55,11 @@ def table_one_row(engine):
     return Table('test_pybigquery.sample_one_row', MetaData(bind=engine), autoload=True)
 
 
+@pytest.fixture(scope='session')
+def table_dml(engine):
+    return Table('test_pybigquery.sample_dml', MetaData(bind=engine), autoload=True)
+
+
 @pytest.fixture(scope='session')
 def session(engine):
     Session = sessionmaker(bind=engine)
@@ -170,4 +188,23 @@ def test_joins(session, table, table_one_row):
 def test_querying_wildcard_tables(engine):
     table = Table('bigquery-public-data.noaa_gsod.gsod*', MetaData(bind=engine), autoload=True)
     rows = table.select().limit(1).execute().first()
-    assert len(rows) > 0
+    assert len(rows) > 0
+
+
+def test_dml(engine, session, table_dml):
+    # test insert
+    engine.execute(table_dml.insert(ONE_ROW_CONTENTS_DML))
+    result = table_dml.select().execute().fetchall()
+    assert len(result) == 1
+
+    # test update
+    session.query(table_dml)\
+        .filter(table_dml.c.string == 'test')\
+        .update({'string': 'updated_row'}, synchronize_session=False)
+    updated_result = table_dml.select().execute().fetchone()
+    assert updated_result['string'] == 'updated_row'
+
+    # test delete
+    session.query(table_dml).filter(table_dml.c.string == 'updated_row').delete(synchronize_session=False)
+    result = table_dml.select().execute().fetchall()
+    assert len(result) == 0