How to use scenario_name method in molecule

Best Python code snippet using molecule_python

test_command.py

Source:test_command.py

...5@pytest.fixture6def scenario_to_test(request):7    return request.param8@pytest.fixture9def scenario_name(request):10    try:11        return request.param12    except AttributeError:13        return None14@pytest.fixture15def driver_name(request):16    return request.param17@pytest.mark.parametrize(18    "scenario_to_test, driver_name, scenario_name",19    [("driver/hetznercloud", "hetznercloud", "default")],20    indirect=["scenario_to_test", "driver_name", "scenario_name"],21)22def test_command_check(scenario_to_test, with_scenario, scenario_name):23    options = {"scenario_name": scenario_name}...

scenarios.py

Source:scenarios.py

1# Copyright 2021 The Chromium Authors. All rights reserved.2# Use of this source code is governed by a BSD-style license that can be3# found in the LICENSE file.4import abc5import subprocess6import jinja27import tempfile8import datetime9import logging10import typing11import os12import utils13import browsers14def GetTemplateFileForBrowser(browser_driver: browsers.BrowserDriver,15                              template_file: str) -> str:16  if browser_driver.name == "safari":17    return f"safari_{template_file}"18  else:19    return template_file20class ScenarioOSADriver(abc.ABC):21  """Base Class encapsulating OSA script driving a scenario, with setup and tear22     down.23  """24  def __init__(self, scenario_name, duration: datetime.timedelta):25    self.name = scenario_name26    self.script_process = None27    self.osa_script = None28    self.duration = duration29  def Launch(self):30    """Starts the driver script.31    """32    assert self.osa_script is not None33    logging.debug(f"Starting scenario {self.name}")34    self.script_process = subprocess.Popen(['osascript', self.osa_script.name])35  def Wait(self):36    """Waits for the script to complete.37    """38    assert self.script_process is not None, "Driver wasn't launched."39    logging.debug(f"Waiting for scenario {self.name}")40    self.script_process.wait()41  def TearDown(self):42    """Terminates the script if currently running and ensures related processes43       are cleaned up.44    """45    logging.debug(f"Tearing down scenario {self.name}")46    if self.script_process:47      utils.TerminateProcess(self.script_process)48    self.osa_script.close()49  @abc.abstractmethod50  def Summary(self):51    """Returns a dictionary describing the scenarios parameters.52    """53    pass54  def IsRunning(self) -> bool:55    """Returns true if the script is currently running.56    """57    return self.script_process.poll() is None58  def _CompileTemplate(self, template_file: str, extra_args: typing.Dict):59    """Compiles script `template_file`, feeding `extra_args` into a temporary60       file.61    """62    loader = jinja2.FileSystemLoader(63        os.path.join(os.path.dirname(__file__), "driver_scripts_templates"))64    env = jinja2.Environment(loader=loader)65    template = env.get_template(template_file)66    self.osa_script = tempfile.NamedTemporaryFile('w+t')67    self.osa_script.write(template.render(**extra_args))68    self.osa_script.flush()69    self._args = extra_args70  def Summary(self):71    """Returns a dictionary describing the scenarios parameters.72    """73    return {'name': self.name, **self._args}74class ScenarioWithBrowserOSADriver(ScenarioOSADriver):75  """Specialisation for OSA script that runs with a browser.76  """77  def __init__(self, scenario_name, browser_driver: browsers.BrowserDriver,78               duration: datetime.timedelta):79    super().__init__(f"{browser_driver.name}_{scenario_name}", duration)80    self.browser = browser_driver81  def Launch(self):82    self.browser.Launch()83    super().Launch()84  def TearDown(self):85    super().TearDown()86    self.browser.TearDown()87  def Summary(self):88    """Returns a dictionary describing the scenarios parameters.89    """90    return {**super().Summary(), 'browser': self.browser.Summary()}91  def _CompileTemplate(self, template_file, extra_args: typing.Dict):92    return super()._CompileTemplate(template_file, {93        "browser": self.browser.process_name,94        **extra_args95    })96class IdleScenario(ScenarioOSADriver):97  """Scenario that lets the system idle.98  """99  def __init__(self, duration: datetime.timedelta, scenario_name="idle"):100    super().__init__(scenario_name, duration)101    self._CompileTemplate("idle", {102        "delay": duration.total_seconds(),103    })104class IdleOnSiteScenario(ScenarioWithBrowserOSADriver):105  """Scenario that lets a browser idle on a web page.106  """107  def __init__(self, browser_driver: browsers.BrowserDriver,108               duration: datetime.timedelta, site_url: str, scenario_name):109    super().__init__(scenario_name, browser_driver, duration)110    self._CompileTemplate(111        GetTemplateFileForBrowser(browser_driver, "idle_on_site"), {112            "idle_site": site_url,113            "delay": duration.total_seconds(),114        })115  @staticmethod116  def Wiki(browser_driver: browsers.BrowserDriver,117           duration: datetime.timedelta):118    return IdleOnSiteScenario(browser_driver, duration,119                              "http://www.wikipedia.com/wiki/Alessandro_Volta",120                              "idle_on_wiki")121  @staticmethod122  def Youtube(browser_driver: browsers.BrowserDriver,123              duration: datetime.timedelta):124    return IdleOnSiteScenario(125        browser_driver, duration,126        "https://www.youtube.com/watch?v=9EE_ICC_wFw?autoplay=1",127        "idle_on_youtube")128class ZeroWindowScenario(ScenarioWithBrowserOSADriver):129  """Scenario that lets a browser idle with no window.130  """131  def __init__(self,132               browser_driver: browsers.BrowserDriver,133               duration: datetime.timedelta,134               scenario_name="zero_window"):135    super().__init__(scenario_name, browser_driver, duration)136    self._CompileTemplate(137        GetTemplateFileForBrowser(browser_driver, "zero_window"), {138            "delay": duration.total_seconds(),139        })140class NavigationScenario(ScenarioWithBrowserOSADriver):141  """Scenario that has a browser navigating on web pages in a loop.142  """143  NAVIGATED_SITES = [144      "https://amazon.com",145      "https://www.amazon.com/s?k=computer&ref=nb_sb_noss_2",146      "https://google.com", "https://www.google.com/search?q=computers",147      "https://www.youtube.com",148      "https://www.youtube.com/results?search_query=computers",149      "https://docs.google.com/document/d/1Ll-8Nvo6JlhzKEttst8GHWCc7_A8Hluy2fX99cy4Sfg/edit?usp=sharing"150  ]151  def __init__(self,152               browser_driver: browsers.BrowserDriver,153               navigation_duration: datetime.timedelta,154               navigation_cycles: int,155               sites=NAVIGATED_SITES,156               scenario_name="navigation"):157    super().__init__(scenario_name, browser_driver,158                     navigation_duration * navigation_cycles * len(sites))159    self._CompileTemplate(160        GetTemplateFileForBrowser(browser_driver, "navigation"), {161            "per_navigation_delay": navigation_duration.total_seconds(),162            "navigation_cycles": navigation_cycles,163            "sites": ",".join([f'"{site}"' for site in sites])164        })165class MeetScenario(ScenarioWithBrowserOSADriver):166  """Scenario that has the browser join a Google Meet room.167  """168  def __init__(self,169               browser_driver: browsers.BrowserDriver,170               duration: datetime.timedelta,171               meeting_id: int,172               scenario_name="meet"):173    super().__init__(scenario_name, browser_driver, duration)174    self._CompileTemplate(GetTemplateFileForBrowser(browser_driver, "meet"), {175        "delay": duration.total_seconds(),176        "meeting_id": meeting_id177    })178def MakeScenarioDriver(scenario_name,179                       browser_driver: browsers.BrowserDriver,180                       meet_meeting_id=None) -> ScenarioOSADriver:181  """Creates scenario driver by name.182  Args:183    scenario_name: Identifier for the scenario to create. Supported scenarios184      are: meet, idle_on_wiki, idle_on_youtube, navigation, zero_window and185      idle.186    browser_driver: Browser the scenario is created with.187    meet_meeting_id: Optional meeting id used for meet scenario.188  """189  if "idle" == scenario_name:190    return IdleScenario(datetime.timedelta(minutes=60))191  if not browser_driver:192    return None193  if "meet" == scenario_name:194    return MeetScenario(browser_driver,195                        datetime.timedelta(minutes=60),196                        meeting_id=meet_meeting_id)197  if "idle_on_wiki" == scenario_name:198    return IdleOnSiteScenario.Wiki(browser_driver,199                                   datetime.timedelta(minutes=60))200  if "idle_on_youtube" == scenario_name:201    return IdleOnSiteScenario.Youtube(browser_driver,202                                      datetime.timedelta(minutes=60))203  if "navigation" == scenario_name:204    return NavigationScenario(205        browser_driver,206        navigation_duration=datetime.timedelta(seconds=15),207        navigation_cycles=70)208  if "zero_window" == scenario_name:209    return ZeroWindowScenario(browser_driver, datetime.timedelta(minutes=60))...

conftest.py

Source:conftest.py

1import os2import shutil3import pexpect4import pkg_resources5import pytest6from molecule import logger, util7from ..conftest import change_dir_to8LOG = logger.get_logger(__name__)9IS_TRAVIS = os.getenv("TRAVIS") and os.getenv("CI")10def _env_vars_exposed(env_vars, env=os.environ):11    """Check if environment variables are exposed and populated."""12    for env_var in env_vars:13        if env_var not in os.environ:14            return False15        return os.environ[env_var] != ""16@pytest.fixture17def with_scenario(request, scenario_to_test, driver_name, scenario_name, skip_test):18    scenario_directory = os.path.join(19        os.path.dirname(util.abs_path(__file__)),20        os.path.pardir,21        "scenarios",22        scenario_to_test,23    )24    with change_dir_to(scenario_directory):25        yield26        if scenario_name:27            msg = "CLEANUP: Destroying instances for all scenario(s)"28            LOG.info(msg)29            cmd = ["molecule", "destroy", "--driver-name", driver_name, "--all"]30            pytest.helpers.run_command(cmd)31@pytest.fixture32def skip_test(request, driver_name):33    msg_tmpl = (34        "Ignoring '{}' tests for now"35        if driver_name == "delegated"36        else "Skipped '{}' not supported"37    )38    support_checks_map = {39        "hetznercloud": lambda: min_ansible("2.8") and supports_hetznercloud()40    }41    try:42        check_func = support_checks_map[driver_name]43        if not check_func():44            pytest.skip(msg_tmpl.format(driver_name))45    except KeyError:46        pass47@pytest.helpers.register48def idempotence(scenario_name):49    cmd = ["molecule", "create", "--scenario-name", scenario_name]50    pytest.helpers.run_command(cmd)51    cmd = ["molecule", "converge", "--scenario_name", scenario_name]52    pytest.helpers.run_command(cmd)53    cmd = ["molecule", "--scenario_name", scenario_name]54    pytest.helpers.run_command(cmd)55@pytest.helpers.register56def init_role(temp_dir, driver_name):57    cmd = ["molecule", "init", "role", "test-init", "--driver-name", driver_name]58    pytest.helpers.run_command(cmd)59    role_directory = os.path.join(temp_dir.strpath, "test-init")60    pytest.helpers.metadata_lint_update(role_directory)61    with change_dir_to(role_directory):62        cmd = ["molecule", "test", "--all"]63        pytest.helpers.run_command(cmd)64@pytest.helpers.register65def init_scenario(temp_dir, driver_name):66    cmd = ["molecule", "init", "role", "test-init", "--driver-name", driver_name]67    pytest.helpers.run_command(cmd)68    role_directory = os.path.join(temp_dir.strpath, "test-init")69    pytest.helpers.metadata_lint_update(role_directory)70    with change_dir_to(role_directory):71        molecule_directory = pytest.helpers.molecule_directory()72        scenario_directory = os.path.join(molecule_directory, "test-scenario")73        cmd = [74            "molecule",75            "init",76            "scenario",77            "test-scenario",78            "--role-name",79            "test-init",80            "--driver-name",81            driver_name,82        ]83        pytest.helpers.run_command(cmd)84        assert os.path.isdir(scenario_directory)85        cmd = ["molecule", "test", "--scenario-name", "test-scenario", "--all"]86        pytest.helpers.run_command(cmd)87@pytest.helpers.register88def metadata_lint_update(role_directory):89    ansible_lint_src = os.path.join(90        os.path.dirname(util.abs_path(__file__)), ".ansible-lint"91    )92    shutil.copy(ansible_lint_src, role_directory)93    with change_dir_to(role_directory):94        cmd = ["ansible-lint", "."]95    pytest.helpers.run_command(cmd)96@pytest.helpers.register97def list(x):98    cmd = ["molecule", "list"]99    out = pytest.helpers.run_command(cmd, log=False)100    out = out.stdout.decode("utf-8")101    out = util.strip_ansi_color(out)102    for l in x.splitlines():103        assert l in out104@pytest.helpers.register105def list_with_format_plain(x):106    cmd = ["molecule", "list", "--format", "plain"]107    result = util.run_command(cmd)108    out = util.strip_ansi_color(result.stdout)109    for l in x.splitlines():110        assert l in out111@pytest.helpers.register112def login(login_args, scenario_name="default"):113    cmd = ["molecule", "destroy", "--scenario-name", scenario_name]114    pytest.helpers.run_command(cmd)115    cmd = ["molecule", "create", "--scenario-name", scenario_name]116    pytest.helpers.run_command(cmd)117    for instance, regexp in login_args:118        if len(login_args) > 1:119            child_cmd = "molecule login --host {} --scenario-name {}".format(120                instance, scenario_name121            )122        else:123            child_cmd = "molecule login --scenario-name {}".format(scenario_name)124        child = pexpect.spawn(child_cmd)125        child.expect(regexp)126        child.sendline("exit")127@pytest.helpers.register128def test(driver_name, scenario_name="default", parallel=False):129    cmd = ["molecule", "test", "--scenario-name", scenario_name]130    if driver_name != "delegated":131        if scenario_name is None:132            cmd.append("--all")133        if parallel:134            cmd.append("--parallel")135    pytest.helpers.run_command(cmd)136@pytest.helpers.register137def verify(scenario_name="default"):138    cmd = ["molecule", "create", "--scenario-name", scenario_name]139    pytest.helpers.run_command(cmd)140    cmd = ["molecule", "converge", "--scenario-name", scenario_name]141    pytest.helpers.run_command(cmd)142    cmd = ["molecule", "verify", "--scenario-name", scenario_name]143    pytest.helpers.run_command(cmd)144def min_ansible(version):145    """Ensure current Ansible is newer than a given a minimal one."""146    try:147        from ansible.release import __version__148        return pkg_resources.parse_version(__version__) >= pkg_resources.parse_version(149            version150        )151    except ImportError as exception:152        LOG.error("Unable to parse Ansible version", exc_info=exception)153        return False154@pytest.helpers.register155def supports_hetznercloud():156    pytest.importorskip("hcloud")157    env_vars = ("HCLOUD_TOKEN",)...

run.py

Source:run.py

1# -*- coding: utf-8 -*-2import logging, sys3import inspect4import config as cf5from python.network.network import Network6from python.utils.tracer import *7from python.routing.direct_communication import *8from python.routing.mte import *9from python.routing.leach import *10from python.routing.fcm import *11from python.network.aggregation_model import *12logging.basicConfig(stream=sys.stderr, level=logging.INFO)13if __name__ == '__main__':14  network  = Network()15  aggregation_function = zero_cost_aggregation16  #aggregation_function = linear_cost_aggregation(0.5)17  traces = {}18  scenario_names = {}19  for scenario in cf.scenarios:20    if type(scenario) is str:21      exec(scenario)22      continue23    network.reset()24    routing_topology, optimization, aggregation, nickname = scenario25    if nickname:26      scenario_name = nickname27    else:28      if optimization:29        scenario_name = routing_topology+' + '+optimization30      else:31        scenario_name = routing_topology32    if scenario_name in scenario_names:33      scenario_names[scenario_name] += 134      scenario_name += " (" + str(scenario_names[scenario_name]) + ")"35    else:36      scenario_names[scenario_name] = 137    routing_protocol_class          = eval(routing_topology)38    network.routing_protocol        = routing_protocol_class()39    if optimization:40      sleep_scheduler_class         = eval(optimization)41      not_class_msg = 'optimization does not hold the name of a class'42      assert inspect.isclass(sleep_scheduler_class), not_class_msg43      network.sleep_scheduler_class = sleep_scheduler_class44    aggregation_function = aggregation + '_cost_aggregation'45    network.set_aggregation_function(eval(aggregation_function))46    logging.info(scenario_name + ': running scenario...')47    traces[scenario_name] = network.simulate()48  if cf.TRACE_COVERAGE:...

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.