KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use assert_truthy method in Testify
Best Python code snippet using Testify_python
test_checkpoint.py
Source:test_checkpoint.py 
...41 with mock.patch(patch_string, new=lambda: self.pool):42 chkpt = TritonCheckpointerTest(43 stream_name, client_name=CLIENT_NAME)44 last = chkpt.last_sequence_number(shard_id)45 assert_truthy(last is None)46 seq_no = '1234'47 chkpt.checkpoint(shard_id, seq_no)48 last = chkpt.last_sequence_number(shard_id)49 assert_truthy(last == seq_no)50 seq_no = '456'51 chkpt.checkpoint(shard_id, seq_no)52 last = chkpt.last_sequence_number(shard_id)53 assert_truthy(last == seq_no)54 def test_new_unicode_checkpoint(self):55 stream_name = u'test1'56 shard_id = u'shardId-000000000000'57 patch_string = u'triton.checkpoint.get_triton_connection_pool'58 with mock.patch(patch_string, new=lambda: self.pool):59 chkpt = TritonCheckpointerTest(60 stream_name, client_name=ascii_to_unicode_str(CLIENT_NAME))61 last = chkpt.last_sequence_number(shard_id)62 assert_truthy(last is None)63 seq_no = u'1234'64 chkpt.checkpoint(shard_id, seq_no)65 last = chkpt.last_sequence_number(shard_id)66 assert_truthy(last == seq_no)67 seq_no = u'456'68 chkpt.checkpoint(shard_id, seq_no)69 last = chkpt.last_sequence_number(shard_id)70 assert_truthy(last == seq_no)71def generate_raw_record(seq_no):72 data = base64.b64encode(msgpack.packb({'value': True}))73 raw_record = {'SequenceNumber': seq_no, 'Data': data}74 return raw_record75def generate_unicode_raw_record(seq_no):76 data = base64.b64encode(msgpack.packb({u'value': True, u'test_üñîçøâé_ké¥_å®å®': u'test_üñîçøâé_âÃ¥l_å®å®'}))77 raw_record = {u'SequenceNumber': seq_no, u'Data': data}78 return raw_record79def get_records(n, offset=0, *args, **kwargs):80 n = int(n)81 offset = int(offset)82 records = [generate_raw_record(str(i + offset)) for i in range(n)]83 return {84 'NextShardIterator': n + offset,85 'MillisBehindLatest': 0,86 'Records': records87 }88def get_unicode_records(n, offset=0, *args, **kwargs):89 n = int(n)90 offset = int(offset)91 records = [generate_unicode_raw_record(str(i + offset)) for i in range(n)]92 return {93 u'NextShardIterator': n + offset,94 u'MillisBehindLatest': 0,95 u'Records': records96 }97def get_batches_of_10_records(98 iter_val, *args, **kwargs99):100 recs = get_records(10, iter_val)101 return recs102def get_batches_of_10_unicode_records(103 iter_val, *args, **kwargs104):105 recs = get_unicode_records(10, iter_val)106 return recs107class StreamIteratorCheckpointTest(TestCase):108 """Test end to end checkpoint functionality"""109 @setup110 def setup_db(self):111 self.tempfile = tempfile.NamedTemporaryFile()112 self.tempfile_name = self.tempfile.name113 self.conn = sqlite3.connect(self.tempfile_name)114 self.pool = SqlitePool(self.conn)115 checkpoint.init_db(lambda: self.pool)116 def test_iterator_checkpoint(self):117 stream_name = 'test1'118 shard_id = 'shardId-000000000000'119 pool_patch = 'triton.checkpoint.get_triton_connection_pool'120 chkpt_patch = 'triton.stream.TritonCheckpointer'121 with mock.patch(pool_patch, new=lambda: self.pool):122 with mock.patch(chkpt_patch, new=TritonCheckpointerTest):123 s = turtle.Turtle()124 s.name = stream_name125 def get_20_records(*args, **kwargs):126 return get_records(20)127 s.conn.get_records = get_20_records128 i = stream.StreamIterator(s, shard_id, stream.ITER_TYPE_LATEST)129 i._iter_value = 1130 for j in range(10):131 val = i.next()132 i.checkpoint()133 last_chkpt_seqno = i.checkpointer.last_sequence_number(134 shard_id)135 assert_truthy(val.seq_num == last_chkpt_seqno)136 def test_unicode_iterator_checkpoint(self):137 stream_name = u'test1_üñîçødé'138 shard_id = u'shardId-üñîçødé-000000000000'139 pool_patch = u'triton.checkpoint.get_triton_connection_pool'140 chkpt_patch = u'triton.stream.TritonCheckpointer'141 with mock.patch(pool_patch, new=lambda: self.pool):142 with mock.patch(chkpt_patch, new=TritonCheckpointerTest):143 s = turtle.Turtle()144 s.name = stream_name145 def get_20_records(*args, **kwargs):146 return get_unicode_records(20)147 s.conn.get_records = get_20_records148 i = stream.StreamIterator(s, shard_id, stream.ITER_TYPE_LATEST)149 i._iter_value = 1150 for j in range(10):151 val = i.next()152 i.checkpoint()153 last_chkpt_seqno = i.checkpointer.last_sequence_number(154 shard_id)155 assert_truthy(val.seq_num == last_chkpt_seqno)156 def test_combined_iterator_checkpoint(self):157 stream_name = 'test2'158 shard_id0 = 'shardId-000000000000'159 shard_id1 = 'shardId-000000000001'160 pool_patch = 'triton.checkpoint.get_triton_connection_pool'161 chkpt_patch = 'triton.stream.TritonCheckpointer'162 with mock.patch(pool_patch, new=lambda: self.pool):163 with mock.patch(chkpt_patch, new=TritonCheckpointerTest):164 s = turtle.Turtle()165 s.name = stream_name166 s.conn.get_records = get_batches_of_10_records167 i1 = stream.StreamIterator(168 s, shard_id0, stream.ITER_TYPE_LATEST)169 i1._iter_value = 0170 i2 = stream.StreamIterator(171 s, shard_id1, stream.ITER_TYPE_LATEST)172 i2._iter_value = 100173 combined_i = stream.CombinedStreamIterator([i1, i2])174 for j in range(25):175 val = combined_i.next()176 combined_i.checkpoint()177 last_chkpts = [178 this_i.checkpointer.last_sequence_number(this_i.shard_id)179 for this_i in [i1, i2]180 ]181 assert_truthy(val.seq_num in last_chkpts)182 # because combined iterator uses a set of iterators,183 # we neet to test both cases184 if int(val.seq_num) > 100:185 assert_truthy('9' in last_chkpts)186 else:187 assert_truthy('109' in last_chkpts)188 def test_stream_get_iterator_from_checkpoint(self):189 stream_name = 'test3'190 shard_id = 'shardId-000000000000'191 pool_patch = 'triton.checkpoint.get_triton_connection_pool'192 chkpt_patch = 'triton.stream.TritonCheckpointer'193 with mock.patch(pool_patch, new=lambda: self.pool):194 with mock.patch(chkpt_patch, new=TritonCheckpointerTest):195 c = turtle.Turtle()196 s = stream.Stream(c, stream_name, 'p_key')197 s._shard_ids = [shard_id, ]198 s.conn.get_records = get_batches_of_10_records199 def get_shard_iterator(*args, **kwargs):200 return {'ShardIterator': 0}201 s.conn.get_shard_iterator = get_shard_iterator202 ci = s.build_iterator_from_checkpoint()203 for j in range(10):204 val = ci.next()205 i = ci.iterators[0]206 assert_truthy(i.iterator_type == i.fallback_iterator_type)207 ci.checkpoint()208 ci = s.build_iterator_from_checkpoint()209 i = ci.iterators[0]210 i._iter_value = i.checkpointer.last_sequence_number(211 shard_id)212 assert_truthy(val.seq_num == i._iter_value)213 for j in range(10):214 val = ci.next()215 #NOTE: iter_value function is not being called since216 #we explicitly set the iterator above217 #https://github.com/postmates/triton-python/commit/7441b5d46d639e7ee2783c43b6a0bfd4adb6c0d6#diff-c1624f14b4890c5253e57cdf3fee2366R92218 #TODO: rewrite test to properly instantiate iterator219 #assert_truthy(...
test_verify.py
Source:test_verify.py
1# -*- coding: utf-8 -*-2import pytest3import pydash4import verify as v5from verify import expect, ensure6from .fixtures import (7 METHOD_CALL_CASES,8 METHOD_RAISE_CASES,9 METHOD_ALIAS_CASES,10 raises_assertion,11 assert_truthy,12 make_parametrize_id13)14@pytest.mark.parametrize('value,assertions', [15 (5, (v.Greater(4), v.Less(6))),16])17def test_expect_multiple_assertions(value, assertions):18 """Test that Expect handles multiple assertions."""19 assert expect(value, *assertions)20@pytest.mark.parametrize('value,predicates', [21 (True, (pydash.is_boolean, pydash.identity)),22])23def test_expect_predicates(value, predicates):24 """Test that Expect handles multiple predicates that returns boolean25 values.26 """27 assert expect(value, *predicates)28@pytest.mark.parametrize('value,predicates', [29 (True, (pydash.is_boolean, pydash.is_number)),30 (True, (pydash.is_integer, pydash.identity)),31])32def test_expect_predicates_raises(value, predicates):33 """Test that Expect handles multiple predicates that returns boolean34 values.35 """36 with raises_assertion():37 assert expect(value, *predicates)38def test_expect_predicates_return_none():39 assert expect(True, assert_truthy)40def test_expect_chaining():41 assert expect(True).Boolean()(assert_truthy)42 assert expect(True, v.Boolean(), assert_truthy).Truthy()43def test_expect_chain_method_proxy():44 for method in [method for method in dir(v)45 if method[0].isupper() or method[:2] in ('to', 'is')]:46 assert getattr(v, method) is getattr(expect(None), method).assertion47@pytest.mark.parametrize('method', [48 'nosuchmethod',49 'expect'50])51def test_expect_chain_invalid_method(method):52 with pytest.raises(AttributeError):53 getattr(expect(None), method)54@pytest.mark.parametrize('meth,value,arg',55 METHOD_CALL_CASES,56 ids=make_parametrize_id)57def test_assert_method(meth, value, arg):58 """Test that method passes when evaluated for comparables."""59 assert expect(value, meth(*arg.args, **arg.kargs))60 assert meth(value, *arg.args, **arg.kargs)61@pytest.mark.parametrize('meth,value,arg',62 METHOD_RAISE_CASES,63 ids=make_parametrize_id)64def test_assert_raises(meth, value, arg):65 """Test that method raises an assertion error when evaluated for66 comparables.67 """68 with raises_assertion() as exc:69 expect(value, meth(*arg.args, **arg.kargs))70 with raises_assertion() as exc:71 meth(value, *arg.args, **arg.kargs)72 with raises_assertion() as exc:73 opts = arg.kargs.copy()74 opts.update({'msg': 'TEST CUSTOM MESSAGE'})75 meth(value, *arg.args, **opts)76 assert opts['msg'] in str(exc.value)77@pytest.mark.parametrize('obj,alias',78 METHOD_ALIAS_CASES,79 ids=make_parametrize_id)80def test_aliases(obj, alias):...
settings.py
Source:settings.py
...17sender_email = os.getenv('MULLE_EMAIL', default="mulle@meck.net")18db_uri = os.getenv('DB_URI', default=None)19clone_dir = os.getenv("MULLE_CLONE_DIR", default="/tmp/mullemeck")20db_uri = db_uri if db_uri else 'sqlite:///dev.db'21def assert_truthy(value, error_message):22 """23 Identical to the regular `assert` statement, but allows for splitting24 across multiple lines because of the parentheses.25 """26 assert value, error_message27def validate_environment_variables():28 """29 Throws an `AssertionError` if any of the required environment variables30 are not correctly provided.31 """32 assert os.access(os.path.dirname(clone_dir), os.W_OK), \33 "Application does not have access to that path"34 assert_truthy(35 github_secret, 'GITHUB_SECRET environment variable required.')36 assert_truthy(...
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 Testify 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?
