How to use append_events method in localstack
Best Python code snippet using localstack_python
test_client.py
Source:test_client.py 
...21 self.assertEqual(22 cm.exception.args[0].details(), "failed to connect to all addresses"23 )24 with self.assertRaises(ServiceUnavailable) as cm:25 esdb_client.append_events(str(uuid4()), expected_position=None, events=[])26 self.assertEqual(27 cm.exception.args[0].details(), "failed to connect to all addresses"28 )29 def test_handle_deadline_exceeded_error(self) -> None:30 class DeadlineExceededRpcError(RpcError, Call):31 def initial_metadata(self) -> None:32 pass33 def trailing_metadata(self) -> None:34 pass35 def code(self) -> StatusCode:36 return StatusCode.DEADLINE_EXCEEDED37 def details(self) -> None:38 pass39 def is_active(self) -> None:40 pass41 def time_remaining(self) -> None:42 pass43 def cancel(self) -> None:44 pass45 def add_callback(self, callback: Any) -> None:46 pass47 with self.assertRaises(GrpcError) as cm:48 raise handle_rpc_error(DeadlineExceededRpcError()) from None49 self.assertEqual(cm.exception.__class__, DeadlineExceeded)50 def test_handle_unavailable_error(self) -> None:51 class UnavailableRpcError(RpcError, Call):52 def initial_metadata(self) -> None:53 pass54 def trailing_metadata(self) -> None:55 pass56 def code(self) -> StatusCode:57 return StatusCode.UNAVAILABLE58 def details(self) -> None:59 pass60 def is_active(self) -> None:61 pass62 def time_remaining(self) -> None:63 pass64 def cancel(self) -> None:65 pass66 def add_callback(self, callback: Any) -> None:67 pass68 with self.assertRaises(GrpcError) as cm:69 raise handle_rpc_error(UnavailableRpcError()) from None70 self.assertEqual(cm.exception.__class__, ServiceUnavailable)71 def test_handle_other_call_error(self) -> None:72 class OtherRpcError(RpcError, Call):73 def initial_metadata(self) -> None:74 pass75 def trailing_metadata(self) -> None:76 pass77 def code(self) -> int:78 return -179 def details(self) -> None:80 pass81 def is_active(self) -> None:82 pass83 def time_remaining(self) -> None:84 pass85 def cancel(self) -> None:86 pass87 def add_callback(self, callback: Any) -> None:88 pass89 with self.assertRaises(GrpcError) as cm:90 raise handle_rpc_error(OtherRpcError()) from None91 self.assertEqual(cm.exception.__class__, GrpcError)92 def test_handle_non_call_rpc_error(self) -> None:93 # Check non-Call errors are handled.94 class MyRpcError(RpcError):95 pass96 msg = "some non-Call error"97 with self.assertRaises(GrpcError) as cm:98 raise handle_rpc_error(MyRpcError(msg)) from None99 self.assertEqual(cm.exception.__class__, GrpcError)100 self.assertIsInstance(cm.exception.args[0], MyRpcError)101 def test_stream_not_found_exception(self) -> None:102 esdb_client = EsdbClient("localhost:2113")103 stream_name = str(uuid4())104 with self.assertRaises(StreamNotFound):105 list(esdb_client.read_stream_events(stream_name))106 with self.assertRaises(StreamNotFound):107 list(esdb_client.read_stream_events(stream_name, backwards=True))108 with self.assertRaises(StreamNotFound):109 list(esdb_client.read_stream_events(stream_name, position=1))110 with self.assertRaises(StreamNotFound):111 list(112 esdb_client.read_stream_events(stream_name, position=1, backwards=True)113 )114 with self.assertRaises(StreamNotFound):115 list(esdb_client.read_stream_events(stream_name, limit=10))116 with self.assertRaises(StreamNotFound):117 list(esdb_client.read_stream_events(stream_name, backwards=True, limit=10))118 with self.assertRaises(StreamNotFound):119 list(esdb_client.read_stream_events(stream_name, position=1, limit=10))120 with self.assertRaises(StreamNotFound):121 list(122 esdb_client.read_stream_events(123 stream_name, position=1, backwards=True, limit=10124 )125 )126 def test_stream_append_and_read_without_occ(self) -> None:127 client = EsdbClient("localhost:2113")128 stream_name = str(uuid4())129 event1 = NewEvent(type="Snapshot", data=b"{}", metadata=b"{}")130 # Append new event.131 client.append_events(stream_name, expected_position=-1, events=[event1])132 events = list(client.read_stream_events(stream_name, backwards=True, limit=1))133 self.assertEqual(len(events), 1)134 self.assertEqual(events[0].type, "Snapshot")135 def test_stream_append_and_read_with_occ(self) -> None:136 client = EsdbClient("localhost:2113")137 stream_name = str(uuid4())138 # Check stream not found.139 with self.assertRaises(StreamNotFound):140 list(client.read_stream_events(stream_name))141 # Check stream position is None.142 self.assertEqual(client.get_stream_position(stream_name), None)143 # Check get error when attempting to append empty list to position 1.144 with self.assertRaises(ExpectedPositionError) as cm:145 client.append_events(stream_name, expected_position=1, events=[])146 self.assertEqual(cm.exception.args[0], f"Stream '{stream_name}' does not exist")147 # Append empty list of events.148 commit_position1 = client.append_events(149 stream_name, expected_position=None, events=[]150 )151 self.assertIsInstance(commit_position1, int)152 # Check stream still not found.153 with self.assertRaises(StreamNotFound):154 list(client.read_stream_events(stream_name))155 # Check stream position is None.156 self.assertEqual(client.get_stream_position(stream_name), None)157 # Check get error when attempting to append new event to position 1.158 event1 = NewEvent(type="OrderCreated", data=b"{}", metadata=b"{}")159 with self.assertRaises(ExpectedPositionError) as cm:160 client.append_events(stream_name, expected_position=1, events=[event1])161 self.assertEqual(cm.exception.args[0], f"Stream '{stream_name}' does not exist")162 # Append new event.163 commit_position2 = client.append_events(164 stream_name, expected_position=None, events=[event1]165 )166 # Todo: Why isn't this +1?167 # self.assertEqual(commit_position2 - commit_position1, 1)168 self.assertEqual(commit_position2 - commit_position1, 126)169 # Check stream position is 0.170 self.assertEqual(client.get_stream_position(stream_name), 0)171 # Read the stream forwards from the start (expect one event).172 events = list(client.read_stream_events(stream_name))173 self.assertEqual(len(events), 1)174 self.assertEqual(events[0].type, "OrderCreated")175 # Check we can't append another new event at initial position.176 event2 = NewEvent(type="OrderUpdated", data=b"{}", metadata=b"{}")177 with self.assertRaises(ExpectedPositionError) as cm:178 client.append_events(stream_name, expected_position=None, events=[event2])179 self.assertEqual(cm.exception.args[0], "Current position is 0")180 # Append another event.181 commit_position3 = client.append_events(182 stream_name, expected_position=0, events=[event2]183 )184 # Check stream position is 1.185 self.assertEqual(client.get_stream_position(stream_name), 1)186 # NB: Why isn't this +1? because it's "disk position" :-|187 # self.assertEqual(commit_position3 - commit_position2, 1)188 self.assertEqual(commit_position3 - commit_position2, 142)189 # Read the stream (expect two events in 'forwards' order).190 events = list(client.read_stream_events(stream_name))191 self.assertEqual(len(events), 2)192 self.assertEqual(events[0].type, "OrderCreated")193 self.assertEqual(events[1].type, "OrderUpdated")194 # Read the stream backwards from the end.195 events = list(client.read_stream_events(stream_name, backwards=True))196 self.assertEqual(len(events), 2)197 self.assertEqual(events[1].type, "OrderCreated")198 self.assertEqual(events[0].type, "OrderUpdated")199 # Read the stream forwards from position 1.200 events = list(client.read_stream_events(stream_name, position=1))201 self.assertEqual(len(events), 1)202 self.assertEqual(events[0].type, "OrderUpdated")203 # Read the stream backwards from position 0.204 events = list(205 client.read_stream_events(stream_name, position=0, backwards=True)206 )207 self.assertEqual(len(events), 1)208 self.assertEqual(events[0].type, "OrderCreated")209 # Read the stream forwards from start with limit.210 events = list(client.read_stream_events(stream_name, limit=1))211 self.assertEqual(len(events), 1)212 self.assertEqual(events[0].type, "OrderCreated")213 # Read the stream backwards from end with limit.214 events = list(client.read_stream_events(stream_name, backwards=True, limit=1))215 self.assertEqual(len(events), 1)216 self.assertEqual(events[0].type, "OrderUpdated")217 # Check we can't append another new event at second position.218 event3 = NewEvent(type="OrderDeleted", data=b"{}", metadata=b"{}")219 with self.assertRaises(ExpectedPositionError) as cm:220 client.append_events(stream_name, expected_position=0, events=[event3])221 self.assertEqual(cm.exception.args[0], "Current position is 1")222 # Append another new event.223 commit_position4 = client.append_events(224 stream_name, expected_position=1, events=[event3]225 )226 # Check stream position is 2.227 self.assertEqual(client.get_stream_position(stream_name), 2)228 # NB: Why isn't this +1? because it's "disk position" :-|229 # self.assertEqual(commit_position4 - commit_position3, 1)230 self.assertEqual(commit_position4 - commit_position3, 142)231 # Read the stream forwards from start (expect three events).232 events = list(client.read_stream_events(stream_name))233 self.assertEqual(len(events), 3)234 self.assertEqual(events[0].type, "OrderCreated")235 self.assertEqual(events[1].type, "OrderUpdated")236 self.assertEqual(events[2].type, "OrderDeleted")237 # Read the stream backwards from end (expect three events).238 events = list(client.read_stream_events(stream_name, backwards=True))239 self.assertEqual(len(events), 3)240 self.assertEqual(events[2].type, "OrderCreated")241 self.assertEqual(events[1].type, "OrderUpdated")242 self.assertEqual(events[0].type, "OrderDeleted")243 # Read the stream forwards from position with limit.244 events = list(client.read_stream_events(stream_name, position=1, limit=1))245 self.assertEqual(len(events), 1)246 self.assertEqual(events[0].type, "OrderUpdated")247 # Read the stream backwards from position withm limit.248 events = list(249 client.read_stream_events(stream_name, position=1, backwards=True, limit=1)250 )251 self.assertEqual(len(events), 1)252 self.assertEqual(events[0].type, "OrderUpdated")253 def test_timeout_stream_append_and_read(self) -> None:254 client = EsdbClient("localhost:2113")255 # Append three events.256 stream_name1 = str(uuid4())257 event1 = NewEvent(258 type="OrderCreated",259 data=b"{}",260 metadata=b"{}",261 )262 event2 = NewEvent(263 type="OrderUpdated",264 data=b"{}",265 metadata=b"{}",266 )267 event3 = NewEvent(268 type="OrderDeleted",269 data=b"{}",270 metadata=b"{}",271 )272 client.append_events(273 stream_name=stream_name1,274 expected_position=None,275 events=[event1, event2],276 )277 # Timeout appending new event.278 with self.assertRaises(DeadlineExceeded):279 client.append_events(280 stream_name1, expected_position=1, events=[event3], timeout=0281 )282 # Timeout reading stream.283 with self.assertRaises(DeadlineExceeded):284 list(client.read_stream_events(stream_name1, timeout=0))285 def test_read_all_events(self) -> None:286 esdb_client = EsdbClient("localhost:2113")287 num_old_events = len(list(esdb_client.read_all_events()))288 event1 = NewEvent(type="OrderCreated", data=b"{}", metadata=b"{}")289 event2 = NewEvent(type="OrderUpdated", data=b"{}", metadata=b"{}")290 event3 = NewEvent(type="OrderDeleted", data=b"{}", metadata=b"{}")291 # Append new events.292 stream_name1 = str(uuid4())293 commit_position1 = esdb_client.append_events(294 stream_name1, expected_position=None, events=[event1, event2, event3]295 )296 stream_name2 = str(uuid4())297 commit_position2 = esdb_client.append_events(298 stream_name2, expected_position=None, events=[event1, event2, event3]299 )300 # Check we can read forwards from the start.301 events = list(esdb_client.read_all_events())302 self.assertEqual(len(events) - num_old_events, 6)303 self.assertEqual(events[-1].stream_name, stream_name2)304 self.assertEqual(events[-1].type, "OrderDeleted")305 self.assertEqual(events[-2].stream_name, stream_name2)306 self.assertEqual(events[-2].type, "OrderUpdated")307 self.assertEqual(events[-3].stream_name, stream_name2)308 self.assertEqual(events[-3].type, "OrderCreated")309 self.assertEqual(events[-4].stream_name, stream_name1)310 self.assertEqual(events[-4].type, "OrderDeleted")311 # Check we can read backwards from the end.312 events = list(esdb_client.read_all_events(backwards=True))313 self.assertEqual(len(events) - num_old_events, 6)314 self.assertEqual(events[0].stream_name, stream_name2)315 self.assertEqual(events[0].type, "OrderDeleted")316 self.assertEqual(events[1].stream_name, stream_name2)317 self.assertEqual(events[1].type, "OrderUpdated")318 self.assertEqual(events[2].stream_name, stream_name2)319 self.assertEqual(events[2].type, "OrderCreated")320 self.assertEqual(events[3].stream_name, stream_name1)321 self.assertEqual(events[3].type, "OrderDeleted")322 # Check we can read forwards from commit position 1.323 events = list(esdb_client.read_all_events(position=commit_position1))324 self.assertEqual(len(events), 4)325 self.assertEqual(events[0].stream_name, stream_name1)326 self.assertEqual(events[0].type, "OrderDeleted")327 self.assertEqual(events[1].stream_name, stream_name2)328 self.assertEqual(events[1].type, "OrderCreated")329 self.assertEqual(events[2].stream_name, stream_name2)330 self.assertEqual(events[2].type, "OrderUpdated")331 self.assertEqual(events[3].stream_name, stream_name2)332 self.assertEqual(events[3].type, "OrderDeleted")333 # Check we can read forwards from commit position 2.334 events = list(esdb_client.read_all_events(position=commit_position2))335 self.assertEqual(len(events), 1)336 self.assertEqual(events[0].stream_name, stream_name2)337 self.assertEqual(events[0].type, "OrderDeleted")338 # Check we can read backwards from commit position 1.339 # NB backwards here doesn't include event at commit position, otherwise340 # first event would an OrderDeleted event, and we get an OrderUpdated.341 events = list(342 esdb_client.read_all_events(position=commit_position1, backwards=True)343 )344 self.assertEqual(len(events) - num_old_events, 2)345 self.assertEqual(events[0].stream_name, stream_name1)346 self.assertEqual(events[0].type, "OrderUpdated")347 self.assertEqual(events[1].stream_name, stream_name1)348 self.assertEqual(events[1].type, "OrderCreated")349 # Check we can read backwards from commit position 2.350 # NB backwards here doesn't include event at commit position.351 events = list(352 esdb_client.read_all_events(position=commit_position2, backwards=True)353 )354 self.assertEqual(len(events) - num_old_events, 5)355 self.assertEqual(events[0].stream_name, stream_name2)356 self.assertEqual(events[0].type, "OrderUpdated")357 self.assertEqual(events[1].stream_name, stream_name2)358 self.assertEqual(events[1].type, "OrderCreated")359 self.assertEqual(events[2].stream_name, stream_name1)360 self.assertEqual(events[2].type, "OrderDeleted")361 # Check we can read forwards from the start with limit.362 events = list(esdb_client.read_all_events(limit=3))363 self.assertEqual(len(events), 3)364 # Check we can read backwards from the end with limit.365 events = list(esdb_client.read_all_events(backwards=True, limit=3))366 self.assertEqual(len(events), 3)367 self.assertEqual(events[0].stream_name, stream_name2)368 self.assertEqual(events[0].type, "OrderDeleted")369 self.assertEqual(events[1].stream_name, stream_name2)370 self.assertEqual(events[1].type, "OrderUpdated")371 self.assertEqual(events[2].stream_name, stream_name2)372 self.assertEqual(events[2].type, "OrderCreated")373 # Check we can read forwards from commit position 1 with limit.374 events = list(esdb_client.read_all_events(position=commit_position1, limit=3))375 self.assertEqual(len(events), 3)376 self.assertEqual(events[0].stream_name, stream_name1)377 self.assertEqual(events[0].type, "OrderDeleted")378 self.assertEqual(events[1].stream_name, stream_name2)379 self.assertEqual(events[1].type, "OrderCreated")380 self.assertEqual(events[2].stream_name, stream_name2)381 self.assertEqual(events[2].type, "OrderUpdated")382 # Check we can read backwards from commit position 2 with limit.383 events = list(384 esdb_client.read_all_events(385 position=commit_position2, backwards=True, limit=3386 )387 )388 self.assertEqual(len(events), 3)389 self.assertEqual(events[0].stream_name, stream_name2)390 self.assertEqual(events[0].type, "OrderUpdated")391 self.assertEqual(events[1].stream_name, stream_name2)392 self.assertEqual(events[1].type, "OrderCreated")393 self.assertEqual(events[2].stream_name, stream_name1)394 self.assertEqual(events[2].type, "OrderDeleted")395 def test_timeout_read_all_events(self) -> None:396 esdb_client = EsdbClient("localhost:2113")397 event1 = NewEvent(type="OrderCreated", data=b"{}", metadata=b"{}")398 event2 = NewEvent(type="OrderUpdated", data=b"{}", metadata=b"{}")399 event3 = NewEvent(type="OrderDeleted", data=b"{}", metadata=b"{}")400 # Append new events.401 stream_name1 = str(uuid4())402 esdb_client.append_events(403 stream_name1, expected_position=None, events=[event1, event2, event3]404 )405 stream_name2 = str(uuid4())406 esdb_client.append_events(407 stream_name2, expected_position=None, events=[event1, event2, event3]408 )409 # Timeout reading all events.410 with self.assertRaises(DeadlineExceeded):411 list(esdb_client.read_all_events(timeout=0.001))412 def test_read_all_filter_include(self) -> None:413 client = EsdbClient("localhost:2113")414 event1 = NewEvent(type="OrderCreated", data=b"{}", metadata=b"{}")415 event2 = NewEvent(type="OrderUpdated", data=b"{}", metadata=b"{}")416 event3 = NewEvent(type="OrderDeleted", data=b"{}", metadata=b"{}")417 # Append new events.418 stream_name1 = str(uuid4())419 client.append_events(420 stream_name1, expected_position=None, events=[event1, event2, event3]421 )422 # Read only OrderCreated.423 events = list(client.read_all_events(filter_include=("OrderCreated",)))424 types = set([e.type for e in events])425 self.assertEqual(types, {"OrderCreated"})426 # Read only OrderCreated and OrderDeleted.427 events = list(428 client.read_all_events(filter_include=("OrderCreated", "OrderDeleted"))429 )430 types = set([e.type for e in events])431 self.assertEqual(types, {"OrderCreated", "OrderDeleted"})432 def test_read_all_filter_exclude(self) -> None:433 client = EsdbClient("localhost:2113")434 event1 = NewEvent(type="OrderCreated", data=b"{}", metadata=b"{}")435 event2 = NewEvent(type="OrderUpdated", data=b"{}", metadata=b"{}")436 event3 = NewEvent(type="OrderDeleted", data=b"{}", metadata=b"{}")437 # Append new events.438 stream_name1 = str(uuid4())439 client.append_events(440 stream_name1, expected_position=None, events=[event1, event2, event3]441 )442 # Exclude OrderCreated.443 events = list(client.read_all_events(filter_exclude=("OrderCreated",)))444 types = set([e.type for e in events])445 self.assertNotIn("OrderCreated", types)446 self.assertIn("OrderUpdated", types)447 self.assertIn("OrderDeleted", types)448 # Exclude OrderCreated and OrderDeleted.449 events = list(450 client.read_all_events(filter_exclude=("OrderCreated", "OrderDeleted"))451 )452 types = set([e.type for e in events])453 self.assertNotIn("OrderCreated", types)454 self.assertIn("OrderUpdated", types)455 self.assertNotIn("OrderDeleted", types)456 def test_subscribe_all_events_default_filter(self) -> None:457 client = EsdbClient("localhost:2113")458 event1 = NewEvent(type="OrderCreated", data=b"{a}", metadata=b"{}")459 event2 = NewEvent(type="OrderUpdated", data=b"{b}", metadata=b"{}")460 event3 = NewEvent(type="OrderDeleted", data=b"{c}", metadata=b"{}")461 # Append new events.462 stream_name1 = str(uuid4())463 client.append_events(464 stream_name1, expected_position=None, events=[event1, event2, event3]465 )466 # Subscribe to all events, from the start.467 subscription = client.subscribe_all_events()468 # Iterate over the first three events.469 events = []470 for event in subscription:471 events.append(event)472 if len(events) == 3:473 break474 # Get the current commit position.475 commit_position = client.get_commit_position()476 # Subscribe from the current commit position.477 subscription = client.subscribe_all_events(position=commit_position)478 # Append three more events.479 event1 = NewEvent(type="OrderCreated", data=b"{}", metadata=b"{}")480 event2 = NewEvent(type="OrderUpdated", data=b"{}", metadata=b"{}")481 event3 = NewEvent(type="OrderDeleted", data=b"{}", metadata=b"{}")482 stream_name2 = str(uuid4())483 client.append_events(484 stream_name2, expected_position=None, events=[event1, event2, event3]485 )486 # Check the stream name of the newly received events.487 events = []488 for event in subscription:489 self.assertEqual(event.stream_name, stream_name2)490 events.append(event)491 self.assertIn(event.type, ["OrderCreated", "OrderUpdated", "OrderDeleted"])492 if len(events) == 3:493 break494 def test_subscribe_all_events_no_filter(self) -> None:495 client = EsdbClient("localhost:2113")496 # Append new events.497 event1 = NewEvent(type="OrderCreated", data=b"{}", metadata=b"{}")498 event2 = NewEvent(type="OrderUpdated", data=b"{}", metadata=b"{}")499 event3 = NewEvent(type="OrderDeleted", data=b"{}", metadata=b"{}")500 stream_name1 = str(uuid4())501 client.append_events(502 stream_name1, expected_position=None, events=[event1, event2, event3]503 )504 # Subscribe from the current commit position.505 subscription = client.subscribe_all_events(506 filter_exclude=[],507 filter_include=[],508 )509 # Expect to get system events.510 for event in subscription:511 if event.type.startswith("$"):512 break513 def test_subscribe_all_events_include_filter(self) -> None:514 client = EsdbClient("localhost:2113")515 # Append new events.516 event1 = NewEvent(type="OrderCreated", data=b"{}", metadata=b"{}")517 event2 = NewEvent(type="OrderUpdated", data=b"{}", metadata=b"{}")518 event3 = NewEvent(type="OrderDeleted", data=b"{}", metadata=b"{}")519 stream_name1 = str(uuid4())520 client.append_events(521 stream_name1, expected_position=None, events=[event1, event2, event3]522 )523 # Subscribe from the beginning.524 subscription = client.subscribe_all_events(525 filter_exclude=[],526 filter_include=["OrderCreated"],527 )528 # Expect to only get "OrderCreated" events.529 count = 0530 for event in subscription:531 if not event.type.startswith("OrderCreated"):532 self.fail(f"Include filter is broken: {event.type}")533 count += 1534 break535 self.assertEqual(count, 1)536 def test_subscribe_all_events_from_commit_position_zero(self) -> None:537 client = EsdbClient("localhost:2113")538 # Append new events.539 event1 = NewEvent(type="OrderCreated", data=b"{}", metadata=b"{}")540 event2 = NewEvent(type="OrderUpdated", data=b"{}", metadata=b"{}")541 event3 = NewEvent(type="OrderDeleted", data=b"{}", metadata=b"{}")542 stream_name1 = str(uuid4())543 client.append_events(544 stream_name1, expected_position=None, events=[event1, event2, event3]545 )546 # Subscribe from the beginning.547 subscription = client.subscribe_all_events()548 # Expect to only get "OrderCreated" events.549 count = 0550 for _ in subscription:551 count += 1552 break553 self.assertEqual(count, 1)554 def test_timeout_subscribe_all_events(self) -> None:555 client = EsdbClient("localhost:2113")556 # Append new events.557 event1 = NewEvent(type="OrderCreated", data=b"{}", metadata=b"{}")558 event2 = NewEvent(type="OrderUpdated", data=b"{}", metadata=b"{}")559 event3 = NewEvent(type="OrderDeleted", data=b"{}", metadata=b"{}")560 stream_name1 = str(uuid4())561 client.append_events(562 stream_name1, expected_position=None, events=[event1, event2, event3]563 )564 # Subscribe from the beginning.565 subscription = client.subscribe_all_events(timeout=0.5)566 # Expect to only get "OrderCreated" events.567 count = 0568 with self.assertRaises(DeadlineExceeded):569 for _ in subscription:570 count += 1...
test_run_process.py
Source:test_run_process.py
...27 @mock.patch(f'{base.__name__}.save')28 def test_call(self, mock_save):29 """Calling the process runs all steps in order."""30 saved_events = []31 def append_events(*events, submission_id=None):32 for event in events:33 saved_events.append((event, submission_id))34 mock_save.side_effect = append_events35 proc = self.FooProcess(self.submission_id)36 runner = base.ProcessRunner(proc)37 runner.run(mock.MagicMock(event=mock.MagicMock(some_value=1)))38 self.assertEqual(saved_events[0][0].status,39 process.Process.Status.PENDING)40 self.assertEqual(saved_events[0][1], self.submission_id)41 self.assertEqual(saved_events[1][0].status,42 process.Process.Status.IN_PROGRESS)43 self.assertEqual(saved_events[1][0].step, 'step_a')44 self.assertEqual(saved_events[1][1], self.submission_id)45 self.assertEqual(saved_events[2][0].status,46 process.Process.Status.IN_PROGRESS)47 self.assertEqual(saved_events[2][0].step, 'step_b')48 self.assertEqual(saved_events[2][1], self.submission_id)49 self.assertEqual(saved_events[3][0].status,50 process.Process.Status.SUCCEEDED)51 self.assertEqual(saved_events[3][0].step, 'step_c')52 self.assertEqual(saved_events[3][1], self.submission_id)53 @mock.patch(f'{process.__name__}.base.AddProcessStatus', mock.MagicMock)54 @mock.patch(f'{base.__name__}.save')55 def test_failing_process(self, mock_save):56 """Calling the process runs all steps in order, but one fails."""57 saved_events = []58 def append_events(*events, submission_id=None):59 for event in events:60 saved_events.append((event, submission_id))61 mock_save.side_effect = append_events62 proc = self.FooProcess(self.submission_id)63 runner = base.ProcessRunner(proc)64 runner.run(mock.MagicMock(event=mock.MagicMock(some_value=5)))65 self.assertEqual(saved_events[0][0].status,66 process.Process.Status.PENDING)67 self.assertEqual(saved_events[0][1], self.submission_id)68 self.assertEqual(saved_events[1][0].status,69 process.Process.Status.IN_PROGRESS)70 self.assertEqual(saved_events[1][0].step, 'step_a')71 self.assertEqual(saved_events[1][1], self.submission_id)72 self.assertEqual(saved_events[2][0].status,...
calendar_info.py
Source:calendar_info.py
...41 self.month = month42 self.year = year43 self.events = []44 self.holidays = []45 self.append_events(local_holidays)46 self.append_events(local_events)47 self.append_events(google_holidays)48 self.append_events(google_events)49 def append_events(self, events=None):50 if events is None:51 return52 for event in events:53 self.append_event(event)54 def append_event(self, event=None):55 if event is None:56 return57 elif not event.is_in_year_and_month(self.year, self.month):58 return59 elif event.is_holiday():60 if already_exists(self.holidays, event):61 return62 self.holidays.append(event)63 else:...
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?
