# Copyright 2015 Google Inc.

#

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

class Test__do_nothing_page_start(unittest.TestCase):

def _call_fut(self, iterator, page, response):

from google.cloud.iterator import _do_nothing_page_start

return _do_nothing_page_start(iterator, page, response)

def test_do_nothing(self):

result = self._call_fut(None, None, None)

self.assertIsNone(result)

class TestPage(unittest.TestCase):

@staticmethod

def _get_target_class():

from google.cloud.iterator import Page

return Page

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

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

def test_constructor(self):

parent = object()

item_to_value = object()

page = self._make_one(parent, (1, 2, 3), item_to_value)

self.assertIs(page._parent, parent)

self.assertEqual(page._num_items, 3)

self.assertEqual(page._remaining, 3)

self.assertIs(page._item_to_value, item_to_value)

def test_num_items_property(self):

page = self._make_one(None, (), None)

num_items = 42

page._num_items = num_items

self.assertEqual(page.num_items, num_items)

def test_remaining_property(self):

page = self._make_one(None, (), None)

remaining = 1337

page._remaining = remaining

self.assertEqual(page.remaining, remaining)

def test___iter__(self):

page = self._make_one(None, (), None)

self.assertIs(iter(page), page)

def test_iterator_calls__item_to_value(self):

import six

class Parent(object):

calls = 0

def item_to_value(self, item):

self.calls += 1

return item

parent = Parent()

page = self._make_one(parent, (10, 11, 12),

Parent.item_to_value)

page._remaining = 100

self.assertEqual(parent.calls, 0)

self.assertEqual(page.remaining, 100)

self.assertEqual(six.next(page), 10)

self.assertEqual(parent.calls, 1)

self.assertEqual(page.remaining, 99)

self.assertEqual(six.next(page), 11)

self.assertEqual(parent.calls, 2)

self.assertEqual(page.remaining, 98)

self.assertEqual(six.next(page), 12)

self.assertEqual(parent.calls, 3)

self.assertEqual(page.remaining, 97)

class TestIterator(unittest.TestCase):

@staticmethod

def _get_target_class():

from google.cloud.iterator import Iterator

return Iterator

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

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

def test_constructor(self):

connection = _Connection()

client = _Client(connection)

item_to_value = object()

token = 'ab13nceor03'

max_results = 1337

iterator = self._make_one(client, item_to_value, page_token=token,

max_results=max_results)

self.assertFalse(iterator._started)

self.assertIs(iterator.client, client)

self.assertIs(iterator._item_to_value, item_to_value)

self.assertEqual(iterator.max_results, max_results)

# Changing attributes.

self.assertEqual(iterator.page_number, 0)

self.assertEqual(iterator.next_page_token, token)

self.assertEqual(iterator.num_results, 0)

def test_pages_property(self):

iterator = self._make_one(None, None)

self.assertFalse(iterator._started)

mock_iter = object()

incremented = []

def page_iter(increment):

incremented.append(increment)

return mock_iter

iterator._page_iter = page_iter

self.assertIs(iterator.pages, mock_iter)

self.assertEqual(incremented, [True])

# Check the side-effect.

self.assertTrue(iterator._started)

def test_pages_property_started(self):

import types

iterator = self._make_one(None, None)

self.assertIsInstance(iterator.pages, types.GeneratorType)

# Make sure we cannot restart.

with self.assertRaises(ValueError):

getattr(iterator, 'pages')

def test_pages_property_items_started(self):

import types

iterator = self._make_one(None, None)

self.assertIsInstance(iter(iterator), types.GeneratorType)

with self.assertRaises(ValueError):

getattr(iterator, 'pages')

@staticmethod

def _do_nothing(parent, value):

return parent, value

def test__items_iter(self):

import types

import six

from google.cloud.iterator import Page

# Items to be returned.

item1 = 17

item2 = 100

item3 = 211

# Make pages from mock responses

parent = object()

page1 = Page(parent, (item1, item2), self._do_nothing)

page2 = Page(parent, (item3,), self._do_nothing)

iterator = self._make_one(None, None)

# Fake the page iterator on the object.

incremented = []

def page_iter(increment):

incremented.append(increment)

return iter((page1, page2))

iterator._page_iter = page_iter

items_iter = iterator._items_iter()

# Make sure it is a generator.

self.assertIsInstance(items_iter, types.GeneratorType)

# Consume items and check the state of the iterator.

self.assertEqual(iterator.num_results, 0)

self.assertEqual(six.next(items_iter), (parent, item1))

self.assertEqual(iterator.num_results, 1)

self.assertEqual(six.next(items_iter), (parent, item2))

self.assertEqual(iterator.num_results, 2)

self.assertEqual(six.next(items_iter), (parent, item3))

self.assertEqual(iterator.num_results, 3)

with self.assertRaises(StopIteration):

six.next(items_iter)

# Make sure our page_iter() was called correctly.

self.assertEqual(incremented, [False])

def test___iter__(self):

iterator = self._make_one(None, None)

self.assertFalse(iterator._started)

incremented = []

def page_iter(increment):

incremented.append(increment)

return iter(())

iterator._page_iter = page_iter

self.assertEqual(list(iterator), [])

# Check the side-effect.

self.assertTrue(iterator._started)

def test___iter___started(self):

import types

iterator = self._make_one(None, None)

self.assertIsInstance(iter(iterator), types.GeneratorType)

with self.assertRaises(ValueError):

iter(iterator)

def test___iter___pages_started(self):

import types

iterator = self._make_one(None, None)

self.assertIsInstance(iterator.pages, types.GeneratorType)

with self.assertRaises(ValueError):

iter(iterator)

def test__next_page_virtual(self):

iterator = self._make_one(None, None)

with self.assertRaises(NotImplementedError):

iterator._next_page()

class TestHTTPIterator(unittest.TestCase):

@staticmethod

def _get_target_class():

from google.cloud.iterator import HTTPIterator

return HTTPIterator

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

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

def test_constructor(self):

from google.cloud.iterator import _do_nothing_page_start

connection = _Connection()

client = _Client(connection)

path = '/foo'

iterator = self._make_one(client, path, None)

self.assertFalse(iterator._started)

self.assertIs(iterator.client, client)

self.assertEqual(iterator.path, path)

self.assertIsNone(iterator._item_to_value)

self.assertEqual(iterator._items_key, 'items')

self.assertIsNone(iterator.max_results)

self.assertEqual(iterator.extra_params, {})

self.assertIs(iterator._page_start, _do_nothing_page_start)

# Changing attributes.

self.assertEqual(iterator.page_number, 0)

self.assertIsNone(iterator.next_page_token)

self.assertEqual(iterator.num_results, 0)

def test_constructor_w_extra_param_collision(self):

connection = _Connection()

client = _Client(connection)

path = '/foo'

extra_params = {'pageToken': 'val'}

with self.assertRaises(ValueError):

self._make_one(client, path, None, extra_params=extra_params)

def test_pages_iter_empty_then_another(self):

import six

from google.cloud._testing import _Monkey

from google.cloud import iterator as MUT

items_key = 'its-key'

iterator = self._make_one(None, None, None, items_key=items_key)

# Fake the next page class.

fake_page = MUT.Page(None, (), None)

page_args = []

def dummy_response():

return {}

def dummy_page_class(*args):

page_args.append(args)

return fake_page

iterator._get_next_page_response = dummy_response

pages_iter = iterator.pages

with _Monkey(MUT, Page=dummy_page_class):

page = six.next(pages_iter)

self.assertIs(page, fake_page)

self.assertEqual(

page_args, [(iterator, (), iterator._item_to_value)])

def test_iterate(self):

import six

path = '/foo'

key1 = 'key1'

key2 = 'key2'

item1, item2 = object(), object()

ITEMS = {key1: item1, key2: item2}

def item_to_value(iterator, item): # pylint: disable=unused-argument

return ITEMS[item['name']]

connection = _Connection(

{'items': [{'name': key1}, {'name': key2}]})

client = _Client(connection)

iterator = self._make_one(client, path=path,

item_to_value=item_to_value)

self.assertEqual(iterator.num_results, 0)

items_iter = iter(iterator)

val1 = six.next(items_iter)

self.assertEqual(val1, item1)

self.assertEqual(iterator.num_results, 1)

val2 = six.next(items_iter)

self.assertEqual(val2, item2)

self.assertEqual(iterator.num_results, 2)

with self.assertRaises(StopIteration):

six.next(items_iter)

kw, = connection._requested

self.assertEqual(kw['method'], 'GET')

self.assertEqual(kw['path'], path)

self.assertEqual(kw['query_params'], {})

def test__has_next_page_new(self):

connection = _Connection()

client = _Client(connection)

path = '/foo'

iterator = self._make_one(client, path, None)

self.assertTrue(iterator._has_next_page())

def test__has_next_page_w_number_no_token(self):

connection = _Connection()

client = _Client(connection)

path = '/foo'

iterator = self._make_one(client, path, None)

iterator.page_number = 1

self.assertFalse(iterator._has_next_page())

def test__has_next_page_w_number_w_token(self):

connection = _Connection()

client = _Client(connection)

path = '/foo'

token = 'token'

iterator = self._make_one(client, path, None)

iterator.page_number = 1

iterator.next_page_token = token

self.assertTrue(iterator._has_next_page())

def test__has_next_page_w_max_results_not_done(self):

iterator = self._make_one(None, None, None, max_results=3,

page_token='definitely-not-none')

iterator.page_number = 1

self.assertLess(iterator.num_results, iterator.max_results)

self.assertTrue(iterator._has_next_page())

def test__has_next_page_w_max_results_done(self):

iterator = self._make_one(None, None, None, max_results=3)

iterator.page_number = 1

iterator.num_results = iterator.max_results

self.assertFalse(iterator._has_next_page())

def test__get_query_params_no_token(self):

connection = _Connection()

client = _Client(connection)

path = '/foo'

iterator = self._make_one(client, path, None)

self.assertEqual(iterator._get_query_params(), {})

def test__get_query_params_w_token(self):

connection = _Connection()

client = _Client(connection)

path = '/foo'

token = 'token'

iterator = self._make_one(client, path, None)

iterator.next_page_token = token

self.assertEqual(iterator._get_query_params(),

{'pageToken': token})

def test__get_query_params_w_max_results(self):

connection = _Connection()

client = _Client(connection)

path = '/foo'

max_results = 3

iterator = self._make_one(client, path, None,

max_results=max_results)

iterator.num_results = 1

local_max = max_results - iterator.num_results

self.assertEqual(iterator._get_query_params(),

{'maxResults': local_max})

def test__get_query_params_extra_params(self):

connection = _Connection()

client = _Client(connection)

path = '/foo'

extra_params = {'key': 'val'}

iterator = self._make_one(client, path, None,

extra_params=extra_params)

self.assertEqual(iterator._get_query_params(), extra_params)

def test__get_query_params_w_token_and_extra_params(self):

connection = _Connection()

client = _Client(connection)

path = '/foo'

token = 'token'

extra_params = {'key': 'val'}

iterator = self._make_one(client, path, None,

extra_params=extra_params)

iterator.next_page_token = token

expected_query = extra_params.copy()

expected_query.update({'pageToken': token})

self.assertEqual(iterator._get_query_params(), expected_query)

def test__get_next_page_response_new_no_token_in_response(self):

path = '/foo'

token = 'token'

key1 = 'key1'

key2 = 'key2'

connection = _Connection({'items': [{'name': key1}, {'name': key2}],

'nextPageToken': token})

client = _Client(connection)

iterator = self._make_one(client, path, None)

response = iterator._get_next_page_response()

self.assertEqual(response['items'], [{'name': key1}, {'name': key2}])

kw, = connection._requested

self.assertEqual(kw['method'], 'GET')

self.assertEqual(kw['path'], path)

self.assertEqual(kw['query_params'], {})

def test__get_next_page_response_with_post(self):

path = '/foo'

returned = {'green': 'eggs', 'ham': 55}

connection = _Connection(returned)

client = _Client(connection)

iterator = self._make_one(client, path, None)

iterator._HTTP_METHOD = 'POST'

response = iterator._get_next_page_response()

self.assertEqual(response, returned)

self.assertEqual(len(connection._requested), 1)

called_kwargs = connection._requested[0]

self.assertEqual(called_kwargs, {

'method': iterator._HTTP_METHOD,

'path': path,

'data': {},

})

def test__get_next_page_bad_http_method(self):

path = '/foo'

client = _Client(None)

iterator = self._make_one(client, path, None)

iterator._HTTP_METHOD = 'NOT-A-VERB'

with self.assertRaises(ValueError):

iterator._get_next_page_response()

class TestGAXIterator(unittest.TestCase):

@staticmethod

def _get_target_class():

from google.cloud.iterator import GAXIterator

return GAXIterator

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

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

def test_constructor(self):

client = _Client(None)

token = 'zzzyy78kl'

page_iter = SimpleIter(token)

item_to_value = object()

max_results = 1337

iterator = self._make_one(client, page_iter, item_to_value,

max_results=max_results)

self.assertFalse(iterator._started)

self.assertIs(iterator.client, client)

self.assertIs(iterator._item_to_value, item_to_value)

self.assertEqual(iterator.max_results, max_results)

self.assertIs(iterator._gax_page_iter, page_iter)

# Changing attributes.

self.assertEqual(iterator.page_number, 0)

self.assertEqual(iterator.next_page_token, token)

self.assertEqual(iterator.num_results, 0)

@staticmethod

def _do_nothing(parent, value):

return parent, value

def test__next_page(self):

from google.cloud._testing import _GAXPageIterator

from google.cloud.iterator import Page

# Make a mock ``google.gax.PageIterator``

page_items = (29, 31) # Items for just one page.

page_token = '2sde98ds2s0hh'

page_iter = _GAXPageIterator(page_items, page_token=page_token)

# Wrap the GAX iterator.

iterator = self._make_one(None, page_iter, self._do_nothing)

page = iterator._next_page()

# First check the page token.

self.assertEqual(iterator.next_page_token, page_token)

# Then check the page.

self.assertIsInstance(page, Page)

# _do_nothing will throw the iterator in front.

expected = zip((iterator, iterator), page_items)

self.assertEqual(list(page), list(expected))

def test__next_page_empty(self):

from google.cloud._testing import _GAXPageIterator

# Make a mock ``google.gax.PageIterator``

page_iter = _GAXPageIterator()

# Wrap the GAX iterator.

iterator = self._make_one(None, page_iter, self._do_nothing)

page = iterator._next_page()

self.assertIsNone(page)

self.assertIsNone(iterator.next_page_token)

def test_iterate(self):

import six

from google.cloud._testing import _GAXPageIterator

item1 = object()

item2 = object()

item3 = object()

token1 = 'smkdme30e2e32r'

token2 = '39cm9csl123dck'

# Make a mock ``google.gax.PageIterator``

page1 = (item1,)

page2 = (item2, item3)

page_iter = _GAXPageIterator(page1, page2, page_token=token1)

iterator = self._make_one(None, page_iter, self._do_nothing)

self.assertEqual(iterator.num_results, 0)

items_iter = iter(iterator)

val1 = six.next(items_iter)

self.assertEqual(val1, (iterator, item1))

self.assertEqual(iterator.num_results, 1)

self.assertEqual(iterator.next_page_token, token1)

# Before grabbing the next page, hot-swap the token

# on the ``page_iter``.

page_iter.page_token = token2

# Grab the next item (which will cause the next page).

val2 = six.next(items_iter)

self.assertEqual(val2, (iterator, item2))

self.assertEqual(iterator.num_results, 2)

self.assertEqual(iterator.next_page_token, token2)

# Grab the final item from the final / current page.

val3 = six.next(items_iter)

self.assertEqual(val3, (iterator, item3))

self.assertEqual(iterator.num_results, 3)

# Make sure the token did not change.

self.assertEqual(iterator.next_page_token, token2)

with self.assertRaises(StopIteration):

six.next(items_iter)

class _Connection(object):

def __init__(self, *responses):

self._responses = responses

self._requested = []

def api_request(self, **kw):

self._requested.append(kw)

response, self._responses = self._responses[0], self._responses[1:]

return response

class _Client(object):

def __init__(self, connection):

self._connection = connection

class SimpleIter(object):

def __init__(self, page_token=None):

self.page_token = page_token