# Copyright 2020 Google LLC All rights reserved.

#

# 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 sys

import unittest

class TestUtils(unittest.TestCase):

skip_condition = sys.version_info[0] < 3

skip_message = "Subtests are not supported in Python 2"

@unittest.skipIf(skip_condition, skip_message)

def test_PeekIterator(self):

from google.cloud.spanner_dbapi.utils import PeekIterator

cases = [

("list", [1, 2, 3, 4, 6, 7], [1, 2, 3, 4, 6, 7]),

("iter_from_list", iter([1, 2, 3, 4, 6, 7]), [1, 2, 3, 4, 6, 7]),

("tuple", ("a", 12, 0xFF), ["a", 12, 0xFF]),

("iter_from_tuple", iter(("a", 12, 0xFF)), ["a", 12, 0xFF]),

("no_args", (), []),

]

for name, data_in, expected in cases:

with self.subTest(name=name):

pitr = PeekIterator(data_in)

actual = list(pitr)

self.assertEqual(actual, expected)

@unittest.skipIf(skip_condition, "Python 2 has an outdated iterator definition")

def test_peekIterator_list_rows_converted_to_tuples(self):

from google.cloud.spanner_dbapi.utils import PeekIterator

# Cloud Spanner returns results in lists e.g. [result].

# PeekIterator is used by BaseCursor in its fetch* methods.

# This test ensures that anything passed into PeekIterator

# will be returned as a tuple.

pit = PeekIterator([["a"], ["b"], ["c"], ["d"], ["e"]])

got = list(pit)

want = [("a",), ("b",), ("c",), ("d",), ("e",)]

self.assertEqual(got, want, "Rows of type list must be returned as tuples")

seventeen = PeekIterator([[17]])

self.assertEqual(list(seventeen), [(17,)])

pit = PeekIterator([["%", "%d"]])

self.assertEqual(next(pit), ("%", "%d"))

pit = PeekIterator([("Clark", "Kent")])

self.assertEqual(next(pit), ("Clark", "Kent"))

@unittest.skipIf(skip_condition, "Python 2 has an outdated iterator definition")

def test_peekIterator_nonlist_rows_unconverted(self):

from google.cloud.spanner_dbapi.utils import PeekIterator

pi = PeekIterator(["a", "b", "c", "d", "e"])

got = list(pi)

want = ["a", "b", "c", "d", "e"]

self.assertEqual(got, want, "Values should be returned unchanged")

@unittest.skipIf(skip_condition, skip_message)

def test_backtick_unicode(self):

from google.cloud.spanner_dbapi.utils import backtick_unicode

cases = [

("SELECT (1) as foo WHERE 1=1", "SELECT (1) as foo WHERE 1=1"),

("SELECT (1) as föö", "SELECT (1) as `föö`"),

("SELECT (1) as `föö`", "SELECT (1) as `föö`"),

("SELECT (1) as `föö` `umläut", "SELECT (1) as `föö` `umläut"),

("SELECT (1) as `föö", "SELECT (1) as `föö"),

]

for sql, want in cases:

with self.subTest(sql=sql):

got = backtick_unicode(sql)

self.assertEqual(got, want)

@unittest.skipIf(skip_condition, skip_message)

def test_StreamedManyResultSets(self):

from google.cloud.spanner_dbapi.utils import StreamedManyResultSets

cases = [

("iter_from_list", iter([1, 2, 3, 4, 6, 7]), [1, 2, 3, 4, 6, 7]),

("iter_from_tuple", iter(("a", 12, 0xFF)), ["a", 12, 0xFF]),

]

for name, data_in, expected in cases:

with self.subTest(name=name):

stream_result = StreamedManyResultSets()

stream_result._iterators.append(data_in)

actual = list(stream_result)

self.assertEqual(actual, expected)