Best Python code snippet using localstack_python
test_response.py
Source:test_response.py
...26 # Given27 stub_response = MOCK_RESPONSE28 # When29 # Then30 assert_that(stub_response, is_response().with_status_code(200))31 assert_that(stub_response, not_(is_response().with_status_code(201)))32 assert_that(is_response().with_status_code(200), has_string("response with status_code: <200>"))33 assert_that(34 is_response().with_status_code(201),35 mismatches_with(stub_response, contains_string("was response with status code: was <200>")),36 )37 assert_that(38 is_response().with_status_code(200),39 matches_with(stub_response, "was response with status code: was <200>"),40 )41def test_response_matcher_body():42 # Given43 stub_response = MOCK_RESPONSE44 # When45 # Then46 assert_that(stub_response, is_response().with_body("sausages"))47 assert_that(stub_response, not_(is_response().with_body("chips")))48 assert_that(is_response().with_body("chips"), has_string("response with body: 'chips'"))49 assert_that(50 is_response().with_body("chips"),51 mismatches_with(stub_response, contains_string("was response with body: was 'sausages'")),52 )53 assert_that(54 is_response().with_body("sausages"),55 matches_with(stub_response, contains_string("was response with body: was 'sausages'")),56 )57def test_response_matcher_content():58 # Given59 stub_response = MOCK_RESPONSE60 # When61 # Then62 assert_that(stub_response, is_response().with_content(b"content"))63 assert_that(stub_response, not_(is_response().with_content(b"chips")))64 assert_that(65 str(is_response().with_content(b"content")),66 contains_string("response with content: <b'content'>"),67 )68 assert_that(69 is_response().with_content(b"chips"),70 mismatches_with(71 stub_response, contains_string("was response with content: was <b'content'>")72 ),73 )74 assert_that(75 is_response().with_content(b"content"),76 matches_with(stub_response, contains_string("was response with content: was <b'content'>")),77 )78def test_response_matcher_json():79 # Given80 stub_response = MOCK_RESPONSE81 # When82 # Then83 assert_that(stub_response, is_response().with_json({"a": "b"}))84 assert_that(stub_response, not_(is_response().with_json({"a": "c"})))85 assert_that(86 str(is_response().with_json({"a": "b"})),87 contains_string("response with json: <{'a': 'b'}>"),88 )89 assert_that(90 is_response().with_json([1, 2, 4]),91 mismatches_with(stub_response, contains_string("was response with json: was <{'a': 'b'}>")),92 )93 assert_that(94 is_response().with_json({"a": "b"}),95 matches_with(stub_response, contains_string("was response with json: was <{'a': 'b'}>")),96 )97def test_response_matcher_headers():98 # Given99 response = MOCK_RESPONSE100 # When101 # Then102 assert_that(response, is_response().with_headers({"key": "value"}))103 assert_that(response, not_(is_response().with_headers({"key": "nope"})))104 assert_that(105 str(is_response().with_headers({"key": "value"})),106 contains_string("response with headers: <{'key': 'value'}"),107 )108 assert_that(109 is_response().with_headers({"key": "nope"}),110 mismatches_with(111 response, contains_string("was response with headers: was <{'key': 'value'}")112 ),113 )114 assert_that(115 is_response().with_headers({"key": "value"}),116 matches_with(response, contains_string("was response with headers: was <{'key': 'value'}")),117 )118def test_response_matcher_cookies():119 # Given120 response = MOCK_RESPONSE121 # When122 # Then123 assert_that(response, is_response().with_cookies({"name": "value"}))124 assert_that(response, not_(is_response().with_cookies({"name": "nope"})))125 assert_that(126 str(is_response().with_cookies({"name": "value"})),127 contains_string("response with cookies: <{'name': 'value'}"),128 )129 assert_that(130 is_response().with_cookies({"name": "nope"}),131 mismatches_with(132 response, contains_string("was response with cookies: was <{'name': 'value'}")133 ),134 )135 assert_that(136 is_response().with_cookies({"name": "value"}),137 matches_with(138 response, contains_string("was response with cookies: was <{'name': 'value'}")139 ),140 )141def test_response_matcher_elapsed():142 # Given143 response = MOCK_RESPONSE144 # When145 # Then146 assert_that(response, is_response().with_elapsed(timedelta(seconds=1)))147 assert_that(response, not_(is_response().with_elapsed(timedelta(seconds=60))))148 assert_that(149 str(is_response().with_elapsed(timedelta(seconds=1))),150 contains_string("response with elapsed: <0:00:01>"),151 )152 assert_that(153 is_response().with_elapsed(timedelta(seconds=60)),154 mismatches_with(response, contains_string("was response with elapsed: was <0:00:01>")),155 )156 assert_that(157 is_response().with_elapsed(timedelta(seconds=1)),158 matches_with(response, contains_string("was response with elapsed: was <0:00:01>")),159 )160def test_response_matcher_history_and_url():161 # Given162 response = MOCK_RESPONSE163 # When164 # Then165 assert_that(166 response,167 is_response().with_history(168 contains_exactly(169 is_response().with_url(is_url().with_path("/path1")),170 is_response().with_url(is_url().with_path("/path2")),171 )172 ),173 )174 assert_that(175 response,176 not_(177 is_response().with_history(178 contains_exactly(179 is_response().with_url(is_url().with_path("/path1")),180 is_response().with_url(is_url().with_path("/path3")),181 )182 )183 ),184 )185 assert_that(186 str(187 is_response().with_history(188 contains_exactly(189 is_response().with_url(is_url().with_path("/path1")),190 is_response().with_url(is_url().with_path("/path2")),191 )192 )193 ),194 contains_string(195 "response with history: a sequence containing "196 "[response with url: URL with path: '/path1', response with url: URL with path: '/path2']"197 ),198 )199 assert_that(200 is_response().with_history(201 contains_exactly(202 is_response().with_url(is_url().with_path("/path1")),203 is_response().with_url(is_url().with_path("/path3")),204 )205 ),206 mismatches_with(207 response,208 contains_string(209 "was response with history: item 1: was response with url: was URL with path: was </path2>"210 ),211 ),212 )213def test_response_matcher_url():214 # Given215 response = MOCK_RESPONSE216 # When217 # Then218 assert_that(response, is_response().with_url(is_url().with_path("/path0")))219 assert_that(response, not_(is_response().with_url(is_url().with_path("/nope"))))220 assert_that(221 str(is_response().with_url(is_url().with_path("/path0"))),222 contains_string("response with url: URL with path: '/path0'"),223 )224 assert_that(225 is_response().with_url(is_url().with_path("/nope")),226 mismatches_with(227 response, contains_string("was response with url: was URL with path: was </path0>")228 ),229 )230 assert_that(231 is_response().with_url(is_url().with_path("/path0")),232 matches_with(233 response, contains_string("was response with url: was URL with path: was </path0>")234 ),235 )236def test_response_matcher_encoding():237 # Given238 stub_response = MOCK_RESPONSE239 # When240 # Then241 assert_that(stub_response, is_response().with_encoding("utf-8"))242 assert_that(stub_response, not_(is_response().with_encoding("ISO-8859-1")))243 assert_that(is_response().with_encoding("utf-8"), has_string("response with encoding: 'utf-8'"))244 assert_that(245 is_response().with_encoding("ISO-8859-1"),246 mismatches_with(stub_response, contains_string("was response with encoding: was 'utf-8'")),247 )248 assert_that(249 is_response().with_encoding("utf-8"),250 matches_with(stub_response, contains_string("was response with encoding: was 'utf-8'")),251 )252def test_response_matcher_invalid_json():253 # Given254 stub_response = mock.MagicMock(255 status_code=200, text="body", content=b"content", headers={"key": "value"}256 )257 type(stub_response).json = mock.PropertyMock(side_effect=ValueError)258 # When259 # Then260 assert_that(stub_response, not_(is_response().with_json([1, 2, 4])))261 assert_that(262 is_response().with_json([1, 2, 4]),263 mismatches_with(stub_response, contains_string("was response with json: was <None>")),264 )265def test_redirect_to():266 # Given267 stub_response = mock.MagicMock(268 status_code=301, headers={"Location": a_url().with_path("/sausages").build()}269 )270 # When271 # Then272 assert_that(stub_response, redirects_to(is_url().with_path("/sausages")))273 assert_that(stub_response, not_(redirects_to(is_url().with_path("/bacon"))))274 assert_that(275 redirects_to(is_url().with_path("/sausages")),276 has_string("redirects to URL with path: '/sausages'"),277 )278def test_response_matcher_builder():279 # Given280 stub_response = MOCK_RESPONSE281 matcher = (282 is_response()283 .with_status_code(200)284 .and_body("sausages")285 .and_content(b"content")286 .and_json(has_entries(a="b"))287 .and_headers(has_entries(key="value"))288 .and_cookies(has_entries(name="value"))289 .and_elapsed(between(timedelta(seconds=1), timedelta(minutes=1)))290 .and_history(291 contains_exactly(292 is_response().with_url(is_url().with_path("/path1")),293 is_response().with_url(is_url().with_path("/path2")),294 )295 )296 .and_url(is_url().with_path("/path0"))297 .and_encoding("utf-8")298 )299 mismatcher = is_response().with_body("kale").and_status_code(404)300 # When301 # Then302 assert_that(stub_response, matcher)303 assert_that(stub_response, not_(mismatcher))304 assert_that(305 matcher,306 has_string(307 "response with "308 "status_code: <200> "309 "body: 'sausages' "310 "content: <b'content'> "311 "json: a dictionary containing {'a': 'b'} "312 "headers: a dictionary containing {'key': 'value'} "313 "cookies: a dictionary containing {'name': 'value'} "...
Microservices.py
Source:Microservices.py
1# Logger2 3import random 4import json 5from MyMQTT import * 6import time 7from dataclasses import dataclass8from dataclasses_json import dataclass_json, config9from os.path import exists10from typing import List11class Raspberry_Pie_Logger: 12 def __init__(self,topic,clientID,broker,port): 13 self.clientID=clientID 14 self.topic=topic 15 self.client=MyMQTT(clientID,broker,port,self) 16 def run(self): 17 self.client.start() 18 print('{} has started'.format(self.clientID)) 19 self.client.mySubscribe(self.topic) 20 def end(self): 21 self.client.stop() 22 print('{} has stopped'.format(self.clientID)) 23 def emergencyHealthCheck(self, topic, msg):24 # print("Emergency Health Check")25 # decompose massage26 massage=json.loads(msg) 27 name = massage['e']['n'] 28 value = massage['e']['v'] 29 # unit = massage['e']['unit'] 30 # time = massage['e']['timestamp'] 31 # source = massage['bn'].split('/')32 # patientID = source[-2]33 # sensorID =source[-1]34 35 is_response = False36 if (name == "oxygen"):37 if (value >= 95 and value <= 100):38 # do nothing all right 39 pass 40 elif (value > 85 and value <= 94):41 # #hypoxia42 # print("hypoxia: " +str(name)+" "+str( value))43 massage['e']['n'] = "Hypoxia"44 is_response = True45 elif (value >= 0 and value <= 84):46 #sensor problem47 # print("sensor problem : " +name)48 massage['e']['n'] ="Sensor Error"49 is_response = True50 else:51 #sensor problem52 # print("sensor problem : " +name )53 massage['e']['n'] = "Value Error"54 is_response = True55 elif (name == "room temperature"):56 if (value>= 20.0 and value<= 22.0):57 #do nothing all right Normal Room58 pass59 elif (value> 22.0 and value< 50.0):60 #Hot Room61 # print("hypoxia: " +name +" "+ value)62 massage['e']['n'] = "Hot Room"63 # is_response = True64 elif (value < 20.0 and value > -10.0):65 #cold Room66 # print("sensor problem : " +name )67 massage['e']['n'] = "Cold Room"68 # is_response = True69 else:70 #sensor problem71 # print("sensor problem : " +name )72 massage['e']['n'] = "Value Error"73 is_response = True74 elif (name == "heartrate"):75 if (value>= 60 and value<= 100):76 #do nothing all right Normal heart rate77 pass78 elif (value> 100):79 #Hot Room80 # print("hypoxia: " +name +" "+ value)81 massage['e']['n'] = "Tachycardia" 82 is_response = True83 elif (value< 60):84 #cold Room85 # print("sensor problem : " +name )86 massage['e']['n'] = "Bradycardia"87 is_response = True88 else:89 #sensor problem90 # print("sensor problem : " +name )91 massage['e']['n'] = "Value Error"92 is_response = True93 elif (name == "body temperature"):94 if (value>= 35.6 and value<= 37.4):95 #do nothing all right Normal Room96 pass97 elif (value>= 37.5 and value<= 39.4):98 #Hot Room99 # print("hypoxia: " +name +" "+ value)100 massage['e']['n'] = "Fever"101 is_response = True102 elif (value>= 39.5 and value<= 42):103 #cold Room104 # print("sensor problem : " +name )105 massage['e']['n'] = "High Fever" 106 is_response = True 107 elif (value>= 33.0 and value<= 35.5):108 #cold Room109 # print("sensor problem : " +name )110 massage['e']['n'] = "Hypothermia"111 is_response = True112 else:113 #sensor problem114 # print("sensor problem : " +name )115 massage['e']['n'] = "Value Error"116 is_response = True117 #118 if is_response == True:119 massage['e']['n']= "Warning/" +massage['e']['n']120 massage['e']['t'] = time.time()121 # json.dumps(massage)122 return topic+"/Warning", massage123 else:124 return None, None125 def roomAdjustments(self, topic, msg):126 # print("Emergency Health Check")127 # decompose massage128 massage=json.loads(msg) 129 name = massage['e']['n'] 130 value = massage['e']['v'] 131 # unit = massage['e']['unit'] 132 # time = massage['e']['timestamp'] 133 source = massage['bn'].split('/')134 patientID = source[-2]135 # sensorID =source[-1]136 is_response = False137 response_massage = { 138 'bn': patientID +"/" +"RoomCommand", 139 'e': 140 {'n': '', 'v': '', 't': '', 'u': ''}141 } 142 if (name == "room temperature"):143 if (value>= 20.0 and value<= 22.0):144 #do nothing all right Normal Room145 pass146 elif (value> 22.1 and value<= 25.0):147 #warm Room148 massage['e']['n']= "HeatingSystem"149 massage['e']['v'] = 0150 massage['e']['u'] = "bool Off(0)/On(0)"151 is_response = True152 elif (value> 25.0):153 #Hot Room154 massage['e']['n']= "Windows"155 massage['e']['v'] = 0156 massage['e']['u'] = "bool (Open(0)/Close(1)"157 is_response = True158 elif (value< 19.9 and value>= 17):159 #chill Room160 massage['e']['n']= "Windows"161 massage['e']['v'] = 1162 massage['e']['u'] = "bool (Open(0)/Close(1)"163 is_response = True164 elif (value< 17):165 #cold Room166 massage['e']['n']= "HeatingSystem"167 massage['e']['v'] = 1168 massage['e']['u'] = "bool Off(0)/On(0)"169 is_response = True170 else:171 #sensor problem172 pass173 # print("sensor problem : " +name )174 # massage['e']['n'] = " (Value Error)"175 # is_response = True176 #177 if is_response == True:178 response_massage['e']['t'] = time.strftime("%d-%m-%Y %H:%M:%S")179 # json.dumps(response_massage)180 t_source = topic.split('/')181 t_source.pop() #remove sensor id182 topic = "/".join(t_source)183 return topic+"/RoomCommand", massage184 else:185 return None, None186 def notify(self,topic,msg): 187 massage=json.loads(msg) 188 name = massage['e']['n'] 189 value = massage['e']['v'] 190 unit = massage['e']['u'] 191 time = massage['e']['t'] 192 # patientID = source[-2]193 # sensorID =source[-1]194 print(f'{topic} {name}: {round(value,2)}{unit} at {time}.') 195 #process massege196 t_source = topic.split('/')197 # CALL microservices here198 if t_source[-1] != "Warning" and t_source[-1] != "RoomCommand":199 response_topic, response = self.emergencyHealthCheck(topic, msg)200 if response is not None:201 self.client.myPublish(response_topic, response)202 response_topic, response = self.roomAdjustments(topic, msg)203 if response is not None:204 self.client.myPublish(response_topic, response)205 206if __name__ == '__main__': 207 conf=json.load(open("settings.json")) 208 broker = conf["broker"] 209 port = conf["port"] 210 baseTopic = conf["baseTopic"]211 topic = baseTopic + "/#" 212 RP_Logger = Raspberry_Pie_Logger(topic,"Logger",broker,port) 213 RP_Logger.run() 214 while True: 215 time.sleep(1) ...
packet.py
Source:packet.py
...36 # Opposite of the getter, add the flag bits back onto the sequence number int37 # Woe be unto those that use ints bigger than 32 bits for their sequence numbers38 self._sequence = seq_num | ((self.is_response << 30) + (self.is_client << 31))39 @property40 def is_response(self):41 return self._get_bit_truthiness(30)42 @is_response.setter43 def is_response(self, is_response):44 self._set_bit(is_response, 30)45 @property46 def is_client(self):47 return self._get_bit_truthiness(31)48 @is_client.setter49 def is_client(self, is_client):50 self._set_bit(is_client, 31)51 @property52 def size(self):53 # Public getter for size, but no public setter54 return self._size55 @property56 def _size(self):57 return int(unpack_from('I', self._buffer, 4)[0])...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!