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

How to use __scenario method in Molotov

Best Python code snippet using molotov_python

blife.py

Source:blife.py

1import gym2import time3import random4import numpy as np5from gym import error, spaces, utils6from gym.utils import seeding7from .sensors import Sensors8from .scenarios import WhiteNoiseScenario9class BatteryLifetimeEnv(gym.Env):10    """11    Description:12        The blife environment simulates the behavior of a re-chargable13        lithium battery while the controlled device samples white noise14        and delivery it out in streaming or batch processes.15        The controlled device is a Raspberry Pi 3.16    Source:17        This environment and its documentation is available at18        https://github.com/lcarnevale/blife19    Observation:20        Type: Box(4)21        Num     Observation                 Min     Max22        0       Battery Voltage             0 V     5 V23        1       Battery Current             0 mA    2000 mA24        2       Network Outbound Traffic    0       +inf25        3       Buffer Size                 0       10026    Actions:27        Type: Discrete(3)28        Num     Action29        0       do nothing30        1       sample any 0.2s, deliver any 0.2s (streaming)31        2       sample any 1s, deliver any 1s (streaming)32        3       sample any 0.2s, deliver any 5s (batch)33        4       sample ant 1s, deliver any 5s (batch)34    Reward:35        TODO36    Starting State:37        Configure and start the white noise scenario. 38    Episode Termination:39        TODO        40    """41    def __init__(self, battery_capacity=2000, num_discrete_actions=5):42        self.__step_counter = 043        self.__battery_capacity = battery_capacity44        min_battery_voltage = 0         #V45        max_battery_voltage = 5         #V46        min_battery_current = 0         #mA47        max_battery_current = 2000      #mA48        min_net_outbound = 0            #Bytes49        max_net_outbound = 1000000000   #Bytes50        min_buffer_size = 0             #Units51        max_buffer_size = 100           #Units52        self.__perform_action = [53            self.__action_do_nothing,54            self.__action_streaming_one,55            self.__action_streaming_two,56            self.__action_batch_one,57            self.__action_batch_two58        ]59        self.__sensors = Sensors()60        self.__max_expected_runtime = 0.061        low = np.array([62            min_battery_voltage, min_battery_current, 63            min_net_outbound, min_buffer_size], 64            dtype = np.float1665        )66        high = np.array([67            max_battery_voltage, max_battery_current,68            max_net_outbound, max_buffer_size],69            dtype = np.float1670        )71        self.action_space = spaces.Discrete(num_discrete_actions)72        self.observation_space = spaces.Box(low, high, dtype = np.float16)73    def reset(self):74        """Reset the state of the environment to an initial state.75        Returns:76            The observations space.77            See the class description for more details.78        """79        self.__configure_experiment()80        self.__reset_experiment()81        return self.__observe()82    def __configure_experiment(self):83        self.__scenario = WhiteNoiseScenario(84            'broker.mqttdashboard.com',85            '/fcrlab/distinsys/lcarnevale',86        )87    88    def __reset_experiment(self):89        self.__scenario.reset()90    def __observe(self):91        """92        Returns:93            The observations space wrapped within a numpy array.94            See the class description for more details.95        """96        observations = self.__next_observation()97        voltage = observations[0]98        current = observations[1]99        self.__power_last_mean = current * voltage100        return np.array(observations)101    def __next_observation(self):102        """Observe the environment.103        Returns:104            The observations space.105            See the class description for more details.106        """107        return [108            self.__sensors.get_bus_voltage(),109            self.__sensors.get_bus_current(),110            self.__sensors.get_net_bytes_sent(),111            self.__scenario.get_queue_size()112        ]113    def step(self, action):114        """Execute one time step within the environment.115        Returns:116        """117        done = False118        119        # perform action120        self.__valuate_action(action)121        self.__perform_action[action]()122        123        # collect observation124        observation = self.__sub_observe()125        voltage_mean = observation[0]126        current_mean = observation[1]127        observation = np.array(observation)128        # calculate reward129        power_mean = voltage_mean * current_mean130        reward = self.__reward_function(power_mean)131        # verify the termination condition132        expected_runtime = self.__battery_capacity / current_mean133        self.__set_max_expected_runtime(expected_runtime)134        done = self.__termination_function(expected_runtime)135        # print("\t\ttermination: %.3fh (runtime) >= 8h is %s" % (expected_runtime, done))136        137        self.__power_last_mean = power_mean138        return observation, reward, done, {}139    140    def __valuate_action(self, action):141        err_msg = "%r (%s) invalid" % (action, type(action))142        assert self.action_space.contains(action), err_msg143    # action 0144    def __action_do_nothing(self):145        # print('do nothing')146        pass147    # action 1148    def __action_streaming_one(self):149        # print("sample any 1s, deliver any 1s (streaming)")150        self.__scenario.set_transmission('streaming')151        self.__scenario.set_rate(0.05)152    # action 2153    def __action_streaming_two(self):154        # print("sample any 10s, deliver any 10s (streaming)")155        self.__scenario.set_transmission('streaming')156        self.__scenario.set_rate(1)157    # action 3158    def __action_batch_one(self):159        self.__scenario.set_transmission('batch')160        self.__scenario.set_rate(0.05)161    # action 4162    def __action_batch_two(self):163        self.__scenario.set_transmission('batch')164        self.__scenario.set_rate(1)165    def __sub_observe(self, observation_rate=1, observation_window=15):166        """Observe the environment under specific conditions.167        """168        voltage_sub_history = list()169        current_sub_history = list()170        net_bytes_sent_sub_history = list()171        queue_size_sub_history = list()172        for _ in range(observation_window):173            observations = self.__next_observation()174            voltage_sub_history.append(observations[0])175            current_sub_history.append(observations[1])176            net_bytes_sent_sub_history.append(observations[2])177            queue_size_sub_history.append(observations[3])178            time.sleep(observation_rate)179        return [180            np.array(voltage_sub_history).mean(),181            np.array(current_sub_history).mean(),182            np.array(net_bytes_sent_sub_history).mean(),183            np.array(queue_size_sub_history).mean()184        ]185    def __reward_function(self, power_mean):186        """Calculate the reward.187        Returns:188            float representing the energy delta in mJ189        """190        energy_delta = - (power_mean - self.__power_last_mean) * 1191        return energy_delta192    def __set_max_expected_runtime(self, expected_runtime):193        if expected_runtime > self.__max_expected_runtime:194            self.__max_expected_runtime = expected_runtime195    def __termination_function(self, expected_runtime, target_expected_runtime=8.5):196        """Calculate the termination197        With the battery capacity and average current consumption,198        you can compute the expected runtime of your project by 199        solving the equation:200        Battery capacity (in mAh) / Average current consumption 201            (in mA) = Hours of expected runtime202        I expect the runtime is at least one day.203        204        Returns:205            bool representing the termination validation206        """207        return expected_runtime >= target_expected_runtime208    def close(self):...

jb_behave_formatter.py

Source:jb_behave_formatter.py

1# coding=utf-82"""3Behave formatter that supports TC4"""5import datetime6import traceback7from collections import deque8from distutils import version9from behave.formatter.base import Formatter10from behave.model import Step, Feature, Scenario11from behave.model_core import Status12from behave import __version__ as behave_version13from teamcity.messages import TeamcityServiceMessages14def _step_name(step):15    assert isinstance(step, Step)16    return step.keyword + " " + step.name.strip()17def _suite_name(suite):18    return suite.name.strip()19class TeamcityFormatter(Formatter):20    """21    Stateful TC reporter.22    Since we can't fetch all steps from the very beginning (even skipped tests are reported)23    we store tests and features on each call.24    To hook into test reporting override _report_suite_started and/or _report_test_started25    """26    def __init__(self, stream_opener, config):27        super(TeamcityFormatter, self).__init__(stream_opener, config)28        assert version.LooseVersion(behave_version) >= version.LooseVersion("1.2.6"), "Only 1.2.6+ is supported"29        self._messages = TeamcityServiceMessages()30        self.__feature = None31        self.__scenario = None32        self.__steps = deque()33        self.__scenario_opened = False34        self.__feature_opened = False35        self.__test_start_time = None36    def feature(self, feature):37        assert isinstance(feature, Feature)38        assert not self.__feature, "Prev. feature not closed"39        self.__feature = feature40    def scenario(self, scenario):41        assert isinstance(scenario, Scenario)42        self.__scenario = scenario43        self.__scenario_opened = False44        self.__steps.clear()45    def step(self, step):46        assert isinstance(step, Step)47        self.__steps.append(step)48    def match(self, match):49        if not self.__feature_opened:50            self._report_suite_started(self.__feature, _suite_name(self.__feature))51            self.__feature_opened = True52        if not self.__scenario_opened:53            self._report_suite_started(self.__scenario, _suite_name(self.__scenario))54            self.__scenario_opened = True55        assert self.__steps, "No steps left"56        step = self.__steps.popleft()57        self._report_test_started(step, _step_name(step))58        self.__test_start_time = datetime.datetime.now()59    def _report_suite_started(self, suite, suite_name):60        """61        :param suite: behave suite62        :param suite_name: suite name that must be reported, be sure to use it instead of suite.name63        """64        self._messages.testSuiteStarted(suite_name)65    def _report_test_started(self, test, test_name):66        """67        Suite name is always stripped, be sure to strip() it too68        :param test: behave test69        :param test_name: test name that must be reported, be sure to use it instead of test.name70        """71        self._messages.testStarted(test_name)72    def result(self, step):73        assert isinstance(step, Step)74        step_name = _step_name(step)75        if step.status == Status.failed:76            try:77                error = traceback.format_exc(step.exc_traceback)78                if error != step.error_message:79                    self._messages.testStdErr(step_name, error)80            except Exception:81                pass  # exception shall not prevent error message82            self._messages.testFailed(step_name, message=step.error_message)83        if step.status == Status.undefined:84            self._messages.testFailed(step_name, message="Undefined")85        if step.status == Status.skipped:86            self._messages.testIgnored(step_name)87        self._messages.testFinished(step_name, testDuration=datetime.datetime.now() - self.__test_start_time)88        if not self.__steps:89            self.__close_scenario()90        elif step.status in [Status.failed, Status.undefined]:91            # Broken background/undefined step stops whole scenario92            reason = "Undefined step" if step.status == Status.undefined else "Prev. step failed"93            self.__skip_rest_of_scenario(reason)94    def __skip_rest_of_scenario(self, reason):95        while self.__steps:96            step = self.__steps.popleft()97            self._report_test_started(step, _step_name(step))98            self._messages.testIgnored(_step_name(step),99                                       message="{0}. Rest part of scenario is skipped".format(reason))100            self._messages.testFinished(_step_name(step))101        self.__close_scenario()102    def __close_scenario(self):103        if self.__scenario:104            self._messages.testSuiteFinished(_suite_name(self.__scenario))105            self.__scenario = None106    def eof(self):107        self.__skip_rest_of_scenario("")108        self._messages.testSuiteFinished(_suite_name(self.__feature))109        self.__feature = None...

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.