KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use upsert method in responses
Best Python code snippet using responses
test_types.py
Source:test_types.py 
1# Licensed to the Apache Software Foundation (ASF) under one or more2# contributor license agreements. See the NOTICE file distributed with3# this work for additional information regarding copyright ownership.4# The ASF licenses this file to You under the Apache License, Version 2.05# (the "License"); you may not use this file except in compliance with6# the License. You may obtain a copy of the License at7#8# http://www.apache.org/licenses/LICENSE-2.09#10# Unless required by applicable law or agreed to in writing, software11# distributed under the License is distributed on an "AS IS" BASIS,12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.13# See the License for the specific language governing permissions and14# limitations under the License.15import sys16import unittest17import datetime18import phoenixdb19from decimal import Decimal20from phoenixdb.tests import DatabaseTestCase21class TypesTest(DatabaseTestCase):22 def checkIntType(self, type_name, min_value, max_value):23 self.createTable("phoenixdb_test_tbl1", "id integer primary key, val {}".format(type_name))24 with self.conn.cursor() as cursor:25 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (1, 1)")26 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (2, NULL)")27 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (3, ?)", [1])28 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (4, ?)", [None])29 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (5, ?)", [min_value])30 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (6, ?)", [max_value])31 cursor.execute("SELECT id, val FROM phoenixdb_test_tbl1 ORDER BY id")32 self.assertEqual(cursor.description[1].type_code, phoenixdb.NUMBER)33 self.assertEqual(cursor.fetchall(), [[1, 1], [2, None], [3, 1], [4, None], [5, min_value], [6, max_value]])34 self.assertRaises(35 self.conn.DatabaseError, cursor.execute,36 "UPSERT INTO phoenixdb_test_tbl1 VALUES (100, {})".format(min_value - 1))37 self.assertRaises(38 self.conn.DatabaseError, cursor.execute,39 "UPSERT INTO phoenixdb_test_tbl1 VALUES (100, {})".format(max_value + 1))40 # XXX The server silently truncates the values41# self.assertRaises(self.conn.DatabaseError, cursor.execute, "UPSERT INTO phoenixdb_test_tbl1 VALUES (100, ?)", [min_value - 1])42# self.assertRaises(self.conn.DatabaseError, cursor.execute, "UPSERT INTO phoenixdb_test_tbl1 VALUES (100, ?)", [max_value + 1])43 def test_integer(self):44 self.checkIntType("integer", -2147483648, 2147483647)45 def test_unsigned_int(self):46 self.checkIntType("unsigned_int", 0, 2147483647)47 def test_bigint(self):48 self.checkIntType("bigint", -9223372036854775808, 9223372036854775807)49 def test_unsigned_long(self):50 self.checkIntType("unsigned_long", 0, 9223372036854775807)51 def test_tinyint(self):52 self.checkIntType("tinyint", -128, 127)53 def test_unsigned_tinyint(self):54 self.checkIntType("unsigned_tinyint", 0, 127)55 def test_smallint(self):56 self.checkIntType("smallint", -32768, 32767)57 def test_unsigned_smallint(self):58 self.checkIntType("unsigned_smallint", 0, 32767)59 def checkFloatType(self, type_name, min_value, max_value):60 self.createTable("phoenixdb_test_tbl1", "id integer primary key, val {}".format(type_name))61 with self.conn.cursor() as cursor:62 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (1, 1)")63 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (2, NULL)")64 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (3, ?)", [1])65 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (4, ?)", [None])66 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (5, ?)", [min_value])67 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (6, ?)", [max_value])68 cursor.execute("SELECT id, val FROM phoenixdb_test_tbl1 ORDER BY id")69 self.assertEqual(cursor.description[1].type_code, phoenixdb.NUMBER)70 rows = cursor.fetchall()71 self.assertEqual([r[0] for r in rows], [1, 2, 3, 4, 5, 6])72 self.assertEqual(rows[0][1], 1.0)73 self.assertEqual(rows[1][1], None)74 self.assertEqual(rows[2][1], 1.0)75 self.assertEqual(rows[3][1], None)76 self.assertAlmostEqual(rows[4][1], min_value)77 self.assertAlmostEqual(rows[5][1], max_value)78 def test_float(self):79 self.checkFloatType("float", -3.4028234663852886e+38, 3.4028234663852886e+38)80 def test_unsigned_float(self):81 self.checkFloatType("unsigned_float", 0, 3.4028234663852886e+38)82 def test_double(self):83 self.checkFloatType("double", -1.7976931348623158E+308, 1.7976931348623158E+308)84 def test_unsigned_double(self):85 self.checkFloatType("unsigned_double", 0, 1.7976931348623158E+308)86 def test_decimal(self):87 self.createTable("phoenixdb_test_tbl1", "id integer primary key, val decimal(8,3)")88 with self.conn.cursor() as cursor:89 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (1, 33333.333)")90 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (2, NULL)")91 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (3, ?)", [33333.333])92 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (4, ?)", [Decimal('33333.333')])93 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (5, ?)", [None])94 cursor.execute("SELECT id, val FROM phoenixdb_test_tbl1 ORDER BY id")95 self.assertEqual(cursor.description[1].type_code, phoenixdb.NUMBER)96 rows = cursor.fetchall()97 self.assertEqual([r[0] for r in rows], [1, 2, 3, 4, 5])98 self.assertEqual(rows[0][1], Decimal('33333.333'))99 self.assertEqual(rows[1][1], None)100 self.assertEqual(rows[2][1], Decimal('33333.333'))101 self.assertEqual(rows[3][1], Decimal('33333.333'))102 self.assertEqual(rows[4][1], None)103 self.assertRaises(104 self.conn.DatabaseError, cursor.execute,105 "UPSERT INTO phoenixdb_test_tbl1 VALUES (100, ?)", [Decimal('1234567890')])106 self.assertRaises(107 self.conn.DatabaseError, cursor.execute,108 "UPSERT INTO phoenixdb_test_tbl1 VALUES (101, ?)", [Decimal('123456.789')])109 def test_boolean(self):110 self.createTable("phoenixdb_test_tbl1", "id integer primary key, val boolean")111 with self.conn.cursor() as cursor:112 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (1, TRUE)")113 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (2, FALSE)")114 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (3, NULL)")115 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (4, ?)", [True])116 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (5, ?)", [False])117 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (6, ?)", [None])118 cursor.execute("SELECT id, val FROM phoenixdb_test_tbl1 ORDER BY id")119 self.assertEqual(cursor.description[1].type_code, phoenixdb.BOOLEAN)120 self.assertEqual(cursor.fetchall(), [[1, True], [2, False], [3, None], [4, True], [5, False], [6, None]])121 def test_time(self):122 self.createTable("phoenixdb_test_tbl1", "id integer primary key, val time")123 with self.conn.cursor() as cursor:124 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (1, '1970-01-01 12:01:02')")125 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (2, NULL)")126 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (3, ?)", [phoenixdb.Time(12, 1, 2)])127 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (4, ?)", [datetime.time(12, 1, 2)])128 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (5, ?)", [None])129 cursor.execute("SELECT id, val FROM phoenixdb_test_tbl1 ORDER BY id")130 self.assertEqual(cursor.fetchall(), [131 [1, datetime.time(12, 1, 2)],132 [2, None],133 [3, datetime.time(12, 1, 2)],134 [4, datetime.time(12, 1, 2)],135 [5, None],136 ])137 @unittest.skip("https://issues.apache.org/jira/browse/CALCITE-797")138 def test_time_full(self):139 self.createTable("phoenixdb_test_tbl1", "id integer primary key, val time")140 with self.conn.cursor() as cursor:141 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (1, '2015-07-12 13:01:02.123')")142 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (2, ?)", [datetime.datetime(2015, 7, 12, 13, 1, 2, 123000)])143 cursor.execute("SELECT id, val FROM phoenixdb_test_tbl1 ORDER BY id")144 self.assertEqual(cursor.fetchall(), [145 [1, datetime.datetime(2015, 7, 12, 13, 1, 2, 123000)],146 [2, datetime.datetime(2015, 7, 12, 13, 1, 2, 123000)],147 ])148 def test_date(self):149 self.createTable("phoenixdb_test_tbl1", "id integer primary key, val date")150 with self.conn.cursor() as cursor:151 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (1, '2015-07-12 00:00:00')")152 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (3, ?)", [phoenixdb.Date(2015, 7, 12)])153 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (4, ?)", [datetime.date(2015, 7, 12)])154 cursor.execute("SELECT id, val FROM phoenixdb_test_tbl1 ORDER BY id")155 self.assertEqual(cursor.fetchall(), [156 [1, datetime.date(2015, 7, 12)],157 [3, datetime.date(2015, 7, 12)],158 [4, datetime.date(2015, 7, 12)],159 ])160 @unittest.skip("https://issues.apache.org/jira/browse/CALCITE-798")161 def test_date_full(self):162 self.createTable("phoenixdb_test_tbl1", "id integer primary key, val date")163 with self.conn.cursor() as cursor:164 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (1, '2015-07-12 13:01:02.123')")165 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (2, ?)", [datetime.datetime(2015, 7, 12, 13, 1, 2, 123000)])166 cursor.execute("SELECT id, val FROM phoenixdb_test_tbl1 ORDER BY id")167 self.assertEqual(cursor.fetchall(), [168 [1, datetime.datetime(2015, 7, 12, 13, 1, 2, 123000)],169 [2, datetime.datetime(2015, 7, 12, 13, 1, 2, 123000)],170 ])171 def test_date_null(self):172 self.createTable("phoenixdb_test_tbl1", "id integer primary key, val date")173 with self.conn.cursor() as cursor:174 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (1, NULL)")175 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (2, ?)", [None])176 cursor.execute("SELECT id, val FROM phoenixdb_test_tbl1 ORDER BY id") # raises NullPointerException on the server177 self.assertEqual(cursor.fetchall(), [178 [1, None],179 [2, None],180 ])181 def test_timestamp(self):182 self.createTable("phoenixdb_test_tbl1", "id integer primary key, val timestamp")183 with self.conn.cursor() as cursor:184 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (1, '2015-07-12 13:01:02.123')")185 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (2, NULL)")186 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (3, ?)", [phoenixdb.Timestamp(2015, 7, 12, 13, 1, 2)])187 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (4, ?)", [datetime.datetime(2015, 7, 12, 13, 1, 2, 123000)])188 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (5, ?)", [None])189 cursor.execute("SELECT id, val FROM phoenixdb_test_tbl1 ORDER BY id")190 self.assertEqual(cursor.fetchall(), [191 [1, datetime.datetime(2015, 7, 12, 13, 1, 2, 123000)],192 [2, None],193 [3, datetime.datetime(2015, 7, 12, 13, 1, 2)],194 [4, datetime.datetime(2015, 7, 12, 13, 1, 2, 123000)],195 [5, None],196 ])197 @unittest.skip("https://issues.apache.org/jira/browse/CALCITE-796")198 def test_timestamp_full(self):199 self.createTable("phoenixdb_test_tbl1", "id integer primary key, val timestamp")200 with self.conn.cursor() as cursor:201 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (1, '2015-07-12 13:01:02.123456789')")202 cursor.execute("SELECT id, val FROM phoenixdb_test_tbl1 ORDER BY id")203 self.assertEqual(cursor.fetchall(), [204 [1, datetime.datetime(2015, 7, 12, 13, 1, 2, 123456789)],205 ])206 def test_varchar(self):207 self.createTable("phoenixdb_test_tbl1", "id integer primary key, val varchar")208 with self.conn.cursor() as cursor:209 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (1, 'abc')")210 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (2, NULL)")211 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (3, ?)", ['abc'])212 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (4, ?)", [None])213 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (5, '')")214 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (6, ?)", [''])215 cursor.execute("SELECT id, val FROM phoenixdb_test_tbl1 ORDER BY id")216 self.assertEqual(cursor.fetchall(), [[1, 'abc'], [2, None], [3, 'abc'], [4, None], [5, None], [6, None]])217 def test_varchar_very_long(self):218 self.createTable("phoenixdb_test_tbl1", "id integer primary key, val varchar")219 with self.conn.cursor() as cursor:220 value = '1234567890' * 1000221 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (1, ?)", [value])222 cursor.execute("SELECT id, val FROM phoenixdb_test_tbl1 ORDER BY id")223 self.assertEqual(cursor.fetchall(), [[1, value]])224 def test_varchar_limited(self):225 self.createTable("phoenixdb_test_tbl1", "id integer primary key, val varchar(2)")226 with self.conn.cursor() as cursor:227 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (1, 'ab')")228 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (2, NULL)")229 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (3, ?)", ['ab'])230 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (4, ?)", [None])231 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (5, '')")232 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (6, ?)", [''])233 cursor.execute("SELECT id, val FROM phoenixdb_test_tbl1 ORDER BY id")234 self.assertEqual(cursor.fetchall(), [[1, 'ab'], [2, None], [3, 'ab'], [4, None], [5, None], [6, None]])235 self.assertRaises(self.conn.DataError, cursor.execute, "UPSERT INTO phoenixdb_test_tbl1 VALUES (100, 'abc')")236 def test_char_null(self):237 self.createTable("phoenixdb_test_tbl1", "id integer primary key, val char(2)")238 with self.conn.cursor() as cursor:239 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (2, NULL)")240 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (4, ?)", [None])241 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (5, '')")242 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (6, ?)", [''])243 cursor.execute("SELECT id, val FROM phoenixdb_test_tbl1 ORDER BY id")244 self.assertEqual(cursor.fetchall(), [[2, None], [4, None], [5, None], [6, None]])245 self.assertRaises(self.conn.DataError, cursor.execute, "UPSERT INTO phoenixdb_test_tbl1 VALUES (100, 'abc')")246 def test_char(self):247 self.createTable("phoenixdb_test_tbl1", "id integer primary key, val char(2)")248 with self.conn.cursor() as cursor:249 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (1, 'ab')")250 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (2, ?)", ['ab'])251 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (3, 'a')")252 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (4, ?)", ['b'])253 cursor.execute("SELECT id, val FROM phoenixdb_test_tbl1 ORDER BY id")254 self.assertEqual(cursor.fetchall(), [[1, 'ab'], [2, 'ab'], [3, 'a'], [4, 'b']])255 self.assertRaises(self.conn.DataError, cursor.execute, "UPSERT INTO phoenixdb_test_tbl1 VALUES (100, 'abc')")256 def test_binary(self):257 self.createTable("phoenixdb_test_tbl1", "id integer primary key, val binary(2)")258 with self.conn.cursor() as cursor:259 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (1, 'ab')")260 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (2, ?)", [phoenixdb.Binary(b'ab')])261 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (3, '\x01\x00')")262 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (4, ?)", [phoenixdb.Binary(b'\x01\x00')])263 cursor.execute("SELECT id, val FROM phoenixdb_test_tbl1 ORDER BY id")264 self.assertEqual(cursor.fetchall(), [265 [1, b'ab'],266 [2, b'ab'],267 [3, b'\x01\x00'],268 [4, b'\x01\x00'],269 ])270 def test_binary_all_bytes(self):271 self.createTable("phoenixdb_test_tbl1", "id integer primary key, val binary(256)")272 with self.conn.cursor() as cursor:273 if sys.version_info[0] < 3:274 value = ''.join(map(chr, range(256)))275 else:276 value = bytes(range(256))277 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (1, ?)", [phoenixdb.Binary(value)])278 cursor.execute("SELECT id, val FROM phoenixdb_test_tbl1 ORDER BY id")279 self.assertEqual(cursor.fetchall(), [[1, value]])280 @unittest.skip("https://issues.apache.org/jira/browse/CALCITE-1050 https://issues.apache.org/jira/browse/PHOENIX-2585")281 def test_array(self):282 self.createTable("phoenixdb_test_tbl1", "id integer primary key, val integer[]")283 with self.conn.cursor() as cursor:284 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (1, ARRAY[1, 2])")285 cursor.execute("UPSERT INTO phoenixdb_test_tbl1 VALUES (2, ?)", [[2, 3]])286 cursor.execute("SELECT id, val FROM phoenixdb_test_tbl1 ORDER BY id")287 self.assertEqual(cursor.fetchall(), [288 [1, [1, 2]],289 [2, [2, 3]],...
test_upsert_validator_vote.py
Source:test_upsert_validator_vote.py
1# Copyright BigchainDB GmbH and BigchainDB contributors2# SPDX-License-Identifier: (Apache-2.0 AND CC-BY-4.0)3# Code is Apache-2.0 and docs are CC-BY-4.04import pytest5import codecs6from bigchaindb.elections.election import Election7from bigchaindb.tendermint_utils import public_key_to_base648from bigchaindb.upsert_validator import ValidatorElection9from bigchaindb.common.exceptions import AmountError10from bigchaindb.common.crypto import generate_key_pair11from bigchaindb.common.exceptions import ValidationError12from bigchaindb.common.transaction_mode_types import BROADCAST_TX_COMMIT13from bigchaindb.elections.vote import Vote14from tests.utils import generate_block, gen_vote15pytestmark = [pytest.mark.execute]16@pytest.mark.bdb17def test_upsert_validator_valid_election_vote(b_mock, valid_upsert_validator_election, ed25519_node_keys):18 b_mock.store_bulk_transactions([valid_upsert_validator_election])19 input0 = valid_upsert_validator_election.to_inputs()[0]20 votes = valid_upsert_validator_election.outputs[0].amount21 public_key0 = input0.owners_before[0]22 key0 = ed25519_node_keys[public_key0]23 election_pub_key = ValidatorElection.to_public_key(valid_upsert_validator_election.id)24 vote = Vote.generate([input0],25 [([election_pub_key], votes)],26 election_id=valid_upsert_validator_election.id)\27 .sign([key0.private_key])28 assert vote.validate(b_mock)29@pytest.mark.bdb30def test_upsert_validator_valid_non_election_vote(b_mock, valid_upsert_validator_election, ed25519_node_keys):31 b_mock.store_bulk_transactions([valid_upsert_validator_election])32 input0 = valid_upsert_validator_election.to_inputs()[0]33 votes = valid_upsert_validator_election.outputs[0].amount34 public_key0 = input0.owners_before[0]35 key0 = ed25519_node_keys[public_key0]36 election_pub_key = ValidatorElection.to_public_key(valid_upsert_validator_election.id)37 # Ensure that threshold conditions are now allowed38 with pytest.raises(ValidationError):39 Vote.generate([input0],40 [([election_pub_key, key0.public_key], votes)],41 election_id=valid_upsert_validator_election.id)\42 .sign([key0.private_key])43@pytest.mark.bdb44def test_upsert_validator_delegate_election_vote(b_mock, valid_upsert_validator_election, ed25519_node_keys):45 alice = generate_key_pair()46 b_mock.store_bulk_transactions([valid_upsert_validator_election])47 input0 = valid_upsert_validator_election.to_inputs()[0]48 votes = valid_upsert_validator_election.outputs[0].amount49 public_key0 = input0.owners_before[0]50 key0 = ed25519_node_keys[public_key0]51 delegate_vote = Vote.generate([input0],52 [([alice.public_key], 3), ([key0.public_key], votes-3)],53 election_id=valid_upsert_validator_election.id)\54 .sign([key0.private_key])55 assert delegate_vote.validate(b_mock)56 b_mock.store_bulk_transactions([delegate_vote])57 election_pub_key = ValidatorElection.to_public_key(valid_upsert_validator_election.id)58 alice_votes = delegate_vote.to_inputs()[0]59 alice_casted_vote = Vote.generate([alice_votes],60 [([election_pub_key], 3)],61 election_id=valid_upsert_validator_election.id)\62 .sign([alice.private_key])63 assert alice_casted_vote.validate(b_mock)64 key0_votes = delegate_vote.to_inputs()[1]65 key0_casted_vote = Vote.generate([key0_votes],66 [([election_pub_key], votes-3)],67 election_id=valid_upsert_validator_election.id)\68 .sign([key0.private_key])69 assert key0_casted_vote.validate(b_mock)70@pytest.mark.bdb71def test_upsert_validator_invalid_election_vote(b_mock, valid_upsert_validator_election, ed25519_node_keys):72 b_mock.store_bulk_transactions([valid_upsert_validator_election])73 input0 = valid_upsert_validator_election.to_inputs()[0]74 votes = valid_upsert_validator_election.outputs[0].amount75 public_key0 = input0.owners_before[0]76 key0 = ed25519_node_keys[public_key0]77 election_pub_key = ValidatorElection.to_public_key(valid_upsert_validator_election.id)78 vote = Vote.generate([input0],79 [([election_pub_key], votes+1)],80 election_id=valid_upsert_validator_election.id)\81 .sign([key0.private_key])82 with pytest.raises(AmountError):83 assert vote.validate(b_mock)84@pytest.mark.bdb85def test_valid_election_votes_received(b_mock, valid_upsert_validator_election, ed25519_node_keys):86 alice = generate_key_pair()87 b_mock.store_bulk_transactions([valid_upsert_validator_election])88 assert valid_upsert_validator_election.get_commited_votes(b_mock) == 089 input0 = valid_upsert_validator_election.to_inputs()[0]90 votes = valid_upsert_validator_election.outputs[0].amount91 public_key0 = input0.owners_before[0]92 key0 = ed25519_node_keys[public_key0]93 # delegate some votes to alice94 delegate_vote = Vote.generate([input0],95 [([alice.public_key], 4), ([key0.public_key], votes-4)],96 election_id=valid_upsert_validator_election.id)\97 .sign([key0.private_key])98 b_mock.store_bulk_transactions([delegate_vote])99 assert valid_upsert_validator_election.get_commited_votes(b_mock) == 0100 election_public_key = ValidatorElection.to_public_key(valid_upsert_validator_election.id)101 alice_votes = delegate_vote.to_inputs()[0]102 key0_votes = delegate_vote.to_inputs()[1]103 alice_casted_vote = Vote.generate([alice_votes],104 [([election_public_key], 2), ([alice.public_key], 2)],105 election_id=valid_upsert_validator_election.id)\106 .sign([alice.private_key])107 assert alice_casted_vote.validate(b_mock)108 b_mock.store_bulk_transactions([alice_casted_vote])109 # Check if the delegated vote is count as valid vote110 assert valid_upsert_validator_election.get_commited_votes(b_mock) == 2111 key0_casted_vote = Vote.generate([key0_votes],112 [([election_public_key], votes-4)],113 election_id=valid_upsert_validator_election.id)\114 .sign([key0.private_key])115 assert key0_casted_vote.validate(b_mock)116 b_mock.store_bulk_transactions([key0_casted_vote])117 assert valid_upsert_validator_election.get_commited_votes(b_mock) == votes - 2118@pytest.mark.bdb119def test_valid_election_conclude(b_mock, valid_upsert_validator_election, ed25519_node_keys):120 # Node 0: cast vote121 tx_vote0 = gen_vote(valid_upsert_validator_election, 0, ed25519_node_keys)122 # check if the vote is valid even before the election doesn't exist123 with pytest.raises(ValidationError):124 assert tx_vote0.validate(b_mock)125 # store election126 b_mock.store_bulk_transactions([valid_upsert_validator_election])127 # cannot conclude election as not votes exist128 assert not valid_upsert_validator_election.has_concluded(b_mock)129 # validate vote130 assert tx_vote0.validate(b_mock)131 assert not valid_upsert_validator_election.has_concluded(b_mock, [tx_vote0])132 b_mock.store_bulk_transactions([tx_vote0])133 assert not valid_upsert_validator_election.has_concluded(b_mock)134 # Node 1: cast vote135 tx_vote1 = gen_vote(valid_upsert_validator_election, 1, ed25519_node_keys)136 # Node 2: cast vote137 tx_vote2 = gen_vote(valid_upsert_validator_election, 2, ed25519_node_keys)138 # Node 3: cast vote139 tx_vote3 = gen_vote(valid_upsert_validator_election, 3, ed25519_node_keys)140 assert tx_vote1.validate(b_mock)141 assert not valid_upsert_validator_election.has_concluded(b_mock, [tx_vote1])142 # 2/3 is achieved in the same block so the election can be.has_concludedd143 assert valid_upsert_validator_election.has_concluded(b_mock, [tx_vote1, tx_vote2])144 b_mock.store_bulk_transactions([tx_vote1])145 assert not valid_upsert_validator_election.has_concluded(b_mock)146 assert tx_vote2.validate(b_mock)147 assert tx_vote3.validate(b_mock)148 # conclusion can be triggered my different votes in the same block149 assert valid_upsert_validator_election.has_concluded(b_mock, [tx_vote2])150 assert valid_upsert_validator_election.has_concluded(b_mock, [tx_vote2, tx_vote3])151 b_mock.store_bulk_transactions([tx_vote2])152 # Once the blockchain records >2/3 of the votes the election is assumed to be.has_concludedd153 # so any invocation of `.has_concluded` for that election should return False154 assert not valid_upsert_validator_election.has_concluded(b_mock)155 # Vote is still valid but the election cannot be.has_concludedd as it it assmed that it has156 # been.has_concludedd before157 assert tx_vote3.validate(b_mock)158 assert not valid_upsert_validator_election.has_concluded(b_mock, [tx_vote3])159@pytest.mark.abci160def test_upsert_validator(b, node_key, node_keys, ed25519_node_keys):161 if b.get_latest_block()['height'] == 0:162 generate_block(b)163 (node_pub, _) = list(node_keys.items())[0]164 validators = [{'public_key': {'type': 'ed25519-base64', 'value': node_pub},165 'voting_power': 10}]166 latest_block = b.get_latest_block()167 # reset the validator set168 b.store_validator_set(latest_block['height'], validators)169 generate_block(b)170 power = 1171 public_key = '9B3119650DF82B9A5D8A12E38953EA47475C09F0C48A4E6A0ECE182944B24403'172 public_key64 = public_key_to_base64(public_key)173 new_validator = {'public_key': {'value': public_key, 'type': 'ed25519-base16'},174 'node_id': 'some_node_id',175 'power': power}176 voters = ValidatorElection.recipients(b)177 election = ValidatorElection.generate([node_key.public_key],178 voters,179 new_validator, None).sign([node_key.private_key])180 code, message = b.write_transaction(election, BROADCAST_TX_COMMIT)181 assert code == 202182 assert b.get_transaction(election.id)183 tx_vote = gen_vote(election, 0, ed25519_node_keys)184 assert tx_vote.validate(b)185 code, message = b.write_transaction(tx_vote, BROADCAST_TX_COMMIT)186 assert code == 202187 resp = b.get_validators()188 validator_pub_keys = []189 for v in resp:190 validator_pub_keys.append(v['public_key']['value'])191 assert (public_key64 in validator_pub_keys)192 new_validator_set = b.get_validators()193 validator_pub_keys = []194 for v in new_validator_set:195 validator_pub_keys.append(v['public_key']['value'])196 assert (public_key64 in validator_pub_keys)197@pytest.mark.bdb198def test_get_validator_update(b, node_keys, node_key, ed25519_node_keys):199 reset_validator_set(b, node_keys, 1)200 power = 1201 public_key = '9B3119650DF82B9A5D8A12E38953EA47475C09F0C48A4E6A0ECE182944B24403'202 public_key64 = public_key_to_base64(public_key)203 new_validator = {'public_key': {'value': public_key, 'type': 'ed25519-base16'},204 'node_id': 'some_node_id',205 'power': power}206 voters = ValidatorElection.recipients(b)207 election = ValidatorElection.generate([node_key.public_key],208 voters,209 new_validator).sign([node_key.private_key])210 # store election211 b.store_bulk_transactions([election])212 tx_vote0 = gen_vote(election, 0, ed25519_node_keys)213 tx_vote1 = gen_vote(election, 1, ed25519_node_keys)214 tx_vote2 = gen_vote(election, 2, ed25519_node_keys)215 assert not election.has_concluded(b, [tx_vote0])216 assert not election.has_concluded(b, [tx_vote0, tx_vote1])217 assert election.has_concluded(b, [tx_vote0, tx_vote1, tx_vote2])218 assert Election.process_block(b, 4, [tx_vote0]) == []219 assert Election.process_block(b, 4, [tx_vote0, tx_vote1]) == []220 update = Election.process_block(b, 4, [tx_vote0, tx_vote1, tx_vote2])221 assert len(update) == 1222 update_public_key = codecs.encode(update[0].pub_key.data, 'base64').decode().rstrip('\n')223 assert update_public_key == public_key64224 # remove validator225 power = 0226 new_validator = {'public_key': {'value': public_key, 'type': 'ed25519-base16'},227 'node_id': 'some_node_id',228 'power': power}229 voters = ValidatorElection.recipients(b)230 election = ValidatorElection.generate([node_key.public_key],231 voters,232 new_validator).sign([node_key.private_key])233 # store election234 b.store_bulk_transactions([election])235 tx_vote0 = gen_vote(election, 0, ed25519_node_keys)236 tx_vote1 = gen_vote(election, 1, ed25519_node_keys)237 tx_vote2 = gen_vote(election, 2, ed25519_node_keys)238 b.store_bulk_transactions([tx_vote0, tx_vote1])239 update = Election.process_block(b, 9, [tx_vote2])240 assert len(update) == 1241 update_public_key = codecs.encode(update[0].pub_key.data, 'base64').decode().rstrip('\n')242 assert update_public_key == public_key64243 # assert that the public key is not a part of the current validator set244 for v in b.get_validators(10):245 assert not v['public_key']['value'] == public_key64246# ============================================================================247# Helper functions248# ============================================================================249def reset_validator_set(b, node_keys, height):250 validators = []251 for (node_pub, _) in node_keys.items():252 validators.append({'public_key': {'type': 'ed25519-base64',253 'value': node_pub},254 'voting_power': 10})...
test_util.py
Source:test_util.py
1import unittest2from buildpack import util3class M2EEMock:4 class ConfigMock:5 def __init__(self):6 self._conf = {}7 def __init__(self):8 self.config = self.ConfigMock()9class TestUtil(unittest.TestCase):10 def _test_upsert_m2ee_config_equals(11 self,12 value1,13 value2,14 expected_value,15 append=False,16 overwrite=False,17 section="SomeSection",18 key="SomeKey",19 ):20 m2ee = M2EEMock()21 if value1:22 m2ee.config._conf[section] = {key: value1}23 else:24 m2ee.config._conf[section] = {}25 util._upsert_m2ee_config_setting(26 m2ee,27 section,28 key,29 value2,30 append=append,31 overwrite=overwrite,32 )33 assert m2ee.config._conf[section][key] == expected_value34 def test_upsert_m2ee_config_section_insert(self):35 self._test_upsert_m2ee_config_equals(None, "value", "value")36 def test_upsert_m2ee_config_section_overwrite(self):37 self._test_upsert_m2ee_config_equals(38 "value1", "value2", "value2", overwrite=True39 )40 def test_upsert_m2ee_config_append_string(self):41 self._test_upsert_m2ee_config_equals(42 "value1", "value2", "value1value2", append=True43 )44 def test_upsert_m2ee_config_append_int(self):45 self._test_upsert_m2ee_config_equals(1, 2, 3, append=True)46 def test_upsert_m2ee_config_append_dict_without_overwrite(self):47 self._test_upsert_m2ee_config_equals(48 {"key1": "value1", "key2": "value2"},49 {"key2": "value2a", "key3": "value3"},50 {"key1": "value1", "key2": "value2", "key3": "value3"},51 append=True,52 )53 def test_upsert_m2ee_config_append_dict_with_overwrite(self):54 self._test_upsert_m2ee_config_equals(55 {"key1": "value1", "key2": "value2"},56 {"key2": "value2a", "key3": "value3"},57 {"key1": "value1", "key2": "value2a", "key3": "value3"},58 overwrite=True,59 append=True,60 )61 def test_upsert_m2ee_config_append_list(self):62 self._test_upsert_m2ee_config_equals(63 [1, 2, 3, 4],64 [5, 6, 7, 8, 1],65 [1, 2, 3, 4, 5, 6, 7, 8, 1],66 append=True,67 )68 def test_upsert_m2ee_config_append_set(self):69 self._test_upsert_m2ee_config_equals(70 {1, 2, 3, 4},71 {5, 6, 7, 8, 1},72 {1, 2, 3, 4, 5, 6, 7, 8},73 append=True,74 )75 def test_upsert_m2ee_config_overwrite_existing(self):76 m2ee = M2EEMock()77 m2ee.config._conf["SomeSection"] = {"SomeKey": "value1"}78 with self.assertRaises(ValueError):79 util._upsert_m2ee_config_setting(80 m2ee, "SomeSection", "SomeKey", "value2"81 )82 def test_upsert_m2ee_config_append_type_difference(self):83 m2ee = M2EEMock()84 m2ee.config._conf["SomeSection"] = {"SomeKey": "value1"}85 with self.assertRaises(ValueError):86 util._upsert_m2ee_config_setting(87 m2ee, "SomeSection", "SomeKey", 2, append=True88 )89 def test_upsert_javaopts_string(self):90 m2ee = M2EEMock()91 m2ee.config._conf["m2ee"] = {"javaopts": []}92 util.upsert_javaopts(m2ee, "-DSomeOption")93 assert util.get_javaopts(m2ee) == ["-DSomeOption"]94 def test_upsert_javaopts_list(self):95 m2ee = M2EEMock()96 m2ee.config._conf["m2ee"] = {"javaopts": ["-DSomeOption3"]}97 util.upsert_javaopts(m2ee, ["-DSomeOption1", "-DSomeOption2"])98 assert util.get_javaopts(m2ee) == [99 "-DSomeOption3",100 "-DSomeOption1",101 "-DSomeOption2",102 ]103 def test_upsert_custom_environment_vars(self):104 m2ee = M2EEMock()105 m2ee.config._conf["m2ee"] = {106 "custom_environment": {"SOME_VAR": "SOME_VALUE"}107 }108 util.upsert_custom_environment_variable(109 m2ee, "ANOTHER_VAR", "ANOTHER_VALUE"110 )111 util.upsert_custom_environment_variable(112 m2ee, "SOME_VAR", "ANOTHER_VALUE"113 )114 assert util.get_custom_environment_variables(m2ee) == {115 "SOME_VAR": "ANOTHER_VALUE",116 "ANOTHER_VAR": "ANOTHER_VALUE",117 }118 def test_upsert_logging_config_dict(self):119 m2ee = M2EEMock()120 m2ee.config._conf["logging"] = [{"type": "tcpjsonlines1"}]121 util.upsert_logging_config(m2ee, {"type": "tcpjsonlines2"})122 assert m2ee.config._conf["logging"] == [123 {"type": "tcpjsonlines1"},124 {"type": "tcpjsonlines2"},...
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 responses 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?
