KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use data_poller method in yandex-tank
Best Python code snippet using yandex-tank
test_data_poller.py
Source:test_data_poller.py 
1# Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.2#3# Licensed under the Apache License, Version 2.0 (the "License"). You4# may not use this file except in compliance with the License. A copy of5# the License is located at6#7# http://aws.amazon.com/apache2.0/8#9# or in the "license" file accompanying this file. This file is10# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF11# ANY KIND, either express or implied. See the License for the specific12# language governing permissions and limitations under the License.13import datetime14from dateutil import tz15import unittest16import mock17from ebcli.display import data_poller18class TestDataPoller(unittest.TestCase):19 maxDiff = None20 ENVIRONMENT_HEALTH = environment_health = {21 'EnvironmentName': 'health-tests-test-1',22 'HealthStatus': 'Ok',23 'Status': 'Ready',24 'Color': 'Green',25 'Causes': ['Fake cause 1', 'Fake cause 2'],26 'ApplicationMetrics': {27 'RequestCount': 028 },29 'InstancesHealth': {30 'NoData': 0,31 'Unknown': 0,32 'Pending': 1,33 'Ok': 1,34 'Info': 0,35 'Warning': 0,36 'Degraded': 0,37 'Severe': 038 },39 'RefreshedAt': datetime.datetime(2018, 3, 12, 22, 19, 1, tzinfo=tz.tzutc()),40 'ResponseMetadata': {41 'RequestId': 'fc611309-224f-489c-af04-c21e4cc70100',42 'HTTPStatusCode': 200,43 'date': 'Mon, 12 Mar 2018 22:19:04 GMT',44 'RetryAttempts': 045 }46 }47 DESCRIBE_INSTANCES_HEALTH = {48 'InstanceHealthList': [49 {50 'InstanceId': 'i-0f111c68ca2eb9ce2',51 'HealthStatus': 'Severe',52 'Color': 'Red',53 'Causes':[54 'Instance initialization failed.'55 ],56 'LaunchedAt':datetime.datetime(2018, 3, 14, 4, 12, 27, tzinfo=tz.tzutc()),57 'ApplicationMetrics': {58 'Duration': 10,59 'RequestCount': 6,60 'StatusCodes': {61 'Status2xx': 6,62 'Status3xx': 0,63 'Status4xx': 0,64 'Status5xx': 065 },66 'Latency': {67 'P999': 15.6,68 'P99': 15.6,69 'P95': 15.6,70 'P90': 15.6,71 'P85': 14.3,72 'P75': 11.7,73 'P50': 5.2,74 'P10': 0.075 }76 },77 'System': {78 'CPUUtilization': {79 'User': 0.0,80 'System': 0.0,81 'Idle': 100.082 }83 },84 'Deployment': {85 },86 'AvailabilityZone': 'us-east-1d',87 'InstanceType': 't2.micro'88 }89 ],90 'RefreshedAt': datetime.datetime(2018, 3, 14, 0, 8, 46, tzinfo=tz.tzutc()),91 'ResponseMetadata': {92 'RequestId': '26fdf953-1700-4c4f-8009-19fba9b4d2e8',93 'HTTPStatusCode': 200,94 'date': 'Wed, 14 Mar 2018 00: 08:48 GMT',95 'RetryAttempts': 096 }97 }98 GET_HEALTH_DATA_RESULT = {99 'environment': {100 'Cause': 'Fake cause 1',101 'Causes': ['Fake cause 1', 'Fake cause 2'],102 'Color': 'Green',103 'EnvironmentName': 'health-tests-test-1',104 'HealthStatus': 'Ok',105 'RefreshedAt': datetime.datetime(2018, 3, 12, 22, 19, 1, tzinfo=tz.tzutc()),106 'ResponseMetadata': {107 'HTTPStatusCode': 200,108 'RequestId': 'fc611309-224f-489c-af04-c21e4cc70100',109 'RetryAttempts': 0,110 'date': 'Mon, 12 Mar 2018 22:19:04 GMT'111 },112 'Status': 'Ready',113 'Total': 0,114 'requests': 0.0115 },116 'instances': [117 {118 'AvailabilityZone': '1d',119 'Cause': 'Instance initialization failed.',120 'Causes': ['Instance initialization failed.'],121 'Color': 'Red',122 'Deployment': {},123 'HealthStatus': 'Severe',124 'InstanceId': 'i-0f111c68ca2eb9ce2',125 'InstanceType': 't2.micro',126 'LaunchedAt': datetime.datetime(2018, 3, 14, 4, 12, 27, tzinfo=tz.tzutc()),127 'launched': '2018-03-14 04:12:27',128 'load1': '-',129 'load5': '-',130 'requests': 0.0,131 'running': '1 day',132 'status_sort': 0133 }134 ]135 }136 def test_format_float(self):137 self.assertEqual('1.0', data_poller.format_float(flt=1, number_of_places=1))138 self.assertEqual('1.0', data_poller.format_float(flt=1.0, number_of_places=1))139 self.assertEqual('1.0', data_poller.format_float(flt=1.00, number_of_places=1))140 self.assertEqual('1.009', data_poller.format_float(flt=1.009, number_of_places=3))141 self.assertEqual('1.01', data_poller.format_float(flt=1.009, number_of_places=2))142 self.assertEqual('1.0', data_poller.format_float(flt=1.001, number_of_places=1))143 self.assertEqual('1.500', data_poller.format_float(flt=1.4999, number_of_places=3))144 self.assertEqual('1.5', data_poller.format_float(flt=1.51, number_of_places=1))145 self.assertEqual('1.6', data_poller.format_float(flt=1.55, number_of_places=1))146 self.assertEqual('2.0', data_poller.format_float(flt=1.99, number_of_places=1))147 @mock.patch('ebcli.lib.utils.get_local_time_as_string')148 def test_collapse_environment_health_data(self, get_local_time_as_string_mock):149 get_local_time_as_string_mock.return_value = '2018-03-14 04:12:27'150 self.assertEqual(151 {152 'Cause': 'Fake cause 1',153 'Causes': ['Fake cause 1', 'Fake cause 2'],154 'Color': 'Green',155 'Degraded': 0,156 'EnvironmentName': 'health-tests-test-1',157 'HealthStatus': 'Ok',158 'Info': 0,159 'NoData': 0,160 'Ok': 1,161 'Pending': 1,162 'RefreshedAt': datetime.datetime(2018, 3, 12, 22, 19, 1, tzinfo=tz.tzutc()),163 'RequestCount': 0,164 'Severe': 0,165 'Status': 'Ready',166 'Total': 2,167 'Unknown': 0,168 'Warning': 0,169 'requests': 0.0,170 'ResponseMetadata': {171 'RequestId': 'fc611309-224f-489c-af04-c21e4cc70100',172 'HTTPStatusCode': 200,173 'date': 'Mon, 12 Mar 2018 22:19:04 GMT',174 'RetryAttempts': 0175 }176 },177 data_poller.collapse_environment_health_data(TestDataPoller.ENVIRONMENT_HEALTH))178 @mock.patch('ebcli.lib.utils.get_local_time_as_string')179 @mock.patch('ebcli.display.data_poller.format_time_since')180 def test_collapse_instance_health_data(self, format_time_since_mock, get_local_time_as_string_mock):181 self.maxDiff = None182 get_local_time_as_string_mock.return_value = '2018-03-14 04:12:27'183 format_time_since_mock.return_value = '4 hours'184 self.assertEqual(185 [186 {187 'AvailabilityZone': '1d',188 'Cause': 'Instance initialization failed.',189 'Causes': ['Instance initialization failed.'],190 'Color': 'Red',191 'Deployment': {},192 'Duration': 10,193 'HealthStatus': 'Severe',194 'Idle': 100.0,195 'InstanceId': 'i-0f111c68ca2eb9ce2',196 'InstanceType': 't2.micro',197 'LaunchedAt': datetime.datetime(2018, 3, 14, 4, 12, 27, tzinfo=tz.tzutc()),198 'P10': '0.000',199 'P10_sort': 0.0,200 'P50': '5.200',201 'P50_sort': 5.2,202 'P75': '11.700',203 'P75_sort': 11.7,204 'P85': '14.300',205 'P85_sort': 14.3,206 'P90': '15.600*',207 'P90_sort': 15.6,208 'P95': '15.600',209 'P95_sort': 15.6,210 'P99': '15.600*',211 'P999': '15.600',212 'P999_sort': 15.6,213 'P99_sort': 15.6,214 'RequestCount': 6,215 'Status2xx': 6,216 'Status3xx': 0,217 'Status4xx': 0,218 'Status5xx': 0,219 'Status_2xx': '0.0',220 'Status_2xx_sort': 0.0,221 'Status_3xx': '0.0',222 'Status_3xx_sort': 0.0,223 'Status_4xx': '0.0',224 'Status_4xx_sort': 0.0,225 'Status_5xx': '0.0',226 'Status_5xx_sort': 0.0,227 'System': 0.0,228 'User': 0.0,229 'launched': '2018-03-14 04:12:27',230 'load1': '-',231 'load5': '-',232 'requests': 0.6,233 'running': '4 hours',234 'status_sort': 0235 }236 ],237 data_poller.collapse_instance_health_data(TestDataPoller.DESCRIBE_INSTANCES_HEALTH)238 )239 def test_format_time_since(self):240 now = datetime.datetime(2018, 3, 13, 19, 12, 27, tzinfo=tz.tzutc())241 five_seconds_ago = now - datetime.timedelta(seconds=5)242 one_minute_ago = now - datetime.timedelta(minutes=1)243 five_minutes_ago = now - datetime.timedelta(minutes=5)244 one_hour_ago = now - datetime.timedelta(hours=1)245 five_hours_ago = now - datetime.timedelta(hours=5)246 one_day_ago = now - datetime.timedelta(days=1)247 five_days_ago = now - datetime.timedelta(days=5)248 with mock.patch('ebcli.display.data_poller._datetime_utcnow_wrapper') as datetime_now_mock:249 datetime_now_mock.return_value = now250 self.assertEqual('-', data_poller.format_time_since(None))251 self.assertEqual('-', data_poller.format_time_since(''))252 self.assertEqual('5 secs', data_poller.format_time_since(five_seconds_ago))253 self.assertEqual('1 min', data_poller.format_time_since(one_minute_ago))254 self.assertEqual('5 mins', data_poller.format_time_since(five_minutes_ago))255 self.assertEqual('1 hour', data_poller.format_time_since(one_hour_ago))256 self.assertEqual('5 hours', data_poller.format_time_since(five_hours_ago))257 self.assertEqual('1 day', data_poller.format_time_since(one_day_ago))258 self.assertEqual('5 days', data_poller.format_time_since(five_days_ago))259 @mock.patch('ebcli.display.data_poller.format_time_since')260 @mock.patch('ebcli.lib.utils.get_local_time')261 @mock.patch('ebcli.lib.elasticbeanstalk.get_environment_health')262 @mock.patch('ebcli.lib.elasticbeanstalk.get_instance_health')263 def test_get_health_data(264 self,265 get_instance_health_mock,266 get_environment_health_mock,267 get_local_time_mock,268 format_time_since_mock269 ):270 get_environment_health_mock.return_value = TestDataPoller.ENVIRONMENT_HEALTH271 get_instance_health_mock.return_value = TestDataPoller.DESCRIBE_INSTANCES_HEALTH272 format_time_since_mock.return_value = '1 day'273 get_local_time_mock.return_value = datetime.datetime(2018, 3, 14, 4, 12, 27)274 poller = data_poller.DataPoller('some_app_name', 'some_env_name')275 self.assertEqual(276 TestDataPoller.GET_HEALTH_DATA_RESULT,277 poller._get_health_data()278 )279 @mock.patch('ebcli.display.data_poller.DataPoller._get_sleep_time')280 @mock.patch('ebcli.display.data_poller.LOG')281 @mock.patch('ebcli.display.data_poller.DataPoller._get_health_data')282 def test_poll_for_health_data(283 self,284 _get_health_data_mock,285 log_mock,286 get_sleep_time_mock287 ):288 poller = data_poller.DataPoller('some_app_name', 'some_env_name')289 poller._get_health_data = mock.MagicMock(290 side_effect=[291 mock.MagicMock(return_value=TestDataPoller.GET_HEALTH_DATA_RESULT),292 KeyboardInterrupt293 ]294 )295 get_sleep_time_mock.side_effect = [0, 0, KeyboardInterrupt]296 _get_health_data_mock.return_value = 'data'297 poller._poll_for_health_data()298 calls = [299 mock.call('Starting data poller child thread'),300 mock.call("Sleeping for 0 second(s)"),301 mock.call('Exiting due to: KeyboardInterrupt')302 ]...
driver_funcs.py
Source:driver_funcs.py
...106) -> Any:107 m = importlib.import_module(f"tomato.drivers.{driver}")108 func = getattr(m, command)109 return func(address, channel, jobqueue, logger, **kwargs)110def data_poller(111 driver: str,112 jq: multiprocessing.Queue,113 lq: multiprocessing.Queue,114 address: str,115 channel: int,116 device: str,117 root: str,118 loglevel: int,119 kwargs: dict,120) -> None:121 log_worker_config(lq, loglevel)122 log = logging.getLogger()123 pollrate = kwargs.pop("pollrate", 10)124 log.debug(f"in 'data_poller', {pollrate=}")...
test_health.py
Source:test_health.py
1# Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.2#3# Licensed under the Apache License, Version 2.0 (the "License"). You4# may not use this file except in compliance with the License. A copy of5# the License is located at6#7# http://aws.amazon.com/apache2.0/8#9# or in the "license" file accompanying this file. This file is10# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF11# ANY KIND, either express or implied. See the License for the specific12# language governing permissions and limitations under the License.13import os14import shutil15import threading16import mock17import unittest18from ebcli.core.ebcore import EB19from ebcli.core import fileoperations20from ebcli.operations import commonops21from .. import mock_responses22class TestHealth(unittest.TestCase):23 def setUp(self):24 self.root_dir = os.getcwd()25 if os.path.exists('testDir'):26 shutil.rmtree('testDir')27 os.mkdir('testDir')28 os.chdir('testDir')29 def tearDown(self):30 os.chdir(self.root_dir)31 shutil.rmtree('testDir')32 @classmethod33 def tearDownClass(cls):34 import time35 time.sleep(0.5)36 @mock.patch('ebcli.controllers.health.healthops.elasticbeanstalk.describe_configuration_settings')37 @mock.patch('ebcli.display.data_poller.elasticbeanstalk.get_environment_health')38 @mock.patch('ebcli.display.data_poller.elasticbeanstalk.get_instance_health')39 def test_health(40 self,41 get_instance_health_mock,42 get_environment_health_mock,43 describe_configuration_settings_mock44 ):45 fileoperations.create_config_file(46 'my-application',47 'us-west-2',48 'php-7.2'49 )50 commonops.set_environment_for_current_branch('environment-1')51 describe_configuration_settings_mock.return_value = mock_responses.DESCRIBE_CONFIGURATION_SETTINGS_RESPONSE__2['ConfigurationSettings'][0]52 get_environment_health_mock.return_value = mock_responses.DESCRIBE_ENVIRONMENT_HEALTH_RESPONSE53 get_instance_health_mock.return_value = mock_responses.DESCRIBE_INSTANCES_HEALTH_RESPONSE54 app = EB(argv=['health'])55 app.setup()56 app.run()57 @mock.patch('ebcli.controllers.health.healthops.elasticbeanstalk.describe_configuration_settings')58 @mock.patch('ebcli.display.data_poller.elasticbeanstalk.get_environment_health')59 @mock.patch('ebcli.display.data_poller.elasticbeanstalk.get_instance_health')60 def test_health__mono(61 self,62 get_instance_health_mock,63 get_environment_health_mock,64 describe_configuration_settings_mock65 ):66 fileoperations.create_config_file(67 'my-application',68 'us-west-2',69 'php-7.2'70 )71 commonops.set_environment_for_current_branch('environment-1')72 describe_configuration_settings_mock.return_value = mock_responses.DESCRIBE_CONFIGURATION_SETTINGS_RESPONSE__2['ConfigurationSettings'][0]73 get_environment_health_mock.return_value = mock_responses.DESCRIBE_ENVIRONMENT_HEALTH_RESPONSE74 get_instance_health_mock.return_value = mock_responses.DESCRIBE_INSTANCES_HEALTH_RESPONSE75 app = EB(argv=['health', '--mono'])76 app.setup()77 app.run()78 @mock.patch('ebcli.controllers.health.healthops.elasticbeanstalk.describe_configuration_settings')79 @mock.patch('ebcli.display.data_poller.elasticbeanstalk.get_environment_health')80 @mock.patch('ebcli.display.data_poller.elasticbeanstalk.get_instance_health')81 def test_health__refresh(82 self,83 get_instance_health_mock,84 get_environment_health_mock,85 describe_configuration_settings_mock86 ):87 fileoperations.create_config_file(88 'my-application',89 'us-west-2',90 'php-7.2'91 )92 commonops.set_environment_for_current_branch('environment-1')93 describe_configuration_settings_mock.return_value = mock_responses.DESCRIBE_CONFIGURATION_SETTINGS_RESPONSE__2['ConfigurationSettings'][0]94 get_environment_health_mock.return_value = mock_responses.DESCRIBE_ENVIRONMENT_HEALTH_RESPONSE95 get_instance_health_mock.return_value = mock_responses.DESCRIBE_INSTANCES_HEALTH_RESPONSE96 def run_eb_health_refresh():97 app = EB(argv=['health', '--refresh'])98 app.setup()99 app.run()100 t = threading.Thread(target=run_eb_health_refresh)101 t.start()102 t.join(timeout=2)103 @mock.patch('ebcli.controllers.health.healthops.elasticbeanstalk.describe_configuration_settings')104 @mock.patch('ebcli.display.data_poller.elasticbeanstalk.get_environment_health')105 @mock.patch('ebcli.display.data_poller.elasticbeanstalk.get_instance_health')106 def test_health__mono__view_status(107 self,108 get_instance_health_mock,109 get_environment_health_mock,110 describe_configuration_settings_mock111 ):112 fileoperations.create_config_file(113 'my-application',114 'us-west-2',115 'php-7.2'116 )117 commonops.set_environment_for_current_branch('environment-1')118 describe_configuration_settings_mock.return_value = mock_responses.DESCRIBE_CONFIGURATION_SETTINGS_RESPONSE__2['ConfigurationSettings'][0]119 get_environment_health_mock.return_value = mock_responses.DESCRIBE_ENVIRONMENT_HEALTH_RESPONSE120 get_instance_health_mock.return_value = mock_responses.DESCRIBE_INSTANCES_HEALTH_RESPONSE121 app = EB(argv=['health', '--view', 'status'])122 app.setup()123 app.run()124 @mock.patch('ebcli.controllers.health.healthops.elasticbeanstalk.describe_configuration_settings')125 @mock.patch('ebcli.display.data_poller.elasticbeanstalk.get_environment_health')126 @mock.patch('ebcli.display.data_poller.elasticbeanstalk.get_instance_health')127 def test_health__mono__view_request(128 self,129 get_instance_health_mock,130 get_environment_health_mock,131 describe_configuration_settings_mock132 ):133 fileoperations.create_config_file(134 'my-application',135 'us-west-2',136 'php-7.2'137 )138 commonops.set_environment_for_current_branch('environment-1')139 describe_configuration_settings_mock.return_value = mock_responses.DESCRIBE_CONFIGURATION_SETTINGS_RESPONSE__2['ConfigurationSettings'][0]140 get_environment_health_mock.return_value = mock_responses.DESCRIBE_ENVIRONMENT_HEALTH_RESPONSE141 get_instance_health_mock.return_value = mock_responses.DESCRIBE_INSTANCES_HEALTH_RESPONSE142 app = EB(argv=['health', '--view', 'request'])143 app.setup()144 app.run()145 @mock.patch('ebcli.controllers.health.healthops.elasticbeanstalk.describe_configuration_settings')146 @mock.patch('ebcli.display.data_poller.elasticbeanstalk.get_environment_health')147 @mock.patch('ebcli.display.data_poller.elasticbeanstalk.get_instance_health')148 def test_health__mono__view_cpu(149 self,150 get_instance_health_mock,151 get_environment_health_mock,152 describe_configuration_settings_mock153 ):154 fileoperations.create_config_file(155 'my-application',156 'us-west-2',157 'php-7.2'158 )159 commonops.set_environment_for_current_branch('environment-1')160 describe_configuration_settings_mock.return_value = mock_responses.DESCRIBE_CONFIGURATION_SETTINGS_RESPONSE__2['ConfigurationSettings'][0]161 get_environment_health_mock.return_value = mock_responses.DESCRIBE_ENVIRONMENT_HEALTH_RESPONSE162 get_instance_health_mock.return_value = mock_responses.DESCRIBE_INSTANCES_HEALTH_RESPONSE163 app = EB(argv=['health', '--view', 'cpu'])164 app.setup()...
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 yandex-tank 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?
