KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use idempotence method in molecule

Best Python code snippet using molecule_python

test_default_connection_idempotence.py

Source:test_default_connection_idempotence.py

1import os2import time3import unittest4import uuid5import tests.file_utils as file_utils6from onlinepayments.sdk.call_context import CallContext7from onlinepayments.sdk.declined_payment_exception import DeclinedPaymentException8from onlinepayments.sdk.domain.address import Address9from onlinepayments.sdk.domain.amount_of_money import AmountOfMoney10from onlinepayments.sdk.domain.card import Card11from onlinepayments.sdk.domain.card_payment_method_specific_input import CardPaymentMethodSpecificInput12from onlinepayments.sdk.domain.create_payment_request import CreatePaymentRequest13from onlinepayments.sdk.domain.customer import Customer14from onlinepayments.sdk.domain.order import Order15from onlinepayments.sdk.idempotence_exception import IdempotenceException16from tests.unit.server_mock_utils import create_client, create_server_listening17class DefaultConnectionIdempotenceTest(unittest.TestCase):18    """19    Tests that sending the same request twice using DefaultConnection20     has the same effect as only sending one request21    """22    def setUp(self):23        # The server path the request arrived at24        self.request_path = None25        # Stores the idempotence header received at the server side26        self.idempotence_header = "unset"27    def test_idempotence_first_request(self):28        """Test a request with idempotence key where the first request is successful"""29        response_body = read_resource("idempotence_success.json")30        idempotence_key = str(uuid.uuid4())31        call_context = CallContext(idempotence_key)32        request = create_payment_request()33        test_path = "/v2/1/payments"  # relative url through which the request should be sent34        additional_headers \35            = (("Content-Type", "application/json"),36               ("Location", "http://localhost/v2/1/payments/1_1"))37        handler = self.create_handler(response_code=201, body=response_body, additional_headers=additional_headers)38        with create_server_listening(handler) as address:39            with create_client(address) as client:40                response = client.merchant("1").payments().create_payment(request, call_context)41        self.assertIsNotNone(response)42        self.assertIsNotNone(response.payment)43        self.assertIsNotNone(response.payment.id)44        self.assertEqual(test_path, self.request_path, 'Request has arrived at the wrong path')45        self.assertEqual(idempotence_key, self.idempotence_header, "Wrong idempotence key is sent in the request")46        self.assertEqual(idempotence_key, call_context.idempotence_key)47    def test_idempotence_second_request(self):48        """49        Test that the client can successfully handle a response50         indicating a request has been sent prior51        """52        response_body = read_resource("idempotence_success.json")53        idempotence_key = str(uuid.uuid4())54        idempotence_timestamp = str(int(time.time() * 1000))55        call_context = CallContext(idempotence_key)56        request = create_payment_request()57        test_path = "/v2/1/payments"  # relative url through which the request should be sent58        additional_headers \59            = (("Content-Type", "application/json"),60               ("Location", "http://localhost/v2/1/payments/1_1"),61               ("X-GCS-Idempotence-Request-Timestamp", idempotence_timestamp))62        handler = self.create_handler(response_code=201,63                                      body=response_body,64                                      additional_headers=additional_headers)65        with create_server_listening(handler) as address:66            with create_client(address) as client:67                response = client.merchant("1").payments().create_payment(request, call_context)68        self.assertIsNotNone(response)69        self.assertIsNotNone(response.payment)70        self.assertIsNotNone(response.payment.id)71        self.assertEqual(test_path, self.request_path, 'Request has arrived at the wrong path')72        self.assertEqual(idempotence_key, self.idempotence_header, "Wrong idempotence key is sent in the request")73        self.assertEqual(idempotence_key, call_context.idempotence_key)74        self.assertEqual(long(idempotence_timestamp), long(call_context.idempotence_request_timestamp))75    def test_idempotence_first_failure(self):76        """Test a request where a request is rejected without prior requests"""77        response_body = read_resource("idempotence_rejected.json")78        idempotence_key = str(uuid.uuid4())79        call_context = CallContext(idempotence_key)80        request = create_payment_request()81        test_path = "/v2/1/payments"  # relative url through which the request should be sent82        handler = self.create_handler(response_code=402, body=response_body,83                                      additional_headers=(("Content-Type", "application/json"),))84        with create_server_listening(handler) as address:85            with create_client(address) as client:86                with self.assertRaises(DeclinedPaymentException) as exc:87                    client.merchant("1").payments().create_payment(request, call_context)88        self.assertEqual(402, exc.exception.status_code)89        self.assertEqual(response_body, exc.exception.response_body)90        self.assertEqual(test_path, self.request_path,91                         'Request has arrived at the wrong path')92        self.assertEqual(idempotence_key, self.idempotence_header,93                         "Wrong idempotence key is sent in the request")94        self.assertEqual(idempotence_key, call_context.idempotence_key)95        self.assertIsNone(call_context.idempotence_request_timestamp)96    def test_idempotence_second_failure(self):97        """Test a request where a request is rejected with a prior request"""98        response_body = read_resource("idempotence_rejected.json")99        idempotence_key = str(uuid.uuid4())100        idempotence_timestamp = str(int(time.time() * 1000))101        call_context = CallContext(idempotence_key)102        request = create_payment_request()103        # relative url through which the request should be sent104        test_path = "/v2/1/payments"105        additional_headers = (("Content-Type", "application/json"),106                              ("X-GCS-Idempotence-Request-Timestamp", idempotence_timestamp))107        handler = self.create_handler(response_code=402, body=response_body, additional_headers=additional_headers)108        with create_server_listening(handler) as address:109            with create_client(address) as client:110                with self.assertRaises(DeclinedPaymentException) as exc:111                    client.merchant("1").payments().create_payment(request, call_context)112        self.assertEqual(402, exc.exception.status_code)113        self.assertEqual(response_body, exc.exception.response_body)114        self.assertEqual(test_path, self.request_path,115                         'Request has arrived at the wrong path')116        self.assertEqual(idempotence_key, self.idempotence_header,117                         "Wrong idempotence key is sent in the request")118        self.assertEqual(idempotence_key, call_context.idempotence_key)119        self.assertEqual(int(idempotence_timestamp),120                         int(call_context.idempotence_request_timestamp))121    def test_idempotence_duplicate_request(self):122        """Test a request where a request arrived twice"""123        response_body = read_resource("idempotence_duplicate_failure.json")124        idempotence_key = str(uuid.uuid4())125        idempotence_timestamp = str(int(time.time() * 1000))126        call_context = CallContext(idempotence_key)127        request = create_payment_request()128        test_path = "/v2/1/payments"  # relative url through which the request should be sent129        additional_headers = (("Content-Type", "application/json"),130                              ("X-GCS-Idempotence-Request-Timestamp", idempotence_timestamp))131        handler = self.create_handler(response_code=409, body=response_body,132                                      additional_headers=additional_headers)133        with create_server_listening(handler) as address:134            with create_client(address) as client:135                with self.assertRaises(IdempotenceException) as exc:136                    client.merchant("1").payments().create_payment(request, call_context)137        self.assertEqual(409, exc.exception.status_code)138        self.assertEqual(response_body, exc.exception.response_body)139        self.assertEqual(idempotence_key, exc.exception.idempotence_key)140        self.assertEqual(idempotence_key, self.idempotence_header,141                         "Wrong idempotence key is sent in the request")142        self.assertEqual(idempotence_key, call_context.idempotence_key)143        self.assertEqual(test_path, self.request_path,144                         'Request has arrived at the wrong path')145        self.assertEqual(int(idempotence_timestamp),146                         int(call_context.idempotence_request_timestamp))147    def create_handler(self, response_code=200, body='',148                       additional_headers=()):149        """150        Creates a request handler that receives the request on the151         server side152        :param response_code: status code of the desired response153        :param body: the body of the response message to return, it154         should be in json format155        :param additional_headers: additional headers that are added to156         the handler's response157        If the request is sent through the proper path,158         self.request_successful will be set to true, false otherwise159        """160        def handler_func(handler):161            self.idempotence_header = handler.headers['X-GCS-Idempotence-Key']162            # record if the request was sent through the expected path163            self.request_path = handler.path164            handler.protocol_version = 'HTTP/1.1'165            handler.send_response(response_code)166            for header in additional_headers:167                handler.send_header(*header)168            handler.send_header('dummy', None)169            handler.end_headers()170            handler.wfile.write(body)171        return handler_func172def create_payment_request():173    """Convenience method that creates a payment request"""174    amount_of_money = AmountOfMoney()175    amount_of_money.amount = 2345L176    amount_of_money.currency_code = "CAD"177    billing_address = Address()178    billing_address.country_code = "CA"179    customer = Customer()180    customer.billing_address = billing_address181    order = Order()182    order.amount_of_money = amount_of_money183    order.customer = customer184    card = Card()185    card.cvv = "123"186    card.card_number = "4567350000427977"187    card.expiry_date = "1220"188    card_payment_input = CardPaymentMethodSpecificInput()189    card_payment_input.payment_product_id = 1190    card_payment_input.card = card191    body = CreatePaymentRequest()192    body.order = order193    body.card_payment_method_specific_input = card_payment_input194    return body195def read_resource(relative_path):196    return file_utils.read_file(os.path.join("default_implementation",197                                             relative_path))198if __name__ == '__main__':...

call_context.py

Source:call_context.py

1class CallContext:2    """3    A call context can be used to send extra information with a request, and to4    receive extra information from a response.5    6    Please note that this class is not thread-safe. Each request should get its7    own call context instance.8    """9    __idempotence_key = None10    __idempotence_request_timestamp = None11    def __init__(self, idempotence_key=None):12        """13        Sets the idempotence key to use for the next request for which this call14        context is used.15        """16        self.__idempotence_key = idempotence_key17    @property18    def idempotence_key(self):19        """20        :return: The idempotence key.21        """22        return self.__idempotence_key23    @property24    def idempotence_request_timestamp(self):25        """26        :return: The idempotence request timestamp from the response to the last27         request for which this call context was used, or None if no idempotence28         request timestamp was present.29        """30        return self.__idempotence_request_timestamp31    @idempotence_request_timestamp.setter32    def idempotence_request_timestamp(self, idempotence_request_timestamp):33        """34        Sets the idempotence request timestamp.35        This method should only be called by Communicator objects based on the36        response to the request for which this call context was used.37        """...

idempotence.py

Source:idempotence.py

1import sys2import os3VAR_IDEMPOTENCE = u'idempotence'4class CallbackModule(object):5    """6    This callback module performs the idempotence test whenever the 'idempotence' variable is set to True.7    """8    def __init__(self):9        pass10    def playbook_on_stats(self, stats):11        if (u'%s' % VAR_IDEMPOTENCE) in self.playbook.extra_vars.keys() and self.playbook.extra_vars[VAR_IDEMPOTENCE]:12            if len(stats.dark) > 0:13                print ('idempotence test failed! unreachable=%s > 0') % stats.dark14                sys.exit(os.EX_SOFTWARE)15            if len(stats.changed) > 0:16                print ('idempotence test failed! changed=%s > 0') % stats.changed17                sys.exit(os.EX_SOFTWARE)18            if len(stats.failures) > 0:19                print ('idempotence test failed! failures=%s > 0') % stats.failures...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.