Ultra File Manager

Edit: /opt/gsutil/gslib/vendored/boto/tests/unit/mws/test_connection.py

#!/usr/bin/env python
# Copyright (c) 2012 Amazon.com, Inc. or its affiliates.  All Rights Reserved
#
# Permission is hereby granted, free of charge, to any person obtaining a
# copy of this software and associated documentation files (the
# "Software"), to deal in the Software without restriction, including
# without limitation the rights to use, copy, modify, merge, publish, dis-
# tribute, sublicense, and/or sell copies of the Software, and to permit
# persons to whom the Software is furnished to do so, subject to the fol-
# lowing conditions:
#
# The above copyright notice and this permission notice shall be included
# in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
# OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABIL-
# ITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT
# SHALL THE AUTHOR BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
# IN THE SOFTWARE.
#
from boto.mws.connection import MWSConnection, api_call_map, destructure_object
from boto.mws.response import (ResponseElement, GetFeedSubmissionListResult,
                               ResponseFactory)
from boto.exception import BotoServerError

from tests.compat import unittest

from tests.unit import AWSMockServiceTestCase

from mock import MagicMock

class TestMWSConnection(AWSMockServiceTestCase):
    connection_class = MWSConnection
    mws = True

def default_body(self):
        return b"""<?xml version="1.0"?>
<GetFeedSubmissionListResponse xmlns="http://mws.amazonservices.com/
doc/2009-01-01/">
  <GetFeedSubmissionListResult>
    <NextToken>2YgYW55IGNhcm5hbCBwbGVhc3VyZS4=</NextToken>
    <HasNext>true</HasNext>
    <FeedSubmissionInfo>
      <FeedSubmissionId>2291326430</FeedSubmissionId>
      <FeedType>_POST_PRODUCT_DATA_</FeedType>
      <SubmittedDate>2009-02-20T02:10:35+00:00</SubmittedDate>
      <FeedProcessingStatus>_SUBMITTED_</FeedProcessingStatus>
    </FeedSubmissionInfo>
  </GetFeedSubmissionListResult>
  <ResponseMetadata>
    <RequestId>1105b931-6f1c-4480-8e97-f3b467840a9e</RequestId>
  </ResponseMetadata>
</GetFeedSubmissionListResponse>"""

def default_body_error(self):
        return b"""<?xml version="1.0" encoding="UTF-8"?>
<ErrorResponse xmlns="http://mws.amazonaws.com/doc/2009-01-01/">
  
  <Error>
    <Type>Sender</Type>
    <Code>string</Code>
    <Message>string</Message>
    <Detail>
      
      <AnyElement xmlns=""/>
    </Detail>
  </Error>
  <RequestId>string</RequestId>
</ErrorResponse>"""

def test_destructure_object(self):
        # Test that parsing of user input to Amazon input works.
        response = ResponseElement()
        response.C = 'four'
        response.D = 'five'
        inputs = [
            ('A', 'B'), ['B', 'A'], set(['C']),
            False, 'String', {'A': 'one', 'B': 'two'},
            response,
            {'A': 'one', 'B': 'two',
             'C': [{'D': 'four', 'E': 'five'},
                   {'F': 'six', 'G': 'seven'}]},
        ]
        outputs = [
            {'Prefix.1': 'A', 'Prefix.2': 'B'},
            {'Prefix.1': 'B', 'Prefix.2': 'A'},
            {'Prefix.1': 'C'},
            {'Prefix': 'false'}, {'Prefix': 'String'},
            {'Prefix.A': 'one', 'Prefix.B': 'two'},
            {'Prefix.C': 'four', 'Prefix.D': 'five'},
            {'Prefix.A': 'one', 'Prefix.B': 'two',
             'Prefix.C.member.1.D': 'four',
             'Prefix.C.member.1.E': 'five',
             'Prefix.C.member.2.F': 'six',
             'Prefix.C.member.2.G': 'seven'}
        ]
        for user, amazon in zip(inputs, outputs):
            result = {}
            members = user is inputs[-1]
            destructure_object(user, result, prefix='Prefix', members=members)
            self.assertEqual(result, amazon)

def test_decorator_order(self):
        for action, func in api_call_map.items():
            func = getattr(self.service_connection, func)
            decs = [func.__name__]
            while func:
                i = 0
                if not hasattr(func, '__closure__'):
                    func = getattr(func, '__wrapped__', None)
                    continue
                while i < len(func.__closure__):
                    value = func.__closure__[i].cell_contents
                    if hasattr(value, '__call__'):
                        if 'requires' == value.__name__:
                            self.assertTrue(not decs or decs[-1] == 'requires')
                        decs.append(value.__name__)
                    i += 1
                func = getattr(func, '__wrapped__', None)

def test_built_api_call_map(self):
        # Ensure that the map is populated.
        # It starts empty, but the decorators should add to it as they're
        # applied. As of 2013/10/21, there were 52 calls (with more likely
        # to be added), so let's simply ensure there are enough there.
        self.assertTrue(len(api_call_map.keys()) > 50)

def test_method_for(self):
        # First, ensure that the map is in "right enough" state.
        self.assertTrue('GetFeedSubmissionList' in api_call_map)

# Make sure we can find the correct method.
        func = self.service_connection.method_for('GetFeedSubmissionList')
        # Ensure the right name was found.
        self.assertTrue(callable(func))
        ideal = self.service_connection.get_feed_submission_list
        self.assertEqual(func, ideal)

# Check a non-existent action.
        func = self.service_connection.method_for('NotHereNorThere')
        self.assertEqual(func, None)

def test_response_factory(self):
        connection = self.service_connection
        body = self.default_body()
        action = 'GetFeedSubmissionList'
        parser = connection._response_factory(action, connection=connection)
        response = connection._parse_response(parser, 'text/xml', body)
        self.assertEqual(response._action, action)
        self.assertEqual(response.__class__.__name__, action + 'Response')
        self.assertEqual(response._result.__class__,
                         GetFeedSubmissionListResult)

class MyResult(GetFeedSubmissionListResult):
            _hello = '_world'

scope = {'GetFeedSubmissionListResult': MyResult}
        connection._setup_factories([scope])

parser = connection._response_factory(action, connection=connection)
        response = connection._parse_response(parser, 'text/xml', body)
        self.assertEqual(response._action, action)
        self.assertEqual(response.__class__.__name__, action + 'Response')
        self.assertEqual(response._result.__class__, MyResult)
        self.assertEqual(response._result._hello, '_world')
        self.assertEqual(response._result.HasNext, 'true')

def test_get_service_status(self):
        with self.assertRaises(AttributeError) as err:
            self.service_connection.get_service_status()

self.assertTrue('products' in str(err.exception))
        self.assertTrue('inventory' in str(err.exception))
        self.assertTrue('feeds' in str(err.exception))

def test_post_request(self):

self.service_connection._mexe = MagicMock(
            side_effect=
                BotoServerError(500, 'You request has bee throttled', body=self.default_body_error()))

with self.assertRaises(BotoServerError) as err:
            self.service_connection.get_lowest_offer_listings_for_asin(
                MarketplaceId='12345',
                ASINList='ASIN12345',
                condition='Any',
                SellerId='1234',
                excludeme='True')

self.assertTrue('throttled' in str(err.reason))
            self.assertEqual(int(err.status), 200)
            
    def test_sandboxify(self):
        # Create one-off connection class that has self._sandboxed = True
        conn = MWSConnection(https_connection_factory=self.https_connection_factory,
            aws_access_key_id='aws_access_key_id',
            aws_secret_access_key='aws_secret_access_key',
            sandbox=True)
        self.assertEqual(conn._sandboxify('a/bogus/path'), 'a/bogus_Sandbox/path')

if __name__ == '__main__':
    unittest.main()

Ukuran: 8.3 KB