Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use add_service method in localstack
Best Python code snippet using localstack_python
test_coordinator.py
Source:test_coordinator.py 
...11 def test_random_coordinator_1_service(self):12 service1 = Mock()13 service1.execite_query.return_value = {"status": "success", "payload": {"answer": "Service1"}} 14 coordinator = RandomResultServiceCoordinator()15 coordinator.add_service(service1)16 result = coordinator.execute_query("Question")17 self.assertIsNotNone(result)18 self.assertEqual(result, {'status': 'success', 'payload': {'answer': 'Service1'}})19 result = coordinator.execute_query("Question")20 self.assertIsNotNone(result)21 self.assertEqual(result, {'status': 'success', 'payload': {'answer': 'Service1'}})22 result = coordinator.execute_query("Question")23 self.assertIsNotNone(result)24 self.assertEqual(result, {'status': 'success', 'payload': {'answer': 'Service1'}})25 def test_random_coordinator_n_services(self):26 service1 = Mock()27 service1.execite_query.return_value = {"status": "success", "payload": {"answer": "Service1"}} 28 service2 = Mock()29 service2.execite_query.return_value = {"status": "success", "payload": {"answer": "Service2"}} 30 service3 = Mock()31 service3.execite_query.return_value = {"status": "success", "payload": {"answer": "Service3"}} 32 coordinator = RandomResultServiceCoordinator()33 coordinator.add_service(service1)34 coordinator.add_service(service2)35 coordinator.add_service(service3)36 result = coordinator.execute_query("Question")37 self.assertIsNotNone(result)38 self.assertTrue(result in [{'status': 'success', 'payload': {'answer': 'Service1'}},39 {'status': 'success', 'payload': {'answer': 'Service2'}},40 {'status': 'success', 'payload': {'answer': 'Service3'}}])41 result = coordinator.execute_query("Question")42 self.assertIsNotNone(result)43 self.assertTrue(result in [{'status': 'success', 'payload': {'answer': 'Service1'}},44 {'status': 'success', 'payload': {'answer': 'Service2'}},45 {'status': 'success', 'payload': {'answer': 'Service3'}}])46 result = coordinator.execute_query("Question")47 self.assertIsNotNone(result)48 self.assertTrue(result in [{'status': 'success', 'payload': {'answer': 'Service1'}},49 {'status': 'success', 'payload': {'answer': 'Service2'}},50 {'status': 'success', 'payload': {'answer': 'Service3'}}])51 def test_random_coordinator_1_services_skip_failure(self):52 service1 = Mock()53 service1.execite_query.return_value = {"status": "failure", "payload": {"answer": "Service1"}}54 coordinator = RandomResultServiceCoordinator()55 coordinator.add_service(service1)56 result = coordinator.execute_query("Question")57 self.assertIsNone(result)58 def test_random_coordinator_1_services_no_skip_failure(self):59 service1 = Mock()60 service1.execite_query.return_value = {"status": "failure", "payload": {}}61 coordinator = RandomResultServiceCoordinator()62 coordinator.add_service(service1)63 result = coordinator.execute_query("Question", skip_failures=False)64 self.assertIsNotNone(result)65 self.assertEqual(result, {"status": "failure", "payload": {}})66 def test_try_next_on_failure_one_succeed(self):67 service1 = Mock()68 service1.execite_query.return_value = {"status": "failure", "payload": {"answer": "Service1"}}69 service1.name = "Service1"70 service2 = Mock()71 service2.execite_query.return_value = {"status": "failure", "payload": {"answer": "Service2"}}72 service2.name = "Service2"73 service3 = Mock()74 service3.execite_query.return_value = {"status": "success", "payload": {"answer": "Service3"}}75 service3.name = "Service3"76 coordinator = RandomResultServiceCoordinator()77 coordinator.add_service(service1)78 coordinator.add_service(service2)79 coordinator.add_service(service3)80 result = coordinator.execute_query("Question", skip_failures=True, try_next_on_failure=True)81 self.assertIsNotNone(result)82 self.assertEqual(result, {'status': 'success', 'payload': {'answer': 'Service3'}})83 def test_try_next_on_failure_all_fail(self):84 service1 = Mock()85 service1.execite_query.return_value = {"status": "failure", "payload": {"answer": "Service1"}}86 service1.name = "Service1"87 service2 = Mock()88 service2.execite_query.return_value = {"status": "failure", "payload": {"answer": "Service2"}}89 service2.name = "Service2"90 service3 = Mock()91 service3.execite_query.return_value = {"status": "failure", "payload": {"answer": "Service3"}}92 service3.name = "Service3"93 coordinator = RandomResultServiceCoordinator()94 coordinator.add_service(service1)95 coordinator.add_service(service2)96 coordinator.add_service(service3)97 result = coordinator.execute_query("Question", skip_failures=True, try_next_on_failure=True)98 self.assertIsNone(result)99class ServiceRoundRobinResultCoordinatorTests(unittest.TestCase):100 def test_round_robin_coordinator_no_services(self):101 coordinator = RoundRobinResultServiceCoordinator()102 result = coordinator.execute_query("Question")103 self.assertIsNone(result)104 def test_round_robin_coordinator_1_service(self):105 service1 = Mock()106 service1.execite_query.return_value = {"status": "success", "payload": {"answer": "Service1"}} 107 coordinator = RoundRobinResultServiceCoordinator()108 coordinator.add_service(service1)109 result = coordinator.execute_query("Question")110 self.assertIsNotNone(result)111 self.assertEqual(result, {'status': 'success', 'payload': {'answer': 'Service1'}})112 result = coordinator.execute_query("Question")113 self.assertIsNotNone(result)114 self.assertEqual(result, {'status': 'success', 'payload': {'answer': 'Service1'}})115 result = coordinator.execute_query("Question")116 self.assertIsNotNone(result)117 self.assertEqual(result, {'status': 'success', 'payload': {'answer': 'Service1'}})118 def test_round_robin_coordinator_n_services(self):119 service1 = Mock()120 service1.execite_query.return_value = {"status": "success", "payload": {"answer": "Service1"}} 121 service2 = Mock()122 service2.execite_query.return_value = {"status": "success", "payload": {"answer": "Service2"}} 123 service3 = Mock()124 service3.execite_query.return_value = {"status": "success", "payload": {"answer": "Service3"}} 125 coordinator = RoundRobinResultServiceCoordinator()126 coordinator.add_service(service1)127 coordinator.add_service(service2)128 coordinator.add_service(service3)129 result = coordinator.execute_query("Question")130 self.assertIsNotNone(result)131 self.assertEqual(result, {'status': 'success', 'payload': {'answer': 'Service1'}})132 result = coordinator.execute_query("Question")133 self.assertIsNotNone(result)134 self.assertEqual(result, {'status': 'success', 'payload': {'answer': 'Service2'}})135 result = coordinator.execute_query("Question")136 self.assertIsNotNone(result)137 self.assertEqual(result, {'status': 'success', 'payload': {'answer': 'Service3'}})138 result = coordinator.execute_query("Question")139 self.assertIsNotNone(result)140 self.assertEqual(result, {'status': 'success', 'payload': {'answer': 'Service1'}})141 def test_try_next_on_failure_one_succeed(self):142 service1 = Mock()143 service1.execite_query.return_value = {"status": "failure", "payload": {"answer": "Service1"}}144 service1.name = "Service1"145 service2 = Mock()146 service2.execite_query.return_value = {"status": "failure", "payload": {"answer": "Service2"}}147 service2.name = "Service2"148 service3 = Mock()149 service3.execite_query.return_value = {"status": "success", "payload": {"answer": "Service3"}}150 service3.name = "Service3"151 coordinator = RoundRobinResultServiceCoordinator()152 coordinator.add_service(service1)153 coordinator.add_service(service2)154 coordinator.add_service(service3)155 result = coordinator.execute_query("Question", skip_failures=True, try_next_on_failure=True)156 self.assertIsNotNone(result)157 self.assertEqual(result, {'status': 'success', 'payload': {'answer': 'Service3'}})158 def test_try_next_on_failure_all_fail(self):159 service1 = Mock()160 service1.execite_query.return_value = {"status": "failure", "payload": {"answer": "Service1"}}161 service1.name = "Service1"162 service2 = Mock()163 service2.execite_query.return_value = {"status": "failure", "payload": {"answer": "Service2"}}164 service2.name = "Service2"165 service3 = Mock()166 service3.execite_query.return_value = {"status": "failure", "payload": {"answer": "Service3"}}167 service3.name = "Service3"168 coordinator = RoundRobinResultServiceCoordinator()169 coordinator.add_service(service1)170 coordinator.add_service(service2)171 coordinator.add_service(service3)172 result = coordinator.execute_query("Question", skip_failures=True, try_next_on_failure=True)173 self.assertIsNone(result)174class ServiceAllResultsCoordinatorTests(unittest.TestCase):175 def test_all_coordinator_no_services(self):176 coordinator = AllResultsServiceCoordinator()177 result = coordinator.execute_query("Question")178 self.assertIsNone(result)179 def test_all_coordinator_1_service(self):180 service1 = Mock()181 service1.execite_query.return_value = {"status": "success", "payload": {"answer": "Service1"}} 182 coordinator = AllResultsServiceCoordinator()183 coordinator.add_service(service1)184 result = coordinator.execute_query("Question")185 self.assertIsNotNone(result)186 self.assertEqual(result, [{"status": "success", "payload": {"answer": "Service1"}}])187 result = coordinator.execute_query("Question")188 self.assertIsNotNone(result)189 self.assertEqual(result, [{"status": "success", "payload": {"answer": "Service1"}} ])190 def test_all_coordinator_n_services(self):191 service1 = Mock()192 service1.execite_query.return_value = {"status": "success", "payload": {"answer": "Service1"}}193 service2 = Mock()194 service2.execite_query.return_value = {"status": "success", "payload": {"answer": "Service2"}} 195 service3 = Mock()196 service3.execite_query.return_value = {"status": "success", "payload": {"answer": "Service3"}} 197 coordinator = AllResultsServiceCoordinator()198 coordinator.add_service(service1)199 coordinator.add_service(service2)200 coordinator.add_service(service3)201 result = coordinator.execute_query("Question")202 self.assertIsNotNone(result)203 self.assertEqual(result, [{'status': 'success', 'payload': {'answer': 'Service1'}},204 {'status': 'success', 'payload': {'answer': 'Service2'}},205 {'status': 'success', 'payload': {'answer': 'Service3'}}])206 def test_all_coordinator_1_service_skip_failures(self):207 service1 = Mock()208 service1.execite_query.return_value = {"status": "failure", "payload": {}}209 coordinator = AllResultsServiceCoordinator()210 coordinator.add_service(service1)211 result = coordinator.execute_query("Question", skip_failures=True)212 self.assertIsNotNone(result)213 self.assertEqual(result, [])214 def test_all_coordinator_1_service_no_skip_failures(self):215 service1 = Mock()216 service1.execite_query.return_value = {"status": "failure", "payload": {}}217 coordinator = AllResultsServiceCoordinator()218 coordinator.add_service(service1)219 result = coordinator.execute_query("Question", skip_failures=False)220 self.assertIsNotNone(result)221 self.assertEqual(result, [{"status": "failure", "payload": {}}])222 def test_all_coordinator_n_services_skip_failures(self):223 service1 = Mock()224 service1.execite_query.return_value = {"status": "failure", "payload": {}}225 service2 = Mock()226 service2.execite_query.return_value = {"status": "success", "payload": {"answer": "Service2"}}227 service3 = Mock()228 service3.execite_query.return_value = {"status": "success", "payload": {"answer": "Service3"}}229 coordinator = AllResultsServiceCoordinator()230 coordinator.add_service(service1)231 coordinator.add_service(service2)232 coordinator.add_service(service3)233 result = coordinator.execute_query("Question", skip_failures=True)234 self.assertIsNotNone(result)235 self.assertEqual(result, [{'status': 'success', 'payload': {'answer': 'Service2'}},236 {'status': 'success', 'payload': {'answer': 'Service3'}}])237 def test_all_coordinator_n_services_no_skip_failures(self):238 service1 = Mock()239 service1.execite_query.return_value = {"status": "failure", "payload": {}}240 service2 = Mock()241 service2.execite_query.return_value = {"status": "success", "payload": {"answer": "Service2"}}242 service3 = Mock()243 service3.execite_query.return_value = {"status": "success", "payload": {"answer": "Service3"}}244 coordinator = AllResultsServiceCoordinator()245 coordinator.add_service(service1)246 coordinator.add_service(service2)247 coordinator.add_service(service3)248 result = coordinator.execute_query("Question", skip_failures=False)249 self.assertIsNotNone(result)250 self.assertEqual(result, [{"status": "failure", "payload": {}},251 {'status': 'success', 'payload': {'answer': 'Service2'}},...
test_conf.py
Source:test_conf.py
...33 assert config.name == NAME34def test_equality(config):35 assert config == config36 atv2 = conf.AppleTV(ADDRESS_1, NAME)37 atv2.add_service(conf.AirPlayService(IDENTIFIER_1, PORT_1))38 assert config != atv239def test_add_services_and_get(config):40 config.add_service(DMAP_SERVICE)41 config.add_service(MRP_SERVICE)42 config.add_service(AIRPLAY_SERVICE)43 services = config.services44 assert len(services), 345 assert DMAP_SERVICE in services46 assert MRP_SERVICE in services47 assert AIRPLAY_SERVICE in services48 assert config.get_service(Protocol.DMAP) == DMAP_SERVICE49 assert config.get_service(Protocol.MRP) == MRP_SERVICE50 assert config.get_service(Protocol.AirPlay) == AIRPLAY_SERVICE51def test_identifier_order(config):52 assert config.identifier is None53 config.add_service(DMAP_SERVICE)54 assert config.identifier == IDENTIFIER_155 config.add_service(MRP_SERVICE)56 assert config.identifier == IDENTIFIER_257 config.add_service(AIRPLAY_SERVICE)58 assert config.identifier == IDENTIFIER_259def test_add_airplay_service(config):60 config.add_service(AIRPLAY_SERVICE)61 airplay = config.get_service(Protocol.AirPlay)62 assert airplay.protocol == Protocol.AirPlay63 assert airplay.port == PORT_164def test_main_service_no_service(config):65 with pytest.raises(exceptions.NoServiceError):66 config.main_service()67def test_main_service_airplay_no_service(config):68 config.add_service(AIRPLAY_SERVICE)69 with pytest.raises(exceptions.NoServiceError):70 config.main_service()71def test_main_service_get_service(config):72 config.add_service(DMAP_SERVICE)73 assert config.main_service() == DMAP_SERVICE74 config.add_service(MRP_SERVICE)75 assert config.main_service() == MRP_SERVICE76def test_main_service_override_protocol(config):77 config.add_service(DMAP_SERVICE)78 config.add_service(MRP_SERVICE)79 assert config.main_service(protocol=DMAP_SERVICE.protocol) == DMAP_SERVICE80def test_set_credentials_for_missing_service(config):81 assert not config.set_credentials(Protocol.DMAP, "dummy")82def test_set_credentials(config):83 config.add_service(DMAP_SERVICE)84 assert config.get_service(Protocol.DMAP).credentials is None85 config.set_credentials(Protocol.DMAP, "dummy")86 assert config.get_service(Protocol.DMAP).credentials == "dummy"87def test_empty_device_info(config):88 config = conf.AppleTV(ADDRESS_1, NAME)89 device_info = config.device_info90 assert device_info.operating_system == OperatingSystem.Unknown91 assert device_info.version is None92 assert device_info.build_number is None93 assert device_info.model == DeviceModel.Unknown94 assert device_info.mac is None95def test_tvos_device_info(config):96 config.add_service(MRP_SERVICE)97 config.add_service(AIRPLAY_SERVICE)98 device_info = config.device_info99 assert device_info.operating_system == OperatingSystem.TvOS100 assert device_info.version == "8.0.0"101 assert device_info.build_number == "17K795"102 assert device_info.model == DeviceModel.Gen4K103 assert device_info.mac == "FF:EE:DD:CC:BB:AA"104def test_tvos_device_info_no_airplay(config):105 config.add_service(MRP_SERVICE)106 device_info = config.device_info107 assert device_info.operating_system == OperatingSystem.TvOS108 assert device_info.version == "13.3.1"109 assert device_info.build_number == "17K795"110 assert device_info.model == DeviceModel.Gen2111 assert device_info.mac == "FF:EE:DD:CC:BB:AA"112def test_legacy_device_info(config):113 config.add_service(DMAP_SERVICE)114 config.add_service(AIRPLAY_SERVICE)115 device_info = config.device_info116 assert device_info.operating_system == OperatingSystem.Legacy117 assert device_info.version == "8.0.0"118 assert device_info.build_number is None119 assert device_info.model == DeviceModel.Gen4K120 assert device_info.mac == "AA:BB:CC:DD:EE:FF"121def test_ready_dmap(config):122 assert not config.ready123 config.add_service(AIRPLAY_SERVICE)124 assert not config.ready125 config.add_service(DMAP_SERVICE)126 assert config.ready127def test_ready_mrp(config):128 assert not config.ready129 config.add_service(AIRPLAY_SERVICE)130 assert not config.ready131 config.add_service(MRP_SERVICE)132 assert config.ready133# Name collisions on the network results in _X being added to the identifier,134# which should be stripped135def test_dmap_identifier_strip():136 service = conf.DmapService("abcd_2", "dummy")137 assert service.identifier == "abcd"138# This test is a bit strange and couples to protocol specific services,139# but it's mainly to exercise string as that is important. Might refactor140# this in the future.141def test_to_str(config):142 config.add_service(conf.DmapService(IDENTIFIER_1, "LOGIN_ID"))143 config.add_service(conf.MrpService(IDENTIFIER_2, PORT_2))144 # Check for some keywords to not lock up format too much145 output = str(config)146 assert ADDRESS_1 in output147 assert NAME in output148 assert "LOGIN_ID" in output149 assert str(PORT_2) in output150 assert "3689" in output...
test_add_service.py
Source:test_add_service.py
1"""2Tests brewblox_ctl.commands.add_service3"""4import pytest5from brewblox_ctl.commands import add_service6from brewblox_ctl.testing import check_sudo, invoke7TESTED = add_service.__name__8@pytest.fixture(autouse=True)9def m_getgid(mocker):10 m = mocker.patch(TESTED + '.getgid')11 m.return_value = 100012 return m13@pytest.fixture(autouse=True)14def m_getuid(mocker):15 m = mocker.patch(TESTED + '.getuid')16 m.return_value = 100017 return m18@pytest.fixture19def m_utils(mocker):20 m = mocker.patch(TESTED + '.utils', autospec=True)21 m.optsudo.return_value = 'SUDO '22 return m23@pytest.fixture24def m_sh(mocker):25 m = mocker.patch(TESTED + '.sh', autospec=True)26 m.side_effect = check_sudo27 return m28@pytest.fixture29def m_choose(mocker):30 m = mocker.patch(TESTED + '.choose_device')31 m.side_effect = lambda _1, _2: {32 'id': '280038000847343337373738',33 'host': '192.168.0.55',34 }35 return m36@pytest.fixture37def m_find_by_host(mocker):38 m = mocker.patch(TESTED + '.find_device_by_host')39 m.side_effect = lambda _1: {40 'id': '280038000847343337373738',41 'host': '192.168.0.55',42 }43 return m44def test_discover_spark(m_utils, mocker):45 m_discover = mocker.patch(TESTED + '.list_devices', autospec=True)46 m_utils.read_compose.return_value = {'services': {}}47 invoke(add_service.discover_spark)48 m_discover.assert_called_with('all', {'services': {}})49 m_utils.read_compose.side_effect = FileNotFoundError50 invoke(add_service.discover_spark)51 m_discover.assert_called_with('all', {})52def test_add_spark(m_utils, m_sh, mocker, m_choose, m_find_by_host):53 m_utils.read_compose.side_effect = lambda: {'services': {}}54 invoke(add_service.add_spark, '--name testey --discover-now --discovery wifi --command "--do-stuff"')55 invoke(add_service.add_spark, input='testey\n')56 m_utils.confirm.return_value = False57 invoke(add_service.add_spark, '-n testey')58 m_choose.side_effect = lambda _1, _2=None: None59 invoke(add_service.add_spark, '--name testey --discovery wifi', _err=True)60 invoke(add_service.add_spark, '--name testey --device-host 1234')61 invoke(add_service.add_spark, '--name testey --device-id 12345 --simulation')62 invoke(add_service.add_spark, '--name testey --simulation')63def test_add_spark_force(m_utils, m_sh, mocker, m_choose):64 m_utils.confirm.return_value = False65 m_utils.read_compose.side_effect = lambda: {'services': {'testey': {}}}66 invoke(add_service.add_spark, '--name testey', _err=True)67 invoke(add_service.add_spark, '--name testey --force')68def test_spark_overwrite(m_utils, m_sh, m_choose, mocker):69 m_utils.read_compose.side_effect = lambda: {70 'services': {71 'testey': {72 'image': 'brewblox/brewblox-devcon-spark:develop'73 }}}74 invoke(add_service.add_spark, '--name testey --force')75 assert m_utils.warn.call_count == 076 invoke(add_service.add_spark, '--name new-testey')77 assert m_utils.warn.call_count > 078def test_add_tilt(m_utils, m_sh, mocker):79 m_utils.read_compose.side_effect = lambda: {'services': {}}80 invoke(add_service.add_tilt)81 assert m_sh.call_count == 282 m_sh.reset_mock()83 m_utils.confirm.return_value = False84 invoke(add_service.add_tilt)85 assert m_sh.call_count == 186 m_sh.reset_mock()87 m_utils.read_compose.side_effect = lambda: {'services': {'tilt': {}}}88 invoke(add_service.add_tilt, _err=True)89 assert m_sh.call_count == 090def test_add_tilt_force(m_utils, m_sh, mocker, m_choose):91 m_utils.confirm.return_value = False92 m_utils.read_compose.side_effect = lambda: {'services': {'tilt': {}}}93 invoke(add_service.add_tilt, _err=True)94 invoke(add_service.add_tilt, '--force')95def test_add_plaato(m_utils, m_sh, mocker):96 m_utils.read_compose.side_effect = lambda: {'services': {}}97 invoke(add_service.add_plaato, '--name testey --token x')98 invoke(add_service.add_plaato, input='testey\ntoken\n')99 assert m_sh.call_count == 2100 m_sh.reset_mock()101 m_utils.confirm.return_value = False102 invoke(add_service.add_plaato, '-n testey --token x')103 assert m_sh.call_count == 0104def test_add_plaato_force(m_utils, m_sh, mocker, m_choose):105 m_utils.confirm.return_value = False106 m_utils.read_compose.side_effect = lambda: {'services': {'testey': {}}}107 invoke(add_service.add_plaato, '--name testey --token x', _err=True)108 invoke(add_service.add_plaato, '--name testey --token x --force')109def test_add_node_red(m_utils, m_sh, mocker):110 m_utils.read_compose.side_effect = lambda: {'services': {}}111 invoke(add_service.add_node_red)112 assert m_sh.call_count == 2113 m_sh.reset_mock()114 m_utils.confirm.return_value = False115 invoke(add_service.add_node_red)116 assert m_sh.call_count == 1117def test_add_node_red_other_uid(m_utils, m_sh, mocker, m_getuid):118 m_getuid.return_value = 1001119 m_utils.read_compose.side_effect = lambda: {'services': {}}120 invoke(add_service.add_node_red)121 assert m_sh.call_count == 3122def test_add_node_red_force(m_utils, m_sh, mocker, m_choose):123 m_utils.confirm.return_value = False124 m_utils.read_compose.side_effect = lambda: {'services': {'node-red': {}}}125 invoke(add_service.add_node_red, _err=True)...
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:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
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.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
