| 531 | |
| 532 | |
| 533 | class StubbedSession(botocore.session.Session): |
| 534 | def __init__(self, *args, **kwargs): |
| 535 | super().__init__(*args, **kwargs) |
| 536 | self._cached_clients = {} |
| 537 | self._client_stubs = {} |
| 538 | |
| 539 | @property |
| 540 | def cached_clients(self): |
| 541 | return self._cached_clients |
| 542 | |
| 543 | @property |
| 544 | def client_stubs(self): |
| 545 | return self._client_stubs |
| 546 | |
| 547 | def create_client(self, service_name, *args, **kwargs): |
| 548 | if service_name not in self._cached_clients: |
| 549 | client = self._create_stubbed_client(service_name, *args, **kwargs) |
| 550 | self._cached_clients[service_name] = client |
| 551 | return self._cached_clients[service_name] |
| 552 | |
| 553 | def _create_stubbed_client(self, service_name, *args, **kwargs): |
| 554 | client = super().create_client(service_name, *args, **kwargs) |
| 555 | stubber = Stubber(client) |
| 556 | self._client_stubs[service_name] = stubber |
| 557 | return client |
| 558 | |
| 559 | def stub(self, service_name, *args, **kwargs): |
| 560 | if service_name not in self._client_stubs: |
| 561 | self.create_client(service_name, *args, **kwargs) |
| 562 | return self._client_stubs[service_name] |
| 563 | |
| 564 | def activate_stubs(self): |
| 565 | for stub in self._client_stubs.values(): |
| 566 | stub.activate() |
| 567 | |
| 568 | def verify_stubs(self): |
| 569 | for stub in self._client_stubs.values(): |
| 570 | stub.assert_no_pending_responses() |
| 571 | |
| 572 | |
| 573 | class FreezeTime(contextlib.ContextDecorator): |
no outgoing calls