KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use set_failures method in autotest
Best Python code snippet using autotest_python
test_client.py
Source:test_client.py 
1import mock2import pytest3import settings4from queue import Queue, Empty5from pycernan.avro import BaseDummyClient, DummyClient6from pycernan.avro.tcp_conn_pool import TCPConnectionPool, _DefunctConnection, EmptyPoolException7from pycernan.avro.exceptions import SchemaParseException, DatumTypeException, EmptyBatchException8USER_SCHEMA = {9 "namespace": "example.avro",10 "type": "record",11 "name": "User",12 "fields": [13 {"name": "name", "type": "string"},14 {"name": "favorite_number", "type": ["int", "null"]},15 {"name": "favorite_color", "type": ["string", "null"]}16 ]17}18class FakeSocket(object):19 def __init__(self, set_failures=0, close_failures=0):20 self.set_failures = set_failures21 self.close_failures = close_failures22 self.call_args_list = []23 def settimeout(self, *args, **kwargs):24 self.call_args_list.append(('settimeout', mock.call(*args, **kwargs)))25 if self.set_failures > 0:26 self.set_failures -= 127 raise Exception("settimeout Exception")28 def close(self):29 self.call_args_list.append(('close', mock.call()))30 if self.close_failures > 0:31 self.close_failures -= 132 raise Exception("close Exception")33 def _reset_mock(self):34 self.call_args_list = []35class ForcedException(Exception):36 pass37class TestTCPConnectionPool():38 def test_pool_value_errors(self):39 with pytest.raises(ValueError):40 TCPConnectionPool('foobar', 80, -1, 1, 1)41 @mock.patch.object(TCPConnectionPool, '_create_connection', return_value=None, autospec=True)42 def test_pool_creation(self, connection_mock):43 size = 1044 pool = TCPConnectionPool('foobar', 80, size, 3, 1)45 assert pool.pool is not None46 for _ in range(size):47 assert pool.pool.get_nowait() == _DefunctConnection48 # Pool should be drained now.49 with pytest.raises(Empty):50 pool.pool.get_nowait()51 @mock.patch('pycernan.avro.tcp_conn_pool.socket.create_connection', autospec=True)52 def test_connection(self, connect_mock):53 mock_sock = FakeSocket()54 host, port = ('foobar', 9000)55 connect_timeout, recv_timeout = (1, 1)56 connect_mock.return_value = mock_sock57 pool = TCPConnectionPool(host, port, 1, connect_timeout, recv_timeout)58 with pool.connection() as sock:59 assert sock == mock_sock60 assert mock_sock.call_args_list == [('settimeout', mock.call(recv_timeout))]61 assert connect_mock.call_args_list == [mock.call((host, port), timeout=connect_timeout)]62 # After using a connection, it should be checked-in/cached for others to use63 connect_mock.return_value = FakeSocket()64 assert not pool.pool.empty()65 with pool.connection() as sock:66 assert sock == mock_sock67 # When we are at capacity, then callers will block68 # waiting for a connection to be returned.69 with pytest.raises(EmptyPoolException):70 with pool.connection(_block=False):71 assert False # Should not be reached72 @mock.patch('pycernan.avro.tcp_conn_pool.socket.create_connection', autospec=True)73 def test_connection_on_exception_closes_connections(self, connect_mock):74 pool = TCPConnectionPool('foobar', 8080, 1, 1, 1)75 # Connections are closed when application layer Exceptions occur.76 mock_sock = FakeSocket()77 connect_mock.return_value = mock_sock78 with pytest.raises(ForcedException):79 with pool.connection() as sock:80 raise ForcedException()81 assert mock_sock.call_args_list[-1] == ('close', mock.call())82 # Repeat the same experiment, this time simulating exception on close.83 # The semantics should match, namely, the exception is tolerated and84 # users get back the application level Exception.85 mock_sock.close_failures = 1086 mock_sock.call_args_list = []87 with pytest.raises(ForcedException):88 with pool.connection() as sock:89 raise ForcedException()90 assert mock_sock.call_args_list[-1] == ('close', mock.call())91 @mock.patch.object(TCPConnectionPool, '_create_connection', side_effect=ForcedException("Oops"), autospec=True)92 def test_create_connection_reraises(self, create_connection_mock):93 pool = TCPConnectionPool('foobar', 80, 10, 1, 1)94 with pytest.raises(ForcedException):95 with pool.connection():96 assert False # Should not be reached.97 # No connections become cached.98 assert pool.pool.qsize() == 099 assert pool.size == 0100 @mock.patch('pycernan.avro.tcp_conn_pool.socket.create_connection', autospec=True)101 def test_exceptions_result_in_defunct_connections(self, connect_mock):102 size = 10103 pool = TCPConnectionPool('foobar', 80, size, 1, 1)104 mock_sock = FakeSocket()105 connect_mock.return_value = mock_sock106 with pytest.raises(ForcedException):107 with pool.connection():108 raise ForcedException()109 assert pool.pool.qsize() == size110 assert pool.pool.get_nowait() == _DefunctConnection111 @mock.patch('pycernan.avro.tcp_conn_pool.socket.create_connection', autospec=True)112 def test_defunct_connections_are_regenerated(self, connect_mock):113 pool = TCPConnectionPool('foobar', 80, 10, 1, 1)114 mock_sock = FakeSocket()115 connect_mock.return_value = mock_sock116 pool.pool = Queue()117 pool.pool.put(_DefunctConnection)118 with pool.connection() as conn:119 assert conn is mock_sock120 @mock.patch('pycernan.avro.tcp_conn_pool.socket.create_connection', autospec=True)121 def test_regenerating_connections_tolerates_exceptions(self, connect_mock):122 num_failures = 20123 pool = TCPConnectionPool('foobar', 80, 10, 1, 1)124 mock_sock = FakeSocket(set_failures=num_failures)125 connect_mock.return_value = mock_sock126 # Simulate a prior failure127 pool.pool = Queue()128 pool.size = 1129 pool.pool._put(_DefunctConnection)130 # Try a number of failed operations.131 # Each time the defunct connection should be returned to the queue132 # for subsequent calls to reattempt.133 for i in range(num_failures):134 with pytest.raises(Exception):135 with pool.connection():136 assert False # Should not be reached137 assert pool.pool.qsize() == pool.size == 1138 assert pool.pool.get_nowait() is _DefunctConnection139 pool.pool._put(_DefunctConnection)140 # Failures are exhausted, we should be able to now141 # regenerate a valid context.142 with pool.connection():143 pass144 assert pool.pool.qsize() == 1145 assert pool.pool.get_nowait() is mock_sock146 @mock.patch('pycernan.avro.tcp_conn_pool.socket.create_connection', autospec=True)147 def test_closing_tolerates_close_exceptions(self, create_mock):148 expected_sock = FakeSocket(close_failures=10)149 create_mock.return_value = expected_sock150 pool = TCPConnectionPool('foobar', 80, 10, 1, 1)151 with pool.connection():152 pass153 pool.closeall()154 assert expected_sock.call_args_list[-1] == ('close', mock.call())155@pytest.mark.parametrize("avro_file", settings.test_data)156def test_publish_file(avro_file):157 c = DummyClient()158 c.publish_file(avro_file)159@mock.patch('pycernan.avro.client.serialize', autospec=True)160@pytest.mark.parametrize('ephemeral', [True, False])161def test_publish(m_serialize, ephemeral):162 expected_serialize_result = mock.MagicMock()163 m_serialize.return_value = expected_serialize_result164 user = {165 'name': 'Foo Bar Matic',166 'favorite_number': 24,167 'favorite_color': 'Nonyabusiness',168 }169 c = DummyClient()170 with mock.patch.object(c, 'publish_blob', autospec=True) as m_publish_blob:171 c.publish(USER_SCHEMA, [user], ephemeral_storage=ephemeral, kwarg1='one', kwarg2='two')172 assert m_serialize.call_args_list == [173 mock.call(USER_SCHEMA, [user], ephemeral)174 ]175 assert m_publish_blob.call_args_list == [176 mock.call(expected_serialize_result, kwarg1='one', kwarg2='two')177 ]178def test_publish_bad_schema():179 schema = {}180 user = {}181 c = DummyClient()182 with pytest.raises(SchemaParseException):183 c.publish(schema, [user])184def test_publish_bad_datum_empty():185 user = {}186 c = DummyClient()187 with pytest.raises(DatumTypeException):188 c.publish(USER_SCHEMA, [user])189def test_publish_empty_batch():190 c = DummyClient()191 with pytest.raises(EmptyBatchException):192 c.publish(USER_SCHEMA, [])193@mock.patch('pycernan.avro.tcp_conn_pool.socket.create_connection', autospec=True)194def test_creating_a_client_connects_a_socket(m_connect):195 connect_timeout = 1196 publish_timeout = 50197 expected_sock = FakeSocket()198 m_connect.return_value = expected_sock199 client = BaseDummyClient(200 host='some fake host',201 port=31337,202 connect_timeout=connect_timeout,203 publish_timeout=publish_timeout)204 with client.pool.connection():205 assert expected_sock.call_args_list == [('settimeout', mock.call(publish_timeout))]206 assert m_connect.call_args_list == [mock.call(('some fake host', 31337), timeout=connect_timeout)]207@mock.patch('pycernan.avro.tcp_conn_pool.socket.create_connection', autospec=True)208def test_closing_the_client_closes_the_socket_and_clears_it(m_connect):209 expected_sock = FakeSocket()210 m_connect.return_value = expected_sock211 client = BaseDummyClient(212 host='some fake host',213 port=31337,214 connect_timeout=666,215 publish_timeout=999)216 with client.pool.connection():217 pass # Generates a connection218 assert client.pool is not None219 assert client.pool.pool.qsize()220 # Close calls should gracefully handle defunct connections too221 client.pool._put(_DefunctConnection)222 expected_sock._reset_mock()223 client.close()224 assert expected_sock.call_args_list == [('close', mock.call())]225 assert client.pool is not None226 assert client.pool.pool.qsize() == 0227@mock.patch('pycernan.avro.tcp_conn_pool.socket.create_connection', autospec=True)228def test_closing_the_client_with_no_socket_does_not_crash(m_connect):229 expected_sock = FakeSocket()230 m_connect.return_value = expected_sock231 client = BaseDummyClient(232 host='some fake host',233 port=31337,234 connect_timeout=666,235 publish_timeout=999)...
plot_svhn.py
Source:plot_svhn.py
1#2# Copyright (C) 2020 by The Board of Trustees of Stanford University3# This program is free software: you can redistribute it and/or modify it under4# the terms of the Modified BSD-3 License as published by the Open Source5# Initiative.6# If you use this program in your research, we request that you reference the7# Illusion paper, and that you send us a citation of your work.8# This program is distributed in the hope that it will be useful, but WITHOUT ANY9# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A10# PARTICULAR PURPOSE. See the BSD-3 License for more details.11# You should have received a copy of the Modified BSD-3 License along with this12# program. If not, see <https://opensource.org/licenses/BSD-3-Clause>.13#14import matplotlib.pyplot as plt15import numpy as np16N = 817M = 204818def get_failures(array, expected):19 failures = np.zeros((array.shape[0],16))20 for i in range(array.shape[0]):21 for j in range(16):22 if int(get_bit(array[i],j))!=expected:23 failures[i,j] = 124 return failures.reshape(-1)25def get_bit(uint16_word, index):26 word = np.uint16(uint16_word) # force just in case27 s = bin(word)[2:]28 if len(s)<16:29 s = "0"*(16-len(s))+s30 return s[index]31def get_masks(prefix,time="2010"):#19710"):32 zeros = [[] for i in range(N)]33 f = open(prefix + "/" + time + "trace_10c_0.txt",'r')34 for i in range(N):35 f.readline() # skip the "physical chip #" print36 for k in range(M/16): #while True:37 row = f.readline()38 row = row.split(' ') # ',' for csv39 row = [int(x,16) for x in row]40 zeros[i].extend(row)41 f.close()42 forced_zeros_memory = np.array(zeros, dtype=np.int16).view(dtype=np.uint16)43 ones = [[] for i in range(N)]44 f = open(prefix + "/" + time + "trace_10c_1.txt",'r')45 for i in range(N):46 f.readline() # skip the "physical chip #" print47 for k in range(M/16): #while True:48 row = f.readline()49 row = row.split(' ') # ',' for csv50 row = [int(x,16) for x in row]51 ones[i].extend(row)52 f.close()53 forced_ones_memory = np.array(ones, dtype=np.int16).view(dtype=np.uint16)54 return forced_ones_memory, forced_zeros_memory55bit_failures = np.zeros((8,2048*16))56start_addr_tmp = 057data_length = M58fig, axs = plt.subplots(2,8,sharex=True,sharey=True,figsize=(14,10))59cmap = "gnuplot" #"coolwarm_r"#"RdYlGn"60start_addr_tmp = 10#M-6461data_length = 6462mask_ones = np.zeros((8,2048*2),dtype=np.int8)63mask_zeros = np.zeros((8,2048*2),dtype=np.int8)64i=065f = open("../c_models_test/c_svhn/activation_mask_NR_TENYEAR.c","r")66data = f.readline()67full_data = ''68while ';' not in data:69 data = f.readline()70 full_data += data71mask_zeros[i,10:10+data_length*2] 72arr = eval("["+full_data[1:-3]+"]")73mask_zeros[i,10:10+data_length*2] = arr74data = f.readline()75full_data = ''76while ';' not in data:77 data = f.readline()78 full_data += data79arr = eval("["+full_data[1:-3]+"]")80mask_ones[i,10:10+data_length*2] = arr81f.close() 82start_addr_tmp = 1083for i in range(8):84 if (i==0):85 set_failures = get_failures(mask_ones[i,start_addr_tmp:start_addr_tmp+data_length],1)86 reset_failures = get_failures(mask_zeros[i,start_addr_tmp:start_addr_tmp+data_length],0)87 bit_failures[i,100*16:100*16+16*data_length] = np.logical_or(set_failures,reset_failures)88 #bit_failures[i,0:16*data_length] = np.logical_or(set_failures,reset_failures)89 #print(bit_failures.sum())90 axs[0,i].imshow(1-bit_failures[i].reshape((128,16*16)),extent=(0,16,0,128),aspect="auto", cmap=cmap ,vmin=0,vmax=1)91start_addr_tmp = 092data_length = M93bit_failures = np.zeros((8,2048*16))94mask_ones,mask_zeros = get_masks("../data/processed/svhn_de_clean")95mask_ones0,mask_zeros0 = get_masks("../data/processed/svhn_de_clean","0")96for i in range(8):97 set_failures = get_failures(mask_ones[i,start_addr_tmp:start_addr_tmp+data_length],1)98 reset_failures = get_failures(mask_zeros[i,start_addr_tmp:start_addr_tmp+data_length],0)99 bit_failures[i,0:16*data_length] = np.logical_or(set_failures,reset_failures)100 set_failures = get_failures(mask_ones0[i,start_addr_tmp:start_addr_tmp+data_length],1)101 reset_failures = get_failures(mask_zeros0[i,start_addr_tmp:start_addr_tmp+data_length],0)102 bit_failures[i,0:16*data_length] -= np.logical_or(set_failures,reset_failures)103 axs[1,i].imshow(1-bit_failures[i].reshape((128,16*16)),extent=(0,16,0,128),aspect="auto", cmap=cmap,vmin=0,vmax=1)104for i in range(8):105 axs[1,i].set_xlabel("Chip " +str(i),fontsize=18)106 for j in range(2):107 axs[j,i].set_xticks(np.arange(0,16,4))108 axs[j,i].set_xticklabels(np.arange(0,16*16,4*16))109 #axs[j,i].set_xlim((0,16))110axs[0,0].set_ylabel("Without any\nResiliency Technique",fontsize=18)111axs[1,0].set_ylabel("With\nDistributed ENDURER",fontsize=18)112fig.suptitle("Permanent Bit Failures - 10-Years: SVHN",fontsize=28)...
plot_d2nn.py
Source:plot_d2nn.py
1#2# Copyright (C) 2020 by The Board of Trustees of Stanford University3# This program is free software: you can redistribute it and/or modify it under4# the terms of the Modified BSD-3 License as published by the Open Source5# Initiative.6# If you use this program in your research, we request that you reference the7# Illusion paper, and that you send us a citation of your work.8# This program is distributed in the hope that it will be useful, but WITHOUT ANY9# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A10# PARTICULAR PURPOSE. See the BSD-3 License for more details.11# You should have received a copy of the Modified BSD-3 License along with this12# program. If not, see <https://opensource.org/licenses/BSD-3-Clause>.13#14import matplotlib.pyplot as plt15import numpy as np16N = 817M = 204818def get_failures(array, expected):19 failures = np.zeros((array.shape[0],16))20 for i in range(array.shape[0]):21 for j in range(16):22 if int(get_bit(array[i],j))!=expected:23 failures[i,j] = 124 return failures.reshape(-1)25def get_bit(uint16_word, index):26 word = np.uint16(uint16_word) # force just in case27 s = bin(word)[2:]28 if len(s)<16:29 s = "0"*(16-len(s))+s30 return s[index]31def get_masks(prefix,time="19710"):32 zeros = [[] for i in range(N)]33 f = open(prefix + "/" + time + "trace_10c_0.txt",'r')34 for i in range(N):35 f.readline() # skip the "physical chip #" print36 for k in range(M/16): #while True:37 row = f.readline()38 row = row.split(' ') # ',' for csv39 row = [int(x,16) for x in row]40 zeros[i].extend(row)41 f.close()42 forced_zeros_memory = np.array(zeros, dtype=np.int16).view(dtype=np.uint16)43 ones = [[] for i in range(N)]44 f = open(prefix + "/" + time + "trace_10c_1.txt",'r')45 for i in range(N):46 f.readline() # skip the "physical chip #" print47 for k in range(M/16): #while True:48 row = f.readline()49 row = row.split(' ') # ',' for csv50 row = [int(x,16) for x in row]51 ones[i].extend(row)52 f.close()53 forced_ones_memory = np.array(ones, dtype=np.int16).view(dtype=np.uint16)54 return forced_ones_memory, forced_zeros_memory55bit_failures = np.zeros((8,2048*16))56start_addr_tmp = 057data_length = M58fig, axs = plt.subplots(2,8,sharex=True,sharey=True,figsize=(14,10))59cmap = "gnuplot"60start_addr_tmp = M-102461data_length = 21662mask_ones,mask_zeros = get_masks("../data/processed/d2nn_no_resilience_clean","59130")63for i in range(8):64 if i==0:65 set_failures = get_failures(mask_ones[i,start_addr_tmp:start_addr_tmp+data_length],1)66 reset_failures = get_failures(mask_zeros[i,start_addr_tmp:start_addr_tmp+data_length],0)67 bit_failures[i,100*16:100*16+16*data_length] = np.logical_or(set_failures,reset_failures)68 axs[0,i].imshow(1-bit_failures[i].reshape((128,16*16)),extent=(0,16,0,128),aspect="auto", cmap=cmap ,vmin=0,vmax=1)69start_addr_tmp = 070data_length = M71bit_failures = np.zeros((8,2048*16))72mask_ones,mask_zeros = get_masks("../data/processed/d2nn_de_clean","59130")73mask_ones0,mask_zeros0 = get_masks("../data/processed/d2nn_de_clean","0")74for i in range(8):75 set_failures = get_failures(mask_ones[i,start_addr_tmp:start_addr_tmp+data_length],1)76 reset_failures = get_failures(mask_zeros[i,start_addr_tmp:start_addr_tmp+data_length],0)77 bit_failures[i,0:16*data_length] = np.logical_or(set_failures,reset_failures)78 set_failures = get_failures(mask_ones0[i,start_addr_tmp:start_addr_tmp+data_length],1)79 reset_failures = get_failures(mask_zeros0[i,start_addr_tmp:start_addr_tmp+data_length],0)80 bit_failures[i,0:16*data_length] -= np.logical_or(set_failures,reset_failures)81 axs[1,i].imshow(1-bit_failures[i].reshape((128,16*16)),extent=(0,16,0,128),aspect="auto", cmap=cmap,vmin=0,vmax=1)82for i in range(8):83 axs[1,i].set_xlabel("Chip " +str(i),fontsize=18)84 for j in range(2):85 axs[j,i].set_xticks(np.arange(0,16,4))86 axs[j,i].set_xticklabels(np.arange(0,16*16,4*16))87 #axs[j,i].set_xlim((0,16))88axs[0,0].set_ylabel("Without any\nResiliency Technique",fontsize=18)89axs[1,0].set_ylabel("With\nDistributed ENDURER",fontsize=18)90fig.suptitle("Permanent Bit Failures - 10-Years: D2NN",fontsize=28)...
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 autotest 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?
