How to use generate_runbook method in lisa

Best Python code snippet using lisa_python

test_testsuite.py

Source:test_testsuite.py Github

copy

Full Screen

...140 def generate_suite_instance(self) -> MockTestSuite:141 case_results = generate_cases_result()142 self.case_results = case_results[:2]143 suite_metadata = case_results[0].runtime_data.metadata.suite144 runbook = generate_runbook(is_single_env=True, local=True, remote=True)145 envs = load_environments(runbook)146 self.default_env = list(envs.values())[0]147 assert self.default_env148 test_suite = MockTestSuite(149 metadata=suite_metadata,150 )151 return test_suite152 def setUp(self) -> None:153 cleanup_cases_metadata()154 def test_expanded_nodespace(self) -> None:155 cases = generate_cases_metadata()156 for case in cases:157 self.assertIsNotNone(case.requirement)158 assert case.requirement.environment...

Full Screen

Full Screen

test_platform.py

Source:test_platform.py Github

copy

Full Screen

1# Copyright (c) Microsoft Corporation.2# Licensed under the MIT license.3from dataclasses import dataclass, field4from typing import Any, List, Optional, Type, Union5from unittest.case import TestCase6from dataclasses_json import dataclass_json7import lisa8from lisa import schema9from lisa.environment import (10 Environment,11 Environments,12 EnvironmentStatus,13 LisaException,14 constants,15 load_environments,16)17from lisa.feature import Feature18from lisa.platform_ import Platform, load_platform19from lisa.util import ResourceAwaitableException, plugin_manager20from lisa.util.logger import Logger21from selftests.test_environment import generate_runbook as generate_env_runbook22@dataclass23class MockPlatformTestData:24 prepared_envs: List[str] = field(default_factory=list)25 deployed_envs: List[str] = field(default_factory=list)26 deleted_envs: List[str] = field(default_factory=list)27@dataclass_json()28@dataclass29class MockPlatformSchema:30 # for other UT to set value31 return_prepared: bool = True32 deploy_success: bool = True33 deployed_status: EnvironmentStatus = EnvironmentStatus.Deployed34 wait_more_resource_error: bool = False35class MockPlatform(Platform):36 def __init__(self, runbook: schema.Platform) -> None:37 super().__init__(runbook=runbook)38 self.test_data = MockPlatformTestData()39 # prevent real calls40 for plugin_name, _ in plugin_manager.list_name_plugin():41 plugin_manager.unregister(name=plugin_name)42 @classmethod43 def type_name(cls) -> str:44 return constants.PLATFORM_MOCK45 @classmethod46 def supported_features(cls) -> List[Type[Feature]]:47 return []48 def set_test_config(49 self,50 return_prepared: bool = True,51 deploy_success: bool = True,52 deployed_status: EnvironmentStatus = EnvironmentStatus.Deployed,53 wait_more_resource_error: bool = False,54 ) -> None:55 self.initialize()56 self._mock_runbook.return_prepared = return_prepared57 self._mock_runbook.deploy_success = deploy_success58 self._mock_runbook.deployed_status = deployed_status59 self._mock_runbook.wait_more_resource_error = wait_more_resource_error60 def _initialize(self, *args: Any, **kwargs: Any) -> None:61 self._mock_runbook: MockPlatformSchema = self.runbook.get_extended_runbook(62 MockPlatformSchema, constants.PLATFORM_MOCK63 )64 def _prepare_environment(self, environment: Environment, log: Logger) -> bool:65 self.test_data.prepared_envs.append(environment.name)66 requirements = environment.runbook.nodes_requirement67 if self._mock_runbook.return_prepared and requirements:68 min_capabilities: List[schema.NodeSpace] = []69 for node_space in requirements:70 min_capabilities.append(node_space.generate_min_capability(node_space))71 environment.runbook.nodes_requirement = min_capabilities72 return self._mock_runbook.return_prepared73 def _deploy_environment(self, environment: Environment, log: Logger) -> None:74 if self._mock_runbook.wait_more_resource_error:75 raise ResourceAwaitableException("any", "wait more resource")76 if not self._mock_runbook.deploy_success:77 raise LisaException("mock deploy failed")78 if self._mock_runbook.return_prepared and environment.runbook.nodes_requirement:79 requirements = environment.runbook.nodes_requirement80 for node_space in requirements:81 environment.create_node_from_requirement(node_requirement=node_space)82 for node in environment.nodes.list():83 node._is_initialized = True84 self.test_data.deployed_envs.append(environment.name)85 if self._mock_runbook.deployed_status not in [86 EnvironmentStatus.Deployed,87 EnvironmentStatus.Connected,88 ]:89 raise LisaException(90 f"expected status is {self._mock_runbook.deployed_status}, "91 f"deployment should be failed"92 )93 def _delete_environment(self, environment: Environment, log: Logger) -> None:94 self.test_data.deleted_envs.append(environment.name)95 self.delete_called = True96def generate_platform(97 keep_environment: Optional[Union[str, bool]] = False,98 admin_password: str = "do not use for real",99 admin_key_file: str = "",100) -> MockPlatform:101 runbook_data = {102 constants.TYPE: constants.PLATFORM_MOCK,103 "keep_environment": keep_environment,104 "admin_password": admin_password,105 "admin_private_key_file": admin_key_file,106 }107 runbook = schema.load_by_type(schema.Platform, runbook_data)108 platform = load_platform([runbook])109 platform.initialize()110 try:111 assert isinstance(platform, MockPlatform), f"actual: {type(platform)}"112 except AssertionError:113 # as UT imported from tests package, instead of from lisa.tests package114 # ignore by assign type from current package115 platform = MockPlatform(runbook)116 return platform117def generate_environments() -> Environments:118 envs_runbook = generate_env_runbook(local=True, requirement=True)119 envs = load_environments(envs_runbook)120 return envs121class PlatformTestCase(TestCase):122 def setUp(self) -> None:123 lisa.environment._global_environment_id = 0124 def test_prepared_env_not_success_with_exception(self) -> None:125 platform = generate_platform()126 platform.set_test_config(return_prepared=False)127 envs = generate_environments()128 self.assertEqual(2, len(envs))129 with self.assertRaises(LisaException) as cm:130 [platform.prepare_environment(env) for env in envs.values()]131 self.assertEqual(132 "no capability found for environment: Environment("133 "name='customized_0', topology='subnet', nodes_raw=[{'type': 'local', "134 "'capability': {'core_count': {'min': 4}}}], nodes_requirement=None, "135 "_original_nodes_requirement=None)",136 str(cm.exception),137 )138 def test_prepared_env_success(self) -> None:139 platform = generate_platform()140 platform.set_test_config(return_prepared=True)141 envs = generate_environments()142 self.assertEqual(2, len(envs))143 prepared_environments = [144 platform.prepare_environment(env) for env in envs.values()145 ]146 self.assertEqual(2, len(prepared_environments))147 def test_prepared_env_sorted_predefined_first(self) -> None:148 platform = generate_platform()149 platform.set_test_config()150 envs = generate_environments()151 # verify init as expected152 self.assertListEqual(["customized_0", "customized_1"], [x for x in envs])153 self.assertListEqual([True, True], [x.is_predefined for x in envs.values()])154 # verify stable sort155 envs["customized_1"].is_predefined = False156 prepared_environments = [157 platform.prepare_environment(env) for env in envs.values()158 ]159 prepared_environments.sort(key=lambda x: (not x.is_predefined, x.cost))160 self.assertListEqual(161 ["customized_0", "customized_1"], [x.name for x in prepared_environments]162 )163 self.assertListEqual(164 [True, False], [x.is_predefined for x in prepared_environments]165 )166 # verify reverse sort167 envs["customized_0"].is_predefined = False168 envs["customized_1"].is_predefined = True169 prepared_environments = [170 platform.prepare_environment(env) for env in envs.values()171 ]172 prepared_environments.sort(key=lambda x: (not x.is_predefined, x.cost))173 self.assertListEqual(174 ["customized_1", "customized_0"],175 [x.name for x in prepared_environments],176 )177 self.assertListEqual(178 [True, False], [x.is_predefined for x in prepared_environments]179 )180 def test_prepared_env_sorted_by_cost(self) -> None:181 platform = generate_platform()182 envs = generate_environments()183 platform.set_test_config()184 self.assertListEqual(["customized_0", "customized_1"], [x for x in envs])185 self.assertListEqual([0, 0], [x.cost for x in envs.values()])186 envs["customized_0"].cost = 1187 envs["customized_1"].cost = 2188 prepared_environments = [189 platform.prepare_environment(env) for env in envs.values()190 ]191 prepared_environments.sort(key=lambda x: (not x.is_predefined, x.cost))192 self.assertListEqual(193 ["customized_0", "customized_1"], [x.name for x in prepared_environments]194 )195 self.assertListEqual([1, 2], [x.cost for x in prepared_environments])196 envs["customized_0"].cost = 2197 envs["customized_1"].cost = 1198 prepared_environments = [199 platform.prepare_environment(env) for env in envs.values()200 ]201 prepared_environments.sort(key=lambda x: (not x.is_predefined, x.cost))202 self.assertListEqual(203 ["customized_1", "customized_0"], [x.name for x in prepared_environments]204 )205 self.assertListEqual([1, 2], [x.cost for x in prepared_environments])206 def test_prepared_env_deploy_failed(self) -> None:207 platform = generate_platform()208 envs = generate_environments()209 platform.set_test_config(deploy_success=False)210 for env in envs.values():211 with self.assertRaises(LisaException) as cm:212 platform.deploy_environment(env)213 self.assertEqual("mock deploy failed", str(cm.exception))214 def test_prepared_env_deleted_not_ready(self) -> None:215 platform = generate_platform()216 envs = generate_environments()217 platform.set_test_config()218 for env in envs.values():219 platform.deploy_environment(env)220 self.assertEqual(EnvironmentStatus.Deployed, env.status)221 platform.delete_environment(env)...

Full Screen

Full Screen

test_environment.py

Source:test_environment.py Github

copy

Full Screen

...87 return CUSTOM_REMOTE88 @classmethod89 def type_schema(cls) -> Type[schema.TypedSchema]:90 return CustomRemoteNodeSchema91def generate_runbook(92 is_single_env: bool = False,93 local: bool = False,94 remote: bool = False,95 requirement: bool = False,96 local_remote_node_extensions: bool = False,97) -> schema.EnvironmentRoot:98 environments: List[Any] = []99 nodes: List[Any] = []100 if local:101 nodes.append(102 {103 constants.TYPE: constants.ENVIRONMENTS_NODES_LOCAL,104 constants.ENVIRONMENTS_NODES_CAPABILITY: {"core_count": {"min": 4}},105 }106 )107 if remote:108 nodes.append(109 {110 constants.TYPE: constants.ENVIRONMENTS_NODES_REMOTE,111 constants.ENVIRONMENTS_NODES_REMOTE_ADDRESS: "internal_address",112 constants.ENVIRONMENTS_NODES_REMOTE_PORT: 22,113 "public_address": "public_address",114 "public_port": 10022,115 constants.ENVIRONMENTS_NODES_REMOTE_USERNAME: "name_of_user",116 constants.ENVIRONMENTS_NODES_REMOTE_PASSWORD: "do_not_use_it",117 }118 )119 if requirement:120 nodes.append(121 {122 constants.TYPE: constants.ENVIRONMENTS_NODES_REQUIREMENT,123 "node_count": 2,124 "core_count": 8,125 "disk": {"data_disk_count": {"min": 1}},126 "network_interface": {127 "type": "NetworkInterface",128 "data_path": {129 "is_allow_set": True,130 "items": ["Sriov", "Synthetic"],131 },132 "nic_count": {"min": 1, "max": 1, "max_inclusive": True},133 "max_nic_count": 4,134 },135 }136 )137 if local_remote_node_extensions:138 nodes.extend(139 [140 {141 constants.TYPE: CUSTOM_LOCAL,142 constants.ENVIRONMENTS_NODES_CAPABILITY: {"core_count": {"min": 4}},143 "custom_local_field": CUSTOM_LOCAL,144 },145 {146 constants.TYPE: CUSTOM_REMOTE,147 constants.ENVIRONMENTS_NODES_REMOTE_ADDRESS: "internal_address",148 constants.ENVIRONMENTS_NODES_REMOTE_PORT: 22,149 "public_address": "public_address",150 "public_port": 10022,151 constants.ENVIRONMENTS_NODES_REMOTE_USERNAME: "name_of_user",152 constants.ENVIRONMENTS_NODES_REMOTE_PASSWORD: "do_not_use_it",153 "custom_remote_field": CUSTOM_REMOTE,154 },155 ]156 )157 if is_single_env:158 environments = [{"nodes": nodes}]159 else:160 for n in nodes:161 environments.append({"nodes": [n]})162 data = {constants.ENVIRONMENTS: environments}163 return schema.load_by_type(schema.EnvironmentRoot, data)164class EnvironmentTestCase(TestCase):165 def setUp(self) -> None:166 lisa.environment._global_environment_id = 0167 def test_load_empty_runbook(self) -> None:168 envs = load_environments(None)169 self.assertEqual(0, len(envs))170 self.assertEqual(False, envs.warn_as_error)171 def test_create_from_runbook_split(self) -> None:172 runbook = generate_runbook(local=True, remote=True)173 envs = load_environments(runbook)174 self.assertEqual(2, len(envs))175 for env in envs.values():176 for n in env.nodes.list():177 # mock initializing178 n._is_initialized = True179 self.assertEqual(1, len(env.nodes))180 def test_create_from_runbook_merged(self) -> None:181 runbook = generate_runbook(is_single_env=True, local=True, remote=True)182 envs = load_environments(runbook)183 self.assertEqual(1, len(envs))184 for env in envs.values():185 for n in env.nodes.list():186 # mock initializing187 n._is_initialized = True188 self.assertEqual(2, len(env.nodes))189 def test_create_from_runbook_cap(self) -> None:190 runbook = generate_runbook(local=True, requirement=True)191 envs = load_environments(runbook)192 self.assertEqual(2, len(envs))193 env = envs.get("customized_0")194 assert env195 for n in env.nodes.list():196 # mock initializing197 n._is_initialized = True198 self.assertEqual(search_space.IntRange(min=4), n.capability.core_count)199 self.assertIsNone(n.capability.disk)200 # check from env capability201 env_cap = env.capability202 self.assertEqual(1, len(env_cap.nodes))203 self.assertEqual(search_space.IntRange(min=4), env_cap.nodes[0].core_count)204 self.assertIsNone(env_cap.nodes[0].disk)205 # test pure node_requirement206 env = envs.get("customized_1")207 assert env208 env_cap = env.capability209 assert env_cap.nodes[0].disk210 assert env_cap.nodes[0].network_interface211 self.assertEqual(2, len(env_cap.nodes))212 self.assertEqual(8, env_cap.nodes[0].core_count)213 self.assertEqual(214 search_space.IntRange(min=1), env_cap.nodes[0].disk.data_disk_count215 )216 self.assertEqual(217 search_space.IntRange(min=1, max=1),218 env_cap.nodes[0].network_interface.nic_count,219 )220 def test_create_from_requirement(self) -> None:221 requirement = simple_requirement(min_count=2)222 env_requirement = requirement.environment223 assert env_requirement224 envs = load_environments(None)225 env = envs.get_or_create(requirement=env_requirement)226 assert env227 self.assertEqual(1, len(envs))228 requirement = simple_requirement(min_count=2)229 env_requirement = requirement.environment230 assert env_requirement231 env = envs.get_or_create(requirement=env_requirement)232 self.assertEqual(1, len(envs), "get or create again won't create new")233 assert env234 self.assertEqual(0, len(env.nodes))235 self.assertSequenceEqual([], env.runbook.nodes)236 assert env.runbook.nodes_requirement237 self.assertEqual(2, len(env.runbook.nodes_requirement))238 def test_create_from_custom_local_remote(self) -> None:239 runbook = generate_runbook(240 local_remote_node_extensions=True, is_single_env=True241 )242 envs = load_environments(runbook)243 self.assertEqual(1, len(envs))244 for env in envs.values():245 done: int = 0246 for n in env.nodes.list():247 if n.type_name() == CUSTOM_LOCAL:248 l_n = cast(CustomLocalNode, n)249 self.assertEqual(l_n.custom_local_field, CUSTOM_LOCAL)250 done += 1251 elif n.type_name() == CUSTOM_REMOTE:252 r_n = cast(CustomRemoteNode, n)253 self.assertEqual(r_n.custom_remote_field, CUSTOM_REMOTE)...

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 lisa 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