Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

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.