How to use test_ip method in Airtest
Best Python code snippet using Airtest
test_links.py
Source:test_links.py 
1from spec.python import db_connection2import pytest3import web4import sam.pages.links5from sam import constants6from sam import errors7import json8import urllib9db = db_connection.db10sub_id = db_connection.default_sub11ds_full = db_connection.dsid_default12keys = [u'bytes', u'dst_end', u'dst_start', u'links', u'packets', u'protocols', u'src_end', u'src_start']13keys_p = [u'bytes', u'dst_end', u'dst_start', u'links', u'packets', u'port', u'protocols', u'src_end', u'src_start']14def test_decode_get():15 with db_connection.env(mock_input=True, login_active=False, mock_session=True):16 p = sam.pages.links.Links()17 good_data = {18 'address': '110,110.20,110.20.30,110.20.30.40',19 'filter': '180',20 'tstart': '1',21 'tend': str(2**31 - 1),22 'protocol': 'ALL',23 'ds': 'ds{}'.format(ds_full),24 'flat': 'true'25 }26 actual = p.decode_get_request(good_data)27 expected = {28 'protocol': None,29 'addresses': ['110', '110.20', '110.20.30', '110.20.30.40'],30 'ds': 1,31 'tend': 2147483647,32 'tstart': 1,33 'port': 180,34 'flat': True35 }36 assert actual == expected37 min_data = {38 'address': '50',39 'ds': 'ds{}'.format(ds_full),40 }41 actual = p.decode_get_request(min_data)42 expected = {43 'protocol': None,44 'addresses': ['50'],45 'ds': 1,46 'tstart': 1,47 'tend': 2147483647,48 'port': None,49 'flat': False50 }51 assert actual == expected52 data_no_address = {53 'ds': '{}'.format(ds_full),54 }55 with pytest.raises(errors.MalformedRequest):56 actual = p.decode_get_request(data_no_address)57 assert actual == expected58 data_no_ds = {59 'address': '50',60 }61 with pytest.raises(errors.MalformedRequest):62 actual = p.decode_get_request(data_no_ds)63 assert actual == expected64def test_filter_protocol_blank():65 qstring = '/links?address=10,50,110,150,59&filter=&protocol=&tstart=1521498000&tend=1521498300&ds=ds1'66 with db_connection.env(mock_session=True):67 app = web.application(constants.urls)68 req = app.request(qstring)69 assert req.status == "200 OK"70 data = json.loads(req.data)71 assert set(data.keys()) == {u'150', u'59', u'110', u'10', u'50'}72def test_empty_request():73 with db_connection.env(mock_session=True):74 app = web.application(constants.urls)75 req = app.request('/links', 'GET')76 assert req.status == "200 OK"77 assert req.headers['Content-Type'] == "application/json"78 data = json.loads(req.data)79 assert 'result' in set(data.keys())80 assert data['result'] == 'failure'81def test_simple_request8():82 app = web.application(constants.urls)83 test_ip = '110'84 with db_connection.env(mock_session=True):85 input_data = {'address': test_ip, 'ds': ds_full}86 GET_data = urllib.urlencode(input_data)87 req = app.request('/links?{0}'.format(GET_data), 'GET')88 assert req.status == "200 OK"89 assert req.headers['Content-Type'] == "application/json"90 data = json.loads(req.data)91 assert data.keys() == [test_ip]92 assert sorted(data[test_ip].keys()) == ['inputs', 'outputs']93 assert sorted(data[test_ip]['inputs'][0].keys()) == keys94 assert sorted(data[test_ip]['outputs'][0].keys()) == keys95def test_simple_request16():96 app = web.application(constants.urls)97 test_ip = '110.20'98 with db_connection.env(mock_session=True):99 input_data = {'address': test_ip, 'ds': ds_full}100 GET_data = urllib.urlencode(input_data)101 req = app.request('/links?{0}'.format(GET_data), 'GET')102 assert req.status == "200 OK"103 assert req.headers['Content-Type'] == "application/json"104 data = json.loads(req.data)105 assert data.keys() == [test_ip]106 assert sorted(data[test_ip].keys()) == ['inputs', 'outputs']107 assert sorted(data[test_ip]['inputs'][0].keys()) == keys108 assert sorted(data[test_ip]['outputs'][0].keys()) == keys109def test_simple_request24():110 app = web.application(constants.urls)111 test_ip = '110.20.30'112 with db_connection.env(mock_session=True):113 input_data = {'address': test_ip, 'ds': ds_full}114 GET_data = urllib.urlencode(input_data)115 req = app.request('/links?{0}'.format(GET_data), 'GET')116 assert req.status == "200 OK"117 assert req.headers['Content-Type'] == "application/json"118 data = json.loads(req.data)119 assert data.keys() == [test_ip]120 assert sorted(data[test_ip].keys()) == ['inputs', 'outputs']121 assert sorted(data[test_ip]['inputs'][0].keys()) == keys122 assert sorted(data[test_ip]['outputs'][0].keys()) == keys123def test_simple_request32():124 app = web.application(constants.urls)125 test_ip = '110.20.30.40'126 with db_connection.env(mock_session=True):127 input_data = {'address': test_ip, 'ds': ds_full}128 GET_data = urllib.urlencode(input_data)129 req = app.request('/links?{0}'.format(GET_data), 'GET')130 assert req.status == "200 OK"131 assert req.headers['Content-Type'] == "application/json"132 data = json.loads(req.data)133 assert data.keys() == [test_ip]134 assert sorted(data[test_ip].keys()) == ['inputs', 'outputs']135 assert sorted(data[test_ip]['inputs'][0].keys()) == keys_p136 assert sorted(data[test_ip]['outputs'][0].keys()) == keys_p137def test_ports():138 app = web.application(constants.urls)139 test_ip = '50.60.70.80'140 with db_connection.env(mock_session=True):141 input_data = {'address': test_ip, 'ds': ds_full}142 GET_data = urllib.urlencode(input_data)143 req = app.request('/links?{0}'.format(GET_data), 'GET')144 assert req.headers['Content-Type'] == "application/json"145 data = json.loads(req.data)146 assert len(data[test_ip]['inputs']) == 14147 assert len(data[test_ip]['outputs']) == 15148 input_data = {'address': test_ip, 'ds': ds_full, 'filter': '160'}149 GET_data = urllib.urlencode(input_data)150 req = app.request('/links?{0}'.format(GET_data), 'GET')151 assert req.headers['Content-Type'] == "application/json"152 data = json.loads(req.data)153 assert len(data[test_ip]['inputs']) == 3154 assert len(data[test_ip]['outputs']) == 2155 input_data = {'address': test_ip, 'ds': ds_full, 'filter': '128'}156 GET_data = urllib.urlencode(input_data)157 req = app.request('/links?{0}'.format(GET_data), 'GET')158 assert req.headers['Content-Type'] == "application/json"159 data = json.loads(req.data)160 assert len(data[test_ip]['inputs']) == 0161 assert len(data[test_ip]['outputs']) == 2162 input_data = {'address': test_ip, 'ds': ds_full, 'filter': '360'}163 GET_data = urllib.urlencode(input_data)164 req = app.request('/links?{0}'.format(GET_data), 'GET')165 assert req.headers['Content-Type'] == "application/json"166 data = json.loads(req.data)167 assert len(data[test_ip]['inputs']) == 2168 assert len(data[test_ip]['outputs']) == 0169 input_data = {'address': test_ip, 'ds': ds_full, 'filter': '1040'}170 GET_data = urllib.urlencode(input_data)171 req = app.request('/links?{0}'.format(GET_data), 'GET')172 assert req.headers['Content-Type'] == "application/json"173 data = json.loads(req.data)174 assert len(data[test_ip]['inputs']) == 0175 assert len(data[test_ip]['outputs']) == 0176def test_timerange():177 app = web.application(constants.urls)178 mkt = db_connection.make_timestamp179 time_all = (1, 2 ** 31 - 1)180 time_crop = (mkt('2017-3-21 6:13'), mkt('2017-3-24 13:30'))181 time_tiny = (mkt('2017-3-23 6:13'), mkt('2017-3-23 13:30'))182 test_ip = '50.60.70.80'183 with db_connection.env(mock_session=True):184 input_data = {'address': test_ip, 'ds': ds_full, 'tstart': time_all[0], 'tend': time_all[1]}185 GET_data = urllib.urlencode(input_data)186 req = app.request('/links?{0}'.format(GET_data), 'GET')187 assert req.status == "200 OK"188 assert req.headers['Content-Type'] == "application/json"189 data = json.loads(req.data)190 assert len(data[test_ip]['inputs']) == 14191 assert len(data[test_ip]['outputs']) == 15192 test_ip = '50.60.70.80'193 input_data = {'address': test_ip, 'ds': ds_full, 'tstart': time_crop[0], 'tend': time_crop[1]}194 GET_data = urllib.urlencode(input_data)195 req = app.request('/links?{0}'.format(GET_data), 'GET')196 assert req.status == "200 OK"197 assert req.headers['Content-Type'] == "application/json"198 data = json.loads(req.data)199 assert len(data[test_ip]['inputs']) == 14200 assert len(data[test_ip]['outputs']) == 15201 test_ip = '50.60.70.80'202 input_data = {'address': test_ip, 'ds': ds_full, 'tstart': time_tiny[0], 'tend': time_tiny[1]}203 GET_data = urllib.urlencode(input_data)204 req = app.request('/links?{0}'.format(GET_data), 'GET')205 assert req.status == "200 OK"206 assert req.headers['Content-Type'] == "application/json"207 data = json.loads(req.data)208 assert len(data[test_ip]['inputs']) == 1...
test_overlay.py
Source:test_overlay.py
1import copy2import glob3import os4import pytest5import pynq6from .mock_devices import MockMemoryMappedDevice7THIS_DIR = os.path.dirname(__file__)8HWH_PATH = os.path.join(THIS_DIR, 'data', '2016.4')9HWH_FILES = [os.path.basename(f)10 for f in glob.glob(os.path.join(HWH_PATH, "*.hwh"))]11@pytest.fixture12def device():13 device = MockMemoryMappedDevice('mmap_device')14 yield device15 device.close()16@pytest.mark.parametrize('hwh_file', HWH_FILES)17def test_complete_description(hwh_file):18 parser = pynq.pl_server.hwh_parser._HWHZynq(19 os.path.join(HWH_PATH, hwh_file)20 )21 description = pynq.overlay._complete_description(22 parser.ip_dict, parser.hierarchy_dict, False, parser.mem_dict, "device"23 )24 assert description['ip'].keys() == parser.ip_dict.keys()25 for ipname, desc in description['ip'].items():26 assert desc['device'] == 'device'27 assert desc['driver'] == pynq.DefaultIP28 assert desc.items() >= parser.ip_dict[ipname].items()29 for hiername, desc in description['hierarchies'].items():30 assert desc['device'] == 'device'31 # assert desc['driver'] == pynq.overlay.DefaultHierarchy32 assert desc['driver'] == pynq.overlay.DocumentHierarchy33simple_description = {34 'ip': {35 'test_ip': {'phys_addr': 0x80000000, 'addr_range': 65536,36 'type': 'xilinx.com:test:test_ip:1.0', 'registers': {},37 'parameters': {}, 'fullpath': 'test_ip'},38 },39 'hierarchies': {},40 'gpio': {},41 'interrupts': {},42}43hier_description = {44 'ip': {45 'hier/test_ip': {46 'phys_addr': 0x80000000, 'addr_range': 65536,47 'type': 'xilinx.com:test:test_ip:1.0', 'registers': {},48 'parameters': {}, 'fullpath': 'hier/test_ip'},49 },50 'hierarchies': {'hier': {51 'ip': {52 'test_ip': {53 'phys_addr': 0x80000000, 'addr_range': 65536,54 'type': 'xilinx.com:test:test_ip:1.0', 'registers': {},55 'parameters': {}, 'fullpath': 'hier/test_ip'},56 },57 'hierarchies': {},58 'gpio': {},59 'interrupts': {},60 }},61 'gpio': {},62 'interrupts': {},63}64def _copy_assign(description, ignore_version=False, device='device'):65 completed = copy.deepcopy(description)66 pynq.overlay._assign_drivers(completed, ignore_version, device)67 return completed68def test_ip_unregister():69 class TestDriver(pynq.DefaultIP):70 bindto = ['xilinx.com:test:test_ip:1.0']71 desc = _copy_assign(simple_description)72 assert desc['ip']['test_ip']['driver'] == TestDriver73 TestDriver.unregister()74 desc = _copy_assign(simple_description)75 assert desc['ip']['test_ip']['driver'] == pynq.DefaultIP76def test_ip_driver_replace():77 class TestDriver(pynq.DefaultIP):78 bindto = ['xilinx.com:test:test_ip:1.0']79 desc = _copy_assign(simple_description)80 assert desc['ip']['test_ip']['driver'] == TestDriver81 class TestDriver2(pynq.DefaultIP):82 bindto = ['xilinx.com:test:test_ip:1.0']83 desc = _copy_assign(simple_description)84 assert desc['ip']['test_ip']['driver'] == TestDriver285 TestDriver2.unregister()86 TestDriver.unregister()87 desc = _copy_assign(simple_description)88 assert desc['ip']['test_ip']['driver'] == pynq.DefaultIP89def test_ip_version_mismatch():90 class TestDriver(pynq.DefaultIP):91 bindto = ['xilinx.com:test:test_ip:1.1']92 with pytest.warns(UserWarning):93 desc = _copy_assign(simple_description)94 assert desc['ip']['test_ip']['driver'] == pynq.DefaultIP95 TestDriver.unregister()96def test_ip_version_ignore():97 class TestDriver(pynq.DefaultIP):98 bindto = ['xilinx.com:test:test_ip:1.1']99 desc = _copy_assign(simple_description, ignore_version=True)100 assert desc['ip']['test_ip']['driver'] == TestDriver101 TestDriver.unregister()102def test_ip_unsupported_configuration(device):103 class TestDriver(pynq.DefaultIP):104 bindto = ['xilinx.com:test:test_ip:1.0']105 def __init__(self, description):106 raise pynq.UnsupportedConfiguration('Test Error')107 desc = _copy_assign(simple_description, device=device)108 assert desc['ip']['test_ip']['driver'] == TestDriver109 ip_map = pynq.overlay._IPMap(desc)110 with pytest.warns(UserWarning):111 driver = ip_map.test_ip112 assert type(driver) is pynq.DefaultIP113 TestDriver.unregister()114def test_ip_supported_configuration(device):115 class TestDriver(pynq.DefaultIP):116 bindto = ['xilinx.com:test:test_ip:1.0']117 desc = _copy_assign(simple_description, device=device)118 assert desc['ip']['test_ip']['driver'] == TestDriver119 ip_map = pynq.overlay._IPMap(desc)120 driver = ip_map.test_ip121 assert type(driver) is TestDriver 122 TestDriver.unregister()123 124def test_hierarchy_bind():125 class HierarchyDriver(pynq.DefaultHierarchy):126 @staticmethod127 def checkhierarchy(description):128 return 'test_ip' in description['ip']129 desc = _copy_assign(hier_description)130 assert desc['hierarchies']['hier']['driver'] == HierarchyDriver131 HierarchyDriver.unregister()132 desc = _copy_assign(hier_description)133 assert desc['hierarchies']['hier']['driver'] == \134 pynq.overlay.DocumentHierarchy135 # assert desc['hierarchies']['hier']['driver'] == pynq.DefaultHierarchy136def test_hierarchy_replace():137 class HierarchyDriver1(pynq.DefaultHierarchy):138 @staticmethod139 def checkhierarchy(description):140 return 'test_ip' in description['ip']141 class HierarchyDriver2(pynq.DefaultHierarchy):142 @staticmethod143 def checkhierarchy(description):144 return 'test_ip' in description['ip']145 desc = _copy_assign(hier_description)146 assert desc['hierarchies']['hier']['driver'] == HierarchyDriver2147 HierarchyDriver2.unregister()148 desc = _copy_assign(hier_description)149 assert desc['hierarchies']['hier']['driver'] == HierarchyDriver1150 HierarchyDriver1.unregister()151 desc = _copy_assign(hier_description)152 assert desc['hierarchies']['hier']['driver'] == \153 pynq.overlay.DocumentHierarchy154 # assert desc['hierarchies']['hier']['driver'] == pynq.DefaultHierarchy155INTERRUPT_IP_DESCRIPTION = {156 'test_ip': {'phys_addr': 0x80000000, 'addr_range': 65536,157 'type': 'xilinx.com:test:test_ip:1.0', 'registers': {},158 'fullpath': 'test_ip',159 'interrupts': {'test_interrupt': {160 'parent': '', 'index': 0,161 'fullpath': 'test_ip/test_interrupt'}}162 }163}164class MockInterrupt:165 pass166def working_interrupt(*args, **kwargs):167 return MockInterrupt()168def broken_interrupt(*args, **kwargs):169 raise ValueError("IRQ not connected")170def test_interrupt_attribute(device, monkeypatch):171 monkeypatch.setattr(pynq.overlay, 'Interrupt', working_interrupt)172 desc = copy.deepcopy(INTERRUPT_IP_DESCRIPTION['test_ip'])173 desc['device'] = device174 ip = pynq.overlay.DefaultIP(desc)175 assert type(ip.test_interrupt) == MockInterrupt176def test_broken_interrupt_attribute(device, monkeypatch):177 monkeypatch.setattr(pynq.overlay, 'Interrupt', broken_interrupt)178 desc = copy.deepcopy(INTERRUPT_IP_DESCRIPTION['test_ip'])179 desc['device'] = device180 with pytest.warns(UserWarning):181 ip = pynq.overlay.DefaultIP(desc)...
test_rate_limiter.py
Source:test_rate_limiter.py
1import time2import unittest3from datetime import datetime4from unittest.mock import Mock, patch5from app.ip_analyzer.rate_limiter.limiter import IPCacheRateLimiter6class MockRedis:7 def __init__(self):8 self.storage = {}9 self.keys_ttl = {}10 def get(self, key):11 # Check if we have this key:12 value = self.storage.get(key)13 key_ttl = self.keys_ttl.get(key)14 if not key_ttl:15 return value16 current_time = int(time.time())17 if key_ttl["ttl"] + key_ttl["set_at"] <= current_time:18 # The key is expired!19 return None20 return value21 def expire(self, key, ttl):22 current_time = int(time.time())23 self.keys_ttl[key] = {24 "ttl": ttl,25 "set_at": current_time26 }27 def incr(self, key, count: int):28 if key not in self.storage:29 self.storage[key] = count30 return count31 self.storage[key] += count32 return self.storage[key]33mock_time = Mock()34class TestRateLimiter(unittest.TestCase):35 def setUp(self):36 self.mock_redis = MockRedis()37 self.limiter = IPCacheRateLimiter(self.mock_redis, 3, 2)38 mock_time.return_value = time.mktime(datetime(2020, 1, 1, 1, 1, 1).timetuple())39 @patch('time.time', mock_time)40 def test_simple_rate_limit_case(self):41 test_ip = '1.1.1.1'42 self.assertEqual(self.limiter.can_be_served(test_ip), True)43 self.assertEqual(self.limiter.can_be_served(test_ip), True)44 self.assertEqual(self.limiter.can_be_served(test_ip), True)45 # Rate limit!46 self.assertEqual(self.limiter.can_be_served(test_ip), False)47 # Check the storage state:48 self.assertDictEqual(self.mock_redis.storage, {'1.1.1.1#1577833261': 4})49 self.assertDictEqual(self.mock_redis.keys_ttl, {'1.1.1.1#1577833261': {'set_at': 1577833261, 'ttl': 4}})50 @patch('time.time', mock_time)51 def test_two_clients_rate_limit_case(self):52 test_ip = '1.1.1.1'53 second_ip = '1.1.1.2'54 self.assertEqual(self.limiter.can_be_served(test_ip), True)55 self.assertEqual(self.limiter.can_be_served(second_ip), True)56 self.assertEqual(self.limiter.can_be_served(test_ip), True)57 self.assertEqual(self.limiter.can_be_served(test_ip), True)58 # Rate limit!59 self.assertEqual(self.limiter.can_be_served(test_ip), False)60 # Second ip should not get rate limit:61 self.assertEqual(self.limiter.can_be_served(second_ip), True)62 # Check the storage state:63 self.assertDictEqual(self.mock_redis.storage, {'1.1.1.1#1577833261': 4, '1.1.1.2#1577833261': 2})64 self.assertDictEqual(self.mock_redis.keys_ttl, {'1.1.1.1#1577833261': {'set_at': 1577833261, 'ttl': 4},65 '1.1.1.2#1577833261': {'set_at': 1577833261, 'ttl': 4}})66 @patch('time.time', mock_time)67 def test_rate_limit_done_case(self):68 test_ip = '1.1.1.1'69 self.assertEqual(self.limiter.can_be_served(test_ip), True)70 self.assertEqual(self.limiter.can_be_served(test_ip), True)71 self.assertEqual(self.limiter.can_be_served(test_ip), True)72 # Rate limit!73 self.assertEqual(self.limiter.can_be_served(test_ip), False)74 # move 3 sec forward - limiting should be over now!75 mock_time.return_value = time.mktime(datetime(2020, 1, 1, 1, 1, 4).timetuple())76 self.assertEqual(self.limiter.can_be_served(test_ip), True)77 # Check the storage state:78 self.assertDictEqual(self.mock_redis.storage, {'1.1.1.1#1577833261': 4, '1.1.1.1#1577833264': 1})79 self.assertDictEqual(self.mock_redis.keys_ttl, {'1.1.1.1#1577833261': {'set_at': 1577833261, 'ttl': 4},80 '1.1.1.1#1577833264': {'set_at': 1577833264, 'ttl': 4}})81 @patch('time.time', mock_time)82 def test_sliding_window_alg(self):83 test_ip = '1.1.1.1'84 # req 1 + 2 on sec 185 self.assertEqual(self.limiter.can_be_served(test_ip), True)86 self.assertEqual(self.limiter.can_be_served(test_ip), True)87 # Move one sec forward:88 # req 3 on sec 289 mock_time.return_value = time.mktime(datetime(2020, 1, 1, 1, 1, 2).timetuple())90 self.assertEqual(self.limiter.can_be_served(test_ip), True)91 # req 4 on sec 2 - should get rate limit92 self.assertEqual(self.limiter.can_be_served(test_ip), False)93 # Move one sec forward - The window should include now only 2 req94 mock_time.return_value = time.mktime(datetime(2020, 1, 1, 1, 1, 3).timetuple())95 self.assertEqual(self.limiter.can_be_served(test_ip), True)96 # Check the storage state:97 self.assertDictEqual(self.mock_redis.storage,98 {'1.1.1.1#1577833261': 2, '1.1.1.1#1577833262': 2, '1.1.1.1#1577833263': 1})99 self.assertDictEqual(self.mock_redis.keys_ttl, {'1.1.1.1#1577833261': {'set_at': 1577833261, 'ttl': 4},100 '1.1.1.1#1577833262': {'set_at': 1577833262, 'ttl': 4},101 '1.1.1.1#1577833263': {'set_at': 1577833263, 'ttl': 4}})102if __name__ == '__main__':...
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 Airtest 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?
