KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use switch_context method in pytest-cov

Best Python code snippet using pytest-cov

old_test_tasks.py

Source:old_test_tasks.py

...43    hand_maker_commitment_func_mock = mocker.patch.object(SwapCommitment, 'hand_maker_commitment_msg', autospect=True)44    maker_commitment_task = CommitmentTask(factory.swaps, factory.refund_queue, factory.message_queue,45                                           message_broker, commitment_service_address)46    maker_commitment_task.start()47    switch_context()48    message_broker.send(commitment_service_address, maker_commitment_msg_signed)49    switch_context()50    assert len(swaps) == 151    hand_maker_commitment_func_mock.assert_called_once_with(maker_commitment_msg_signed)52def test_maker_commitment_task_fail(mocker, swaps, factory, maker_commitment_msg_signed, message_broker,53                                    commitment_service_account):54    commitment_service_address = commitment_service_account.address55    offer_id = maker_commitment_msg_signed.offer_id56    hand_maker_commitment_func_mock = mocker.patch.object(SwapCommitment, 'hand_maker_commitment_msg', autospect=True)57    # TODO generalise the filled swaps in fixture58    assert len(swaps) == 059    # set a non-finished Swap a the offer_id60    existing_swap = factory.make_swap(offer_id)61    assert len(swaps) == 162    maker_commitment_task = CommitmentTask(factory.swaps, factory.refund_queue, factory.message_queue,63                                           message_broker, commitment_service_address)64    maker_commitment_task.start()65    # try to send a commitment at the existing offer_id66    message_broker.send(commitment_service_address, maker_commitment_msg_signed)67    switch_context()68    # the only swap registered should still be the existing swap69    assert len(swaps) == 170    assert swaps[offer_id] is existing_swap71    assert not hand_maker_commitment_func_mock.called72def test_taker_commitment_task(mocker, swaps, factory, taker_commitment_msg_signed, commitment_service_account, message_broker,73                               ):74    commitment_service_address = commitment_service_account.address75    offer_id = taker_commitment_msg_signed.offer_id76    hand_taker_commitment_func_mock = mocker.patch.object(SwapCommitment, 'hand_taker_commitment_msg', autospect=True)77    assert len(swaps) == 078    swap = factory.make_swap(offer_id)79    assert len(swaps) == 180    taker_commitment_task = CancellationRequestTask(swaps, message_broker, commitment_service_address)81    taker_commitment_task.start()82    switch_context()83    # try to send a commitment at the existing offer_id84    message_broker.send(commitment_service_address, taker_commitment_msg_signed)85    switch_context()86    # the only swap registered should still be the existing swap87    assert len(swaps) == 188    assert swaps[offer_id] is swap89    hand_taker_commitment_func_mock.assert_called_once_with(taker_commitment_msg_signed)90def test_swap_execution_task(mocker, swaps, factory, swap_execution_msg, message_broker, commitment_service_account,91                             maker_account):92    commitment_service_address = commitment_service_account.address93    swap_execution_msg.sign(maker_account.privatekey)94    offer_id = swap_execution_msg.offer_id95    hand_swap_execution_func_mock = mocker.patch.object(SwapCommitment, 'hand_swap_execution_msg', autospect=True)96    assert len(swaps) == 097    swap = factory.make_swap(offer_id)98    assert len(swaps) == 199    swap_execution_task = SwapExecutionTask(swaps, message_broker, commitment_service_address)100    swap_execution_task.start()101    switch_context()102    # try to send a commitment at the existing offer_id103    message_broker.send(commitment_service_address, swap_execution_msg)104    switch_context()105    # the only swap registered should still be the existing swap106    assert len(swaps) == 1107    assert swaps[offer_id] is swap108    hand_swap_execution_func_mock.assert_called_once_with(swap_execution_msg)109def test_transfer_received_task(mocker, swaps, factory, commitment_service_trader_client, commitment_service_account,110                                maker_account, maker_trader_client):111    commitment_service_address = commitment_service_account.address112    offer_id = 123113    assert len(swaps) == 0114    swap = factory.make_swap(offer_id)115    assert len(swaps) == 1116    hand_transfer_receipt_func_mock = mocker.patch.object(SwapCommitment, 'hand_transfer_receipt', autospect=True)117    # set static time118    time_func = mocker.patch.object(timestamp, 'time', autospect=True)119    time_func.return_value = 1120    maker_commitment_task = TransferReceivedTask(swaps, commitment_service_trader_client)121    maker_commitment_task.start()122    switch_context()123    maker_trader_client.transfer(commitment_service_address, 1, identifier=offer_id)124    switch_context()125    hand_transfer_receipt_func_mock.assert_called_once()126    args, kwargs = hand_transfer_receipt_func_mock.call_args127    transfer_receipt = args[0]128    assert transfer_receipt.sender == maker_account.address129    assert transfer_receipt.amount == 1130    assert transfer_receipt.identifier == offer_id131    assert transfer_receipt.timestamp == 1132def test_refund_task(mocker, maker_account, refund_queue, commitment_service_trader_client):133    sender_address = maker_account.address134    receipt = TransferReceipt(sender=sender_address, amount=1, identifier=123, received_timestamp=1)135    refund = Refund(receipt, priority=1, claim_fee=False)136    trader_client_transfer_mock = mocker.patch.object(TraderClientMock, 'transfer', autospect=True, return_value=True)137    refund_task = RefundTask(commitment_service_trader_client, refund_queue, fee_rate=0.1)138    refund_task.start()139    switch_context()140    refund_queue.put(refund)141    assert len(refund_queue) == 1142    switch_context()143    assert len(refund_queue) == 0144    trader_client_transfer_mock.assert_called_once_with(sender_address, 1, 123)145def test_refund_task_claim_fee(mocker, maker_account, refund_queue, commitment_service_trader_client):146    transfer_amount = 1147    fee_rate = 0.1148    sender_address = maker_account.address149    offer_id = 123150    receipt = TransferReceipt(sender=sender_address, amount=transfer_amount, identifier=offer_id,151                              received_timestamp=1)152    refund = Refund(receipt, priority=1, claim_fee=True)153    trader_client_transfer_mock = mocker.patch.object(TraderClientMock, 'transfer', autospect=True, return_value=True)154    refund_task = RefundTask(commitment_service_trader_client, refund_queue, fee_rate=fee_rate)155    refund_task.start()156    switch_context()157    refund_queue.put(refund)158    switch_context()159    expected_amount = transfer_amount - (transfer_amount * fee_rate)160    trader_client_transfer_mock.assert_called_once_with(sender_address, expected_amount, 123)161def test_message_sender_task_send(mocker, message_broker, message_queue, maker_account, commitment_service_account ):162    def sign_func(msg):163        msg.sign(commitment_service_account.privatekey)164    # use some arbitrary message and receiver:165    message = messages.SwapCompleted(123, 1)166    receiver = maker_account.address167    message_broker_send_mock = mocker.patch.object(MessageBroker, 'send', autospect=True)168    message_sender_task = MessageSenderTask(message_broker, message_queue, sign_func=sign_func)169    message_sender_task.start()170    switch_context()171    data = (message, receiver)172    message_queue.put(data)173    assert len(message_queue) == 1174    switch_context()175    assert len(message_queue) == 0176    message_broker_send_mock.assert_called_once_with(topic=receiver, message=message)177def test_message_sender_task_broadcast(mocker, message_broker, message_queue, commitment_service_account):178    def sign_func(msg):179        msg.sign(commitment_service_account.privatekey)180    # use some arbitrary message and receiver:181    message = messages.SwapCompleted(123, 1)182    receiver = None183    message_broker_broadcast_mock = mocker.patch.object(MessageBroker, 'broadcast', autospect=True)184    message_sender_task = MessageSenderTask(message_broker, message_queue, sign_func=sign_func)185    message_sender_task.start()186    switch_context()187    data = (message, receiver)188    message_queue.put(data)189    assert len(message_queue) == 1190    switch_context()191    assert len(message_queue) == 0...

alert_utils.py

Source:alert_utils.py

...17    dashboard_json = {}18    if grafana_org_utils.get_current_org_name()["name"] == \19            constants.ALERT_ORG:20        dashboard_json = dashboard_utils.get_dashboard(slug)21    elif switch_context(constants.ALERT_ORG):22        dashboard_json = dashboard_utils.get_dashboard(slug)23        # return to main org24        switch_context(constants.MAIN_ORG)25    return dashboard_json26def switch_context(org_name):27    alert_org_id = grafana_org_utils.get_org_id(28        org_name29    )30    switched = grafana_org_utils.switch_context(json.loads(31        alert_org_id)["id"]32    )33    return switched34def delete_alert_dashboard(dashboard_name):35    slug = "alerts-" + str(dashboard_name)[:-1] + "-dashboard"36    dashboard_json = {}37    if grafana_org_utils.get_current_org_name()["name"] == \38            constants.ALERT_ORG:39        dashboard_json = dashboard_utils.delete_dashboard(slug)40    elif switch_context(constants.ALERT_ORG):41        dashboard_json = dashboard_utils.delete_dashboard(slug)42        # return to main org43        switch_context(constants.MAIN_ORG)44    return dashboard_json45def post_dashboard(alert_dashboard):46    resp = None47    if grafana_org_utils.get_current_org_name()["name"] == \48            constants.ALERT_ORG:49        resp = dashboard_utils._post_dashboard(alert_dashboard)50    elif switch_context(constants.ALERT_ORG):51        resp = dashboard_utils._post_dashboard(alert_dashboard)52        # return to main org53        switch_context(constants.MAIN_ORG)54    return resp55def get_alert(alert_id):56    resp = None57    if grafana_org_utils.get_current_org_name()["name"] == \58            constants.ALERT_ORG:59        resp = dashboard_utils.get_alert(alert_id)60    elif switch_context(constants.ALERT_ORG):61        resp = dashboard_utils.get_alert(alert_id)62        # return to main org63        switch_context(constants.MAIN_ORG)64    return resp.json()65def remove_row(alert_dashboard, integration_id, resource_type, resource_name):66    rows = alert_dashboard["dashboard"]["rows"]67    new_rows = []68    flag = True69    for row in rows:70        for target in row["panels"][0]["targets"]:71            if resource_type == "bricks":72                result = parse_target(73                    target['target'],74                    constants.BRICK_TEMPLATE75                )76                hostname = resource_name.split(":")[0].split(77                    "|")[1].replace(".", "_")...

plugin.py

Source:plugin.py

...24    def pytest_collection(self):25        """26        Perform the collection phase for the given session.27        """28        with self.switch_context("collection"):29            yield30    @pytest.hookimpl(hookwrapper=True, tryfirst=True)31    def pytest_runtest_logstart(self, nodeid):32        """33        Called at the start of running the runtest protocol for a single item.34        """35        with self.switch_context("{}|start".format(nodeid)):36            yield37    @pytest.hookimpl(hookwrapper=True, tryfirst=True)38    def pytest_runtest_setup(self, item):39        """40        Called to perform the setup phase for a test item.41        """42        with self.switch_context("{}|setup".format(item.nodeid)):43            yield44    @pytest.hookimpl(hookwrapper=True, tryfirst=True)45    def pytest_runtest_call(self, item):46        """47        Called to run the test for test item (the call phase).48        """49        with self.switch_context("{}|call".format(item.nodeid)):50            yield51    @pytest.hookimpl(hookwrapper=True, tryfirst=True)52    def pytest_runtest_teardown(self, item):53        """54        Called to perform the teardown phase for a test item.55        """56        with self.switch_context("{}|teardown".format(item.nodeid)):57            yield58    @pytest.hookimpl(hookwrapper=True, tryfirst=True)59    def pytest_runtest_logfinish(self, nodeid):60        """61        Called at the end of running the runtest protocol for a single item.62        """63        with self.switch_context("{}|finish".format(nodeid)):64            yield65    @contextmanager66    def switch_context(self, context):67        """68        Update the dynamic context file69        """70        log.debug("Switching coverage context to: %s", context)71        try:72            with atomic_write(self.context_file_path, overwrite=True) as wfh:73                wfh.write(context)74            yield75        finally:76            with atomic_write(self.context_file_path, overwrite=True) as wfh:77                wfh.write("")78@pytest.hookimpl(trylast=True)79def pytest_configure(config):80    """...

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.