# Copyright 2015 Google LLC

#

# Licensed under the Apache License, Version 2.0 (the "License");

# you may not use this file except in compliance with the License.

# You may obtain a copy of the License at

#

# http://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing, software

# distributed under the License is distributed on an "AS IS" BASIS,

# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

# See the License for the specific language governing permissions and

# limitations under the License.

import datetime

import logging

import re

import time

import types

import unittest

import warnings

import mock

import pytest

import google.api_core.exceptions

from test_utils.imports import maybe_fail_import

try:

from google.cloud import bigquery_storage

from google.cloud.bigquery_storage_v1.services.big_query_read.transports import (

grpc as big_query_read_grpc_transport,

)

except ImportError: # pragma: NO COVER

bigquery_storage = None

big_query_read_grpc_transport = None

try:

import pandas

except (ImportError, AttributeError): # pragma: NO COVER

pandas = None

try:

import geopandas

except (ImportError, AttributeError): # pragma: NO COVER

geopandas = None

try:

import pyarrow

import pyarrow.types

except ImportError: # pragma: NO COVER

pyarrow = None

try:

from tqdm import tqdm

except (ImportError, AttributeError): # pragma: NO COVER

tqdm = None

from google.cloud.bigquery.dataset import DatasetReference

def _mock_client():

from google.cloud.bigquery import client

mock_client = mock.create_autospec(client.Client)

mock_client.project = "my-project"

return mock_client

class _SchemaBase(object):

def _verify_field(self, field, r_field):

self.assertEqual(field.name, r_field["name"])

self.assertEqual(field.field_type, r_field["type"])

self.assertEqual(field.mode, r_field.get("mode", "NULLABLE"))

def _verifySchema(self, schema, resource):

r_fields = resource["schema"]["fields"]

self.assertEqual(len(schema), len(r_fields))

for field, r_field in zip(schema, r_fields):

self._verify_field(field, r_field)

class TestEncryptionConfiguration(unittest.TestCase):

KMS_KEY_NAME = "projects/1/locations/us/keyRings/1/cryptoKeys/1"

@staticmethod

def _get_target_class():

from google.cloud.bigquery.table import EncryptionConfiguration

return EncryptionConfiguration

def _make_one(self, *args, **kw):

return self._get_target_class()(*args, **kw)

def test_ctor_defaults(self):

encryption_config = self._make_one()

self.assertIsNone(encryption_config.kms_key_name)

def test_ctor_with_key(self):

encryption_config = self._make_one(kms_key_name=self.KMS_KEY_NAME)

self.assertEqual(encryption_config.kms_key_name, self.KMS_KEY_NAME)

class TestTableReference(unittest.TestCase):

@staticmethod

def _get_target_class():

from google.cloud.bigquery.table import TableReference

return TableReference

def _make_one(self, *args, **kw):

return self._get_target_class()(*args, **kw)

def test_ctor_defaults(self):

dataset_ref = DatasetReference("project_1", "dataset_1")

table_ref = self._make_one(dataset_ref, "table_1")

self.assertEqual(table_ref.dataset_id, dataset_ref.dataset_id)

self.assertEqual(table_ref.table_id, "table_1")

def test_to_api_repr(self):

dataset_ref = DatasetReference("project_1", "dataset_1")

table_ref = self._make_one(dataset_ref, "table_1")

resource = table_ref.to_api_repr()

self.assertEqual(

resource,

{"projectId": "project_1", "datasetId": "dataset_1", "tableId": "table_1"},

)

def test_from_api_repr(self):

from google.cloud.bigquery.table import TableReference

dataset_ref = DatasetReference("project_1", "dataset_1")

expected = self._make_one(dataset_ref, "table_1")

got = TableReference.from_api_repr(

{"projectId": "project_1", "datasetId": "dataset_1", "tableId": "table_1"}

)

self.assertEqual(expected, got)

def test_from_string(self):

cls = self._get_target_class()

got = cls.from_string("string-project.string_dataset.string_table")

self.assertEqual(got.project, "string-project")

self.assertEqual(got.dataset_id, "string_dataset")

self.assertEqual(got.table_id, "string_table")

def test_from_string_w_prefix(self):

cls = self._get_target_class()

got = cls.from_string("google.com:string-project.string_dataset.string_table")

self.assertEqual(got.project, "google.com:string-project")

self.assertEqual(got.dataset_id, "string_dataset")

self.assertEqual(got.table_id, "string_table")

def test_from_string_legacy_string(self):

cls = self._get_target_class()

with self.assertRaises(ValueError):

cls.from_string("string-project:string_dataset.string_table")

def test_from_string_w_incorrect_prefix(self):

cls = self._get_target_class()

with self.assertRaises(ValueError):

cls.from_string("google.com.string-project.string_dataset.string_table")

def test_from_string_not_fully_qualified(self):

cls = self._get_target_class()

with self.assertRaises(ValueError):

cls.from_string("string_table")

with self.assertRaises(ValueError):

cls.from_string("string_dataset.string_table")

with self.assertRaises(ValueError):

cls.from_string("a.b.c.d")

def test_from_string_with_default_project(self):

cls = self._get_target_class()

got = cls.from_string(

"string_dataset.string_table", default_project="default-project"

)

self.assertEqual(got.project, "default-project")

self.assertEqual(got.dataset_id, "string_dataset")

self.assertEqual(got.table_id, "string_table")

def test_from_string_ignores_default_project(self):

cls = self._get_target_class()

got = cls.from_string(

"string-project.string_dataset.string_table",

default_project="default-project",

)

self.assertEqual(got.project, "string-project")

self.assertEqual(got.dataset_id, "string_dataset")

self.assertEqual(got.table_id, "string_table")

def test___eq___wrong_type(self):

dataset_ref = DatasetReference("project_1", "dataset_1")

table = self._make_one(dataset_ref, "table_1")

other = object()

self.assertNotEqual(table, other)

self.assertEqual(table, mock.ANY)

def test___eq___project_mismatch(self):

dataset = DatasetReference("project_1", "dataset_1")

other_dataset = DatasetReference("project_2", "dataset_1")

table = self._make_one(dataset, "table_1")

other = self._make_one(other_dataset, "table_1")

self.assertNotEqual(table, other)

def test___eq___dataset_mismatch(self):

dataset = DatasetReference("project_1", "dataset_1")

other_dataset = DatasetReference("project_1", "dataset_2")

table = self._make_one(dataset, "table_1")

other = self._make_one(other_dataset, "table_1")

self.assertNotEqual(table, other)

def test___eq___table_mismatch(self):

dataset = DatasetReference("project_1", "dataset_1")

table = self._make_one(dataset, "table_1")

other = self._make_one(dataset, "table_2")

self.assertNotEqual(table, other)

def test___eq___equality(self):

dataset = DatasetReference("project_1", "dataset_1")

table = self._make_one(dataset, "table_1")

other = self._make_one(dataset, "table_1")

self.assertEqual(table, other)

def test___hash__set_equality(self):

dataset = DatasetReference("project_1", "dataset_1")

table1 = self._make_one(dataset, "table1")

table2 = self._make_one(dataset, "table2")

set_one = {table1, table2}

set_two = {table1, table2}

self.assertEqual(set_one, set_two)

def test___hash__not_equals(self):

dataset = DatasetReference("project_1", "dataset_1")

table1 = self._make_one(dataset, "table1")

table2 = self._make_one(dataset, "table2")

set_one = {table1}

set_two = {table2}

self.assertNotEqual(set_one, set_two)

def test___repr__(self):

dataset = DatasetReference("project1", "dataset1")

table1 = self._make_one(dataset, "table1")

expected = (

"TableReference(DatasetReference('project1', 'dataset1'), " "'table1')"

)

self.assertEqual(repr(table1), expected)

def test___str__(self):

dataset = DatasetReference("project1", "dataset1")

table1 = self._make_one(dataset, "table1")

self.assertEqual(str(table1), "project1.dataset1.table1")

class TestTable(unittest.TestCase, _SchemaBase):

PROJECT = "prahj-ekt"

DS_ID = "dataset-name"

TABLE_NAME = "table-name"

KMS_KEY_NAME = "projects/1/locations/us/keyRings/1/cryptoKeys/1"

@staticmethod

def _get_target_class():

from google.cloud.bigquery.table import Table

return Table

def _make_one(self, *args, **kw):

if len(args) == 0:

dataset = DatasetReference(self.PROJECT, self.DS_ID)

table_ref = dataset.table(self.TABLE_NAME)

args = (table_ref,)

return self._get_target_class()(*args, **kw)

def _setUpConstants(self):

import datetime

from google.cloud._helpers import UTC

self.WHEN_TS = 1437767599.006

self.WHEN = datetime.datetime.utcfromtimestamp(self.WHEN_TS).replace(tzinfo=UTC)

self.ETAG = "ETAG"

self.TABLE_FULL_ID = "%s:%s.%s" % (self.PROJECT, self.DS_ID, self.TABLE_NAME)

self.RESOURCE_URL = "http://example.com/path/to/resource"

self.NUM_BYTES = 12345

self.NUM_ROWS = 67

self.NUM_EST_BYTES = 1234

self.NUM_EST_ROWS = 23

def _make_resource(self):

self._setUpConstants()

return {

"creationTime": self.WHEN_TS * 1000,

"tableReference": {

"projectId": self.PROJECT,

"datasetId": self.DS_ID,

"tableId": self.TABLE_NAME,

},

"schema": {

"fields": [

{"name": "full_name", "type": "STRING", "mode": "REQUIRED"},

{"name": "age", "type": "INTEGER", "mode": "REQUIRED"},

]

},

"etag": "ETAG",

"id": self.TABLE_FULL_ID,

"lastModifiedTime": self.WHEN_TS * 1000,

"location": "US",

"selfLink": self.RESOURCE_URL,

"numRows": self.NUM_ROWS,

"numBytes": self.NUM_BYTES,

"type": "TABLE",

"streamingBuffer": {

"estimatedRows": str(self.NUM_EST_ROWS),

"estimatedBytes": str(self.NUM_EST_BYTES),

"oldestEntryTime": self.WHEN_TS * 1000,

},

"externalDataConfiguration": {

"sourceFormat": "CSV",

"csvOptions": {"allowJaggedRows": True, "encoding": "encoding"},

},

"labels": {"x": "y"},

}

def _verifyReadonlyResourceProperties(self, table, resource):

if "creationTime" in resource:

self.assertEqual(table.created, self.WHEN)

else:

self.assertIsNone(table.created)

if "etag" in resource:

self.assertEqual(table.etag, self.ETAG)

else:

self.assertIsNone(table.etag)

if "numRows" in resource:

self.assertEqual(table.num_rows, self.NUM_ROWS)

else:

self.assertIsNone(table.num_rows)

if "numBytes" in resource:

self.assertEqual(table.num_bytes, self.NUM_BYTES)

else:

self.assertIsNone(table.num_bytes)

if "selfLink" in resource:

self.assertEqual(table.self_link, self.RESOURCE_URL)

else:

self.assertIsNone(table.self_link)

if "streamingBuffer" in resource:

self.assertEqual(table.streaming_buffer.estimated_rows, self.NUM_EST_ROWS)

self.assertEqual(table.streaming_buffer.estimated_bytes, self.NUM_EST_BYTES)

self.assertEqual(table.streaming_buffer.oldest_entry_time, self.WHEN)

else:

self.assertIsNone(table.streaming_buffer)

self.assertEqual(table.full_table_id, self.TABLE_FULL_ID)

self.assertEqual(

table.table_type, "TABLE" if "view" not in resource else "VIEW"

)

def _verifyResourceProperties(self, table, resource):

self._verifyReadonlyResourceProperties(table, resource)

if "expirationTime" in resource:

self.assertEqual(table.expires, self.EXP_TIME)

else:

self.assertIsNone(table.expires)

self.assertEqual(table.description, resource.get("description"))

self.assertEqual(table.friendly_name, resource.get("friendlyName"))

self.assertEqual(table.location, resource.get("location"))

if "view" in resource:

self.assertEqual(table.view_query, resource["view"]["query"])

self.assertEqual(

table.view_use_legacy_sql, resource["view"].get("useLegacySql", True)

)

else:

self.assertIsNone(table.view_query)

self.assertIsNone(table.view_use_legacy_sql)

if "schema" in resource:

self._verifySchema(table.schema, resource)

else:

self.assertEqual(table.schema, [])

if "externalDataConfiguration" in resource:

edc = table.external_data_configuration

self.assertEqual(edc.source_format, "CSV")

self.assertEqual(edc.options.allow_jagged_rows, True)

if "labels" in resource:

self.assertEqual(table.labels, {"x": "y"})

else:

self.assertEqual(table.labels, {})

if "encryptionConfiguration" in resource:

self.assertIsNotNone(table.encryption_configuration)

self.assertEqual(

table.encryption_configuration.kms_key_name,

resource["encryptionConfiguration"]["kmsKeyName"],

)

else:

self.assertIsNone(table.encryption_configuration)

def test_ctor(self):

dataset = DatasetReference(self.PROJECT, self.DS_ID)

table_ref = dataset.table(self.TABLE_NAME)

table = self._make_one(table_ref)

self.assertEqual(table.table_id, self.TABLE_NAME)

self.assertEqual(table.project, self.PROJECT)

self.assertEqual(table.dataset_id, self.DS_ID)

self.assertEqual(table.reference.table_id, self.TABLE_NAME)

self.assertEqual(table.reference.project, self.PROJECT)

self.assertEqual(table.reference.dataset_id, self.DS_ID)

self.assertEqual(

table.path,

"/projects/%s/datasets/%s/tables/%s"

% (self.PROJECT, self.DS_ID, self.TABLE_NAME),

)

self.assertEqual(table.schema, [])

self.assertIsNone(table.created)

self.assertIsNone(table.etag)

self.assertIsNone(table.modified)

self.assertIsNone(table.num_bytes)

self.assertIsNone(table.num_rows)

self.assertIsNone(table.self_link)

self.assertIsNone(table.full_table_id)

self.assertIsNone(table.table_type)

self.assertIsNone(table.description)

self.assertIsNone(table.expires)

self.assertIsNone(table.friendly_name)

self.assertIsNone(table.location)

self.assertIsNone(table.view_query)

self.assertIsNone(table.view_use_legacy_sql)

self.assertIsNone(table.external_data_configuration)

self.assertEqual(table.labels, {})

self.assertIsNone(table.encryption_configuration)

self.assertIsNone(table.time_partitioning)

self.assertIsNone(table.clustering_fields)

def test_ctor_w_schema(self):

from google.cloud.bigquery.schema import SchemaField

dataset = DatasetReference(self.PROJECT, self.DS_ID)

table_ref = dataset.table(self.TABLE_NAME)

full_name = SchemaField("full_name", "STRING", mode="REQUIRED")

age = SchemaField("age", "INTEGER", mode="REQUIRED")

table = self._make_one(table_ref, schema=[full_name, age])

self.assertEqual(table.schema, [full_name, age])

def test_ctor_string(self):

table = self._make_one("some-project.some_dset.some_tbl")

self.assertEqual(table.project, "some-project")

self.assertEqual(table.dataset_id, "some_dset")

self.assertEqual(table.table_id, "some_tbl")

def test_ctor_tablelistitem(self):

from google.cloud.bigquery.table import Table, TableListItem

import datetime

from google.cloud._helpers import _millis, UTC

self.WHEN_TS = 1437767599.125

self.EXP_TIME = datetime.datetime(2015, 8, 1, 23, 59, 59, tzinfo=UTC)

project = "test-project"

dataset_id = "test_dataset"

table_id = "coffee_table"

resource = {

"creationTime": self.WHEN_TS * 1000,

"expirationTime": _millis(self.EXP_TIME),

"kind": "bigquery#table",

"id": "{}:{}.{}".format(project, dataset_id, table_id),

"tableReference": {

"projectId": project,

"datasetId": dataset_id,

"tableId": table_id,

},

"friendlyName": "Mahogany Coffee Table",

"type": "TABLE",

"timePartitioning": {

"type": "DAY",

"field": "mycolumn",

"expirationMs": "10000",

},

"labels": {"some-stuff": "this-is-a-label"},

"clustering": {"fields": ["string"]},

}

table_list_item = TableListItem(resource)

table = Table(table_list_item)

self.assertIsNone(table.created)

self.assertEqual(table.reference.project, project)

self.assertEqual(table.reference.dataset_id, dataset_id)

self.assertEqual(table.reference.table_id, table_id)

def test_ctor_string_wo_project_id(self):

with pytest.raises(ValueError):

# Project ID is missing.

self._make_one("some_dset.some_tbl")

def test_num_bytes_getter(self):

dataset = DatasetReference(self.PROJECT, self.DS_ID)

table_ref = dataset.table(self.TABLE_NAME)

table = self._make_one(table_ref)

# Check with no value set.

self.assertIsNone(table.num_bytes)

num_bytes = 1337

# Check with integer value set.

table._properties = {"numBytes": num_bytes}

self.assertEqual(table.num_bytes, num_bytes)

# Check with a string value set.

table._properties = {"numBytes": str(num_bytes)}

self.assertEqual(table.num_bytes, num_bytes)

# Check with invalid int value.

table._properties = {"numBytes": "x"}

with self.assertRaises(ValueError):

getattr(table, "num_bytes")

def test_num_rows_getter(self):

dataset = DatasetReference(self.PROJECT, self.DS_ID)

table_ref = dataset.table(self.TABLE_NAME)

table = self._make_one(table_ref)

# Check with no value set.

self.assertIsNone(table.num_rows)

num_rows = 42

# Check with integer value set.

table._properties = {"numRows": num_rows}

self.assertEqual(table.num_rows, num_rows)

# Check with a string value set.

table._properties = {"numRows": str(num_rows)}

self.assertEqual(table.num_rows, num_rows)

# Check with invalid int value.

table._properties = {"numRows": "x"}

with self.assertRaises(ValueError):

getattr(table, "num_rows")

def test__eq__wrong_type(self):

table = self._make_one("project_foo.dataset_bar.table_baz")

class TableWannabe:

pass

not_a_table = TableWannabe()

not_a_table._properties = table._properties

assert table != not_a_table # Can't fake it.

def test__eq__same_table_basic(self):

table_1 = self._make_one("project_foo.dataset_bar.table_baz")

table_2 = self._make_one("project_foo.dataset_bar.table_baz")

assert table_1 == table_2

def test__eq__same_table_multiple_properties(self):

from google.cloud.bigquery import SchemaField

table_1 = self._make_one("project_foo.dataset_bar.table_baz")

table_1.require_partition_filter = True

table_1.labels = {"first": "one", "second": "two"}

table_1.schema = [

SchemaField("name", "STRING", "REQUIRED"),

SchemaField("age", "INTEGER", "NULLABLE"),

]

table_2 = self._make_one("project_foo.dataset_bar.table_baz")

table_2.require_partition_filter = True

table_2.labels = {"first": "one", "second": "two"}

table_2.schema = [

SchemaField("name", "STRING", "REQUIRED"),

SchemaField("age", "INTEGER", "NULLABLE"),

]

assert table_1 == table_2

def test__eq__same_table_property_different(self):

table_1 = self._make_one("project_foo.dataset_bar.table_baz")

table_1.description = "This is table baz"

table_2 = self._make_one("project_foo.dataset_bar.table_baz")

table_2.description = "This is also table baz"

assert table_1 == table_2 # Still equal, only table reference is important.

def test__eq__different_table(self):

table_1 = self._make_one("project_foo.dataset_bar.table_baz")

table_2 = self._make_one("project_foo.dataset_bar.table_baz_2")

assert table_1 != table_2

def test_hashable(self):

table_1 = self._make_one("project_foo.dataset_bar.table_baz")

table_1.description = "This is a table"

table_1b = self._make_one("project_foo.dataset_bar.table_baz")

table_1b.description = "Metadata is irrelevant for hashes"

assert hash(table_1) == hash(table_1b)

def test_schema_setter_non_sequence(self):

dataset = DatasetReference(self.PROJECT, self.DS_ID)

table_ref = dataset.table(self.TABLE_NAME)

table = self._make_one(table_ref)

with self.assertRaises(TypeError):

table.schema = object()

def test_schema_setter_invalid_field(self):

from google.cloud.bigquery.schema import SchemaField

dataset = DatasetReference(self.PROJECT, self.DS_ID)

table_ref = dataset.table(self.TABLE_NAME)

table = self._make_one(table_ref)

full_name = SchemaField("full_name", "STRING", mode="REQUIRED")

with self.assertRaises(ValueError):

table.schema = [full_name, object()]

def test_schema_setter_valid_fields(self):

from google.cloud.bigquery.schema import SchemaField

dataset = DatasetReference(self.PROJECT, self.DS_ID)

table_ref = dataset.table(self.TABLE_NAME)

table = self._make_one(table_ref)

full_name = SchemaField("full_name", "STRING", mode="REQUIRED")

age = SchemaField("age", "INTEGER", mode="REQUIRED")

table.schema = [full_name, age]

self.assertEqual(table.schema, [full_name, age])

def test_schema_setter_invalid_mapping_representation(self):

dataset = DatasetReference(self.PROJECT, self.DS_ID)

table_ref = dataset.table(self.TABLE_NAME)

table = self._make_one(table_ref)

full_name = {"name": "full_name", "type": "STRING", "mode": "REQUIRED"}

invalid_field = {"name": "full_name", "typeooo": "STRING", "mode": "REQUIRED"}

with self.assertRaises(Exception):

table.schema = [full_name, invalid_field]

def test_schema_setter_valid_mapping_representation(self):

from google.cloud.bigquery.schema import SchemaField

dataset = DatasetReference(self.PROJECT, self.DS_ID)

table_ref = dataset.table(self.TABLE_NAME)

table = self._make_one(table_ref)

full_name = {"name": "full_name", "type": "STRING", "mode": "REQUIRED"}

job_status = {

"name": "is_employed",

"type": "STRUCT",

"mode": "NULLABLE",

"fields": [

{"name": "foo", "type": "DATE", "mode": "NULLABLE"},

{"name": "bar", "type": "BYTES", "mode": "REQUIRED"},

],

}

table.schema = [full_name, job_status]

expected_schema = [

SchemaField("full_name", "STRING", mode="REQUIRED"),

SchemaField(

"is_employed",

"STRUCT",

mode="NULLABLE",

fields=[

SchemaField("foo", "DATE", mode="NULLABLE"),

SchemaField("bar", "BYTES", mode="REQUIRED"),

],

),

]

self.assertEqual(table.schema, expected_schema)

def test_props_set_by_server(self):

import datetime

from google.cloud._helpers import UTC

from google.cloud._helpers import _millis

CREATED = datetime.datetime(2015, 7, 29, 12, 13, 22, tzinfo=UTC)

MODIFIED = datetime.datetime(2015, 7, 29, 14, 47, 15, tzinfo=UTC)

TABLE_FULL_ID = "%s:%s.%s" % (self.PROJECT, self.DS_ID, self.TABLE_NAME)

URL = "http://example.com/projects/%s/datasets/%s/tables/%s" % (

self.PROJECT,

self.DS_ID,

self.TABLE_NAME,

)

dataset = DatasetReference(self.PROJECT, self.DS_ID)

table_ref = dataset.table(self.TABLE_NAME)

table = self._make_one(table_ref)

table._properties["creationTime"] = _millis(CREATED)

table._properties["etag"] = "ETAG"

table._properties["lastModifiedTime"] = _millis(MODIFIED)

table._properties["numBytes"] = 12345

table._properties["numRows"] = 66

table._properties["selfLink"] = URL

table._properties["id"] = TABLE_FULL_ID

table._properties["type"] = "TABLE"

self.assertEqual(table.created, CREATED)

self.assertEqual(table.etag, "ETAG")

self.assertEqual(table.modified, MODIFIED)

self.assertEqual(table.num_bytes, 12345)

self.assertEqual(table.num_rows, 66)

self.assertEqual(table.self_link, URL)

self.assertEqual(table.full_table_id, TABLE_FULL_ID)

self.assertEqual(table.table_type, "TABLE")

def test_snapshot_definition_not_set(self):

dataset = DatasetReference(self.PROJECT, self.DS_ID)

table_ref = dataset.table(self.TABLE_NAME)

table = self._make_one(table_ref)

assert table.snapshot_definition is None

def test_snapshot_definition_set(self):

from google.cloud._helpers import UTC

from google.cloud.bigquery.table import SnapshotDefinition

dataset = DatasetReference(self.PROJECT, self.DS_ID)

table_ref = dataset.table(self.TABLE_NAME)

table = self._make_one(table_ref)

table._properties["snapshotDefinition"] = {

"baseTableReference": {

"projectId": "project_x",

"datasetId": "dataset_y",

"tableId": "table_z",

},

"snapshotTime": "2010-09-28T10:20:30.123Z",

}

snapshot = table.snapshot_definition

assert isinstance(snapshot, SnapshotDefinition)

assert snapshot.base_table_reference.path == (

"/projects/project_x/datasets/dataset_y/tables/table_z"

)

assert snapshot.snapshot_time == datetime.datetime(

2010, 9, 28, 10, 20, 30, 123000, tzinfo=UTC

)

def test_description_setter_bad_value(self):

dataset = DatasetReference(self.PROJECT, self.DS_ID)

table_ref = dataset.table(self.TABLE_NAME)

table = self._make_one(table_ref)

with self.assertRaises(ValueError):

table.description = 12345

def test_description_setter(self):

dataset = DatasetReference(self.PROJECT, self.DS_ID)

table_ref = dataset.table(self.TABLE_NAME)

table = self._make_one(table_ref)

table.description = "DESCRIPTION"

self.assertEqual(table.description, "DESCRIPTION")

def test_expires_setter_bad_value(self):

dataset = DatasetReference(self.PROJECT, self.DS_ID)

table_ref = dataset.table(self.TABLE_NAME)

table = self._make_one(table_ref)

with self.assertRaises(ValueError):

table.expires = object()

def test_expires_setter(self):

import datetime

from google.cloud._helpers import UTC

WHEN = datetime.datetime(2015, 7, 28, 16, 39, tzinfo=UTC)

dataset = DatasetReference(self.PROJECT, self.DS_ID)

table_ref = dataset.table(self.TABLE_NAME)

table = self._make_one(table_ref)

table.expires = WHEN

self.assertEqual(table.expires, WHEN)

def test_friendly_name_setter_bad_value(self):

dataset = DatasetReference(self.PROJECT, self.DS_ID)

table_ref = dataset.table(self.TABLE_NAME)

table = self._make_one(table_ref)

with self.assertRaises(ValueError):

table.friendly_name = 12345

def test_friendly_name_setter(self):

dataset = DatasetReference(self.PROJECT, self.DS_ID)

table_ref = dataset.table(self.TABLE_NAME)

table = self._make_one(table_ref)

table.friendly_name = "FRIENDLY"

self.assertEqual(table.friendly_name, "FRIENDLY")

def test_view_query_setter_bad_value(self):

dataset = DatasetReference(self.PROJECT, self.DS_ID)

table_ref = dataset.table(self.TABLE_NAME)

table = self._make_one(table_ref)

with self.assertRaises(ValueError):

table.view_query = 12345

def test_view_query_setter(self):

dataset = DatasetReference(self.PROJECT, self.DS_ID)

table_ref = dataset.table(self.TABLE_NAME)

table = self._make_one(table_ref)

table.view_query = "select * from foo"

self.assertEqual(table.view_query, "select * from foo")

self.assertEqual(table.view_use_legacy_sql, False)

table.view_use_legacy_sql = True

self.assertEqual(table.view_use_legacy_sql, True)

def test_view_query_deleter(self):

dataset = DatasetReference(self.PROJECT, self.DS_ID)

table_ref = dataset.table(self.TABLE_NAME)

table = self._make_one(table_ref)

table.view_query = "select * from foo"

del table.view_query

self.assertIsNone(table.view_query)

self.assertIsNone(table.view_use_legacy_sql)

def test_view_use_legacy_sql_setter_bad_value(self):

dataset = DatasetReference(self.PROJECT, self.DS_ID)

table_ref = dataset.table(self.TABLE_NAME)

table = self._make_one(table_ref)

with self.assertRaises(ValueError):

table.view_use_legacy_sql = 12345

def test_view_use_legacy_sql_setter(self):

dataset = DatasetReference(self.PROJECT, self.DS_ID)

table_ref = dataset.table(self.TABLE_NAME)

table = self._make_one(table_ref)

table.view_use_legacy_sql = True

table.view_query = "select * from foo"

self.assertEqual(table.view_use_legacy_sql, True)

self.assertEqual(table.view_query, "select * from foo")

def test_external_data_configuration_setter(self):

from google.cloud.bigquery.external_config import ExternalConfig

external_config = ExternalConfig("CSV")

dataset = DatasetReference(self.PROJECT, self.DS_ID)

table_ref = dataset.table(self.TABLE_NAME)

table = self._make_one(table_ref)

table.external_data_configuration = external_config

self.assertEqual(

table.external_data_configuration.source_format,

external_config.source_format,

)

def test_external_data_configuration_setter_none(self):

dataset = DatasetReference(self.PROJECT, self.DS_ID)

table_ref = dataset.table(self.TABLE_NAME)

table = self._make_one(table_ref)

table.external_data_configuration = None

self.assertIsNone(table.external_data_configuration)

def test_external_data_configuration_setter_bad_value(self):

dataset = DatasetReference(self.PROJECT, self.DS_ID)

table_ref = dataset.table(self.TABLE_NAME)

table = self._make_one(table_ref)

with self.assertRaises(ValueError):

table.external_data_configuration = 12345

def test_labels_update_in_place(self):

dataset = DatasetReference(self.PROJECT, self.DS_ID)

table_ref = dataset.table(self.TABLE_NAME)

table = self._make_one(table_ref)

del table._properties["labels"] # don't start w/ existing dict

labels = table.labels

labels["foo"] = "bar" # update in place

self.assertEqual(table.labels, {"foo": "bar"})

def test_labels_setter_bad_value(self):

dataset = DatasetReference(self.PROJECT, self.DS_ID)

table_ref = dataset.table(self.TABLE_NAME)

table = self._make_one(table_ref)

with self.assertRaises(ValueError):

table.labels = 12345

def test_mview_query(self):

table = self._make_one()

self.assertIsNone(table.mview_query)

table.mview_query = "SELECT name, SUM(number) FROM dset.tbl GROUP BY 1"

self.assertEqual(

table.mview_query, "SELECT name, SUM(number) FROM dset.tbl GROUP BY 1"

)

del table.mview_query

self.assertIsNone(table.mview_query)

def test_mview_last_refresh_time(self):

table = self._make_one()

self.assertIsNone(table.mview_last_refresh_time)

table._properties["materializedView"] = {

"lastRefreshTime": "1606751842496",

}

self.assertEqual(

table.mview_last_refresh_time,

datetime.datetime(

2020, 11, 30, 15, 57, 22, 496000, tzinfo=datetime.timezone.utc

),

)

def test_mview_enable_refresh(self):

table = self._make_one()

self.assertIsNone(table.mview_enable_refresh)

table.mview_enable_refresh = True

self.assertTrue(table.mview_enable_refresh)

table.mview_enable_refresh = False

self.assertFalse(table.mview_enable_refresh)

table.mview_enable_refresh = None

self.assertIsNone(table.mview_enable_refresh)

def test_mview_refresh_interval(self):

table = self._make_one()

self.assertIsNone(table.mview_refresh_interval)

table.mview_refresh_interval = datetime.timedelta(minutes=30)

self.assertEqual(table.mview_refresh_interval, datetime.timedelta(minutes=30))

self.assertEqual(

table._properties["materializedView"]["refreshIntervalMs"], "1800000"

)

table.mview_refresh_interval = None

self.assertIsNone(table.mview_refresh_interval)

def test_from_string(self):

cls = self._get_target_class()

got = cls.from_string("string-project.string_dataset.string_table")

self.assertEqual(got.project, "string-project")

self.assertEqual(got.dataset_id, "string_dataset")

self.assertEqual(got.table_id, "string_table")

self.assertEqual(

str(got.reference), "string-project.string_dataset.string_table"

)

def test_from_string_legacy_string(self):

cls = self._get_target_class()

with self.assertRaises(ValueError):

cls.from_string("string-project:string_dataset.string_table")

def test_from_string_not_fully_qualified(self):

cls = self._get_target_class()

with self.assertRaises(ValueError):

cls.from_string("string_dataset.string_table")

def test_from_api_repr_missing_identity(self):

self._setUpConstants()

RESOURCE = {}

klass = self._get_target_class()

with self.assertRaises(KeyError):

klass.from_api_repr(RESOURCE)

def test_from_api_repr_bare(self):

self._setUpConstants()

RESOURCE = {

"id": "%s:%s.%s" % (self.PROJECT, self.DS_ID, self.TABLE_NAME),

"tableReference": {

"projectId": self.PROJECT,

"datasetId": self.DS_ID,

"tableId": self.TABLE_NAME,

},

"type": "TABLE",

}

klass = self._get_target_class()

table = klass.from_api_repr(RESOURCE)

self.assertEqual(table.table_id, self.TABLE_NAME)

self._verifyResourceProperties(table, RESOURCE)

def test_from_api_repr_w_properties(self):

import datetime

from google.cloud._helpers import UTC

from google.cloud._helpers import _millis

RESOURCE = self._make_resource()

RESOURCE["view"] = {"query": "select fullname, age from person_ages"}

RESOURCE["type"] = "VIEW"

RESOURCE["location"] = "EU"

self.EXP_TIME = datetime.datetime(2015, 8, 1, 23, 59, 59, tzinfo=UTC)

RESOURCE["expirationTime"] = _millis(self.EXP_TIME)

klass = self._get_target_class()

table = klass.from_api_repr(RESOURCE)

self._verifyResourceProperties(table, RESOURCE)

def test_from_api_repr_w_partial_streamingbuffer(self):

import datetime