Best Python code snippet using fMBT_python
dsl_tests.py
Source:dsl_tests.py
...15 parsed_format = marionette_tg.dsl.parse(mar_format)16 self.assertEquals(parsed_format.get_transport(), "tcp")17 self.assertEquals(parsed_format.get_port(), 80)18 self.assertEquals(19 parsed_format.get_transitions()[0].get_src(), "start")20 self.assertEquals(21 parsed_format.get_transitions()[0].get_dst(), "downstream")22 self.assertEquals(23 parsed_format.get_transitions()[0].get_action_block(), None)24 self.assertEquals(25 parsed_format.get_transitions()[0].get_probability(), float(1.0))26 self.assertEquals(27 parsed_format.get_transitions()[1].get_src(), "downstream")28 self.assertEquals(29 parsed_format.get_transitions()[1].get_dst(), "upstream")30 self.assertEquals(31 parsed_format.get_transitions()[1].get_action_block(), "http_get")32 self.assertEquals(33 parsed_format.get_transitions()[1].get_probability(), float(1.0))34 self.assertEquals(35 parsed_format.get_transitions()[2].get_src(), "upstream")36 self.assertEquals(parsed_format.get_transitions()[2].get_dst(), "end")37 self.assertEquals(38 parsed_format.get_transitions()[2].get_action_block(), "http_ok")39 self.assertEquals(40 parsed_format.get_transitions()[2].get_probability(), float(1.0))41 self.assertEquals(42 parsed_format.get_action_blocks()[0].get_name(), "http_get")43 self.assertEquals(44 parsed_format.get_action_blocks()[0].get_party(), "client")45 self.assertEquals(46 parsed_format.get_action_blocks()[0].get_module(), "fte")47 self.assertEquals(48 parsed_format.get_action_blocks()[0].get_method(), "send")49 self.assertEquals(50 parsed_format.get_action_blocks()[0].get_args(), [51 "^regex\r\n\r\n$", 128])52 self.assertEquals(53 parsed_format.get_action_blocks()[1].get_name(), "http_ok")54 self.assertEquals(55 parsed_format.get_action_blocks()[1].get_party(), "server")56 self.assertEquals(57 parsed_format.get_action_blocks()[1].get_module(), "fte")58 self.assertEquals(59 parsed_format.get_action_blocks()[1].get_method(), "send")60 self.assertEquals(61 parsed_format.get_action_blocks()[1].get_args(), [62 "^regex\r\n\r\n\C*$", 128])63 def test2(self):64 mar_format = """connection(tcp, 80):65 start downstream NULL 1.066 downstream upstream http_get 1.067 upstream end http_ok 1.068 action http_get:69 client fte.send("^regex\r\n\r\n$", 128)70 action http_ok:71 server fte.send("^regex\r\n\r\n\C*$", 128)72 action http_put:73 client fte.send("^regex\r\n\r\n$", 128)"""74 parsed_format = marionette_tg.dsl.parse(mar_format)75 self.assertEquals(parsed_format.get_transport(), "tcp")76 self.assertEquals(parsed_format.get_port(), 80)77 self.assertEquals(78 parsed_format.get_transitions()[0].get_src(), "start")79 self.assertEquals(80 parsed_format.get_transitions()[0].get_dst(), "downstream")81 self.assertEquals(82 parsed_format.get_transitions()[0].get_action_block(), None)83 self.assertEquals(84 parsed_format.get_transitions()[0].get_probability(), float(1.0))85 self.assertEquals(86 parsed_format.get_transitions()[1].get_src(), "downstream")87 self.assertEquals(88 parsed_format.get_transitions()[1].get_dst(), "upstream")89 self.assertEquals(90 parsed_format.get_transitions()[1].get_action_block(), "http_get")91 self.assertEquals(92 parsed_format.get_transitions()[1].get_probability(), float(1.0))93 self.assertEquals(94 parsed_format.get_transitions()[2].get_src(), "upstream")95 self.assertEquals(parsed_format.get_transitions()[2].get_dst(), "end")96 self.assertEquals(97 parsed_format.get_transitions()[2].get_action_block(), "http_ok")98 self.assertEquals(99 parsed_format.get_transitions()[2].get_probability(), float(1.0))100 self.assertEquals(101 parsed_format.get_action_blocks()[0].get_name(), "http_get")102 self.assertEquals(103 parsed_format.get_action_blocks()[0].get_party(), "client")104 self.assertEquals(105 parsed_format.get_action_blocks()[0].get_module(), "fte")106 self.assertEquals(107 parsed_format.get_action_blocks()[0].get_method(), "send")108 self.assertEquals(109 parsed_format.get_action_blocks()[0].get_args(), [110 "^regex\r\n\r\n$", 128])111 self.assertEquals(112 parsed_format.get_action_blocks()[1].get_name(), "http_ok")113 self.assertEquals(114 parsed_format.get_action_blocks()[1].get_party(), "server")115 self.assertEquals(116 parsed_format.get_action_blocks()[1].get_module(), "fte")117 self.assertEquals(118 parsed_format.get_action_blocks()[1].get_method(), "send")119 self.assertEquals(120 parsed_format.get_action_blocks()[1].get_args(), [121 "^regex\r\n\r\n\C*$", 128])122 self.assertEquals(123 parsed_format.get_action_blocks()[2].get_name(), "http_put")124 self.assertEquals(125 parsed_format.get_action_blocks()[2].get_party(), "client")126 self.assertEquals(127 parsed_format.get_action_blocks()[2].get_module(), "fte")128 self.assertEquals(129 parsed_format.get_action_blocks()[2].get_method(), "send")130 self.assertEquals(131 parsed_format.get_action_blocks()[2].get_args(), [132 "^regex\r\n\r\n$", 128])133 def test3(self):134 mar_format = """connection(tcp, 80):135 start downstream NULL 1.0136 downstream upstream http_get 1.0137 upstream end http_ok 1.0138 action http_get:139 client fte.send("^regex\r\n\r\n$", 128)140 action http_ok:141 server fte.send("^regex\r\n\r\n\C*$", 128)142 action http_put:143 client fte.send("^regex\r\n\r\n$", 128)144 action http_notok:145 server fte.send("^regex\r\n\r\n\C*$", 128)"""146 parsed_format = marionette_tg.dsl.parse(mar_format)147 self.assertEquals(parsed_format.get_transport(), "tcp")148 self.assertEquals(parsed_format.get_port(), 80)149 self.assertEquals(150 parsed_format.get_transitions()[0].get_src(), "start")151 self.assertEquals(152 parsed_format.get_transitions()[0].get_dst(), "downstream")153 self.assertEquals(154 parsed_format.get_transitions()[0].get_action_block(), None)155 self.assertEquals(156 parsed_format.get_transitions()[0].get_probability(), float(1.0))157 self.assertEquals(158 parsed_format.get_transitions()[1].get_src(), "downstream")159 self.assertEquals(160 parsed_format.get_transitions()[1].get_dst(), "upstream")161 self.assertEquals(162 parsed_format.get_transitions()[1].get_action_block(), "http_get")163 self.assertEquals(164 parsed_format.get_transitions()[1].get_probability(), float(1.0))165 self.assertEquals(166 parsed_format.get_transitions()[2].get_src(), "upstream")167 self.assertEquals(parsed_format.get_transitions()[2].get_dst(), "end")168 self.assertEquals(169 parsed_format.get_transitions()[2].get_action_block(), "http_ok")170 self.assertEquals(171 parsed_format.get_transitions()[2].get_probability(), float(1.0))172 self.assertEquals(173 parsed_format.get_action_blocks()[0].get_name(), "http_get")174 self.assertEquals(175 parsed_format.get_action_blocks()[0].get_party(), "client")176 self.assertEquals(177 parsed_format.get_action_blocks()[0].get_module(), "fte")178 self.assertEquals(179 parsed_format.get_action_blocks()[0].get_method(), "send")180 self.assertEquals(181 parsed_format.get_action_blocks()[0].get_args(), [182 "^regex\r\n\r\n$", 128])183 self.assertEquals(184 parsed_format.get_action_blocks()[1].get_name(), "http_ok")185 self.assertEquals(186 parsed_format.get_action_blocks()[1].get_party(), "server")187 self.assertEquals(188 parsed_format.get_action_blocks()[1].get_module(), "fte")189 self.assertEquals(190 parsed_format.get_action_blocks()[1].get_method(), "send")191 self.assertEquals(192 parsed_format.get_action_blocks()[1].get_args(), [193 "^regex\r\n\r\n\C*$", 128])194 self.assertEquals(195 parsed_format.get_action_blocks()[2].get_name(), "http_put")196 self.assertEquals(197 parsed_format.get_action_blocks()[2].get_party(), "client")198 self.assertEquals(199 parsed_format.get_action_blocks()[2].get_module(), "fte")200 self.assertEquals(201 parsed_format.get_action_blocks()[2].get_method(), "send")202 self.assertEquals(203 parsed_format.get_action_blocks()[2].get_args(), [204 "^regex\r\n\r\n$", 128])205 self.assertEquals(206 parsed_format.get_action_blocks()[3].get_name(), "http_notok")207 self.assertEquals(208 parsed_format.get_action_blocks()[3].get_party(), "server")209 self.assertEquals(210 parsed_format.get_action_blocks()[3].get_module(), "fte")211 self.assertEquals(212 parsed_format.get_action_blocks()[3].get_method(), "send")213 self.assertEquals(214 parsed_format.get_action_blocks()[3].get_args(), [215 "^regex\r\n\r\n\C*$", 128])216 def test4(self):217 mar_format = """connection(tcp, 8082):218 start handshake NULL 1.0219 handshake upstream upstream_handshake 1.0220 upstream downstream upstream_async 1.0221 downstream upstream downstream_async 1.0222action upstream_handshake:223 client fte.send("^.*$", 128)224action upstream_async:225 client fte.send_async("^.*$", 128)226action downstream_async:227 server fte.send_async("^.*$", 128)228"""229 parsed_format = marionette_tg.dsl.parse(mar_format)230 self.assertEquals(parsed_format.get_transport(), "tcp")231 self.assertEquals(parsed_format.get_port(), 8082)232 self.assertEquals(233 parsed_format.get_transitions()[0].get_src(), "start")234 self.assertEquals(235 parsed_format.get_transitions()[0].get_dst(), "handshake")236 self.assertEquals(237 parsed_format.get_transitions()[0].get_action_block(), None)238 self.assertEquals(239 parsed_format.get_transitions()[0].get_probability(), float(1.0))240 self.assertEquals(241 parsed_format.get_transitions()[1].get_src(), "handshake")242 self.assertEquals(243 parsed_format.get_transitions()[1].get_dst(), "upstream")244 self.assertEquals(245 parsed_format.get_transitions()[1].get_action_block(),246 "upstream_handshake")247 self.assertEquals(248 parsed_format.get_transitions()[1].get_probability(), float(1.0))249 self.assertEquals(250 parsed_format.get_transitions()[2].get_src(), "upstream")251 self.assertEquals(252 parsed_format.get_transitions()[2].get_dst(), "downstream")253 self.assertEquals(254 parsed_format.get_transitions()[2].get_action_block(),255 "upstream_async")256 self.assertEquals(257 parsed_format.get_transitions()[2].get_probability(), float(1.0))258 self.assertEquals(259 parsed_format.get_transitions()[3].get_src(), "downstream")260 self.assertEquals(261 parsed_format.get_transitions()[3].get_dst(), "upstream")262 self.assertEquals(263 parsed_format.get_transitions()[3].get_action_block(),264 "downstream_async")265 self.assertEquals(266 parsed_format.get_transitions()[3].get_probability(), float(1.0))267 self.assertEquals(268 parsed_format.get_action_blocks()[0].get_name(),269 "upstream_handshake")270 self.assertEquals(271 parsed_format.get_action_blocks()[0].get_party(), "client")272 self.assertEquals(273 parsed_format.get_action_blocks()[0].get_module(), "fte")274 self.assertEquals(275 parsed_format.get_action_blocks()[0].get_method(), "send")276 self.assertEquals(277 parsed_format.get_action_blocks()[0].get_args(), ["^.*$", 128])278 self.assertEquals(279 parsed_format.get_action_blocks()[1].get_name(), "upstream_async")280 self.assertEquals(281 parsed_format.get_action_blocks()[1].get_party(), "client")282 self.assertEquals(283 parsed_format.get_action_blocks()[1].get_module(), "fte")284 self.assertEquals(285 parsed_format.get_action_blocks()[1].get_method(), "send_async")286 self.assertEquals(287 parsed_format.get_action_blocks()[1].get_args(), ["^.*$", 128])288 self.assertEquals(289 parsed_format.get_action_blocks()[2].get_name(),290 "downstream_async")291 self.assertEquals(292 parsed_format.get_action_blocks()[2].get_party(), "server")293 self.assertEquals(294 parsed_format.get_action_blocks()[2].get_module(), "fte")295 self.assertEquals(296 parsed_format.get_action_blocks()[2].get_method(), "send_async")297 self.assertEquals(298 parsed_format.get_action_blocks()[2].get_args(), ["^.*$", 128])299 def test5(self):300 mar_format = """connection(tcp, 80):301 start downstream NULL 1.0302 downstream upstream http_get 1.0303 upstream end http_ok 1.0304 action http_get:305 client fte.send("^regex\r\n\r\n$")306 action http_ok:307 server fte.send("^regex\r\n\r\n\C*$")"""308 parsed_format = marionette_tg.dsl.parse(mar_format)309 self.assertEquals(parsed_format.get_transport(), "tcp")310 self.assertEquals(parsed_format.get_port(), 80)311 self.assertEquals(312 parsed_format.get_transitions()[0].get_src(), "start")313 self.assertEquals(314 parsed_format.get_transitions()[0].get_dst(), "downstream")315 self.assertEquals(316 parsed_format.get_transitions()[0].get_action_block(), None)317 self.assertEquals(318 parsed_format.get_transitions()[0].get_probability(), float(1.0))319 self.assertEquals(320 parsed_format.get_transitions()[1].get_src(), "downstream")321 self.assertEquals(322 parsed_format.get_transitions()[1].get_dst(), "upstream")323 self.assertEquals(324 parsed_format.get_transitions()[1].get_action_block(), "http_get")325 self.assertEquals(326 parsed_format.get_transitions()[1].get_probability(), float(1.0))327 self.assertEquals(328 parsed_format.get_transitions()[2].get_src(), "upstream")329 self.assertEquals(parsed_format.get_transitions()[2].get_dst(), "end")330 self.assertEquals(331 parsed_format.get_transitions()[2].get_action_block(), "http_ok")332 self.assertEquals(333 parsed_format.get_transitions()[2].get_probability(), float(1.0))334 def test6(self):335 mar_format = """connection(tcp, 80):336 start do_nothing NULL 1.0337 do_nothing end NULL 1.0338 action http_get:339 client fte.send("^regex1\r\n\r\n$")340 server fte.recv("^regex2\r\n\r\n$")"""341 parsed_format = marionette_tg.dsl.parse(mar_format)342 self.assertEquals(parsed_format.get_transport(), "tcp")343 self.assertEquals(parsed_format.get_port(), 80)344 self.assertEquals(345 parsed_format.get_transitions()[0].get_src(), "start")346 self.assertEquals(347 parsed_format.get_transitions()[0].get_dst(), "do_nothing")348 self.assertEquals(349 parsed_format.get_transitions()[0].get_action_block(), None)350 self.assertEquals(351 parsed_format.get_transitions()[0].get_probability(), float(1.0))352 self.assertEquals(353 parsed_format.get_transitions()[1].get_src(), "do_nothing")354 self.assertEquals(parsed_format.get_transitions()[1].get_dst(), "end")355 self.assertEquals(356 parsed_format.get_transitions()[1].get_action_block(), None)357 self.assertEquals(358 parsed_format.get_transitions()[1].get_probability(), float(1.0))359 self.assertEquals(360 parsed_format.get_action_blocks()[0].get_name(), "http_get")361 self.assertEquals(362 parsed_format.get_action_blocks()[0].get_party(), "client")363 self.assertEquals(364 parsed_format.get_action_blocks()[0].get_module(), "fte")365 self.assertEquals(366 parsed_format.get_action_blocks()[0].get_method(), "send")367 self.assertEquals(368 parsed_format.get_action_blocks()[0].get_args(),369 ["^regex1\r\n\r\n$"])370 self.assertEquals(371 parsed_format.get_action_blocks()[1].get_name(), "http_get")372 self.assertEquals(373 parsed_format.get_action_blocks()[1].get_party(), "server")374 self.assertEquals(375 parsed_format.get_action_blocks()[1].get_module(), "fte")376 self.assertEquals(377 parsed_format.get_action_blocks()[1].get_method(), "recv")378 self.assertEquals(379 parsed_format.get_action_blocks()[1].get_args(),380 ["^regex2\r\n\r\n$"])381 def test7(self):382 mar_format = """connection(tcp, 80):383 start do_nothing NULL 1.0384 do_nothing end NULL 1.0385 start do_err NULL error386 do_err end NULL error387 action http_get:388 client fte.send("^regex1\r\n\r\n$")"""389 parsed_format = marionette_tg.dsl.parse(mar_format)390 self.assertEquals(parsed_format.get_transport(), "tcp")391 self.assertEquals(parsed_format.get_port(), 80)392 self.assertEquals(393 parsed_format.get_transitions()[0].get_src(), "start")394 self.assertEquals(395 parsed_format.get_transitions()[0].get_dst(), "do_nothing")396 self.assertEquals(397 parsed_format.get_transitions()[0].get_action_block(), None)398 self.assertEquals(399 parsed_format.get_transitions()[0].get_probability(), float(1.0))400 self.assertEquals(401 parsed_format.get_transitions()[0].is_error_transition(), False)402 self.assertEquals(403 parsed_format.get_transitions()[1].get_src(), "do_nothing")404 self.assertEquals(parsed_format.get_transitions()[1].get_dst(), "end")405 self.assertEquals(406 parsed_format.get_transitions()[1].get_action_block(), None)407 self.assertEquals(408 parsed_format.get_transitions()[1].get_probability(), float(1.0))409 self.assertEquals(410 parsed_format.get_transitions()[1].is_error_transition(), False)411 self.assertEquals(412 parsed_format.get_transitions()[2].get_src(), "start")413 self.assertEquals(414 parsed_format.get_transitions()[2].get_dst(), "do_err")415 self.assertEquals(416 parsed_format.get_transitions()[2].get_action_block(), None)417 self.assertEquals(418 parsed_format.get_transitions()[2].get_probability(), float(0))419 self.assertEquals(420 parsed_format.get_transitions()[2].is_error_transition(), True)421 self.assertEquals(422 parsed_format.get_transitions()[3].get_src(), "do_err")423 self.assertEquals(424 parsed_format.get_transitions()[3].get_dst(), "end")425 self.assertEquals(426 parsed_format.get_transitions()[3].get_action_block(), None)427 self.assertEquals(428 parsed_format.get_transitions()[3].get_probability(), float(0))429 self.assertEquals(430 parsed_format.get_transitions()[3].is_error_transition(), True)431 def test8(self):432 mar_format = """connection(tcp, 80):433 start do_nothing NULL 1.0434 do_nothing end NULL 1.0435 action http_get:436 client fte.send("^regex1\r\n\r\n$")437 server fte.recv("^regex2\r\n\r\n$") if regex_match_incoming("^regex2.*")"""438 parsed_format = marionette_tg.dsl.parse(mar_format)439 self.assertEquals(440 parsed_format.get_action_blocks()[0].get_name(), "http_get")441 self.assertEquals(442 parsed_format.get_action_blocks()[0].get_party(), "client")443 self.assertEquals(444 parsed_format.get_action_blocks()[0].get_module(), "fte")...
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!!