How to use get_system method in localstack

Best Python code snippet using localstack_python

test_control_node_repository.py

Source:test_control_node_repository.py Github

copy

Full Screen

1import unittest2from mock import MagicMock, call, patch3from bounded_contexts.central_console.domain_services import control_node_repository4from bounded_contexts.central_console.domain_services.abstract_contact_person_repository import \5 AbstractContactPersonRepository6from bounded_contexts.central_console.domain_services.abstract_platform_repository import AbstractPlatformRepository7from bounded_contexts.central_console.domain_services.abstract_server_repository import AbstractServerRepository8from bounded_contexts.central_console.domain_services.abstract_system_repository import AbstractSystemRepository9from bounded_contexts.central_console.domain_services.control_node_repository import CONFIG_SERVER_ID, CONFIG_VERSION10from bounded_contexts.central_console.models.contact_person import ContactPerson11from bounded_contexts.central_console.models.control_node import ControlNode12from bounded_contexts.central_console.models.platform import Platform13from bounded_contexts.central_console.models.server import Server14from bounded_contexts.central_console.models.system import System15from shared_kernel.config.domain_services.abstract_config_repository import AbstractConfigRepository16from shared_kernel.config.models.config import Config17from unit_tests.base_test_case import BaseTestCase18class TestControlNodeRepository(BaseTestCase):19 def setUp(self):20 self.conf_repo = MagicMock(spec=AbstractConfigRepository)21 self.system_repo = MagicMock(spec=AbstractSystemRepository)22 self.server_repo = MagicMock(spec=AbstractServerRepository)23 self.contact_repo = MagicMock(spec=AbstractContactPersonRepository)24 self.platform_repo = MagicMock(spec=AbstractPlatformRepository)25 self.repo = control_node_repository.ControlNodeRepository(26 ControlNode,27 self.conf_repo,28 self.system_repo,29 self.server_repo,30 self.contact_repo,31 self.platform_repo32 )33 @patch.object(control_node_repository, 'get_bytes_from_uuid', autospec=True)34 def test_get_control_node_with_public_ip(self, convert_id_mock):35 server_id_config = MagicMock(spec=Config)36 softversion_config = MagicMock(spec=Config)37 self.system_repo.get_system.return_value = MagicMock(spec=System)38 self.server_repo.get_server.return_value = MagicMock(spec=Server)39 self.conf_repo.get_config.side_effect = [server_id_config, softversion_config]40 self.contact_repo.get_contact_person.return_value = MagicMock(spec=ContactPerson)41 self.set_named_tuple_mock_bool_context_value(self.contact_repo.get_contact_person.return_value, True)42 self.platform_repo.get_platform.return_value = MagicMock(spec=Platform)43 self.set_named_tuple_mock_bool_context_value(self.platform_repo.get_platform.return_value, True)44 self.platform_repo.get_platform.return_value.public_ip = '10.0.0.0'45 actual_node = self.repo.get_control_node()46 self.conf_repo.get_config.assert_has_calls([47 call(CONFIG_SERVER_ID),48 call(CONFIG_VERSION)49 ])50 self.system_repo.get_system.assert_called_once_with()51 self.server_repo.get_server.assert_called_once_with(convert_id_mock.return_value)52 self.contact_repo.get_contact_person.assert_called_once_with()53 self.platform_repo.get_platform.assert_called_once_with(self.system_repo.get_system.return_value.admin_ip)54 self.assertEqual(actual_node.node_id, self.system_repo.get_system.return_value.id)55 self.assertEqual(actual_node.name, self.system_repo.get_system.return_value.name)56 self.assertEqual(actual_node.description, self.server_repo.get_server.return_value.descr)57 self.assertEqual(actual_node.platform, self.platform_repo.get_platform.return_value.name)58 self.assertEqual(actual_node.appliance_type, self.platform_repo.get_platform.return_value.appliance_type)59 self.assertEqual(actual_node.software_version, softversion_config.value)60 self.assertEqual(61 actual_node.intelligence_version,62 self.platform_repo.get_platform.return_value.threat_intelligence_version63 )64 self.assertEqual(actual_node.contact_email, self.contact_repo.get_contact_person.return_value.email)65 self.assertEqual(actual_node.contact_name, self.contact_repo.get_contact_person.return_value.name)66 self.assertEqual(actual_node.admin_ip_address, self.platform_repo.get_platform.return_value.public_ip)67 self.assertEqual(actual_node.vpn_ip_address, self.system_repo.get_system.return_value.vpn_ip)68 @patch.object(control_node_repository, 'get_bytes_from_uuid', autospec=True)69 def test_get_control_node_with_admin_ip(self, convert_id_mock):70 server_id_config = MagicMock(spec=Config)71 softversion_config = MagicMock(spec=Config)72 self.system_repo.get_system.return_value = MagicMock(spec=System)73 self.system_repo.get_system.return_value.admin_ip = '192.168.0.1'74 self.system_repo.get_system.return_value.ha_ip = ''75 self.server_repo.get_server.return_value = MagicMock(spec=Server)76 self.conf_repo.get_config.side_effect = [server_id_config, softversion_config]77 self.contact_repo.get_contact_person.return_value = MagicMock(spec=ContactPerson)78 self.contact_repo.get_contact_person.return_value.__len__ = MagicMock(return_value=1)79 self.platform_repo.get_platform.return_value = MagicMock(spec=Platform)80 self.set_named_tuple_mock_bool_context_value(self.platform_repo.get_platform.return_value, True)81 self.platform_repo.get_platform.return_value.public_ip = None82 actual_node = self.repo.get_control_node()83 self.conf_repo.get_config.assert_has_calls([84 call(CONFIG_SERVER_ID),85 call(CONFIG_VERSION)86 ])87 self.system_repo.get_system.assert_called_once_with()88 self.server_repo.get_server.assert_called_once_with(convert_id_mock.return_value)89 self.contact_repo.get_contact_person.assert_called_once_with()90 self.platform_repo.get_platform.assert_called_once_with(self.system_repo.get_system.return_value.admin_ip)91 self.assertEqual(actual_node.node_id, self.system_repo.get_system.return_value.id)92 self.assertEqual(actual_node.name, self.system_repo.get_system.return_value.name)93 self.assertEqual(actual_node.description, self.server_repo.get_server.return_value.descr)94 self.assertEqual(actual_node.platform, self.platform_repo.get_platform.return_value.name)95 self.assertEqual(actual_node.appliance_type, self.platform_repo.get_platform.return_value.appliance_type)96 self.assertEqual(actual_node.software_version, softversion_config.value)97 self.assertEqual(98 actual_node.intelligence_version,99 self.platform_repo.get_platform.return_value.threat_intelligence_version100 )101 self.assertEqual(actual_node.contact_email, self.contact_repo.get_contact_person.return_value.email)102 self.assertEqual(actual_node.contact_name, self.contact_repo.get_contact_person.return_value.name)103 self.assertEqual(actual_node.admin_ip_address, self.system_repo.get_system.return_value.admin_ip)104 self.assertEqual(actual_node.vpn_ip_address, self.system_repo.get_system.return_value.vpn_ip)105 @patch.object(control_node_repository, 'get_bytes_from_uuid', autospec=True)106 def test_get_control_node_with_ha_ip(self, convert_id_mock):107 server_id_config = MagicMock(spec=Config)108 softversion_config = MagicMock(spec=Config)109 self.system_repo.get_system.return_value = MagicMock(spec=System)110 self.system_repo.get_system.return_value.admin_ip = '192.168.0.1'111 self.system_repo.get_system.return_value.ha_ip = '192.168.0.10'112 self.server_repo.get_server.return_value = MagicMock(spec=Server)113 self.conf_repo.get_config.side_effect = [server_id_config, softversion_config]114 self.contact_repo.get_contact_person.return_value = MagicMock(spec=ContactPerson)115 self.set_named_tuple_mock_bool_context_value(self.contact_repo.get_contact_person.return_value, True)116 self.platform_repo.get_platform.return_value = MagicMock(spec=Platform)117 self.set_named_tuple_mock_bool_context_value(self.platform_repo.get_platform.return_value, True)118 self.platform_repo.get_platform.return_value.public_ip = None119 actual_node = self.repo.get_control_node()120 self.conf_repo.get_config.assert_has_calls([121 call(CONFIG_SERVER_ID),122 call(CONFIG_VERSION)123 ])124 self.system_repo.get_system.assert_called_once_with()125 self.server_repo.get_server.assert_called_once_with(convert_id_mock.return_value)126 self.contact_repo.get_contact_person.assert_called_once_with()127 self.platform_repo.get_platform.assert_called_once_with(self.system_repo.get_system.return_value.admin_ip)128 self.assertEqual(actual_node.node_id, self.system_repo.get_system.return_value.id)129 self.assertEqual(actual_node.name, self.system_repo.get_system.return_value.name)130 self.assertEqual(actual_node.description, self.server_repo.get_server.return_value.descr)131 self.assertEqual(actual_node.platform, self.platform_repo.get_platform.return_value.name)132 self.assertEqual(actual_node.appliance_type, self.platform_repo.get_platform.return_value.appliance_type)133 self.assertEqual(actual_node.software_version, softversion_config.value)134 self.assertEqual(135 actual_node.intelligence_version,136 self.platform_repo.get_platform.return_value.threat_intelligence_version137 )138 self.assertEqual(actual_node.contact_email, self.contact_repo.get_contact_person.return_value.email)139 self.assertEqual(actual_node.contact_name, self.contact_repo.get_contact_person.return_value.name)140 self.assertEqual(actual_node.admin_ip_address, self.system_repo.get_system.return_value.ha_ip)141 self.assertEqual(actual_node.vpn_ip_address, self.system_repo.get_system.return_value.vpn_ip)142 @patch.object(control_node_repository, 'get_bytes_from_uuid', autospec=True)143 def test_get_control_node_no_contact_person(self, convert_id_mock):144 server_id_config = MagicMock(spec=Config)145 softversion_config = MagicMock(spec=Config)146 self.system_repo.get_system.return_value = MagicMock(spec=System)147 self.server_repo.get_server.return_value = MagicMock(spec=Server)148 self.conf_repo.get_config.side_effect = [server_id_config, softversion_config]149 self.contact_repo.get_contact_person.return_value = None150 self.platform_repo.get_platform.return_value = MagicMock(spec=Platform)151 self.set_named_tuple_mock_bool_context_value(self.platform_repo.get_platform.return_value, True)152 self.platform_repo.get_platform.return_value.public_ip = '10.0.0.1'153 actual_node = self.repo.get_control_node()154 self.assertEqual(actual_node.node_id, self.system_repo.get_system.return_value.id)155 self.assertEqual(actual_node.name, self.system_repo.get_system.return_value.name)156 self.assertEqual(actual_node.description, self.server_repo.get_server.return_value.descr)157 self.assertEqual(actual_node.platform, self.platform_repo.get_platform.return_value.name)158 self.assertEqual(actual_node.appliance_type, self.platform_repo.get_platform.return_value.appliance_type)159 self.assertEqual(actual_node.software_version, softversion_config.value)160 self.assertEqual(161 actual_node.intelligence_version,162 self.platform_repo.get_platform.return_value.threat_intelligence_version163 )164 self.assertEqual(actual_node.contact_email, None)165 self.assertEqual(actual_node.contact_name, None)166 self.assertEqual(actual_node.admin_ip_address, self.platform_repo.get_platform.return_value.public_ip)167 self.assertEqual(actual_node.vpn_ip_address, self.system_repo.get_system.return_value.vpn_ip)168 @patch.object(control_node_repository, 'get_bytes_from_uuid', autospec=True)169 def test_get_control_node_no_platform_data(self, convert_id_mock):170 server_id_config = MagicMock(spec=Config)171 softversion_config = MagicMock(spec=Config)172 self.system_repo.get_system.return_value = MagicMock(spec=System)173 self.system_repo.get_system.return_value.ha_ip = ''174 self.server_repo.get_server.return_value = MagicMock(spec=Server)175 self.conf_repo.get_config.side_effect = [server_id_config, softversion_config]176 self.contact_repo.get_contact_person.return_value = MagicMock(spec=ContactPerson)177 self.set_named_tuple_mock_bool_context_value(self.contact_repo.get_contact_person.return_value, True)178 self.platform_repo.get_platform.return_value = None179 actual_node = self.repo.get_control_node()180 self.assertEqual(actual_node.node_id, self.system_repo.get_system.return_value.id)181 self.assertEqual(actual_node.name, self.system_repo.get_system.return_value.name)182 self.assertEqual(actual_node.description, self.server_repo.get_server.return_value.descr)183 self.assertEqual(actual_node.platform, None)184 self.assertEqual(actual_node.appliance_type, None)185 self.assertEqual(actual_node.software_version, softversion_config.value)186 self.assertEqual(actual_node.intelligence_version, None)187 self.assertEqual(actual_node.contact_email, self.contact_repo.get_contact_person.return_value.email)188 self.assertEqual(actual_node.contact_name, self.contact_repo.get_contact_person.return_value.name)189 self.assertEqual(actual_node.admin_ip_address, self.system_repo.get_system.return_value.admin_ip)190 self.assertEqual(actual_node.vpn_ip_address, self.system_repo.get_system.return_value.vpn_ip)191if __name__ == '__main__':...

Full Screen

Full Screen

client_system_api.py

Source:client_system_api.py Github

copy

Full Screen

...8 获取脚本运行的平台9 1.18 新增 获取脚本运行的平台10 :return: int 0：Window；1：IOS；2：Android；-1：其他11 """12 return extra_client_api.get_system(ModName, ModClientSystemName).GetPlatform()13def broadcast_event(event_name, event_data):14 # type: (str, dict) -> None15 """16 本地广播事件，客户端system广播的事件仅客户端system能监听。17 :param event_name: str 事件名18 :param event_data: dict 事件参数19 :return:20 """21 warnings.warn("1.20 BroadcastEvent目前暂时无法使用，需要通过服务端BroadcastToAllClient广播", DeprecationWarning)22 extra_client_api.get_system(ModName, ModClientSystemName).BroadcastEvent(event_name, event_data)23def create_event_data():24 # type: () -> dict25 """26 创建自定义事件的数据，eventData用于发送事件。创建的eventData可以理解为一个dict，可以嵌套赋值dict,list和基本数据类型，但不支持tuple27 :return: dict 事件数据28 """29 warnings.warn("1.20 目前CreateEventData暂时没有特殊功能，可以使用{}代替", DeprecationWarning)30 return extra_client_api.get_system(ModName, ModClientSystemName).CreateEventData()31def define_event(event_name):32 # type: (str) -> None33 """34 定义自定义事件35 :param event_name: str 事件名36 :return:37 """38 extra_client_api.get_system(ModName, ModClientSystemName).DefineEvent(event_name)39def listen_for_event(namespace, system_name, event_name, instance, func, priority=0):40 """41 注册监听某个系统抛出的事件。若监听引擎事件时，namespace和systemName分别为GetEngineNamespace()和GetEngineSystemName()42 :param namespace: str 所监听事件的来源系统的namespace43 :param system_name: str 所监听事件的来源系统的systemName44 :param event_name: str 事件名45 :param instance: instance 回调函数所属的类的实例46 :param func: function 回调函数47 :param priority: int 这个回调函数的优先级。默认值为0，这个数值越大表示被执行的优先级越高，最高为1048 :return:49 """50 extra_client_api.get_system(ModName, ModClientSystemName51 ).ListenForEvent(namespace, system_name, event_name, instance, func, priority)52def notify_to_server(event_name, event_data):53 # type: (str, dict) -> None54 """55 客户端发送事件到服务器56 :param event_name: str 事件名57 :param event_data: dict 事件参数，一般用CreateEventData的返回值58 :return:59 """60 extra_client_api.get_system(ModName, ModClientSystemName).NotifyToServer(event_name, event_data)61def un_define_event(event_name):62 # type: (str) -> None63 """64 取消自定义事件65 :param event_name: str 事件名66 :return:67 """68 extra_client_api.get_system(ModName, ModClientSystemName).UnDefineEvent(event_name)69def un_listen_all_events():70 """71 反注册监听某个系统抛出的所有事件，即不再监听。72 :return:73 """74 extra_client_api.get_system(ModName, ModClientSystemName).UnListenAllEvents()75def un_listen_for_event(namespace, system_name, event_name, instance, func, priority=0):76 """77 反注册监听某个系统抛出的事件，即不再监听。若是引擎事件，则namespace和systemName分别为GetEngineNamespace()和GetEngineSystemName()。与ListenForEvent对应。78 :param namespace: str 所监听事件的来源系统的namespace79 :param system_name: str 所监听事件的来源系统的systemName80 :param event_name: str 事件名81 :param instance: instance 回调函数所属的类的实例82 :param func: function 回调函数83 :param priority: int 这个回调函数的优先级。默认值为0，这个数值越大表示被执行的优先级越高84 :return:85 """86 extra_client_api.get_system(ModName, ModClientSystemName87 ).UnListenForEvent(namespace, system_name, event_name, instance, func, priority)88def destroy_entity(entity_id):89 # type: (str) -> bool90 """91 实体销毁接口92 :param entity_id: str 销毁的实体ID93 :return: bool 是否销毁成功94 """95 return extra_client_api.get_system(ModName, ModClientSystemName).DestroyEntity(entity_id)96def create_engine_effect(path, bind_entity, ani_name):97 # type: (str, str, str) -> int98 """99 用于创建模型挂接特效，具体参见创建特效部分内容。100 :param path: str 特效资源路径，需要加上后缀名（一般是json）101 :param bind_entity: str 绑定实体的Id102 :param ani_name: str 选择使用哪个模型动作的特效103 :return: int或None effectEntityId或者None104 """105 return extra_client_api.get_system(ModName, ModClientSystemName).CreateEngineEffect(path, bind_entity, ani_name)106def create_engine_particle(path, pos=(0, 0, 0)):107 # type: (str, tuple) -> int108 """109 用于创建粒子特效，具体参见创建特效部分内容110 :param path: str 特效资源路径，需要加上后缀名（一般是json）111 :param pos: tuple (float,float,float)创建位置坐标112 :return: int或None particleEntityId或者None113 """114 return extra_client_api.get_system(ModName, ModClientSystemName).CreateEngineParticle(path, pos)115def create_engine_sfx(path, pos=None, rot=None, scale=None):116 # type: (str, tuple, tuple, float) -> int117 """118 用于创建序列帧特效，具体参见创建特效部分内容119 :param path: str 特效资源路径，不用后缀名120 :param pos: tuple (float,float,float)创建位置，可选121 :param rot: tuple (float,float)角度，可选122 :param scale: float 缩放系数，可选123 :return: int或None frameEntityId或者None124 """125 return extra_client_api.get_system(ModName, ModClientSystemName).CreateEngineSfx(path, pos, rot, scale)126def create_engine_sfx_from_editor(path, pos=None, rot=None, scale=None):127 # type: (str, tuple, tuple, float) -> int128 """129 指使用资源包中effects/xxx.json，按照编辑器中编辑好的参数创建序列帧。支持环状序列帧130 :param path: str 特效资源路径，不用后缀名131 :param pos: tuple (float,float,float)创建位置，可选132 :param rot: tuple (float,float)角度，可选133 :param scale: float 缩放系数，可选134 :return: int或None frameEntityId或者None135 """136 return extra_client_api.get_system(ModName, ModClientSystemName).CreateEngineSfxFromEditor(path, pos, rot, scale)137def create_component(entity_id, namespace, name):138 """139 给实体创建组件，与GetComponent类似，如果已经创建会自动直接Get140 :param entity_id: str 该组件属主的实体id141 :param namespace: str 组件的命名空间，registerComponent的namespace142 :param name: str 组件的名字143 :return: instance 组件实例144 """145 return clientApi.CreateComponent(entity_id, namespace, name)146def get_component(entity_id, namespace, name):147 """148 获取实体的组件。一般用来判断某个组件是否创建过，其他情况请使用CreateComponent149 :param entity_id: str 该组件属主的实体id150 :param namespace: str 组件的命名空间，registerComponent的namespace...

Full Screen

Full Screen

configInfo.py

Source:configInfo.py Github

copy

Full Screen

2定义配置类，读取配置文件中的配置信息包括[HTTP][USER][EMAIL][LOG]3"""4from read_config import get_system, get_email, get_log5class SystemConfig:6 super_name = get_system('super_name')7 super_pwd = get_system('super_pwd')8 username = get_system('username')9 password = get_system('password')10 pass_rate = float(get_system('pass_rate'))11 system_name = get_system('system_name')12class EmailConfig:13 on_off = get_email('on_off')14 title = get_email('title')15 message = get_email('message')16 password = get_email('password')17 receiver = get_email('receiver')18 server = get_email('server')19 sender = get_email('sender')20class LogConfig:21 file_name = get_log('file_name')22 backup = int(get_log('backup'))23 console_level = get_log('console_level')...

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 localstack automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

... View sample repo... View DocumentationSign up Free
›
_

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

...
Sign up with GoogleSign up with Email
Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful