How to use test_delay method in Molotov

Best Python code snippet using molotov_python

runs.py

Source:runs.py

...1164        sweep_R[j - smallest_delay_to_test] = tmp[0][0]1165        sweep_Q[j - smallest_delay_to_test] = tmp[2][0]1166    print(sweep_R)1167    print(sweep_Q)1168def run_sweep_test_delay(num_index_cases, trace_delay=0,1169              trace_false_negative=0.0, cases_contacted=0.0,1170              base_reduction=0.0, early_release_threshold=1.e-6, seed=0):1171    smallest_delay_to_test = 01172    largest_delay_to_test = 51173    sweep_R = np.zeros(largest_delay_to_test - smallest_delay_to_test + 1)1174    sweep_Q = np.zeros(largest_delay_to_test - smallest_delay_to_test + 1)1175    for j in range(smallest_delay_to_test, largest_delay_to_test + 1):1176        tmp = build_contact_trees.contact_tracing(num_index_cases, trace=True,1177                              trace_delay=trace_delay,1178                              test_delay=test_delay,1179                              trace_false_negative=trace_false_negative,1180                              cases_contacted=cases_contacted,1181                              ttq=False,1182                              base_reduction=base_reduction,...

test_output_loop.py

Source:test_output_loop.py

1#!/usr/bin/env python3.72# -*- coding: utf-8 -*-3"""4TFC - Onion-routed, endpoint secure messaging system5Copyright (C) 2013-2019  Markus Ottela6This file is part of TFC.7TFC is free software: you can redistribute it and/or modify it under the terms8of the GNU General Public License as published by the Free Software Foundation,9either version 3 of the License, or (at your option) any later version.10TFC is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;11without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR12PURPOSE. See the GNU General Public License for more details.13You should have received a copy of the GNU General Public License14along with TFC. If not, see <https://www.gnu.org/licenses/>.15"""16import base6417import datetime18import threading19import time20import unittest21from typing        import Tuple22from unittest      import mock23from unittest.mock import MagicMock24from src.common.crypto   import blake2b, encrypt_and_sign25from src.common.database import MessageLog26from src.common.encoding import b58encode, bool_to_bytes, int_to_bytes, str_to_bytes27from src.common.statics  import (CH_FILE_RECV, COMMAND, COMMAND_DATAGRAM_HEADER, CONFIRM_CODE_LENGTH, DIR_USER_DATA,28                                 ENABLE, EXIT, FILE_DATAGRAM_HEADER, FILE_KEY_HEADER, INITIAL_HARAC, KEY_EX_ECDHE,29                                 LOCAL_KEY_DATAGRAM_HEADER, MESSAGE, MESSAGE_DATAGRAM_HEADER, ORIGIN_CONTACT_HEADER,30                                 PRIVATE_MESSAGE_HEADER, SYMMETRIC_KEY_LENGTH, UNIT_TEST_QUEUE, US_BYTE, WIN_SELECT,31                                 WIN_UID_COMMAND, WIN_UID_FILE)32from src.transmitter.packet import split_to_assembly_packets33from src.receiver.output_loop import output_loop34from tests.mock_classes import ContactList, Gateway, GroupList, KeyList, MasterKey, nick_to_pub_key, Settings35from tests.utils        import cd_unit_test, cleanup, gen_queue_dict, tear_queues36def rotate_key(key: bytes, harac: int) -> Tuple[bytes, int]:37    """Move to next key in hash ratchet."""38    return blake2b(key + int_to_bytes(harac), digest_size=SYMMETRIC_KEY_LENGTH), harac + 139class TestOutputLoop(unittest.TestCase):40    def setUp(self) -> None:41        """Pre-test actions."""42        self.unit_test_dir = cd_unit_test()43        self.o_sleep       = time.sleep44        time.sleep         = lambda _: None45    def tearDown(self) -> None:46        """Post-test actions."""47        time.sleep = self.o_sleep48        cleanup(self.unit_test_dir)49    @mock.patch("os.system",      return_value=None)50    @mock.patch("tkinter.Tk",     return_value=MagicMock())51    @mock.patch("builtins.input", side_effect=[b58encode(SYMMETRIC_KEY_LENGTH * b"a"),52                                               bytes(CONFIRM_CODE_LENGTH).hex(),53                                               b58encode(SYMMETRIC_KEY_LENGTH * b"a", public_key=True)])54    def test_loop(self, *_) -> None:55        # Setup56        queues     = gen_queue_dict()57        kek        = SYMMETRIC_KEY_LENGTH * b"a"58        conf_code  = bytes(1)59        tx_pub_key = nick_to_pub_key("Bob")60        o_sleep    = self.o_sleep61        test_delay = 0.362        def queue_packet(mk, hk, tx_harac, packet, onion_pub_key=None) -> None:63            """Create encrypted datagram."""64            if onion_pub_key is None:65                header = b""66                queue  = queues[COMMAND_DATAGRAM_HEADER]67                packet = split_to_assembly_packets(packet, COMMAND)[0]68            else:69                header = onion_pub_key + ORIGIN_CONTACT_HEADER70                queue  = queues[MESSAGE_DATAGRAM_HEADER]71                packet = split_to_assembly_packets(packet, MESSAGE)[0]72            encrypted_harac   = encrypt_and_sign(int_to_bytes(tx_harac), hk)73            encrypted_message = encrypt_and_sign(packet, mk)74            encrypted_packet  = header + encrypted_harac + encrypted_message75            queue.put((datetime.datetime.now(), encrypted_packet))76        def queue_delayer() -> None:77            """Place datagrams into queue after delay."""78            o_sleep(test_delay)79            local_harac = INITIAL_HARAC80            tx_harac    = INITIAL_HARAC81            local_hek   = SYMMETRIC_KEY_LENGTH * b"a"82            file_key    = SYMMETRIC_KEY_LENGTH * b"b"83            local_key   = SYMMETRIC_KEY_LENGTH * b"a"84            tx_mk       = SYMMETRIC_KEY_LENGTH * b"a"85            tx_hk       = SYMMETRIC_KEY_LENGTH * b"a"86            # Queue local key packet87            local_key_packet = encrypt_and_sign(local_key + local_hek + conf_code, key=kek)88            queues[LOCAL_KEY_DATAGRAM_HEADER].put((datetime.datetime.now(), local_key_packet))89            o_sleep(test_delay)90            # Select file window91            command = WIN_SELECT + WIN_UID_FILE92            queue_packet(local_key, tx_hk, local_harac, command)93            local_key, local_harac = rotate_key(local_key, local_harac)94            o_sleep(test_delay)95            # Select local window96            command = WIN_SELECT + WIN_UID_COMMAND97            queue_packet(local_key, tx_hk, local_harac, command)98            local_key, local_harac = rotate_key(local_key, local_harac)99            o_sleep(test_delay)100            # A message that goes to buffer101            queue_packet(tx_mk,102                         tx_hk,103                         tx_harac,104                         bool_to_bytes(False) + PRIVATE_MESSAGE_HEADER + b"Hi Bob",105                         tx_pub_key)106            tx_mk, tx_harac = rotate_key(tx_mk, tx_harac)107            # ECDHE keyset for Bob108            command = (KEY_EX_ECDHE109                       + nick_to_pub_key("Bob")110                       + (4 * SYMMETRIC_KEY_LENGTH * b"a")111                       + str_to_bytes("Bob"))112            queue_packet(local_key, tx_hk, local_harac, command)113            local_key, local_harac = rotate_key(local_key, local_harac)114            o_sleep(test_delay)115            # Message for Bob116            queue_packet(tx_mk,117                         tx_hk,118                         tx_harac,119                         bool_to_bytes(False) + PRIVATE_MESSAGE_HEADER + b"Hi Bob",120                         tx_pub_key)121            tx_mk, tx_harac = rotate_key(tx_mk, tx_harac)122            o_sleep(test_delay)123            # Enable file reception for Bob124            command = CH_FILE_RECV + ENABLE.upper() + US_BYTE125            queue_packet(local_key, tx_hk, local_harac, command)126            o_sleep(test_delay)127            # File packet from Bob128            ct     = encrypt_and_sign(b"test", file_key)129            f_hash = blake2b(ct)130            packet = nick_to_pub_key("Bob") + ORIGIN_CONTACT_HEADER + ct131            queues[FILE_DATAGRAM_HEADER].put((datetime.datetime.now(), packet))132            o_sleep(test_delay)133            # File key packet from Bob134            queue_packet(tx_mk,135                         tx_hk,136                         tx_harac,137                         bool_to_bytes(False) + FILE_KEY_HEADER + base64.b85encode(f_hash + file_key),138                         tx_pub_key)139            o_sleep(test_delay)140            # Queue exit message to break the loop141            o_sleep(0.5)142            queues[UNIT_TEST_QUEUE].put(EXIT)143            o_sleep(test_delay)144        threading.Thread(target=queue_delayer).start()145        # Test146        master_key  = MasterKey()147        settings    = Settings()148        message_log = MessageLog(f"{DIR_USER_DATA}{settings.software_operation}_logs", master_key.master_key)149        self.assertIsNone(output_loop(queues, Gateway(), settings, ContactList(), KeyList(), GroupList(),150                                      master_key, message_log, stdin_fd=1, unit_test=True,))151        # Teardown152        tear_queues(queues)153if __name__ == "__main__":...

testing_test2.py

Source:testing_test2.py

1import covasim as cv2import sciris as sc3cv.check_version('0.27.9')4do_plot = 15datafile = sc.thisdir(__file__, '../../example_data.csv')6pars = sc.objdict(7    diag_factor = 1.0,8    n_days=30,9    pop_infected=10010    )11sim = cv.Sim(pars, datafile=datafile)12case = 213testprobdict = {14    0: cv.test_prob(symp_prob=0, asymp_prob=0, test_sensitivity=1.0, loss_prob=0.0, test_delay=0, start_day=0), # works, no diagnoses15    1: cv.test_prob(symp_prob=1, asymp_prob=0, test_sensitivity=1.0, loss_prob=0.0, test_delay=0, start_day=0), # works, ~half diagnosed16    2: cv.test_prob(symp_prob=0, asymp_prob=1, test_sensitivity=1.0, loss_prob=0.0, test_delay=0, start_day=0), # works, most diagnosed (during asymptomatic period)17    3: cv.test_prob(symp_prob=1, asymp_prob=1, test_sensitivity=1.0, loss_prob=0.0, test_delay=0, start_day=0), # works, most diagnosed, slight mismatch due to new infections that day18    4: cv.test_prob(symp_prob=1, asymp_prob=1, test_sensitivity=0.0, loss_prob=0.0, test_delay=0, start_day=0), # works, no diagnoses19    5: cv.test_prob(symp_prob=1, asymp_prob=1, test_sensitivity=0.5, loss_prob=0.0, test_delay=0, start_day=0), # works, ~80% diagnosed20    6: cv.test_prob(symp_prob=1, asymp_prob=1, test_sensitivity=1.0, loss_prob=0.5, test_delay=0, start_day=0), # works, ~80% diagnosed21    7: cv.test_prob(symp_prob=1, asymp_prob=1, test_sensitivity=1.0, loss_prob=0.0, test_delay=0, start_day=0), # works, most diagnosed, slight mismatch due to new infections that day22    8: cv.test_prob(symp_prob=1, asymp_prob=1, test_sensitivity=1.0, loss_prob=0.0, test_delay=5, start_day=0), # works, most diagnosed, with delay, should be about one set of tick marks behind23    9: cv.test_prob(symp_prob=1, asymp_prob=1, test_sensitivity=1.0, loss_prob=0.0, test_delay=0, start_day=7), # works, most diagnosed, with delay24}25sim.update_pars(interventions=testprobdict[case])26to_plot = sc.dcp(cv.default_sim_plots)27to_plot['Total counts'].append('cum_infectious')28to_plot['Daily counts']+= ['new_tests', 'new_infectious']...

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.