Create stub for unit testing

[bgreeley]

It would be nice to have stub methods available for unit testing the cloud storage API just like there are for the mail API, memcache API, etc.

[tkaitchuck]

You should be able to use the API as-is in a unit test. (See the tests included with it) Are you looking for something more?

[bgreeley]

Whenever I run unit tests I get the following error:

WARNING:root:suspended generator urlfetch(context.py:1276) raised AttributeError('NoneType' object has no attribute 'storage')
WARNING:root:suspended generator urlfetch_async(rest_api.py:275) raised AttributeError('NoneType' object has no attribute 'storage')
WARNING:root:suspended generator run(api_utils.py:164) raised AttributeError('NoneType' object has no attribute 'storage')
WARNING:root:suspended generator do_request_async(rest_api.py:198) raised AttributeError('NoneType' object has no attribute 'storage')
WARNING:root:suspended generator do_request_async(storage_api.py:128) raised AttributeError('NoneType' object has no attribute 'storage')

I have set up stub for the URL fetch API:

self.testbed = testbed.Testbed()
self.testbed.activate()
self.testbed.init_urlfetch_stub()

Is there something I'm not doing that I should be?

[bgreeley]

Actually, looks like I needed to add a stub for blobstore as well:

self.testbed.init_blobstore_stub()

That fixed it.

[evanj]

Its possible that this issue should be closed. Unfortunately, commit 19f07688a2 broke this, at least for me.

When I run the tests locally using the previous commit (8f196d4d7c), they all pass. When I run the unit tests with the current master (19f07688a2), I get authentication errors that seem to imply it is trying to access the real Google Cloud Storage:

$ PYTHONPATH=src python test/cloudstorage_test.py

[... lots of omitted output ...]

======================================================================
ERROR: testWriteRead (__main__.CloudStorageTest)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "test/cloudstorage_test.py", line 344, in testWriteRead
    f = cloudstorage.open(TESTFILE, 'w')
  File "/Users/evanjones/appengine-gcs-client/python/src/cloudstorage/cloudstorage_api.py", line 97, in open
    return storage_api.StreamingBuffer(api, filename, content_type, options)
  File "/Users/evanjones/appengine-gcs-client/python/src/cloudstorage/storage_api.py", line 705, in __init__
    body=content)
  File "/Users/evanjones/appengine-gcs-client/python/src/cloudstorage/errors.py", line 128, in check_status
    raise AuthorizationError(msg)
AuthorizationError: Expect status [201] from Google Storage. But got status 401.
Path: '/bucket/testfile'.
Request headers: {'x-goog-resumable': 'start', 'x-goog-api-version': '2', 'accept-encoding': 'gzip, *'}.
Response headers: {'content-length': '131', 'vary': 'Origin', 'server': 'UploadServer', 'x-guploader-uploadid': 'AEnB2UpfGiScxYITfyhCg2fa5zeqqjUF7NXGAjqUFBHivErTlc_qMAWYY692o91UXeISmpvrI3RwT5aJkz6W6FLh6fDP-m8LgA', 'date': 'Wed, 05 Oct 2016 17:13:13 GMT', 'alt-svc': 'quic=":443"; ma=2592000; v="36,35,34,33,32"', 'content-type': 'application/xml; charset=UTF-8', 'www-authenticate': 'Bearer realm="https://accounts.google.com/", error="invalid_token"'}.
Body: "<?xml version='1.0' encoding='UTF-8'?><Error><Code>AuthenticationRequired</Code><Message>Authentication required.</Message></Error>".
Extra info: None.

[justinwhite]

For anyone stumbling on this, the testbed approach works fine. However, I ended up using init_all_stubs because the test server began asking for other [unexpected] stubs, like memcache, besides the urlfetch and blobstore stubs.

Below are the relevant portions of my test:

  import pickle
  import cloudstorage as gcs

  .. 

  def setUp(self):
    self.testbed = testbed.Testbed()
    self.testbed.activate()
    self.testbed.init_all_stubs()

  def tearDown(self):
    self.testbed.deactivate()

  def testFoo(self):
    xs = [('spam', 0), ('ham', 1)]

    # write some sample data to the GCS stub
    gcs_file = gcs.open(GCS_PATH, 'w', content_type='text/plain')
    gcs_file.write(pickle.dumps(xs))
    gcs_file.close()