KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use set_body method in pyresttest
Best Python code snippet using pyresttest_python
py_api_product_facades.py
Source:py_api_product_facades.py 
...30 31 def get_status_code(self):32 return self.__status_code33 34 def set_body(self, value, add_line=False):35 if add_line and type(value) is str:36 self.__body = '(L'+str(sys._getframe().f_back.f_lineno)+') ' + value37 else: 38 self.__body = value39 40 def get_body(self): 41 return self.__body42 43 def get_body_as_dict(self):44 if type(self.__body) is str:45 return {"message": self.__body} 46 else: 47 return self.__body 48 49 @abstractmethod 50 def execute(self): 51 """ Executa o CRUD. """52 pass53 54class PUTProductFacade(CRUDFacade):55 """ 56 Executa os requests de PUT de produto.57 OBS: Sabemos que o FACADE é um grande inimigo do SOLID, mas abri uma exceção para ele, para58 evitar sobrecarregar o consumer com toda a lógica do PUT de produtos. Penso que os riscos59 valem a pena, nesse caso.60 """ 61 62 def execute(self): 63 """ 64 Executa o PUT. 65 Inicializa as propriedades __status_code e __body.66 Retorna:67 bool True/False quanto ao sucesso na execução.68 """69 # Inicia o controle de transações...70 if self.get_db().start_transaction(): 71 # Checagem da estrutura do request...72 checkPUTRequest = prod.CheckProductPUTRequest(body=self.get_body(), db=self.get_db())73 if checkPUTRequest.execute():74 # Inclusão... 75 insertProduct = prod.InsertProduct( schema=checkPUTRequest.get_schema(), 76 request=checkPUTRequest.get_request(), 77 db=self.get_db()78 ) 79 if insertProduct.execute():80 # Tratamento do fabricante do produto...81 checkManufacturerPUTRequest = manu.CheckManufacturerPUTRequest(body=self.get_body(), db=self.get_db())82 if checkManufacturerPUTRequest.execute():83 if checkManufacturerPUTRequest.get_insert_manufacturer():84 # Inclusão do fabricante... 85 insertManufacturer = manu.InsertManufacturer( manufacturer_name=checkManufacturerPUTRequest.get_manufacturer_name(), 86 db=self.get_db()87 ) 88 if insertManufacturer.execute():89 # Reserva a PK do fabricante incluÃdo...90 checkManufacturerPUTRequest.set_manufacturer_id(insertManufacturer.get_primary_key()['id']) 91 # Inclusão do fabricante para o produto... 92 insertProductManufacturer = manu.InsertProductManufacturer( product_id=insertProduct.get_primary_key()['id'],93 manufacturer_id=checkManufacturerPUTRequest.get_manufacturer_id(), 94 db=self.get_db()95 ) 96 if not insertProductManufacturer.execute(): 97 self.set_status_code(insertProductManufacturer.get_error_code())98 self.set_body(insertProductManufacturer.get_error_message(), True)99 else: 100 self.set_status_code(400)101 self.set_body(insertManufacturer.get_error_message(), True) 102 elif checkManufacturerPUTRequest.get_update_manufacturer(): 103 # Atualiza o fabricante do produto... 104 updateManufacturer = manu.UpdateManufacturer( manufacturer_id=checkManufacturerPUTRequest.get_manufacturer_id(),105 manufacturer_name=checkManufacturerPUTRequest.get_manufacturer_name(), 106 db=self.get_db()107 )108 if not updateManufacturer.execute():109 self.set_status_code(updateManufacturer.get_error_code())110 self.set_body(updateManufacturer.get_error_message(), True) 111 else: 112 self.set_status_code(400)113 self.set_body(checkManufacturerPUTRequest.get_error_message(), True) 114 115 # Response...116 if self.get_status_code() == 200: 117 body = checkPUTRequest.get_request()118 body['id'] = insertProduct.get_primary_key()['id']119 body['manufacturer']['id'] = checkManufacturerPUTRequest.get_manufacturer_id()120 body['manufacturer']['name'] = checkManufacturerPUTRequest.get_manufacturer_name() 121 self.set_body(body) # Atualiza122 123 else: 124 self.set_status_code(insertProduct.get_error_code())125 self.set_body(insertProduct.get_error_message(), True) 126 127 else: 128 self.set_status_code(checkPUTRequest.get_error_code())129 self.set_body(checkPUTRequest.get_error_message(), True) 130 131 # Commit & Rollback...132 if self.get_db().in_transaction():133 # Tenta commit...134 if self.get_status_code() == 200:135 if not self.get_db().commit():136 self.set_status_code(self.get_db().get_error_code())137 self.set_body(self.get_db().get_error_message(), True) 138 # 139 # Rollback...140 if self.get_status_code() != 200:141 self.get_db().rollback() 142 # 143 144 else: 145 self.set_status_code(self.get_db().get_error_code())146 self.set_body(self.get_db().get_error_message(), True) 147 #148 return self.get_status_code() == 200149 150class POSTProductFacade(CRUDFacade):151 """ 152 Executa os requests de POST de produto.153 OBS: Sabemos que o FACADE é um grande inimigo do SOLID, mas abri uma exceção para ele, para154 evitar sobrecarregar o consumer com toda a lógica do POST de produtos. Penso que os riscos155 valem a pena, nesse caso.156 """ 157 158 def execute(self): 159 """ 160 Executa o POST. 161 Inicializa as propriedade __status_code e __body.162 Retorna:163 bool True/False quanto ao sucesso na execução.164 """165 # Inicia o controle de transações...166 if self.get_db().start_transaction(): 167 # Checagem da estrutura do request...168 checkPOSTRequest = prod.CheckProductPOSTRequest(body=self.get_body(), db=self.get_db())169 if checkPOSTRequest.execute():170 # Alteração... 171 updateProduct = prod.UpdateProduct( schema=checkPOSTRequest.get_schema(), 172 request=checkPOSTRequest.get_request(), 173 db=self.get_db()174 ) 175 if updateProduct.execute():176 # Tratamento do fabricante do produto...177 checkManufacturerPOSTRequest = manu.CheckManufacturerPOSTRequest(body=self.get_body(), db=self.get_db())178 if checkManufacturerPOSTRequest.execute():179 if checkManufacturerPOSTRequest.get_insert_manufacturer():180 # Inclusão do fabricante... 181 insertManufacturer = manu.InsertManufacturer( manufacturer_name=checkManufacturerPOSTRequest.get_manufacturer_name(), 182 db=self.get_db()183 ) 184 if insertManufacturer.execute():185 # Reserva a PK do fabricante incluÃdo...186 checkManufacturerPOSTRequest.set_manufacturer_id(insertManufacturer.get_primary_key()['id']) 187 # Inclusão do fabricante para o produto... 188 insertProductManufacturer = manu.InsertProductManufacturer( product_id=updateProduct.get_primary_key()['id'],189 manufacturer_id=checkManufacturerPOSTRequest.get_manufacturer_id(), 190 db=self.get_db()191 ) 192 if not insertProductManufacturer.execute(): 193 self.set_status_code(insertProductManufacturer.get_error_code())194 self.set_body(insertProductManufacturer.get_error_message(), True) 195 else: 196 self.set_status_code(insertManufacturer.get_error_code())197 self.set_body(insertManufacturer.get_error_message(), True) 198 elif checkManufacturerPOSTRequest.get_update_manufacturer(): 199 # Atualiza o fabricante do produto... 200 updateManufacturer = manu.UpdateManufacturer( manufacturer_id=checkManufacturerPOSTRequest.get_manufacturer_id(),201 manufacturer_name=checkManufacturerPOSTRequest.get_manufacturer_name(), 202 db=self.get_db()203 )204 if not updateManufacturer.execute(): 205 self.set_status_code(updateManufacturer.get_error_code())206 self.set_body(updateManufacturer.get_error_message(), True) 207 # 208 if self.get_status_code() == 200:209 if checkManufacturerPOSTRequest.get_update_product():210 # Troca de fabricante do produto... 211 updateProductManufacturer = manu.UpdateProductManufacturer( product_id=updateProduct.get_primary_key()['id'],212 manufacturer_id=checkManufacturerPOSTRequest.get_manufacturer_id(), 213 db=self.get_db()214 ) 215 if not updateProductManufacturer.execute(): 216 self.set_status_code(updateProductManufacturer.get_error_code())217 self.set_body(updateProductManufacturer.get_error_message(), True) 218 219 else: 220 self.set_status_code(400)221 self.set_body(checkManufacturerPOSTRequest.get_error_message(), True) 222 223 # Response(json encode)...224 if self.get_status_code() == 200:225 body = checkPOSTRequest.get_request()226 body['id'] = updateProduct.get_primary_key()['id']227 body['manufacturer']['id'] = checkManufacturerPOSTRequest.get_manufacturer_id()228 body['manufacturer']['name'] = checkManufacturerPOSTRequest.get_manufacturer_name() 229 self.set_body(body) # Atualiza230 231 else: 232 self.set_status_code(updateProduct.get_error_code())233 self.set_body(updateProduct.get_error_message(), True) 234 235 else: 236 self.set_status_code(checkPOSTRequest.get_error_code())237 self.set_body(checkPOSTRequest.get_error_message(), True) 238 239 # Commit & Rollback...240 if self.get_db().in_transaction():241 # Tenta commit...242 if self.get_status_code() == 200:243 if not self.get_db().commit():244 self.set_status_code(self.get_db().get_error_code())245 self.set_body(self.get_db().get_error_message(), True) 246 # 247 # Rollback...248 if self.get_status_code() != 200:249 self.get_db().rollback() 250 # 251 252 else: 253 self.set_status_code(self.get_db().get_error_code())254 self.set_body(self.get_db().get_error_message(), True) 255 #256 return self.get_status_code() == 200257 258 259class GETProductFacade(CRUDFacade):260 """ 261 Executa os requestes de GET de produtos.262 OBS: Sabemos que o FACADE é um grande inimigo do SOLID, mas abri uma exceção para ele, para263 evitar sobrecarregar o consumer com toda a lógica do GET de produtos. Penso que os riscos264 valem a pena, nesse caso.265 """ 266 267 def execute(self): 268 """ 269 Executa o GET. 270 Inicializa as propriedade __status_code e __body.271 Retorna:272 bool True/False quanto ao sucesso na execução.273 """ 274 # Checagem da estrutura do request...275 checkGETRequest = prod.CheckProductGETRequest(body=self.get_body(), db=self.get_db())276 if checkGETRequest.execute():277 # Consulta... 278 getProduct = prod.GetProduct( schema=checkGETRequest.get_schema(), 279 request=checkGETRequest.get_request(), 280 db=self.get_db()281 ) 282 if getProduct.execute():283 # Response(json encode)...284 self.set_body(getProduct.get_request())285 else: 286 self.set_status_code(getProduct.get_error_code())287 self.set_body(getProduct.get_error_message(), True) 288 289 else: 290 self.set_status_code(checkGETRequest.get_error_code())291 self.set_body(checkGETRequest.get_error_message(), True) 292 #293 return self.get_status_code() == 200 294class DELETEProductFacade(CRUDFacade):295 """ 296 Executa os requests de DELETE de produto.297 OBS: Sabemos que o FACADE é um grande inimigo do SOLID, mas abri uma exceção para ele, para298 evitar sobrecarregar o consumer com toda a lógica do DELETE de produtos. Penso que os riscos299 valem a pena, nesse caso.300 """ 301 302 def execute(self): 303 """ 304 Executa o DELETE. 305 Inicializa as propriedade __status_code e __body.306 Retorna:307 bool True/False quanto ao sucesso na execução.308 """309 # Inicia o controle de transações...310 if self.get_db().start_transaction(): 311 # Checagem da estrutura do request...312 checkDELRequest = prod.CheckProductDELETERequest(body=self.get_body(), db=self.get_db())313 if checkDELRequest.execute():314 # Exclusão... 315 deleteProduct = prod.DeleteProduct( schema=checkDELRequest.get_schema(), 316 request=checkDELRequest.get_request(), 317 db=self.get_db()318 ) 319 if deleteProduct.execute():320 # Tratamento da exclusão(desativação) da associação produto e fabricante...321 deleteProductManufacturer = manu.DeleteProductManufacturer( product_id=deleteProduct.get_primary_key()['id'],322 db=self.get_db()323 ) 324 if not deleteProductManufacturer.execute(): 325 self.set_status_code(deleteProductManufacturer.get_error_code())326 self.set_body(deleteProductManufacturer.get_error_message(), True) 327 else: 328 # Response(json encode)...329 body = checkDELRequest.get_request()330 body['id'] = deleteProduct.get_primary_key()['id'] 331 self.set_body(body) # Atualiza332 else: 333 self.set_status_code(deleteProduct.get_error_code())334 self.set_body(deleteProduct.get_error_message(), True) 335 336 else: 337 self.set_status_code(checkDELRequest.get_error_code())338 self.set_body(checkDELRequest.get_error_message(), True) 339 340 # Commit & Rollback...341 if self.get_db().in_transaction():342 # Tenta commit...343 if self.get_status_code() == 200:344 if not self.get_db().commit():345 self.set_status_code(self.get_db().get_error_code())346 self.set_body(self.get_db().get_error_message(), True) 347 # 348 # Rollback...349 if self.get_status_code() != 200:350 self.get_db().rollback() 351 # 352 353 else: 354 self.set_status_code(self.get_db().get_error_code())355 self.set_body(self.get_db().get_error_message(), True) 356 #357 return self.get_status_code() == 200 ...
utils.py
Source:utils.py
...22Lagrange: v^2 = G/r * 1/sqrt(3)23"""24def load_solution(sim, name):25 if name == "Euler 1":26 sim.set_body(0, 1, [0, 0, 0], [0, 0, 0])27 sim.set_body(1, 1, [1, 0, 0], [0, np.sqrt(1.25*sim.G), 0])28 sim.set_body(2, 1, [-1, 0, 0], [0, -np.sqrt(1.25*sim.G), 0])29 30 elif name == "Euler 2":31 m1, m2, m3 = sim.masses32 p = Polynomial([-m2-m3, -2*m2-3*m3, -m2-3*m3, 3*m1+m2, 3*m1+2*m2, m1+m2])33 z = sorted(p.roots(), key=lambda x: abs(x.imag))[0].real34 w = np.sqrt((sim.G*m2/z**2 + sim.G*m3/(1+z)**2)/m1)35 r1 = -(m2 + (1+z)*m3)/(m1 + m2 + m3)36 r2, r3 = r1+1, r1+1+z37 sim.set_body(0, m1, [r1, 0, 0], [0, r1*w, 0])38 sim.set_body(1, m2, [r2, 0, 0], [0, r2*w, 0])39 sim.set_body(2, m3, [r3, 0, 0], [0, r3*w, 0])40 41 elif name == "Lagrange":42 m1, m2, m3 = sim.masses43 r_cm = np.array([(m2 + 0.5*m3)/(m1 + m2 + m3), np.sqrt(0.75)*m3/(m1 + m2 + m3), 0])44 r1 = -r_cm45 r2 = [1, 0, 0] - r_cm46 r3 = [0.5, np.sqrt(0.75), 0] - r_cm47 w = np.sqrt(sim.G * (m1 + m2 + m3))48 sim.set_body(0, m1, r1, np.cross(ZHAT, r1*w))49 sim.set_body(1, m2, r2, np.cross(ZHAT, r2*w))50 sim.set_body(2, m3, r3, np.cross(ZHAT, r3*w))51 52 elif name == "Figure-8":53 sim.set_body(0, 1, [0.97000436, -0.24308753, 0], [0.466203685, 0.43236573, 0])54 sim.set_body(1, 1, [-0.97000436, 0.24308753, 0], [0.466203685, 0.43236573, 0])55 sim.set_body(2, 1, [0, 0, 0], [-0.93240737, -0.86473146, 0])56 sim.G = 157 58 elif name == "Custom":...
test_producer.py
Source:test_producer.py
...5def test_producer_send_sync(producer):6 msg = Message('test')7 msg.set_keys('send_sync')8 msg.set_tags('XXX')9 msg.set_body('XXXX')10 ret = producer.send_sync(msg)11 assert ret.status == SendStatus.OK12def test_producer_send_async(producer):13 stop_event = threading.Event()14 errors = []15 def on_success(result):16 stop_event.set()17 if not result.msg_id:18 errors.append(AssertionError('Producer send_async failed'))19 def on_exception(exc):20 stop_event.set()21 errors.append(exc)22 msg = Message('test')23 msg.set_keys('send_async')24 msg.set_tags('XXX')25 msg.set_body('XXXX')26 producer.send_async(msg, on_success, on_exception)27 max_wait = 1028 wait_count = 029 while not stop_event.is_set():30 if wait_count >= max_wait:31 stop_event.set()32 raise Exception('test timed-out')33 time.sleep(1)34 wait_count += 135 if errors:36 raise errors[0]37def test_producer_send_oneway(producer):38 msg = Message('test')39 msg.set_keys('send_oneway')40 msg.set_tags('XXX')41 msg.set_body('XXXX')42 producer.send_oneway(msg)43def test_producer_send_oneway_orderly(producer):44 msg = Message('test')45 msg.set_keys('send_oneway_orderly')46 msg.set_tags('XXX')47 msg.set_body('XXXX')48 producer.send_oneway_orderly(msg, 1)49def test_producer_send_orderly(producer):50 msg = Message('test')51 msg.set_keys('send_orderly')52 msg.set_tags('XXX')53 msg.set_body('XXXX')54 ret = producer.send_orderly(msg, 1)55 assert ret.status == SendStatus.OK56def test_producer_send_batch(producer):57 batch_msg = []58 msg = Message('test')59 msg.set_keys('send_batch_1')60 msg.set_tags('XXX1')61 msg.set_body('XXXX1')62 batch_msg.append(msg)63 msg = Message('test')64 msg.set_keys('send_batch_2')65 msg.set_tags('XXX2')66 msg.set_body('XXXX2')67 batch_msg.append(msg)68 ret = producer.send_batch(batch_msg)...
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 pyresttest 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?
