KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use _create_service method in tempest
Best Python code snippet using tempest_python
test_base.py
Source:test_base.py 
...13 exception = RESTServiceException("Service failed")14 self.assertIsNotNone(exception)15 self.assertEquals('Service failed', str(exception))16class RESTServiceTests(ServiceTestCase):17 def _create_service(self):18 class MockRESTService(RESTService):19 def __init__(self, configuration):20 RESTService.__init__(self, configuration)21 def _response_to_json(self, api, response):22 return response23 configuration = ServiceConfiguration.from_data("rest", "test", "category")24 return MockRESTService(configuration)25 def test_init(self):26 service = self._create_service()27 self.assertIsNotNone(service)28 def test_add_base_payload(self):29 service = self._create_service()30 self.assertIsNotNone(service)31 data = {"response": {}}32 started = datetime.now()33 speed = timedelta(microseconds=3000)34 service._add_base_payload(data, "success", "test", "http://some.service.com", "GET", 3, started, speed)35 self.assertTrue('response' in data)36 self.assertEqual(data['response']['status'], "success")37 self.assertEqual(data['response']['url'], "http://some.service.com")38 self.assertEqual(data['response']['call'], "GET")39 self.assertEqual(data['response']['retries'], 3)40 self.assertEqual(data['response']['started'], started.strftime("%d/%m/%Y, %H:%M:%S"))41 self.assertEqual(data['response']['speed'], "3.0ms")42 self.assertEqual(data['response']['service'], "test")43 self.assertEqual(data['response']['category'], "category")44 def test_create_success_payload(self):45 service = self._create_service()46 self.assertIsNotNone(service)47 started = datetime.now()48 speed = timedelta(microseconds=3000)49 data = service._create_success_payload("test", "http://some.service.com", "GET", 3, started, speed, "response")50 self.assertTrue('response' in data)51 self.assertEqual(data['response']['status'], "success")52 self.assertEqual(data['response']['url'], "http://some.service.com")53 self.assertEqual(data['response']['call'], "GET")54 self.assertEqual(data['response']['retries'], 3)55 self.assertEqual(data['response']['started'], started.strftime("%d/%m/%Y, %H:%M:%S"))56 self.assertEqual(data['response']['speed'], "3.0ms")57 self.assertEqual(data['response']['service'], "test")58 self.assertEqual(data['response']['category'], "category")59 self.assertTrue("payload" in data['response'])60 self.assertEqual("response", data['response']['payload'])61 def test_create_statuscode_failure_payload(self):62 service = self._create_service()63 self.assertIsNotNone(service)64 started = datetime.now()65 speed = timedelta(microseconds=3000)66 mock_response = Mock()67 mock_response.status_code = 50068 data = service._create_statuscode_failure_payload("test", "http://some.service.com", "GET", 3, started, speed, mock_response)69 self.assertTrue('response' in data)70 self.assertEqual(data['response']['status'], "failure")71 self.assertEqual(data['response']['url'], "http://some.service.com")72 self.assertEqual(data['response']['call'], "GET")73 self.assertEqual(data['response']['retries'], 3)74 self.assertEqual(data['response']['started'], started.strftime("%d/%m/%Y, %H:%M:%S"))75 self.assertEqual(data['response']['speed'], "3.0ms")76 self.assertEqual(data['response']['service'], "test")77 self.assertEqual(data['response']['category'], "category")78 self.assertTrue("payload" in data['response'])79 self.assertEqual(data['response']['payload']['type'], 'statusCode')80 self.assertEqual(data['response']['payload']['statusCode'], 500)81 def test_create_http_failure_payload(self):82 service = self._create_service()83 self.assertIsNotNone(service)84 started = datetime.now()85 speed = timedelta(microseconds=3000)86 data = service._create_http_failure_payload("test", "http://some.service.com", "GET", 3, started, speed, "Service crashed")87 self.assertTrue('response' in data)88 self.assertEqual(data['response']['status'], "failure")89 self.assertEqual(data['response']['url'], "http://some.service.com")90 self.assertEqual(data['response']['call'], "GET")91 self.assertEqual(data['response']['retries'], 3)92 self.assertEqual(data['response']['started'], started.strftime("%d/%m/%Y, %H:%M:%S"))93 self.assertEqual(data['response']['speed'], "3.0ms")94 self.assertEqual(data['response']['service'], "test")95 self.assertEqual(data['response']['category'], "category")96 self.assertTrue("payload" in data['response'])97 self.assertEqual(data['response']['payload']['type'], 'http')98 self.assertEqual(data['response']['payload']['httpError'], "Service crashed")99 def test_create_general_failure_payload(self):100 service = self._create_service()101 self.assertIsNotNone(service)102 started = datetime.now()103 speed = timedelta(microseconds=3000)104 data = service._create_general_failure_payload("test", "http://some.service.com", "GET", 3, started, speed,105 RESTServiceException("Something went wrong"))106 self.assertTrue('response' in data)107 self.assertEqual(data['response']['status'], "failure")108 self.assertEqual(data['response']['url'], "http://some.service.com")109 self.assertEqual(data['response']['call'], "GET")110 self.assertEqual(data['response']['retries'], 3)111 self.assertEqual(data['response']['started'], started.strftime("%d/%m/%Y, %H:%M:%S"))112 self.assertEqual(data['response']['speed'], "3.0ms")113 self.assertEqual(data['response']['service'], "test")114 self.assertEqual(data['response']['category'], "category")115 self.assertTrue("payload" in data['response'])116 self.assertEqual(data['response']['payload']['type'], 'general')117 self.assertEqual(data['response']['payload']['error'], "Something went wrong")118 def patch_requests_get(self, url, headers, timeout):119 mock_response = Mock()120 mock_response.status_code = 200121 mock_response.raise_for_status.return_value = True122 return mock_response123 @patch("programy.services.rest.base.RESTService._requests_get", patch_requests_get)124 def test_do_get(self):125 service = self._create_service()126 self.assertIsNotNone(service)127 response, count = service._do_get("http://some.api.com")128 self.assertEqual(0, count)129 self.assertIsNotNone(response)130 @patch("programy.services.rest.base.RESTService._requests_get", patch_requests_get)131 def test_get(self):132 service = self._create_service()133 self.assertIsNotNone(service)134 response = service._get("test", "http://some.api.com")135 self.assertIsNotNone(response)136 self.assertTrue(response['response']['status'], 'success')137 def patch_requests_get_fail(self, url, headers, timeout):138 mock_response = Mock()139 mock_response.status_code = 500140 mock_response.raise_for_status.return_value = True141 return mock_response142 @patch("programy.services.rest.base.RESTService._requests_get", patch_requests_get_fail)143 def test_get_fail(self):144 service = self._create_service()145 self.assertIsNotNone(service)146 response = service._get("test", "http://some.api.com")147 self.assertIsNotNone(response)148 self.assertTrue(response['response']['status'], 'failure')149 def patch_requests_post(self, url, headers, params, timeout):150 mock_response = Mock()151 mock_response.status_code = 200152 mock_response.raise_for_status.return_value = True153 return mock_response154 @patch("programy.services.rest.base.RESTService._requests_post", patch_requests_post)155 def test_do_post(self):156 service = self._create_service()157 self.assertIsNotNone(service)158 response, count = service._do_post("http://some.api.com", {"query": "Hello"})159 self.assertEqual(0, count)160 self.assertIsNotNone(response)161 @patch("programy.services.rest.base.RESTService._requests_post", patch_requests_post)162 def test_post(self):163 service = self._create_service()164 self.assertIsNotNone(service)165 response = service._post("test", "http://some.api.com", {"query": "Hello"})166 self.assertIsNotNone(response)167 self.assertTrue(response['response']['status'], 'success')168 def patch_requests_post_fail(self, url, headers, params, timeout):169 mock_response = Mock()170 mock_response.status_code = 500171 mock_response.raise_for_status.return_value = True172 return mock_response173 @patch("programy.services.rest.base.RESTService._requests_post", patch_requests_post_fail)174 def test_post_failure(self):175 service = self._create_service()176 self.assertIsNotNone(service)177 response = service._post("test", "http://some.api.com", {"query": "Hello"})178 self.assertIsNotNone(response)...
test_service_registry_db_api.py
Source:test_service_registry_db_api.py
...44 super(ServiceRegistryTest, self).setUp()45 self.setup_dummy_db()46 self.addCleanup(self.reset_dummy_db)47 self.ctxt = utils.dummy_context()48 def _create_service(self, **kwargs):49 values = {50 'service_id': 'f9aff81e-bc1f-4119-941d-ad1ea7f31d19',51 'host': 'localhost',52 'binary': 'playnetmano_rm-engine',53 'topic': 'engine',54 }55 values.update(kwargs)56 return db_api.service_create(self.ctxt, **values)57 def test_service_create_get(self):58 service = self._create_service()59 ret_service = db_api.service_get(self.ctxt, service.id)60 self.assertIsNotNone(ret_service)61 self.assertEqual(service.id, ret_service.id)62 self.assertEqual(service.binary, ret_service.binary)63 self.assertEqual(service.host, ret_service.host)64 self.assertEqual(service.topic, ret_service.topic)65 self.assertEqual(service.disabled, ret_service.disabled)66 self.assertEqual(service.disabled_reason, ret_service.disabled_reason)67 self.assertIsNotNone(service.created_at)68 self.assertIsNotNone(service.updated_at)69 def test_service_get_all(self):70 values = []71 for i in range(4):72 values.append({'service_id': uuidutils.generate_uuid(),73 'host': 'host-%s' % i})74 [self._create_service(**val) for val in values]75 services = db_api.service_get_all(self.ctxt)76 self.assertEqual(4, len(services))77 def test_service_update(self):78 old_service = self._create_service()79 old_updated_time = old_service.updated_at80 values = {'host': 'host-updated'}81 new_service = db_api.service_update(self.ctxt, old_service.id, values)82 self.assertEqual('host-updated', new_service.host)83 self.assertGreater(new_service.updated_at, old_updated_time)84 def test_service_update_values_none(self):85 old_service = self._create_service()86 old_updated_time = old_service.updated_at87 new_service = db_api.service_update(self.ctxt, old_service.id)88 self.assertGreater(new_service.updated_at, old_updated_time)89 def test_service_delete(self):90 service = self._create_service()91 db_api.service_delete(self.ctxt, service.id)92 res = db_api.service_get(self.ctxt, service.id)...
di.py
Source:di.py
...12 self._create_service = create_service_function13 self._services = {}14 def get_service(self, user_id: int, necessary_new=False):15 if self._services.get(user_id) is None or necessary_new:16 service = self._create_service()17 setattr(service, "user_tg_id", user_id)18 self._services[user_id] = service19 return service20 return self._services[user_id]21class SingletonServiceContainer:22 r"""Container for service that is in a **single instance**"""23 def __init__(self, create_service_function):24 r"""25 Parameters26 ----------27 create_service_function :28 Function returning a new service instance29 """30 self._create_service = create_service_function31 self._service = None32 def get_service(self, necessary_new=False):33 if self._service is None or necessary_new:34 self._service = self._create_service()35 return self._service...
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 tempest 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?
