How to use _vivify method in molecule

Best Python code snippet using molecule_python

ansible.py

Source:ansible.py Github

copy

Full Screen

...355 instance-2:356 ansible_connection: docker357 :return: str358 """359 dd = self._vivify()360 for platform in self._config.platforms.instances:361 for group in platform.get('groups', ['ungrouped']):362 instance_name = platform['name']363 connection_options = self.connection_options(instance_name)364 molecule_vars = {365 'molecule_file':366 "{{ lookup('env', 'MOLECULE_FILE') }}",367 'molecule_ephemeral_directory':368 "{{ lookup('env', 'MOLECULE_EPHEMERAL_DIRECTORY') }}",369 'molecule_scenario_directory':370 "{{ lookup('env', 'MOLECULE_SCENARIO_DIRECTORY') }}",371 'molecule_yml':372 "{{ lookup('file', molecule_file) | molecule_from_yaml }}",373 'molecule_instance_config':374 "{{ lookup('env', 'MOLECULE_INSTANCE_CONFIG') }}",375 }376 # All group377 dd['all']['hosts'][instance_name] = connection_options378 dd['all']['vars'] = molecule_vars379 # Named group380 dd[group]['hosts'][instance_name] = connection_options381 dd[group]['vars'] = molecule_vars382 # Ungrouped383 dd['ungrouped']['vars'] = {}384 # Children385 for child_group in platform.get('children', []):386 dd[group]['children'][child_group]['hosts'][387 instance_name] = connection_options388 return self._default_to_regular(dd)389 @property390 def inventory_file(self):391 return os.path.join(self._config.scenario.ephemeral_directory,392 'ansible_inventory.yml')393 @property394 def config_file(self):395 return os.path.join(self._config.scenario.ephemeral_directory,396 'ansible.cfg')397 @property398 @util.memoize399 def playbooks(self):400 return ansible_playbooks.AnsiblePlaybooks(self._config)401 @property402 def directory(self):403 return os.path.join(404 os.path.dirname(__file__), os.path.pardir, os.path.pardir,405 'molecule', 'provisioner', 'ansible')406 def connection_options(self, instance_name):407 d = self._config.driver.ansible_connection_options(instance_name)408 return util.merge_dicts(409 d, self._config.config['provisioner']['connection_options'])410 def check(self):411 """412 Executes `ansible-playbook` against the converge playbook with the413 ``--check`` flag and returns None.414 :return: None415 """416 pb = self._get_ansible_playbook(self.playbooks.converge)417 pb.add_cli_arg('check', True)418 pb.execute()419 def converge(self, playbook=None, **kwargs):420 """421 Executes `ansible-playbook` against the converge playbook unless422 specified otherwise and returns a string.423 :param playbook: An optional string containing an absolute path to a424 playbook.425 :param kwargs: An optional keyword arguments.426 :return: str427 """428 if playbook is None:429 pb = self._get_ansible_playbook(self.playbooks.converge, **kwargs)430 else:431 pb = self._get_ansible_playbook(playbook, **kwargs)432 return pb.execute()433 def destroy(self):434 """435 Executes `ansible-playbook` against the destroy playbook and returns436 None.437 :return: None438 """439 pb = self._get_ansible_playbook(self.playbooks.destroy)440 pb.execute()441 def side_effect(self):442 """443 Executes `ansible-playbook` against the side_effect playbook and444 returns None.445 :return: None446 """447 pb = self._get_ansible_playbook(self.playbooks.side_effect)448 pb.execute()449 def create(self):450 """451 Executes `ansible-playbook` against the create playbook and returns452 None.453 :return: None454 """455 pb = self._get_ansible_playbook(self.playbooks.create)456 pb.execute()457 def prepare(self):458 """459 Executes `ansible-playbook` against the prepare playbook and returns460 None.461 :return: None462 """463 pb = self._get_ansible_playbook(self.playbooks.prepare)464 pb.execute()465 def syntax(self):466 """467 Executes `ansible-playbook` against the converge playbook with the468 ``-syntax-check`` flag and returns None.469 :return: None470 """471 pb = self._get_ansible_playbook(self.playbooks.converge)472 pb.add_cli_arg('syntax-check', True)473 pb.execute()474 def verify(self):475 """476 Executes `ansible-playbook` against the verify playbook and returns477 None.478 :return: None479 """480 pb = self._get_ansible_playbook(self.playbooks.verify)481 pb.execute()482 def write_config(self):483 """484 Writes the provisioner's config file to disk and returns None.485 :return: None486 """487 template = util.render_template(488 self._get_config_template(), config_options=self.config_options)489 util.write_file(self.config_file, template)490 def manage_inventory(self):491 """492 Manages inventory for Ansible and returns None.493 :returns: None494 """495 self._write_inventory()496 self._remove_vars()497 if not self.links:498 self._add_or_update_vars()499 else:500 self._link_or_update_vars()501 def abs_path(self, path):502 return util.abs_path(503 os.path.join(self._config.scenario.directory, path))504 def _add_or_update_vars(self):505 """506 Creates host and/or group vars and returns None.507 :returns: None508 """509 for target in ['host_vars', 'group_vars']:510 if target == 'host_vars':511 vars_target = copy.deepcopy(self.host_vars)512 for instance_name, _ in self.host_vars.items():513 if instance_name == 'localhost':514 instance_key = instance_name515 else:516 instance_key = instance_name517 vars_target[instance_key] = vars_target.pop(instance_name)518 elif target == 'group_vars':519 vars_target = self.group_vars520 if vars_target:521 ephemeral_directory = self._config.scenario.ephemeral_directory522 target_vars_directory = os.path.join(ephemeral_directory,523 target)524 if not os.path.isdir(util.abs_path(target_vars_directory)):525 os.mkdir(util.abs_path(target_vars_directory))526 for target in vars_target.keys():527 target_var_content = vars_target[target]528 path = os.path.join(529 util.abs_path(target_vars_directory), target)530 util.write_file(path, util.safe_dump(target_var_content))531 def _write_inventory(self):532 """533 Writes the provisioner's inventory file to disk and returns None.534 :return: None535 """536 self._verify_inventory()537 util.write_file(self.inventory_file, util.safe_dump(self.inventory))538 def _remove_vars(self):539 """540 Remove host and/or group vars and returns None.541 :returns: None542 """543 dirs = [544 os.path.join(self._config.scenario.ephemeral_directory,545 'group_vars'),546 os.path.join(self._config.scenario.ephemeral_directory,547 'host_vars'),548 ]549 for d in dirs:550 if os.path.islink(d):551 os.unlink(d)552 else:553 if os.path.exists(d):554 shutil.rmtree(d)555 def _link_or_update_vars(self):556 """557 Creates or updates the symlink to group_vars and returns None.558 :returns: None559 """560 for d, source in self.links.items():561 target = os.path.join(self._config.scenario.ephemeral_directory, d)562 source = os.path.join(self._config.scenario.directory, source)563 if not os.path.exists(source):564 msg = "The source path '{}' does not exist.".format(source)565 util.sysexit_with_message(msg)566 msg = "Inventory {} linked to {}".format(source, target)567 LOG.info(msg)568 os.symlink(source, target)569 def _get_ansible_playbook(self, playbook, **kwargs):570 """571 Get an instance of AnsiblePlaybook and returns it.572 :param playbook: A string containing an absolute path to a573 provisioner's playbook.574 :param kwargs: An optional keyword arguments.575 :return: object576 """577 return ansible_playbook.AnsiblePlaybook(playbook, self._config,578 **kwargs)579 def _verify_inventory(self):580 """581 Verify the inventory is valid and returns None.582 :return: None583 """584 if not self.inventory:585 msg = ("Instances missing from the 'platform' "586 "section of molecule.yml.")587 util.sysexit_with_message(msg)588 def _get_config_template(self):589 """590 Returns a config template string.591 :return: str592 """593 return """594{% for section, section_dict in config_options.items() -%}595[{{ section }}]596{% for k, v in section_dict.items() -%}597{{ k }} = {{ v }}598{% endfor -%}599{% endfor -%}600""".strip()601 def _vivify(self):602 """603 Returns an autovivification default dict.604 :return: dict605 """606 return collections.defaultdict(self._vivify)607 def _default_to_regular(self, d):608 if isinstance(d, collections.defaultdict):609 d = {k: self._default_to_regular(v) for k, v in d.items()}610 return d611 def _get_plugin_directory(self):612 return os.path.join(self.directory, 'plugins')613 def _get_libraries_directory(self):614 return util.abs_path(615 os.path.join(self._get_plugin_directory(), 'libraries'))...

Full Screen

Full Screen

vivify.py

Source:vivify.py Github

copy

Full Screen

...240 " objects. Attribute types may not be suitable for vivification"241 )242 return setters243 @entry_exit_logging244 def _vivify(245 self,246 instances: Union[Mapping[str, str], str],247 config: Config,248 defaults: Optional[str] = DEFAULTSECT,249 ) -> Mapping[str, Any]:250 """Internal handler for vivifying objects.251 Args:252 instances (Union[Mapping[str, str], str]): Instances name type253 mapping or the name of the instances section in the supplied254 config.255 config (Config): Configuration data for attributes and optionally256 also instances.257 defaults (Optional[str], optional): Name of the default section in258 the config, if used. Defaults to DEFAULTSECT.259 Returns:260 Mapping[str, Any]: All vivified objects.261 """262 # store for vivified objects263 vivified: Mapping[str, Any] = {}264 # store for instances name and type data265 instances_config: Mapping[str, str]266 instances_is_str: bool = isinstance(instances, str)267 # get the instances data268 if instances_is_str:269 if instances in config:270 instances_config = config[instances]271 else:272 instances_config = {}273 else:274 instances_config = instances275 # extract just the attribute configuration data276 if defaults or instances_is_str:277 config = self._extract_configs(278 config=config,279 instances=instances if instances_is_str else None,280 defaults=defaults,281 )282 # create all objects and set the attributes283 vivified = {284 v: self.types[instances_config[v]]()285 for v in config286 if v in instances_config287 }288 seen: set[tuple[str, str]] = set()289 for instance, attribute, value in self._process_attributes(290 instances_config, config, vivified291 ):292 if (instance, attribute) not in seen:293 seen.add((instance, attribute))294 setattr(vivified[instance], attribute, value)295 return vivified296 @entry_exit_logging297 def vivify(298 self,299 instances: Union[Mapping[str, str], str],300 config: Config,301 defaults: Optional[str] = DEFAULTSECT,302 ) -> Mapping[str, Any]:303 """Vivify the specified objects.304 Args:305 instances (Union[Mapping[str, str], str]): Instances name type306 mapping or the name of the instances section in the supplied307 config.308 config (Config): Configuration data for attributes and optionally309 also instances.310 defaults (Optional[str], optional): Name of the default section in311 the config, if used. Defaults to DEFAULTSECT.312 Returns:313 Mapping[str, Any]: All vivified objects.314 """315 try:316 return self._vivify(317 instances=instances, config=config, defaults=defaults318 )319 except VivificationError:320 raise321 except Exception as error:322 raise VivificationError(323 "Error occurred during vivification."...

Full Screen

Full Screen

test_vivify.py

Source:test_vivify.py Github

copy

Full Screen

...190 string: str191 integer: int192 vivifiable: TestVivifiableTuple193 reference: TestVivify194def test_vivify__vivify():195 """Test that the internal vivification function returns correct values."""196 # check for empty data197 assert (198 Vivifier(types=[])._vivify(instances={}, config={}, defaults="") == {}199 )200 assert (201 Vivifier(types=[])._vivify(202 instances="instances", config={}, defaults="DEFAULT"203 )204 == {}205 )206 # check for dict args207 vivified: Mapping[str, Any] = Vivifier(types=[TestVivify])._vivify(208 instances={"a": "TestVivify", "b": "TestVivify"},209 config={210 "a": {211 "string": "test",212 "integer": "1",213 "vivifiable": "1, 2, 3",214 "reference": "b",215 },216 "b": {217 "string": "test",218 "integer": "2",219 "vivifiable": "2",220 "reference": "a",221 },222 },223 )224 assert isinstance(vivified["a"], TestVivify)225 assert vivified["a"].string == "test"226 assert vivified["a"].integer == 1227 assert vivified["a"].vivifiable == (1, 2, 3)228 assert vivified["a"].reference == vivified["b"]229 assert isinstance(vivified["b"], TestVivify)230 assert vivified["b"].string == "test"231 assert vivified["b"].integer == 2232 assert vivified["b"].vivifiable == (2,)233 assert vivified["b"].reference == vivified["a"]234 # check for config parser style args235 config: ConfigParser = ConfigParser()236 config.read_string(237 """238 [main]239 a = TestVivify240 b = TestVivify241 [DEFAULT]242 string = test243 [a]244 integer = 1245 vivifiable = 1, 2, 3246 reference = b247 [b]248 integer = 2249 vivifiable = 2250 reference = a251 """252 )253 vivified = Vivifier(types=[TestVivify])._vivify(254 instances="main", config=config, defaults="DEFAULT"255 )256def test_vivify_vivify():257 """Test vivification."""258 # simple vivification function call test259 assert Vivifier(types=[]).vivify(instances={}, config={}, defaults="") == {}260 # test invalid missing instance for config261 assert (262 Vivifier(types=[])._vivify(263 instances={}, config={"a": {"integer": "A"}}, defaults=""264 )265 == {}266 )267 # test invalid missing config for instance268 assert (269 Vivifier(types=[])._vivify(instances={"a": "A"}, config={}, defaults="")270 == {}271 )272 # check unsupported type273 class TestUnsupportedList(object):274 unsupported: list[int]275 with pytest.raises(VivificationError):276 Vivifier([TestUnsupportedList]).vivify(277 instances={"test_unsupported_list": "TestUnsupportedList"},278 config={"test_unsupported_list": {"unsupported": "test"}},279 defaults="",280 )281 # check unsupported class282 class TestUnsupportedInteger(object):283 def __init__(self) -> None:...

Full Screen

Full Screen

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:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run molecule 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!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful