Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use create_params method in localstack

Best Python code snippet using localstack_python

test_params.py

Source:test_params.py

...7from gocardless import utils, urlbuilder8class ExpiringLimitTestCase(object):9    """superclass factoring out tests for expiring limit param objects"""10    def test_interval_length_is_positive(self):11        pars = self.create_params(10, "1321230", 1, "day")12        with self.assertRaises(ValueError):13            pars = self.create_params(10, "1123210", -1, "day")14    def test_interval_unit_is_valid(self):15        for interval_unit in ["day", "week", "month"]:16            pars = self.create_params(10, 10, "11235432", interval_unit)17        with self.assertRaises(ValueError):18            pars = self.create_params(10, 10, "1432233123", "invalid")19    def _future_date_tester(self, argname):20        invalid_date = datetime.datetime.now() - datetime.timedelta(100)21        valid_date = datetime.datetime.now() + datetime.timedelta(2000)22        par1 = self.create_params(10, 10, "23423421", "day", **{argname:valid_date})23        with self.assertRaises(ValueError):24            par1 = self.create_params(10, 10, "2342341", "day",25                    **{argname:invalid_date})26    def test_expires_at_in_future(self):27        self._future_date_tester("expires_at")28    def test_interval_count_positive(self):29        with self.assertRaises(ValueError):30            self.create_params(10, 10, "merchid", "day", interval_count=-1)31class PreAuthParamsTestCase(ExpiringLimitTestCase, unittest.TestCase):32    def default_args_construct(self, extra_options):33        """34        For testing optional arguments, builds the param object with valid35        required arguments and adds optionl arguments as keywords from36        `extra_options`37        :param extra_options: Extra optional keyword arguments to pass to38        the constructor.39        """40        return urlbuilder.\41                PreAuthorizationParams(12, "3456", 6, "month", **extra_options)42    def create_params(self, *args, **kwargs):43        return urlbuilder.PreAuthorizationParams(*args, **kwargs)44    def test_max_amount_is_positive(self):45        self.assertRaises(ValueError, \46                urlbuilder.PreAuthorizationParams, -1, "1232532", 4, "month")47    def test_interval_length_is_a_positive_integer(self):48        self.assertRaises(ValueError, \49                urlbuilder.PreAuthorizationParams, 12, "!2343", -3, "month")50    def test_interval_unit_is_one_of_accepted(self):51        for unit_type in ["month", "day", "week"]:52            pa = urlbuilder.PreAuthorizationParams(12, "1234", 3, unit_type)53        self.assertRaises(ValueError, \54                urlbuilder.PreAuthorizationParams, 21,"1234", 4, "soem other unit")55    def test_expires_at_is_later_than_now(self):56        earlier = datetime.datetime.now() - datetime.timedelta(1)57        self.assertRaises(ValueError, self.default_args_construct, \58                {"expires_at":earlier})59    def test_interval_count_is_postive_integer(self):60        self.assertRaises(ValueError, self.default_args_construct, \61                {"interval_count":-1})62class PreAuthParamsToDictTestCase(unittest.TestCase):63    def setUp(self):64        self.all_params = {65                "max_amount":12,66                "interval_unit":"day",67                "interval_length":10,68                "merchant_id":"1234435",69                "name":"aname",70                "description":"adesc",71                "interval_count":123,72                "currency":"GBP",73                "expires_at":datetime.datetime.strptime("2020-01-01", "%Y-%m-%d"),74                "calendar_intervals":True75                }76        self.required_keys = [77                "max_amount", "interval_unit", "interval_length", "merchant_id"]78    def create_from_params_dict(self, in_params):79        params = in_params.copy()80        pa = urlbuilder.PreAuthorizationParams(params.pop("max_amount"), \81                params.pop("merchant_id"), \82                params.pop("interval_length"), \83                params.pop("interval_unit"),\84                **params)85        return pa86    def assert_inverse(self, keys):87        params = dict([[k,v] for k,v in six.iteritems(self.all_params) \88                if k in keys])89        pa = self.create_from_params_dict(params)90        self.assertEqual(params, pa.to_dict())91    def test_to_dict_all_params(self):92        self.assert_inverse(list(self.all_params.keys()))93    def test_to_dict_only_required(self):94        self.assert_inverse(self.required_keys)95class BillParamsTestCase(unittest.TestCase):96    def create_params(self, *args, **kwargs):97        return urlbuilder.BillParams(*args, **kwargs)98    def test_amount_is_positive(self):99        params = self.create_params(10, "merchid")100        with self.assertRaises(ValueError):101            par2 = self.create_params(-1, "merchid")102    def test_to_dict_required(self):103        pars = self.create_params(10, "merchid")104        res = pars.to_dict()105        expected = {"amount":10, "merchant_id":"merchid"}106        self.assertEqual(res, expected)107    def test_to_dict_optional(self):108        pars = self.create_params(10, "merchid", name="aname", description="adesc")109        res = pars.to_dict()110        expected = {"amount":10,111                "name":"aname",112                "description":"adesc",113                "merchant_id":"merchid"114                }115        self.assertEqual(res, expected)116    def test_resource_name_is_bills(self):117        pars = urlbuilder.BillParams(10, "merchid")118        self.assertEqual(pars.resource_name, "bills")119class SubscriptionParamsTestCase(ExpiringLimitTestCase, unittest.TestCase):120    def create_params(self, *args, **kwargs):121        return urlbuilder.SubscriptionParams(*args, **kwargs)122    def test_setup_fee(self):123        pars = self.create_params(10, "merchid", 10, "day", setup_fee=20)124        expected = {125                "merchant_id": "merchid",126                "amount": 10,127                "interval_length": 10,128                "interval_unit" : "day",129                "setup_fee": 20130                }131        self.assertEqual(expected, pars.to_dict())132    def test_start_at_in_future(self):133        valid_date = datetime.datetime.now() + datetime.timedelta(200)134        invalid_date = datetime.datetime.now() - datetime.timedelta(100)135        par1 = self.create_params(10,"merchid", 10, "day", start_at=valid_date)136        with self.assertRaises(ValueError):137            par2 = self.create_params(10, "merchid", 10, "day",138                    start_at=invalid_date)139    def test_expires_at_after_start_at(self):140        date1 = datetime.datetime.now() + datetime.timedelta(100)141        date2 = datetime.datetime.now() + datetime.timedelta(200)142        par1 = self.create_params(10, "merchid", 10, "day",143                expires_at=date2, start_at=date1)144        with self.assertRaises(ValueError):145            par2 = self.create_params(10, "merchid", 10, "day",146                    expires_at=date1, start_at=date2)147    def test_to_dict_only_required(self):148        expected = {149                "merchant_id":"merchid",150                "amount":10,151                "interval_length":10,152                "interval_unit":"day"}153        pars = self.create_params(10, "merchid", 10, "day")154        self.assertEqual(expected, pars.to_dict())155    def test_to_dict_all(self):156        start_at = datetime.datetime.now() + datetime.timedelta(1000)157        expires_at =datetime.datetime.now() + datetime.timedelta(2000)158        expected = {159                "merchant_id":"merchid",160                "amount":10,161                "interval_length":10,162                "interval_unit":"day",163                "interval_count":5,164                "start_at":start_at.isoformat()[:-7] + "Z",165                "expires_at":expires_at.isoformat()[:-7] + "Z",166                "name":"aname",167                "description":"adesc",168                }169        par = self.create_params(10, "merchid", 10, "day", start_at=start_at,170                expires_at=expires_at, interval_count=5, name="aname",171                description="adesc")172        self.assertEqual(expected, par.to_dict())173class PrepopDataTestCase(unittest.TestCase):174    def setUp(self):175        self.mock_prepop = {"first_name": "Tom",176                "last_name": "Blomfield",177                "email": "tom@gocardless.com"178                }179    def assert_prepop(self, params):180        self.assertEqual(params.to_dict()["user"], self.mock_prepop)181    def test_bill_params(self):182        params = urlbuilder.BillParams(10, "amerchid", user=self.mock_prepop)183        self.assert_prepop(params)...

fave.py

Source:fave.py

...5        """6        :param url:7        """8        method = self.get_method_name(self.add_article)9        params = self.create_params(locals())10        r = await self.api_request(method, params)11        return m.AddArticle(**r)12    async def add_link(self, link: str = None):13        """14        Add a link to the user faves.15        :param link: Link URL.16        """17        method = self.get_method_name(self.add_link)18        params = self.create_params(locals())19        r = await self.api_request(method, params)20        return m.AddLink(**r)21    async def add_page(self, user_id: int = None, group_id: int = None):22        """23        :param user_id:24        :param group_id:25        """26        method = self.get_method_name(self.add_page)27        params = self.create_params(locals())28        r = await self.api_request(method, params)29        return m.AddPage(**r)30    async def add_post(31        self, owner_id: int = None, id: int = None, access_key: str = None32    ):33        """34        :param owner_id:35        :param id:36        :param access_key:37        """38        method = self.get_method_name(self.add_post)39        params = self.create_params(locals())40        r = await self.api_request(method, params)41        return m.AddPost(**r)42    async def add_product(43        self, owner_id: int = None, id: int = None, access_key: str = None44    ):45        """46        :param owner_id:47        :param id:48        :param access_key:49        """50        method = self.get_method_name(self.add_product)51        params = self.create_params(locals())52        r = await self.api_request(method, params)53        return m.AddProduct(**r)54    async def add_tag(self, name: str = None):55        """56        :param name:57        """58        method = self.get_method_name(self.add_tag)59        params = self.create_params(locals())60        r = await self.api_request(method, params)61        return m.AddTag(**r)62    async def add_video(63        self, owner_id: int = None, id: int = None, access_key: str = None64    ):65        """66        :param owner_id:67        :param id:68        :param access_key:69        """70        method = self.get_method_name(self.add_video)71        params = self.create_params(locals())72        r = await self.api_request(method, params)73        return m.AddVideo(**r)74    async def edit_tag(self, id: int = None, name: str = None):75        """76        :param id:77        :param name:78        """79        method = self.get_method_name(self.edit_tag)80        params = self.create_params(locals())81        r = await self.api_request(method, params)82        return m.EditTag(**r)83    async def get(84        self,85        extended: bool = None,86        item_type: str = None,87        tag_id: int = None,88        offset: int = None,89        count: int = None,90        fields: str = None,91        is_from_snackbar: bool = None,92    ):93        """94        :param extended: '1' â to return additional 'wall', 'profiles', and 'groups' fields. By default: '0'.95        :param item_type:96        :param tag_id: Tag ID.97        :param offset: Offset needed to return a specific subset of users.98        :param count: Number of users to return.99        :param fields:100        :param is_from_snackbar:101        """102        method = self.get_method_name(self.get)103        params = self.create_params(locals())104        r = await self.api_request(method, params)105        return m.Get(**r)106    async def get_pages(107        self,108        offset: int = None,109        count: int = None,110        type: str = None,111        fields: list = None,112        tag_id: int = None,113    ):114        """115        :param offset:116        :param count:117        :param type:118        :param fields:119        :param tag_id:120        """121        method = self.get_method_name(self.get_pages)122        params = self.create_params(locals())123        r = await self.api_request(method, params)124        return m.GetPages(**r)125    async def get_tags(self,):126        """127        """128        method = self.get_method_name(self.get_tags)129        params = self.create_params(locals())130        r = await self.api_request(method, params)131        return m.GetTags(**r)132    async def mark_seen(self,):133        """134        """135        method = self.get_method_name(self.mark_seen)136        params = self.create_params(locals())137        r = await self.api_request(method, params)138        return m.MarkSeen(**r)139    async def remove_article(140        self, owner_id: int = None, article_id: int = None141    ):142        """143        :param owner_id:144        :param article_id:145        """146        method = self.get_method_name(self.remove_article)147        params = self.create_params(locals())148        r = await self.api_request(method, params)149        return m.RemoveArticle(**r)150    async def remove_link(self, link_id: str = None, link: str = None):151        """152        Removes link from the user's faves.153        :param link_id: Link ID (can be obtained by [vk.com/dev/faves.getLinks|faves.getLinks] method).154        :param link: Link URL155        """156        method = self.get_method_name(self.remove_link)157        params = self.create_params(locals())158        r = await self.api_request(method, params)159        return m.RemoveLink(**r)160    async def remove_page(self, user_id: int = None, group_id: int = None):161        """162        :param user_id:163        :param group_id:164        """165        method = self.get_method_name(self.remove_page)166        params = self.create_params(locals())167        r = await self.api_request(method, params)168        return m.RemovePage(**r)169    async def remove_post(self, owner_id: int = None, id: int = None):170        """171        :param owner_id:172        :param id:173        """174        method = self.get_method_name(self.remove_post)175        params = self.create_params(locals())176        r = await self.api_request(method, params)177        return m.RemovePost(**r)178    async def remove_product(self, owner_id: int = None, id: int = None):179        """180        :param owner_id:181        :param id:182        """183        method = self.get_method_name(self.remove_product)184        params = self.create_params(locals())185        r = await self.api_request(method, params)186        return m.RemoveProduct(**r)187    async def remove_tag(self, id: int = None):188        """189        :param id:190        """191        method = self.get_method_name(self.remove_tag)192        params = self.create_params(locals())193        r = await self.api_request(method, params)194        return m.RemoveTag(**r)195    async def reorder_tags(self, ids: list = None):196        """197        :param ids:198        """199        method = self.get_method_name(self.reorder_tags)200        params = self.create_params(locals())201        r = await self.api_request(method, params)202        return m.ReorderTags(**r)203    async def set_page_tags(204        self, user_id: int = None, group_id: int = None, tag_ids: list = None205    ):206        """207        :param user_id:208        :param group_id:209        :param tag_ids:210        """211        method = self.get_method_name(self.set_page_tags)212        params = self.create_params(locals())213        r = await self.api_request(method, params)214        return m.SetPageTags(**r)215    async def set_tags(216        self,217        item_type: str = None,218        item_owner_id: int = None,219        item_id: int = None,220        tag_ids: list = None,221        link_id: str = None,222        link_url: str = None,223    ):224        """225        :param item_type:226        :param item_owner_id:227        :param item_id:228        :param tag_ids:229        :param link_id:230        :param link_url:231        """232        method = self.get_method_name(self.set_tags)233        params = self.create_params(locals())234        r = await self.api_request(method, params)235        return m.SetTags(**r)236    async def track_page_interaction(237        self, user_id: int = None, group_id: int = None238    ):239        """240        :param user_id:241        :param group_id:242        """243        method = self.get_method_name(self.track_page_interaction)244        params = self.create_params(locals())245        r = await self.api_request(method, params)...

get_or_create_mixin.py

Source:get_or_create_mixin.py

1from copy import deepcopy2import arrow3import sqlalchemy as sa4from sqlalchemy import orm5from sqlalchemy.dialects.postgresql import insert as pg_insert6class GetOrCreateMixin:7    @classmethod8    def get_or_create(9        cls,10        create_params: dict = None,11        index_elements=None,12        index_where=None,13        **search_by_params14    ):15        """16        Implements 'insert or create' behavior for SQLAlchemy using PostgreSQL UPSERT17        semantics.18        This is both safe and efficient, since it takes only one db roundtrip but it19        comes with caveats and can't work in all cases.20        Internally it is using SQLAlchemy Postgres dialect, details are here:21        https://docs.sqlalchemy.org/en/latest/dialects/postgresql.html#sqlalchemy.dialects.postgresql.dml.Insert.on_conflict_do_update22        https://docs.sqlalchemy.org/en/latest/orm/persistence_techniques.html#using-postgresql-on-conflict-with-returning-to-return-upserted-orm-objects23        Note:24            This **IS** multi-thread and/or multi-process safe but it will work only25            if there is **unique index** on destination table that can be used as26            arbiter when deciding should object be INSERTed or just SELECTed27        Warning:28            Implementation is PostgreSQL speciffic. Minimal PostgreSQL version is 9.6.29            For other RDBMs see `get_or_create_no_index`.30        Arguments:31            create_params: column names and values that will be used for creating new32                row.33            search_by_params: column names and values that will be used to detect what34                arbiter index to use. Thus, keys in this dict must all be column names35                of some unique index. Also, any key and value tuples found in this dict36                will overwrite ones in ``create_params``.37            index_elements: A sequence consisting of string column names, Column38                objects, or other column expression objects that will be used to39                infer a target index. If not provided, ``search_by_params.keys()`` are40                used.41            index_where: Additional WHERE criterion that can be used to infer a42                conditional target index.43        """44        create_params = deepcopy(create_params or {})45        create_params.update(search_by_params)46        # We need to update at least one attribute, otherwise session.execute might47        # return no objects. Since all our objects have "updated_at", this is probably48        # best one to update.49        create_params["updated_at"] = arrow.utcnow().datetime50        statement = (51            pg_insert(cls)52            .values([create_params])53            .on_conflict_do_update(54                index_elements=list(search_by_params.keys())55                if not index_elements56                else index_elements,57                index_where=index_where,58                set_=create_params,59            )60        ).returning(cls)61        orm_stmt = (62            sa.select(cls)63            .from_statement(statement)64            .execution_options(populate_existing=True)65        )66        return cls.query.session.execute(orm_stmt).scalar_one()67    @classmethod68    def get_or_create_only_id(69        cls,70        create_params: dict = None,71        index_elements=None,72        index_where=None,73        **search_by_params74    ):75        """76        Same as ``get_or_create`` but instead of returning whole ORM instance, it only77        returns ID of created / updated row.78        """79        create_params = deepcopy(create_params or {})80        create_params.update(search_by_params)81        # We need to update at least one attribute, otherwise session.execute might82        # return no objects. Since all our objects have "updated_at", this is probably83        # best one to update.84        create_params["updated_at"] = arrow.utcnow().datetime85        statement = (86            pg_insert(cls)87            .values([create_params])88            .on_conflict_do_update(89                index_elements=list(search_by_params.keys())90                if not index_elements91                else index_elements,92                index_where=index_where,93                set_=create_params,94            )95        ).returning(cls.id)96        return cls.query.session.execute(statement).scalar_one()97    @classmethod98    def get_or_create_no_index(cls, create_params: dict = None, **search_by_params):99        """100        Same as `get_or_create` but much slower since it in worst case requires101        multiple roundtrips to db.102        This is to be used when destination table doesn't have any unique index that103        is required by `get_or_create` to operate.104        Warning:105            This is **NOT** multi-thread or multi-proccess safe implementation of106            db UPSERT.107        """108        # TODO: Decouple this whole thing from Flask-SQLAlchemy109        from Hexa.db import db110        retv = None111        try:112            retv = db.session.query(cls).filter_by(**search_by_params).one()113        except orm.exc.NoResultFound:114            create_params = deepcopy(create_params or {})115            create_params.update(search_by_params)116            created = cls(**create_params)117            try:118                db.session.add(created)119                db.session.flush()120                retv = created121            except sa.exc.IntegrityError as e:122                db.session.rollback()123                if "duplicate key value violates unique constraint" in str(e):124                    retv = db.session.query(cls).filter_by(**search_by_params).one()125                    for k, v in create_params.items():126                        setattr(retv, k, v)127                else:128                    raise...

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.