<s> """<STR_LIT>""" <EOL> import json <EOL> import os <EOL> import luigi <EOL> from luigi . contrib import bigquery <EOL> from contrib import gcs_test <EOL> from nose . plugins . attrib import attr <EOL> PROJECT_ID = gcs_test . PROJECT_ID <EOL> DATASET_ID = os . environ . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> @ attr ( '<STR_LIT>' ) <EOL> class TestLoadTask ( bigquery . BigqueryLoadTask ) : <EOL> source = luigi . Parameter ( ) <EOL> table = luigi . Parameter ( ) <EOL> @ property <EOL> def schema ( self ) : <EOL> return [ <EOL> { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:name>' : '<STR_LIT>' , '<STR_LIT:type>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:name>' : '<STR_LIT>' , '<STR_LIT:type>' : '<STR_LIT>' } , <EOL> ] <EOL> def source_uris ( self ) : <EOL> return [ self . source ] <EOL> def output ( self ) : <EOL> return bigquery . BigqueryTarget ( PROJECT_ID , DATASET_ID , self . table ) <EOL> @ attr ( '<STR_LIT>' ) <EOL> class TestRunQueryTask ( bigquery . BigqueryRunQueryTask ) : <EOL> query = '''<STR_LIT>''' <EOL> table = luigi . Parameter ( ) <EOL> def output ( self ) : <EOL> return bigquery . BigqueryTarget ( PROJECT_ID , DATASET_ID , self . table ) <EOL> @ attr ( '<STR_LIT>' ) <EOL> class BigqueryGcloudTest ( gcs_test . _GCSBaseTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( BigqueryGcloudTest , self ) . setUp ( ) <EOL> self . bq_client = bigquery . BigqueryClient ( gcs_test . CREDENTIALS ) <EOL> self . table = bigquery . BQTable ( project_id = PROJECT_ID , dataset_id = DATASET_ID , <EOL> table_id = self . id ( ) . split ( '<STR_LIT:.>' ) [ - <NUM_LIT:1> ] ) <EOL> self . addCleanup ( self . bq_client . delete_table , self . table ) <EOL> def create_dataset ( self , data = [ ] ) : <EOL> self . bq_client . delete_table ( self . table ) <EOL> text = '<STR_LIT:\n>' . join ( map ( json . dumps , data ) ) <EOL> gcs_file = gcs_test . bucket_url ( self . id ( ) ) <EOL> self . client . put_string ( text , gcs_file ) <EOL> task = TestLoadTask ( source = gcs_file , table = self . table . table_id ) <EOL> task . run ( ) <EOL> def test_table_uri ( self ) : <EOL> intended_uri = "<STR_LIT>" + PROJECT_ID + "<STR_LIT:/>" + DATASET_ID + "<STR_LIT:/>" + self . table . table_id <EOL> self . assertTrue ( self . table . uri == intended_uri ) <EOL> def test_load_and_copy ( self ) : <EOL> self . create_dataset ( [ <EOL> { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:1> } , <EOL> { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:2> } , <EOL> ] ) <EOL> self . assertTrue ( self . bq_client . dataset_exists ( self . table ) ) <EOL> self . assertTrue ( self . bq_client . table_exists ( self . table ) ) <EOL> self . assertIn ( self . table . dataset_id , <EOL> list ( self . bq_client . list_datasets ( self . table . project_id ) ) ) <EOL> self . assertIn ( self . table . table_id , <EOL> list ( self . bq_client . list_tables ( self . table . dataset ) ) ) <EOL> new_table = self . table . _replace ( table_id = self . table . table_id + '<STR_LIT>' ) <EOL> self . bq_client . copy ( <EOL> source_table = self . table , <EOL> dest_table = new_table <EOL> ) <EOL> self . assertTrue ( self . bq_client . table_exists ( new_table ) ) <EOL> self . bq_client . delete_table ( new_table ) <EOL> self . assertFalse ( self . bq_client . table_exists ( new_table ) ) <EOL> def test_run_query ( self ) : <EOL> task = TestRunQueryTask ( table = self . table . table_id ) <EOL> task . _BIGQUERY_CLIENT = self . bq_client <EOL> task . run ( ) <EOL> self . assertTrue ( self . bq_client . table_exists ( self . table ) ) </s>
