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.