KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
