KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use _session method in Molotov
Best Python code snippet using molotov_python
manager.py
Source:manager.py 
1"""Classes to manage connections."""2import logging3from typing import Coroutine, Sequence, Tuple4from aries_cloudagent.protocols.coordinate_mediation.v1_0.manager import (5 MediationManager,6)7from ....cache.base import BaseCache8from ....config.base import InjectionError9from ....connections.base_manager import BaseConnectionManager10from ....connections.models.conn_record import ConnRecord11from ....connections.models.connection_target import ConnectionTarget12from ....connections.util import mediation_record_if_id13from ....core.error import BaseError14from ....core.profile import ProfileSession15from ....messaging.responder import BaseResponder16from ....protocols.routing.v1_0.manager import RoutingManager17from ....storage.error import StorageError, StorageNotFoundError18from ....transport.inbound.receipt import MessageReceipt19from ....wallet.base import BaseWallet, DIDInfo20from ....wallet.crypto import create_keypair, seed_to_did21from ....wallet.error import WalletNotFoundError22from ....wallet.util import bytes_to_b5823from ....multitenant.manager import MultitenantManager24from ...coordinate_mediation.v1_0.models.mediation_record import MediationRecord25from .messages.connection_invitation import ConnectionInvitation26from .messages.connection_request import ConnectionRequest27from .messages.connection_response import ConnectionResponse28from .messages.problem_report import ProblemReportReason29from .models.connection_detail import ConnectionDetail30class ConnectionManagerError(BaseError):31 """Connection error."""32class ConnectionManager(BaseConnectionManager):33 """Class for managing connections."""34 def __init__(self, session: ProfileSession):35 """36 Initialize a ConnectionManager.37 Args:38 session: The profile session for this connection manager39 """40 self._session = session41 self._logger = logging.getLogger(__name__)42 super().__init__(self._session)43 @property44 def session(self) -> ProfileSession:45 """46 Accessor for the current profile session.47 Returns:48 The profile session for this connection manager49 """50 return self._session51 async def create_invitation(52 self,53 my_label: str = None,54 my_endpoint: str = None,55 auto_accept: bool = None,56 public: bool = False,57 multi_use: bool = False,58 alias: str = None,59 routing_keys: Sequence[str] = None,60 recipient_keys: Sequence[str] = None,61 metadata: dict = None,62 mediation_id: str = None,63 with_did: str = None,64 ) -> Tuple[ConnRecord, ConnectionInvitation]:65 """66 Generate new connection invitation.67 This interaction represents an out-of-band communication channel. In the future68 and in practice, these sort of invitations will be received over any number of69 channels such as SMS, Email, QR Code, NFC, etc.70 Structure of an invite message:71 ::72 {73 "@type": "https://didcomm.org/connections/1.0/invitation",74 "label": "Alice",75 "did": "did:bid:QmWbsNYhMrjHiqZDTUTEJs"76 }77 Or, in the case of a peer DID:78 ::79 {80 "@type": "https://didcomm.org/connections/1.0/invitation",81 "label": "Alice",82 "did": "did:peer:oiSqsNYhMrjHiqZDTUthsw",83 "recipient_keys": ["8HH5gYEeNc3z7PYXmd54d4x6qAfCNrqQqEB3nS7Zfu7K"],84 "service_endpoint": "https://example.com/endpoint"85 "routing_keys": ["9EH5gYEeNc3z7PYXmd53d5x6qAfCNrqQqEB4nS7Zfu6K"],86 }87 Args:88 my_label: label for this connection89 my_endpoint: endpoint where other party can reach me90 auto_accept: auto-accept a corresponding connection request91 (None to use config)92 public: set to create an invitation from the public DID93 multi_use: set to True to create an invitation for multiple use94 alias: optional alias to apply to connection for later use95 Returns:96 A tuple of the new `ConnRecord` and `ConnectionInvitation` instances97 """98 # Mediation Record can still be None after this operation if no99 # mediation id passed and no default100 mediation_record = await mediation_record_if_id(101 self._session,102 mediation_id,103 or_default=True,104 )105 keylist_updates = None106 image_url = self._session.context.settings.get("image_url")107 # Multitenancy setup108 multitenant_mgr = self._session.inject(MultitenantManager, required=False)109 wallet_id = self._session.settings.get("wallet.id")110 if not my_label:111 my_label = self._session.settings.get("default_label")112 wallet = self._session.inject(BaseWallet)113 if public:114 if not self._session.settings.get("public_invites"):115 raise ConnectionManagerError("Public invitations are not enabled")116 public_did = await wallet.get_public_did()117 if not public_did:118 raise ConnectionManagerError(119 "Cannot create public invitation with no public DID"120 )121 if multi_use:122 raise ConnectionManagerError(123 "Cannot use public and multi_use at the same time"124 )125 if metadata:126 raise ConnectionManagerError(127 "Cannot use public and set metadata at the same time"128 )129 # FIXME - allow ledger instance to format public DID with prefix?130 invitation = ConnectionInvitation(131 label=my_label, did=f"did:bid:{public_did.did}", image_url=image_url132 )133 # Add mapping for multitenant relaying.134 # Mediation of public keys is not supported yet135 if multitenant_mgr and wallet_id:136 await multitenant_mgr.add_key(137 wallet_id, public_did.verkey, skip_if_exists=True138 )139 return None, invitation140 invitation_mode = ConnRecord.INVITATION_MODE_ONCE141 if multi_use:142 invitation_mode = ConnRecord.INVITATION_MODE_MULTI143 if recipient_keys:144 # TODO: register recipient keys for relay145 # TODO: check that recipient keys are in wallet146 invitation_key = recipient_keys[0] # TODO first key appropriate?147 else:148 # Create and store new invitation key149 invitation_signing_key = await wallet.create_signing_key()150 invitation_key = invitation_signing_key.verkey151 recipient_keys = [invitation_key]152 mediation_mgr = MediationManager(self._session.profile)153 keylist_updates = await mediation_mgr.add_key(154 invitation_key, keylist_updates155 )156 if multitenant_mgr and wallet_id:157 await multitenant_mgr.add_key(wallet_id, invitation_key)158 accept = (159 ConnRecord.ACCEPT_AUTO160 if (161 auto_accept162 or (163 auto_accept is None164 and self._session.settings.get("debug.auto_accept_requests")165 )166 )167 else ConnRecord.ACCEPT_MANUAL168 )169 # Create connection record170 connection = ConnRecord(171 invitation_key=invitation_key, # TODO: determine correct key to use172 their_role=ConnRecord.Role.REQUESTER.rfc160,173 state=ConnRecord.State.INVITATION.rfc160,174 accept=accept,175 invitation_mode=invitation_mode,176 alias=alias,177 with_did=with_did,178 )179 await connection.save(self._session, reason="Created new invitation")180 routing_keys = []181 my_endpoint = my_endpoint or self._session.settings.get("default_endpoint")182 # The base wallet can act as a mediator for all tenants183 if multitenant_mgr and wallet_id:184 base_mediation_record = await multitenant_mgr.get_default_mediator()185 if base_mediation_record:186 routing_keys = base_mediation_record.routing_keys187 my_endpoint = base_mediation_record.endpoint188 # If we use a mediator for the base wallet we don't189 # need to register the key at the subwallet mediator190 # because it only needs to know the key of the base mediator191 # sub wallet mediator -> base wallet mediator -> agent192 keylist_updates = None193 if mediation_record:194 routing_keys = [*routing_keys, *mediation_record.routing_keys]195 my_endpoint = mediation_record.endpoint196 # Save that this invitation was created with mediation197 await connection.metadata_set(198 self._session, "mediation", {"id": mediation_id}199 )200 if keylist_updates:201 responder = self._session.inject(BaseResponder, required=False)202 await responder.send(203 keylist_updates, connection_id=mediation_record.connection_id204 )205 # Create connection invitation message206 # Note: Need to split this into two stages to support inbound routing of invites207 # Would want to reuse create_did_document and convert the result208 invitation = ConnectionInvitation(209 label=my_label,210 recipient_keys=recipient_keys,211 routing_keys=routing_keys,212 endpoint=my_endpoint,213 image_url=image_url,214 )215 await connection.attach_invitation(self._session, invitation)216 if metadata:217 for key, value in metadata.items():218 await connection.metadata_set(self._session, key, value)219 return connection, invitation220 async def receive_invitation(221 self,222 invitation: ConnectionInvitation,223 their_public_did: str = None,224 auto_accept: bool = None,225 alias: str = None,226 mediation_id: str = None,227 mediation_record: MediationRecord = None,228 with_did: str = None,229 ) -> ConnRecord:230 """231 Create a new connection record to track a received invitation.232 Args:233 invitation: The `ConnectionInvitation` to store234 auto_accept: set to auto-accept the invitation (None to use config)235 alias: optional alias to set on the record236 Returns:237 The new `ConnRecord` instance238 """239 if not invitation.did:240 if not invitation.recipient_keys:241 raise ConnectionManagerError("Invitation must contain recipient key(s)")242 if not invitation.endpoint:243 raise ConnectionManagerError("Invitation must contain an endpoint")244 accept = (245 ConnRecord.ACCEPT_AUTO246 if (247 auto_accept248 or (249 auto_accept is None250 and self._session.settings.get("debug.auto_accept_invites")251 )252 )253 else ConnRecord.ACCEPT_MANUAL254 )255 # Create connection record256 connection = ConnRecord(257 invitation_key=invitation.recipient_keys and invitation.recipient_keys[0],258 their_label=invitation.label,259 their_role=ConnRecord.Role.RESPONDER.rfc160,260 state=ConnRecord.State.INVITATION.rfc160,261 accept=accept,262 alias=alias,263 their_public_did=their_public_did,264 with_did=with_did,265 )266 await connection.save(267 self._session,268 reason="Created new connection record from invitation",269 log_params={"invitation": invitation, "their_label": invitation.label},270 )271 # Save the invitation for later processing272 await connection.attach_invitation(self._session, invitation)273 if connection.accept == ConnRecord.ACCEPT_AUTO:274 request = await self.create_request(connection, mediation_id=mediation_id)275 responder = self._session.inject(BaseResponder, required=False)276 if responder:277 await responder.send(request, connection_id=connection.connection_id)278 # refetch connection for accurate state279 connection = await ConnRecord.retrieve_by_id(280 self._session, connection.connection_id281 )282 else:283 self._logger.debug("Connection invitation will await acceptance")284 return connection285 async def create_request(286 self,287 connection: ConnRecord,288 my_label: str = None,289 my_endpoint: str = None,290 mediation_id: str = None,291 ) -> ConnectionRequest:292 """293 Create a new connection request for a previously-received invitation.294 Args:295 connection: The `ConnRecord` representing the invitation to accept296 my_label: My label297 my_endpoint: My endpoint298 Returns:299 A new `ConnectionRequest` message to send to the other agent300 """301 keylist_updates = None302 # Mediation Record can still be None after this operation if no303 # mediation id passed and no default304 mediation_record = await mediation_record_if_id(305 self._session,306 mediation_id,307 or_default=True,308 )309 multitenant_mgr = self._session.inject(MultitenantManager, required=False)310 wallet_id = self._session.settings.get("wallet.id")311 base_mediation_record = None312 if multitenant_mgr and wallet_id:313 base_mediation_record = await multitenant_mgr.get_default_mediator()314 my_info = None315 wallet = self._session.inject(BaseWallet)316 if connection.my_did:317 my_info = await wallet.get_local_did(connection.my_did)318 elif connection.with_did:319 my_info = await wallet.get_local_did(connection.with_did)320 connection.my_did = my_info.did321 else:322 # Create new DID for connection323 my_info = await wallet.create_local_did()324 connection.my_did = my_info.did325 mediation_mgr = MediationManager(self._session.profile)326 keylist_updates = await mediation_mgr.add_key(327 my_info.verkey, keylist_updates328 )329 # Add mapping for multitenant relay330 if multitenant_mgr and wallet_id:331 await multitenant_mgr.add_key(wallet_id, my_info.verkey)332 # Create connection request message333 if my_endpoint:334 my_endpoints = [my_endpoint]335 else:336 my_endpoints = []337 default_endpoint = self._session.settings.get("default_endpoint")338 if default_endpoint:339 my_endpoints.append(default_endpoint)340 my_endpoints.extend(self._session.settings.get("additional_endpoints", []))341 did_doc = await self.create_did_document(342 my_info,343 connection.inbound_connection_id,344 my_endpoints,345 mediation_records=list(346 filter(None, [base_mediation_record, mediation_record])347 ),348 )349 if not my_label:350 my_label = self._session.settings.get("default_label")351 request = ConnectionRequest(352 label=my_label,353 connection=ConnectionDetail(did=connection.my_did, did_doc=did_doc),354 image_url=self._session.settings.get("image_url"),355 )356 # Update connection state357 connection.request_id = request._id358 connection.state = ConnRecord.State.REQUEST.rfc160359 await connection.save(self._session, reason="Created connection request")360 # Notify mediator of keylist changes361 if keylist_updates and mediation_record:362 # send a update keylist message with new recipient keys.363 responder = self._session.inject(BaseResponder, required=False)364 await responder.send(365 keylist_updates, connection_id=mediation_record.connection_id366 )367 return request368 async def receive_request(369 self,370 request: ConnectionRequest,371 receipt: MessageReceipt,372 mediation_id: str = None,373 ) -> ConnRecord:374 """375 Receive and store a connection request.376 Args:377 request: The `ConnectionRequest` to accept378 receipt: The message receipt379 Returns:380 The new or updated `ConnRecord` instance381 """382 ConnRecord.log_state(383 "Receiving connection request",384 {"request": request},385 settings=self._session.settings,386 )387 mediation_mgr = MediationManager(self._session.profile)388 keylist_updates = None389 connection = None390 connection_key = None391 my_info = None392 # Multitenancy setup393 multitenant_mgr = self._session.inject(MultitenantManager, required=False)394 wallet_id = self._session.settings.get("wallet.id")395 wallet = self._session.inject(BaseWallet)396 # Determine what key will need to sign the response397 if receipt.recipient_did_public:398 my_info = await wallet.get_local_did(receipt.recipient_did)399 connection_key = my_info.verkey400 else:401 connection_key = receipt.recipient_verkey402 try:403 connection = await ConnRecord.retrieve_by_invitation_key(404 session=self._session,405 invitation_key=connection_key,406 their_role=ConnRecord.Role.REQUESTER.rfc160,407 )408 except StorageNotFoundError:409 raise ConnectionManagerError(410 "No invitation found for pairwise connection "411 f"in state {ConnRecord.State.INVITATION.rfc160}: "412 "a prior connection request may have updated the connection state"413 )414 invitation = None415 if connection:416 invitation = await connection.retrieve_invitation(self._session)417 connection_key = connection.invitation_key418 ConnRecord.log_state(419 "Found invitation",420 {"invitation": invitation},421 settings=self._session.settings,422 )423 if connection.is_multiuse_invitation:424 wallet = self._session.inject(BaseWallet)425 if connection.with_did is not None:426 my_info = await wallet.get_local_did(connection.with_did)427 conn = await self.find_connection(request.connection.did, my_info.did)428 if conn is not None:429 await conn.delete_record(self._session)430 else:431 my_info = await wallet.create_local_did()432 keylist_updates = await mediation_mgr.add_key(433 my_info.verkey, keylist_updates434 )435 new_connection = ConnRecord(436 invitation_key=connection_key,437 my_did=my_info.did,438 state=ConnRecord.State.INVITATION.rfc160,439 accept=connection.accept,440 their_role=connection.their_role,441 )442 await new_connection.save(443 self._session,444 reason="Received connection request from multi-use invitation DID",445 )446 # Transfer metadata from multi-use to new connection447 # Must come after save so there's an ID to associate with metadata448 for key, value in (449 await connection.metadata_get_all(self._session)450 ).items():451 await new_connection.metadata_set(self._session, key, value)452 connection = new_connection453 # Add mapping for multitenant relay454 if multitenant_mgr and wallet_id:455 await multitenant_mgr.add_key(wallet_id, my_info.verkey)456 else:457 # remove key from mediator keylist458 keylist_updates = await mediation_mgr.remove_key(459 connection_key, keylist_updates460 )461 conn_did_doc = request.connection.did_doc462 if not conn_did_doc:463 raise ConnectionManagerError(464 "No DIDDoc provided; cannot connect to public DID"465 )466 if request.connection.did != conn_did_doc.did:467 raise ConnectionManagerError(468 "Connection DID does not match DIDDoc id",469 error_code=ProblemReportReason.REQUEST_NOT_ACCEPTED,470 )471 await self.store_did_document(conn_did_doc)472 if connection:473 connection.their_label = request.label474 connection.their_did = request.connection.did475 connection.state = ConnRecord.State.REQUEST.rfc160476 await connection.save(477 self._session, reason="Received connection request from invitation"478 )479 elif not self._session.settings.get("public_invites"):480 raise ConnectionManagerError("Public invitations are not enabled")481 else: # request from public did482 my_info = await wallet.create_local_did()483 # send update-keylist message with new recipient keys484 keylist_updates = await mediation_mgr.add_key(485 my_info.verkey, keylist_updates486 )487 # Add mapping for multitenant relay488 if multitenant_mgr and wallet_id:489 await multitenant_mgr.add_key(wallet_id, my_info.verkey)490 connection = ConnRecord(491 invitation_key=connection_key,492 my_did=my_info.did,493 their_role=ConnRecord.Role.RESPONDER.rfc160,494 their_did=request.connection.did,495 their_label=request.label,496 accept=(497 ConnRecord.ACCEPT_AUTO498 if self._session.settings.get("debug.auto_accept_requests")499 else ConnRecord.ACCEPT_MANUAL500 ),501 state=ConnRecord.State.REQUEST.rfc160,502 )503 await connection.save(504 self._session, reason="Received connection request from public DID"505 )506 # Attach the connection request so it can be found and responded to507 await connection.attach_request(self._session, request)508 # Send keylist updates to mediator509 mediation_record = await mediation_record_if_id(self._session, mediation_id)510 if keylist_updates and mediation_record:511 responder = self._session.inject(BaseResponder, required=False)512 await responder.send(513 keylist_updates, connection_id=mediation_record.connection_id514 )515 if connection.accept == ConnRecord.ACCEPT_AUTO:516 response = await self.create_response(connection, mediation_id=mediation_id)517 responder = self._session.inject(BaseResponder, required=False)518 if responder:519 await responder.send_reply(520 response, connection_id=connection.connection_id521 )522 # refetch connection for accurate state523 connection = await ConnRecord.retrieve_by_id(524 self._session, connection.connection_id525 )526 else:527 self._logger.debug("Connection request will await acceptance")528 return connection529 async def create_response(530 self,531 connection: ConnRecord,532 my_endpoint: str = None,533 mediation_id: str = None,534 ) -> ConnectionResponse:535 """536 Create a connection response for a received connection request.537 Args:538 connection: The `ConnRecord` with a pending connection request539 my_endpoint: The endpoint I can be reached at540 mediation_id: The record id for mediation that contains routing_keys and541 service endpoint542 Returns:543 A tuple of the updated `ConnRecord` new `ConnectionResponse` message544 """545 ConnRecord.log_state(546 "Creating connection response",547 {"connection_id": connection.connection_id},548 settings=self._session.settings,549 )550 keylist_updates = None551 mediation_record = await mediation_record_if_id(self._session, mediation_id)552 # Multitenancy setup553 multitenant_mgr = self._session.inject(MultitenantManager, required=False)554 wallet_id = self._session.settings.get("wallet.id")555 base_mediation_record = None556 if multitenant_mgr and wallet_id:557 base_mediation_record = await multitenant_mgr.get_default_mediator()558 if ConnRecord.State.get(connection.state) not in (559 ConnRecord.State.REQUEST,560 ConnRecord.State.RESPONSE,561 ):562 raise ConnectionManagerError(563 "Connection is not in the request or response state"564 )565 request = await connection.retrieve_request(self._session)566 wallet = self._session.inject(BaseWallet)567 if connection.my_did:568 my_info = await wallet.get_local_did(connection.my_did)569 elif connection.with_did is not None:570 my_info = await wallet.get_local_did(connection.with_did)571 conn = await self.find_connection(connection.their_did, my_info.did)572 if conn is not None:573 await conn.delete_record(self._session)574 else:575 my_info = await wallet.create_local_did()576 connection.my_did = my_info.did577 mediation_mgr = MediationManager(self._session.profile)578 keylist_updates = await mediation_mgr.add_key(579 my_info.verkey, keylist_updates580 )581 # Add mapping for multitenant relay582 if multitenant_mgr and wallet_id:583 await multitenant_mgr.add_key(wallet_id, my_info.verkey)584 # Create connection response message585 if my_endpoint:586 my_endpoints = [my_endpoint]587 else:588 my_endpoints = []589 default_endpoint = self._session.settings.get("default_endpoint")590 if default_endpoint:591 my_endpoints.append(default_endpoint)592 my_endpoints.extend(self._session.settings.get("additional_endpoints", []))593 did_doc = await self.create_did_document(594 my_info,595 connection.inbound_connection_id,596 my_endpoints,597 mediation_records=list(598 filter(None, [base_mediation_record, mediation_record])599 ),600 )601 response = ConnectionResponse(602 connection=ConnectionDetail(did=my_info.did, did_doc=did_doc)603 )604 # Assign thread information605 response.assign_thread_from(request)606 response.assign_trace_from(request)607 # Sign connection field using the invitation key608 wallet = self._session.inject(BaseWallet)609 await response.sign_field("connection", connection.invitation_key, wallet)610 # Update connection state611 connection.state = ConnRecord.State.RESPONSE.rfc160612 await connection.save(613 self._session,614 reason="Created connection response",615 log_params={"response": response},616 )617 # Update mediator if necessary618 if keylist_updates and mediation_record:619 responder = self._session.inject(BaseResponder, required=False)620 await responder.send(621 keylist_updates, connection_id=mediation_record.connection_id622 )623 # TODO It's possible the mediation request sent here might arrive624 # before the connection response. This would result in an error condition625 # difficult to accomodate for without modifying handlers for trust ping626 # to ensure the connection is active.627 send_mediation_request = await connection.metadata_get(628 self._session, MediationManager.SEND_REQ_AFTER_CONNECTION629 )630 if send_mediation_request:631 mgr = MediationManager(self._session.profile)632 _record, request = await mgr.prepare_request(connection.connection_id)633 responder = self._session.inject(BaseResponder)634 await responder.send(request, connection_id=connection.connection_id)635 return response636 async def accept_response(637 self, response: ConnectionResponse, receipt: MessageReceipt638 ) -> ConnRecord:639 """640 Accept a connection response.641 Process a ConnectionResponse message by looking up642 the connection request and setting up the pairwise connection.643 Args:644 response: The `ConnectionResponse` to accept645 receipt: The message receipt646 Returns:647 The updated `ConnRecord` representing the connection648 Raises:649 ConnectionManagerError: If there is no DID associated with the650 connection response651 ConnectionManagerError: If the corresponding connection is not652 at the request or response stage653 """654 connection = None655 if response._thread:656 # identify the request by the thread ID657 try:658 connection = await ConnRecord.retrieve_by_request_id(659 self._session, response._thread_id660 )661 except StorageNotFoundError:662 pass663 if not connection and receipt.sender_did:664 # identify connection by the DID they used for us665 try:666 connection = await ConnRecord.retrieve_by_did(667 self._session, receipt.sender_did, receipt.recipient_did668 )669 except StorageNotFoundError:670 pass671 if not connection:672 raise ConnectionManagerError(673 "No corresponding connection request found",674 error_code=ProblemReportReason.RESPONSE_NOT_ACCEPTED,675 )676 if ConnRecord.State.get(connection.state) not in (677 ConnRecord.State.REQUEST,678 ConnRecord.State.RESPONSE,679 ):680 raise ConnectionManagerError(681 f"Cannot accept connection response for connection"682 f" in state: {connection.state}"683 )684 their_did = response.connection.did685 conn_did_doc = response.connection.did_doc686 if not conn_did_doc:687 raise ConnectionManagerError(688 "No DIDDoc provided; cannot connect to public DID"689 )690 if their_did != conn_did_doc.did:691 raise ConnectionManagerError("Connection DID does not match DIDDoc id")692 await self.store_did_document(conn_did_doc)693 connection.their_did = their_did694 connection.state = ConnRecord.State.RESPONSE.rfc160695 await connection.save(self._session, reason="Accepted connection response")696 send_mediation_request = await connection.metadata_get(697 self._session, MediationManager.SEND_REQ_AFTER_CONNECTION698 )699 if send_mediation_request:700 mgr = MediationManager(self._session.profile)701 _record, request = await mgr.prepare_request(connection.connection_id)702 responder = self._session.inject(BaseResponder)703 await responder.send(request, connection_id=connection.connection_id)704 return connection705 async def get_endpoints(self, conn_id: str) -> Tuple[str, str]:706 """707 Get connection endpoints.708 Args:709 conn_id: connection identifier710 Returns:711 Their endpoint for this connection712 """713 connection = await ConnRecord.retrieve_by_id(self._session, conn_id)714 wallet = self._session.inject(BaseWallet)715 my_did_info = await wallet.get_local_did(connection.my_did)716 my_endpoint = my_did_info.metadata.get(717 "endpoint",718 self._session.settings.get("default_endpoint"),719 )720 conn_targets = await self.get_connection_targets(721 connection_id=connection.connection_id,722 connection=connection,723 )724 return (my_endpoint, conn_targets[0].endpoint)725 async def create_static_connection(726 self,727 my_did: str = None,728 my_seed: str = None,729 their_did: str = None,730 their_seed: str = None,731 their_verkey: str = None,732 their_endpoint: str = None,733 their_label: str = None,734 alias: str = None,735 ) -> Tuple[DIDInfo, DIDInfo, ConnRecord]:736 """737 Register a new static connection (for use by the test suite).738 Args:739 my_did: override the DID used in the connection740 my_seed: provide a seed used to generate our DID and keys741 their_did: provide the DID used by the other party742 their_seed: provide a seed used to generate their DID and keys743 their_verkey: provide the verkey used by the other party744 their_endpoint: their URL endpoint for routing messages745 alias: an alias for this connection record746 Returns:747 Tuple: my DIDInfo, their DIDInfo, new `ConnRecord` instance748 """749 wallet = self._session.inject(BaseWallet)750 # Multitenancy setup751 multitenant_mgr = self._session.inject(MultitenantManager, required=False)752 wallet_id = self._session.settings.get("wallet.id")753 base_mediation_record = None754 # seed and DID optional755 my_info = await wallet.create_local_did(my_seed, my_did)756 # must provide their DID and verkey if the seed is not known757 if (not their_did or not their_verkey) and not their_seed:758 raise ConnectionManagerError(759 "Either a verkey or seed must be provided for the other party"760 )761 if not their_did:762 their_did = seed_to_did(their_seed)763 if not their_verkey:764 their_verkey_bin, _ = create_keypair(their_seed.encode())765 their_verkey = bytes_to_b58(their_verkey_bin)766 their_info = DIDInfo(their_did, their_verkey, {})767 # Create connection record768 connection = ConnRecord(769 invitation_mode=ConnRecord.INVITATION_MODE_STATIC,770 my_did=my_info.did,771 their_did=their_info.did,772 their_label=their_label,773 state=ConnRecord.State.COMPLETED.rfc160,774 alias=alias,775 )776 await connection.save(self._session, reason="Created new static connection")777 # Add mapping for multitenant relaying / mediation778 if multitenant_mgr and wallet_id:779 base_mediation_record = await multitenant_mgr.get_default_mediator()780 await multitenant_mgr.add_key(wallet_id, my_info.verkey)781 # Synthesize their DID doc782 did_doc = await self.create_did_document(783 their_info,784 None,785 [their_endpoint],786 mediation_records=[base_mediation_record]787 if base_mediation_record788 else None,789 )790 await self.store_did_document(did_doc)791 return my_info, their_info, connection792 async def find_connection(793 self,794 their_did: str,795 my_did: str = None,796 my_verkey: str = None,797 auto_complete=False,798 ) -> ConnRecord:799 """800 Look up existing connection information for a sender verkey.801 Args:802 their_did: Their DID803 my_did: My DID804 my_verkey: My verkey805 auto_complete: Should this connection automatically be promoted to active806 Returns:807 The located `ConnRecord`, if any808 """809 # self._log_state(810 # "Finding connection",811 # {"their_did": their_did, "my_did": my_did, "my_verkey": my_verkey},812 # )813 connection = None814 if their_did:815 try:816 connection = await ConnRecord.retrieve_by_did(817 self._session, their_did, my_did818 )819 except StorageNotFoundError:820 pass821 if (822 connection823 and ConnRecord.State.get(connection.state) is ConnRecord.State.RESPONSE824 and auto_complete825 ):826 connection.state = ConnRecord.State.COMPLETED.rfc160827 await connection.save(self._session, reason="Connection promoted to active")828 if not connection and my_verkey:829 try:830 connection = await ConnRecord.retrieve_by_invitation_key(831 self._session,832 my_verkey,833 their_role=ConnRecord.Role.REQUESTER.rfc160,834 )835 except StorageError:836 pass837 return connection838 async def find_inbound_connection(self, receipt: MessageReceipt) -> ConnRecord:839 """840 Deserialize an incoming message and further populate the request context.841 Args:842 receipt: The message receipt843 Returns:844 The `ConnRecord` associated with the expanded message, if any845 """846 cache_key = None847 connection = None848 resolved = False849 if receipt.sender_verkey and receipt.recipient_verkey:850 cache_key = (851 f"connection_by_verkey::{receipt.sender_verkey}"852 f"::{receipt.recipient_verkey}"853 )854 cache = self._session.inject(BaseCache, required=False)855 if cache:856 async with cache.acquire(cache_key) as entry:857 if entry.result:858 cached = entry.result859 receipt.sender_did = cached["sender_did"]860 receipt.recipient_did_public = cached["recipient_did_public"]861 receipt.recipient_did = cached["recipient_did"]862 try:863 connection = await ConnRecord.retrieve_by_id(864 self._session, cached["id"]865 )866 except:867 connection = await self.resolve_inbound_connection(receipt)868 if connection:869 cache.release(cache_key)870 resolved = True871 else:872 connection = await self.resolve_inbound_connection(receipt)873 if connection:874 cache_val = {875 "id": connection.connection_id,876 "sender_did": receipt.sender_did,877 "recipient_did": receipt.recipient_did,878 "recipient_did_public": receipt.recipient_did_public,879 }880 await entry.set_result(cache_val, 3600)881 resolved = True882 if not connection and not resolved:883 connection = await self.resolve_inbound_connection(receipt)884 return connection885 async def resolve_inbound_connection(self, receipt: MessageReceipt) -> ConnRecord:886 """887 Populate the receipt DID information and find the related `ConnRecord`.888 Args:889 receipt: The message receipt890 Returns:891 The `ConnRecord` associated with the expanded message, if any892 """893 if receipt.sender_verkey:894 try:895 receipt.sender_did = await self.find_did_for_key(receipt.sender_verkey)896 except StorageNotFoundError:897 self._logger.warning(898 "No corresponding DID found for sender verkey: %s",899 receipt.sender_verkey,900 )901 if receipt.recipient_verkey:902 try:903 wallet = self._session.inject(BaseWallet)904 my_info = await wallet.get_local_did_for_verkey(905 receipt.recipient_verkey906 )907 receipt.recipient_did = my_info.did908 if "public" in my_info.metadata and my_info.metadata["public"] is True:909 receipt.recipient_did_public = True910 except InjectionError:911 self._logger.warning(912 "Cannot resolve recipient verkey, no wallet defined by "913 "context: %s",914 receipt.recipient_verkey,915 )916 except WalletNotFoundError:917 self._logger.warning(918 "No corresponding DID found for recipient verkey: %s",919 receipt.recipient_verkey,920 )921 return await self.find_connection(922 receipt.sender_did, receipt.recipient_did, receipt.recipient_verkey, True923 )924 async def get_connection_targets(925 self, *, connection_id: str = None, connection: ConnRecord = None926 ):927 """Create a connection target from a `ConnRecord`.928 Args:929 connection_id: The connection ID to search for930 connection: The connection record itself, if already available931 """932 if not connection_id:933 connection_id = connection.connection_id934 cache = self._session.inject(BaseCache, required=False)935 cache_key = f"connection_target::{connection_id}"936 if cache:937 async with cache.acquire(cache_key) as entry:938 if entry.result:939 targets = [940 ConnectionTarget.deserialize(row) for row in entry.result941 ]942 else:943 if not connection:944 connection = await ConnRecord.retrieve_by_id(945 self._session, connection_id946 )947 targets = await self.fetch_connection_targets(connection)948 await entry.set_result([row.serialize() for row in targets], 3600)949 else:950 targets = await self.fetch_connection_targets(connection)951 return targets952 async def establish_inbound(953 self,954 connection: ConnRecord,955 inbound_connection_id: str,956 outbound_handler: Coroutine,957 ) -> str:958 """Assign the inbound routing connection for a connection record.959 Returns: the current routing state (request or done)960 """961 # The connection must have a verkey, but in the case of a received962 # invitation we might not have created one yet963 wallet = self._session.inject(BaseWallet)964 if connection.my_did:965 my_info = await wallet.get_local_did(connection.my_did)966 else:967 # Create new DID for connection968 my_info = await wallet.create_local_did()969 connection.my_did = my_info.did970 try:971 router = await ConnRecord.retrieve_by_id(972 self._session, inbound_connection_id973 )974 except StorageNotFoundError:975 raise ConnectionManagerError(976 f"Routing connection not found: {inbound_connection_id}"977 )978 if not router.is_ready:979 raise ConnectionManagerError(980 f"Routing connection is not ready: {inbound_connection_id}"981 )982 connection.inbound_connection_id = inbound_connection_id983 route_mgr = RoutingManager(self._session.profile)984 await route_mgr.send_create_route(985 inbound_connection_id, my_info.verkey, outbound_handler986 )987 connection.routing_state = ConnRecord.ROUTING_STATE_REQUEST988 await connection.save(self._session)989 return connection.routing_state990 async def update_inbound(991 self, inbound_connection_id: str, recip_verkey: str, routing_state: str992 ):993 """Activate connections once a route has been established.994 Looks up pending connections associated with the inbound routing995 connection and marks the routing as complete.996 """997 conns = await ConnRecord.query(998 self._session, {"inbound_connection_id": inbound_connection_id}999 )1000 wallet = self._session.inject(BaseWallet)1001 for connection in conns:1002 # check the recipient key1003 if not connection.my_did:1004 continue1005 conn_info = await wallet.get_local_did(connection.my_did)1006 if conn_info.verkey == recip_verkey:1007 connection.routing_state = routing_state...
session.py
Source:session.py
...109 remote_ua_version = user_agent_ua_major110 return remote_hw_family, remote_hw_model, remote_os_family, remote_os_version, remote_ua_family, remote_ua_version111 except:112 return '', '', '', '', user_agent, ''113 def _new_session(self):114 from ..models import Session115 address = self.request.META.get('REMOTE_ADDR', '')116 geoname = self.request.META.get(self._geoname_header_name, '')117 user_agent = self.request.META.get('HTTP_USER_AGENT', '')118 remote_hw_family, remote_hw_model, remote_os_family, remote_os_version, remote_ua_family, remote_ua_version = self._parse_user_agent(user_agent)119 self._session = Session(120 previous_session=self._session,121 remote_address=address,122 remote_geoname=geoname,123 remote_hw_family=remote_hw_family,124 remote_hw_model=remote_hw_model,125 remote_os_family=remote_os_family,126 remote_os_version=remote_os_version,127 remote_ua_family=remote_ua_family,128 remote_ua_version=remote_ua_version)129 def _get_session(self, now, uuid):130 from ..models import Session131 address = self.request.META.get('REMOTE_ADDR', '')132 geoname = self.request.META.get(self._geoname_header_name, '')133 user_agent = self.request.META.get('HTTP_USER_AGENT', '')134 remote_hw_family, remote_hw_model, remote_os_family, remote_os_version, remote_ua_family, remote_ua_version = self._parse_user_agent(user_agent)135 try:136 self._session = Session.objects.get(137 uuid=uuid,138 valid_from__lte=now,139 valid_till__gte=now,140 remote_geoname=geoname,141 remote_hw_family=remote_hw_family,142 remote_os_family=remote_os_family,143 remote_ua_family=remote_ua_family)144 self._session.remote_hw_model = remote_hw_model145 self._session.remote_os_version = remote_os_version146 self._session.remote_ua_version = remote_ua_version147 except Session.DoesNotExist:148 self._session = Session(149 previous_session=self._session,150 remote_address=address,151 remote_geoname=geoname,152 remote_hw_family=remote_hw_family,153 remote_hw_model=remote_hw_model,154 remote_os_family=remote_os_family,155 remote_os_version=remote_os_version,156 remote_ua_family=remote_ua_family,157 remote_ua_version=remote_ua_version)158 def __init__(self, request):159 from collections import OrderedDict160 from django.conf import settings161 from ..models import Principal162 self._session = None163 self._variables = OrderedDict()164 self._geoname_header_name = getattr(settings, 'TALOS_GEONAME_HEADER', '')165 self.request = request166 self.request.session = self167 self.principal = Principal.objects.get(id=0)168 self.principal._load_authentication_context([])169 def init(self):170 self._new_session()171 self.request.principal = self.principal172 def load(self, uuid):173 from ..models import _tznow174 from ..models import Principal175 from json import loads176 now = _tznow()177 self._get_session(now, uuid)178 self._variables = loads(self._session.variables, cls=CustomJSONDecoder) if self._session.variables else {}179 if self._session.principal:180 self.principal = self._session.principal181 self.principal._inject_authentication_context(182 self._session.evidences,183 self._session.roles,184 self._session.privileges,185 self._session.model_actions)186 authentication_period = (now - self._session.valid_from).total_seconds()187 valid_evidences = []188 for evidence in self.principal._evidences_effective.values():189 if authentication_period < evidence.expiration_period:190 valid_evidences.append(evidence)191 if len(valid_evidences) != len(self.principal._evidences_effective) or len(valid_evidences) == 0:192 if len(valid_evidences) == 0:193 self.principal = Principal.objects.get(id=0)194 self.principal._load_authentication_context(valid_evidences)195 self._session.valid_from = now196 self.request.principal = self.principal197 def save(self):198 from ..models import Principal199 from django.utils.functional import LazyObject200 from json import dumps201 prev_evidences = self._session.evidences202 prev_roles = self._session.roles203 prev_privileges = self._session.privileges204 prev_model_actions = self._session.model_actions205 prev_variables = self._session.variables206 prev_principal = self._session.principal207 self._session.variables = dumps(self._variables, cls=CustomJSONEncoder)208 if self.request.principal and self.request.principal.is_authenticated:209 if issubclass(type(self.request.principal), LazyObject):210 self.request.principal._setup()211 if type(self.request.principal._wrapped) == Principal:212 self._session.principal = self.request.principal._wrapped213 elif type(self.request.principal) == Principal:214 self._session.principal = self.request.principal215 if self._session.principal:216 self._session.evidences, \217 self._session.roles, \218 self._session.privileges, \219 self._session.model_actions = self._session.principal._extract_authentication_context()220 else:221 self._session.evidences = ''222 self._session.roles = ''223 self._session.privileges = ''224 self._session.model_actions = ''225 if (226 (self._session.evidences != prev_evidences) or227 (self._session.roles != prev_roles) or228 (self._session.privileges != prev_privileges) or229 (self._session.model_actions != prev_model_actions) or230 (self._session.variables != prev_variables) or231 (self._session.principal != prev_principal)):232 self._session.save()233 def get(self, key, default=None):234 return self._variables.get(key, default)235 def pop(self, key, default=None):236 if key in self._variables:237 return self._variables.pop(key, default)238 return None239 def setdefault(self, key, default=None):240 if key in self._variables:241 return self._variables[key]242 else:243 self._variables[key] = default244 return default245 def set_test_cookie(self):246 self[self.TEST_COOKIE_NAME] = self.TEST_COOKIE_VALUE247 def test_cookie_worked(self):248 return self.get(self.TEST_COOKIE_NAME) == self.TEST_COOKIE_VALUE249 def delete_test_cookie(self):250 del self[self.TEST_COOKIE_NAME]251 def update(self, other):252 self._variables.update(other)253 def has_key(self, key):254 return key in self.__store255 def keys(self):256 return self._variables.keys()257 def values(self):258 return self._variables.values()259 def items(self):260 return self._variables.items()261 def iterkeys(self):262 return self._variables.iterkeys()263 def itervalues(self):264 return self._variables.itervalues()265 def iteritems(self):266 return self._variables.iteritems()267 def clear(self):268 from collections import OrderedDict269 self._variables = OrderedDict()270 def is_empty(self):271 try:272 return len(self._variables) == 0273 except AttributeError:274 return True275 def flush(self):276 from ..models import _tznow277 from ..models import Principal278 if self._session:279 self._session.valid_till = _tznow()280 self._session.save()281 self._new_session()282 self.principal = Principal.objects.get(id=0)283 self.principal._load_authentication_context([])284 self.request.principal = self.principal285 def cycle_key(self):286 if self._session:287 from uuid import uuid4288 self._session.uuid = uuid4()289 def __contains__(self, key):290 return key in self._variables291 def __getitem__(self, key):292 return self._variables[key]293 def __setitem__(self, key, value):294 if key not in ('_auth_user_backend', '_auth_user_hash', '_auth_user_id'):295 self._variables[key] = value...
test_overthere_session.py
Source:test_overthere_session.py
1#2# Copyright 2020 XEBIALABS3#4# Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:5#6# The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.7#8# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.9#10from __future__ import with_statement11import xlnoserunner12import os13from overtherepy import LocalConnectionOptions, OverthereHost, OverthereHostSession, OverthereSessionLogger, Diff14from com.xebialabs.overthere import OperatingSystemFamily15from java.io import File16from java.lang import Thread17from nose.tools import ok_, eq_18class TestOverthereSession(object):19 # preparing to test20 def setUp(self):21 self._linuxhost = OverthereHost(LocalConnectionOptions(os=OperatingSystemFamily.UNIX))22 self._session = OverthereHostSession(self._linuxhost)23 self._session.logger = OverthereSessionLogger(capture=True)24 # ending the test25 def tearDown(self):26 self._session.close_conn()27 def _clear_logs(self):28 self._session.logger.output_lines = []29 self._session.logger.error_lines = []30 def assert_in_list(self, list, expect):31 r = [s for s in list if str(s) == str(expect)]32 eq_(len(r), 1, "expect [%s] to be in list [%s]" % (expect, str(list)))33 def test_not_windows_host(self):34 ok_(not self._session.is_windows())35 def test_work_dir_cleanup(self):36 workdir = self._session.work_dir()37 ok_(workdir.exists())38 workdir_path = workdir.path39 self._session.close_conn()40 ok_(not os.path.exists(workdir_path))41 def test_read_write_file(self):42 f = self._session.upload_text_content_to_work_dir("some text", "my.txt")43 text = self._session.read_file(f.path)44 eq_("some text", text)45 def test_upload_executable_file(self):46 f = self._session.upload_text_content_to_work_dir("#!/bin/sh\necho hello", "my.sh", executable=True)47 r = self._session.execute(f.path)48 self.assert_in_list(r.stdout, "hello")49 def test_upload_classpath_resource(self):50 f = self._session.upload_classpath_resource_to_work_dir("testfiles/singleline.txt")51 text = self._session.read_file(f.path)52 eq_("some text 1", text)53 def test_read_lines(self):54 f = self._session.upload_classpath_resource_to_work_dir("testfiles/multiline.txt")55 lines = self._session.read_file_lines(f.path)56 eq_(3, len(lines))57 def test_execute(self):58 f = self._session.upload_classpath_resource_to_work_dir("testfiles/echo.sh", executable=True)59 response = self._session.execute([f.path, "ping"])60 eq_(response['rc'], 0)61 eq_(response['stdout'][0], "Hi ping")62 def test_execute_automatic_system_exit_on_failure(self):63 success = False64 try:65 f = self._session.upload_classpath_resource_to_work_dir("testfiles/echo.sh", executable=True)66 self._session.execute([f.path, "ping", "1"])67 except SystemExit:68 success = True69 pass70 eq_(success, True)71 def test_execute_check_success_turned_off(self):72 f = self._session.upload_classpath_resource_to_work_dir("testfiles/echo.sh", executable=True)73 response = self._session.execute("%s ping 1" % f.path, check_success=False)74 eq_(response.rc, 1)75 eq_(response['stdout'][0], "Hi ping")76 eq_(response['stdout'][1], "Exiting with 1")77 def test_execute_check_success_turned_off_and_spaces_in_parameters(self):78 f = self._session.upload_classpath_resource_to_work_dir("testfiles/echo.sh", executable=True)79 response = self._session.execute("%s 'ping pong' 2" % f.path, check_success=False)80 eq_(response.rc, 2)81 eq_(response['stdout'][0], "Hi ping pong")82 eq_(response['stdout'][1], "Exiting with 2")83 def test_with_support(self):84 s = OverthereHostSession(self._linuxhost)85 with s:86 work_dir = s.work_dir().path87 eq_(os.path.exists(work_dir), True)88 eq_(os.path.exists(work_dir), False)89 def _local_file(self, resource):90 url = Thread.currentThread().contextClassLoader.getResource(resource)91 return self._session.local_file(File(url.toURI()))92 def test_mkdirs_on_dir_copy(self):93 target = self._session.work_dir_file("some/path")94 source = self._local_file("testfiles")95 eq_(os.path.exists(target.path), False)96 self._session.copy_to(source, target)97 eq_(os.path.exists(target.path), True)98 eq_(os.path.exists(target.path + "/echo.sh"), True)99 log_info = self._session.logger.output_lines100 eq_(len(log_info), 2)101 eq_(log_info[0], "Creating path %s" % target.path)102 eq_(log_info[1], "Copying %s to %s" % (source.path, target.path))103 def test_mkdirs_on_file_copy(self):104 target = self._session.work_dir_file("some/path/some.txt")105 source = self._local_file("testfiles/echo.sh")106 eq_(os.path.exists(target.path), False)107 self._session.copy_to(source, target)108 eq_(os.path.exists(target.path), True)109 log_info = self._session.logger.output_lines110 eq_(len(log_info), 2)111 eq_(log_info[0], "Creating path %s" % target.parentFile.path)112 eq_(log_info[1], "Copying %s to %s" % (source.path, target.path))113 def test_mkdirs_on_turned_off_on_copy(self):114 target = self._session.work_dir_file("some/path")115 source = self._local_file("testfiles")116 success = False117 try:118 self._session.copy_to(source, target, mkdirs=False)119 except:120 success = True121 eq_(success, True)122 def test_delete_from_when_target_does_not_exit(self):123 target = self._session.work_dir_file("some/path")124 source = None125 self._session.delete_from(source, target)126 log_info = self._session.logger.output_lines127 eq_(len(log_info), 1)128 eq_(log_info[0], "Target [%s] does not exist. No deletion to be performed" % target.path)129 def test_delete_from_file_target(self):130 target = self._session.work_dir_file("some/path/test.txt")131 self._session.copy_text_to_file("xxx", target)132 eq_(os.path.exists(target.path), True)133 source = None134 self._session.delete_from(source, target)135 eq_(os.path.exists(target.path), False)136 log_info = self._session.logger.output_lines137 eq_(len(log_info), 2)138 self.assert_in_list(log_info, "Deleting [%s]" % target.path)139 def test_delete_from_folder_target(self):140 target = self._session.work_dir_file("some/path")141 source = self._local_file("testfiles")142 self._session.copy_to(source, target)143 eq_(os.path.exists(target.path), True)144 self._clear_logs()145 self._session.delete_from(source, target)146 eq_(os.path.exists(target.path), False)147 log_info = self._session.logger.output_lines148 eq_(len(log_info), 5)149 self.assert_in_list(log_info, "Recursively deleting directory [%s/asubdir]" % target.path)150 self.assert_in_list(log_info, "Deleting file [%s/echo.sh]" % target.path)151 self.assert_in_list(log_info, "Deleting file [%s/multiline.txt]" % target.path)152 self.assert_in_list(log_info, "Deleting file [%s/singleline.txt]" % target.path)153 self.assert_in_list(log_info, "Deleting directory [%s]" % target.path)154 def test_delete_from_with_external_resources(self):155 target = self._session.work_dir_file("some/path")156 source = self._local_file("testfiles")157 self._session.copy_to(source, target)158 eq_(os.path.exists(target.path), True)159 external_resource = self._session.work_dir_file("some/path/ext.txt")160 self._session.copy_text_to_file("xxx", external_resource)161 eq_(os.path.exists(external_resource.path), True)162 self._clear_logs()163 self._session.delete_from(source, target)164 eq_(os.path.exists(target.path), True)165 log_info = self._session.logger.output_lines166 eq_(len(log_info), 5)167 self.assert_in_list(log_info, "Recursively deleting directory [%s/asubdir]" % target.path)168 self.assert_in_list(log_info, "Deleting file [%s/echo.sh]" % target.path)169 self.assert_in_list(log_info, "Deleting file [%s/multiline.txt]" % target.path)170 self.assert_in_list(log_info, "Deleting file [%s/singleline.txt]" % target.path)171 self.assert_in_list(log_info, "Target directory [%s] is not shared, but still has content from an external source. Will not delete" % target.path)172 def test_copy_diff(self):173 old = self._local_file("directorycompare/old")174 new = self._local_file("directorycompare/new")175 deployed_old = self._session.work_dir_file("olddeployed")176 self._session.copy_to(old, deployed_old)177 eq_(os.path.exists(deployed_old.path), True)178 self._clear_logs()179 diff = Diff.calculate_diff(old, new)180 self._session.copy_diff(deployed_old.path, diff)181 log_info = self._session.logger.output_lines182 eq_(len(log_info), 17)183 self.assert_in_list(log_info, "3 files to be removed.")184 self.assert_in_list(log_info, "3 new files to be copied.")185 self.assert_in_list(log_info, "2 modified files to be copied.")186 self.assert_in_list(log_info, "Start removal of files...")187 self.assert_in_list(log_info, "Removing file %s/removefile.txt" % deployed_old.path)188 self.assert_in_list(log_info, "Removing directory %s/subdirremove" % deployed_old.path)189 self.assert_in_list(log_info, "Removing file %s/subdirboth/removefile.txt" % deployed_old.path)190 self.assert_in_list(log_info, "Removal of files done.")191 self.assert_in_list(log_info, "Start copying of new files...")192 self.assert_in_list(log_info, "Copying directory %s/subdirnew" % deployed_old.path)193 self.assert_in_list(log_info, "Copying file %s/newfile.txt" % deployed_old.path)194 self.assert_in_list(log_info, "Copying file %s/subdirboth/newfile.txt" % deployed_old.path)195 self.assert_in_list(log_info, "Copying of new files done.")196 self.assert_in_list(log_info, "Start copying of modified files...")197 self.assert_in_list(log_info, "Updating file %s/changedfile.txt" % deployed_old.path)198 self.assert_in_list(log_info, "Updating file %s/subdirboth/changedfile.txt" % deployed_old.path)199 self.assert_in_list(log_info, "Copying of modified files done.")200 def test_execution_ctx_logging(self):201 class ExecutionContext(object):202 def __init__(self):203 self.output_success = False204 self.error_success = False205 def logOutput(self, msg):206 self.output_success = True207 def logError(self, msg):208 self.error_success = True209 exec_ctx = ExecutionContext()210 session = OverthereHostSession(self._linuxhost, execution_context=exec_ctx)211 session.logger.info("Check")212 eq_(exec_ctx.output_success, True)213 session.logger.error("Check")...
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 Molotov 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?
