KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use test1Fail method in green

Best Python code snippet using green

db_nbl.py

Source:db_nbl.py

1import collections, random, sys, traceback2from dbsim import *3class UncommittedClobber(AbortTransaction):4    '''WW conflict with uncommitted write'''5    pass6class InvisibleClobber(AbortTransaction):7    '''WW conflict with committed-but-invisible write'''8    pass9class WaitDepth(AbortTransaction):10    '''Blocking on WW would exceed wait depth limit'''11    pass12class Test1Fail(AbortTransaction):13    '''T1 rw T rw T2 and T1 does not precede T2's timestamp'''14    pass15    16class Test2Fail(AbortTransaction):17    '''T1 -- T rw T2 and T1 does not precede T2's timestamp'''18    pass19class DBWedged(Exception):20    '''A cycle of D-committed transactions has been detected.21    NOT a type of AbortTransaction; none of the victims can abort.22    '''23    pass24    25def make_db(tracker, stats, nrec, tid_watch, rid_watch,26            verbose, **extra_kwargs):27    '''A database model based on blacklist isolation (BI) and generalized28cycle detection29    '''30    class Transaction(object):31        def __init__(self, tid):32            self.tid = tid33            self.deps, self.war, self.depstar = set(),set(),None34            self.reads, self.clobbers = {}, {}35            self.dcommit = self.icommit = self.durable = 036            self.last_write,self.blocked = None,False37            self.stamp = None38    v_readers = collections.defaultdict(set)39    zerotx = Transaction(0)40    zerotx.dcommit = zerotx.icommit = zerotx.dcommit = 141    zerotx.depstar = frozenset()42    in_flight = {0:zerotx}43    44    q,e = not verbose, errlog45    def tx_read(pid, rid, for_update=False):46        Q = q and pid not in tid_watch and rid not in rid_watch47        t = in_flight[pid]48        Q or e('read: pid=%s rid=%s', pid, rid)49        # have I seen this record before?50        dep = t.reads.get(rid, None) or (rid in t.clobbers and pid)51        if dep:52            return dep53        def read_filter(it):54            for i,xmin,_ in it:55                x = in_flight[xmin]56                if x.icommit:57                    # definitely safe to use58                    '''NOTE: I-commit occurs the instant t.icommit becomes59                    True. Afterward, t can set v.t = zerotx for all v it60                    created, allowing t to be deleted.61                    '''62                    Q or e('\tUsing I-committed version %s of rid %s', x.tid, rid)63                    return i64                if x.dcommit:65                    # safe unless we're in the tid's WAR-set66                    if t not in x.depstar:67                        Q or e('\tUsing visible D-committed version %s of rid %s', x.tid, rid)68                        return i69                    q or e('\tSkipping invisible D-committed version: pid=%s', x.tid)70            else:71                assert not 'reachable'72        dep,_ = yield from tracker.on_access(pid, rid, read_filter)73        74        v_readers[rid,dep].add(t)75        q or e('\tNew v.r: %s', ' '.join(map(str, (d.tid for d in v_readers[rid,dep]))))76        clobber,_ = tracker.get_overwriter(rid, dep, False)77        if clobber is not None:78            X = Q and clobber not in tid_watch79            clobber = in_flight[clobber]80            clobber.war.add(t)81            X or e('\tNew WAR for %s via rid %d: %s', clobber.tid, rid,82                   ' '.join(map(str, (d.tid for d in clobber.war))))83        t.reads[rid] = dep84        yield from sys_busy(random.randint(ONE_TICK, 2*ONE_TICK),85                            color='green', title='%s=db[%s]' % (dep, rid))86        return dep87        88    def tx_write(pid, rid):89        Q = q and pid not in tid_watch and rid not in rid_watch90        t = in_flight[pid]91        Q or e('write: pid=%s rid=%d', pid, rid)92        93        # have I written this before?94        if rid in t.clobbers:95            q or e('\talready wrote to this record')96            return97        # do the write (don't depend on the version stamp)98        dep,_ = yield from tracker.on_access(pid, rid, False)99        x = in_flight[dep]100        assert x is not t # can't clobber a version we created101        if x.icommit:102            # definitely safe to clobber103            n = len(v_readers.get((rid,dep), ()))104            q or e('\tClobbering I-committed version with pid=%s and %d reader(s)', x.tid, n)105            pass106        elif x.dcommit:107            # only clobber versions we can actually see108            if t in x.depstar:109                q or e('\tAbort: cannot see latest version from pid=%s', x.tid)110                raise InvisibleClobber111                112            q or e('\tClobbering I-committed version: pid=%s', x.tid)113                114        # tie in my WAR and replace the current version115        v_readers[rid,dep].discard(t) # in case we previously read it116        t.reads.pop(rid, None)117        t.clobbers[rid] = dep118        yield from sys_busy(random.randint(ONE_TICK, 2*ONE_TICK),119                            color='blue', title='%s=db[%s]' % (dep, rid))120    tid_watched = set(tid_watch or [0])121    def tx_create(pid, is_readonly):122        if not tid_watched:123            yield from sys_exit(0)124            125        in_flight[pid] = t = Transaction(pid)126        t.begin = yield from sys_now()127        tracker.on_begin(pid)128        129    def finish(pid):130        t = in_flight[pid]131        then,now = t.begin, (yield from sys_now())132        histo_add(resp_histo, then-now)133    def i_commit(t):134        Q = q and t.tid not in tid_watch135        Q or e('\tI-commit %d', t.tid)136        assert not t.deps137        assert not t.war138        t.icommit = tracker.get_next_stamp()139        # clear dependencies140        commits = set()141        for rid,dep in t.clobbers.items():142            R = Q and rid not in rid_watch143            # I created this version; delete its predecessor144            R or e('\tFinalize version %d of rid=%d', t.tid, rid)145            x,_ = tracker.get_overwriter(rid, dep, False)146            x = in_flight.get(x, None)147            if x and x.deps:148                X = Q and x.tid not in tid_watch149                X or e('\tpid %d I-commit no longer blocked on %d', x.tid, t.tid)150                x.deps.discard(t)151                if not x.deps and not x.war and x.dcommit and not x.icommit:152                    assert not x.icommit153                    assert x not in commits154                    commits.add(x)155                X or e('\t=> remaining deps={%s} war={%s}', 156                       ' '.join(map(str, (d.tid for d in (x.deps or ())))),157                       ' '.join(map(str, (d.tid for d in x.war))))158            for x in v_readers[rid,t.tid]:159                X = Q and x.tid not in tid_watch160                # if x accessed a version I created, I ended up in161                # x.deps. If x then committed, I also ended up in162                # x.war; I may also be in x.war due to it163                # clobbering some version I read, but there's no164                # harm removing myself now (the read will be165                # removed soon enough).166                X or e('\tpid %d I-commit no longer blocked on %d', x.tid, t.tid)167                if x.deps:168                    x.deps.discard(t)169                    if not x.deps and not x.war and x.dcommit and not x.icommit:170                        assert not x.icommit171                        assert x not in commits172                        commits.add(x)173                X or e('\t=> remaining deps={%s} war={%s}', 174                       ' '.join(map(str, (d.tid for d in (x.deps or ())))),175                       ' '.join(map(str, (d.tid for d in x.war))))176        for rid,dep in t.reads.items():177            R = Q and rid not in rid_watch178            # remove myself from the version's read set179            Q or e('\tRemove %d from read set of rid=%d', t.tid, rid)180            v_readers[rid,dep].remove(t)181            x,_ = tracker.get_overwriter(rid,dep,False)182            x = in_flight.get(x, None)183            if x and x.war:184                X = Q and x.tid not in tid_watch185                # my read no longer prevents x from I-committing its clobber of v186                x.war.discard(t)187                X or e('\tpid %d I-commit no longer blocked on WAR %d', x.tid, t.tid)188                X or e('\t=> remaining WAR deps: %s', 189                       ' '.join(map(str, (d.tid for d in x.war))))190                if not x.war and not x.deps and x.dcommit and not x.icommit:191                    assert not x.icommit192                    #bad assertion: new versions could arrive after I D-commit193                    #assert v is db[rid].prev194                    assert x not in commits195                    commits.add(x)196                X or e('\trid=%s pid=%d dcommit=%s WAR={%s} deps={%s}',197                       rid, x.tid, x.dcommit,198                       ' '.join(map(str, (d.tid for d in x.war))),199                       ' '.join(map(str, (d.tid for d in x.deps))))200            201        for x in commits:202            i_commit(x)203        204        if t.durable:205            tracker.on_finalize(t.tid)206    def tx_commit(pid):207        Q = q and pid not in tid_watch208        t = in_flight[pid]209        yield from sys_busy(random.randint(ONE_TICK, 2*ONE_TICK), color='yellow')210        Q or e('Commit %s', pid)211        tid_watched.discard(pid)212        # /// BEGIN CRITICAL SECTION ///213        # construct WAR set (as of D-commit); we'll install it at all214        # clobbered versions after the commit succeeds.215        assert not t.war216        t3_list = set()217        for rid,dep in t.clobbers.items():218            x = in_flight[dep]219            if not x.icommit:220                # still not I-committed221                t.deps.add(x)222            t.war |= v_readers[rid,dep]223                224        for rid,dep in t.reads.items():225            assert t in v_readers[rid,dep]226            x = in_flight[dep]227            if not x.icommit:228                t.deps.add(x)229            x,_ = tracker.get_overwriter(rid,dep,False)230            x = in_flight.get(x, None)231            if x:232                t3_list.add(x)233            234        ds = set(t.war)235        236        # cycle test required?237        if t3_list:238            q or e('\tt3_list={%s}',239                   ' '.join('%s@%s/%s' % (x.tid,x.dcommit,x.stamp) for x in t.deps))240            t3_min = min(x.stamp for x in t3_list)241            if t.war:242                '''Test #1243                244                Abort T if T1 rw T rw T2 exists where T1 and T2 are245                already committed and T1 did not commit before T2's246                timestamp.247                '''248                t1_max = max(x.dcommit for x in t.war)249                if not (t1_max < t3_min):250                    raise Test1Fail251            if t.deps:252                '''Test #2253                254                Abort T if T1 -- T rw T2 exists where T1 and T2 are255                already committed and T1 did not commit before T2's256                timestamp257                '''258                q or e('\tApply test #2 to %d', pid)259                q or e('\tdeps={%s}',260                       ' '.join('%s@%s/%s' % (x.tid,x.dcommit,x.stamp) for x in t3_list))261                t1_max = max(x.dcommit for x in t.deps)262                if not (t1_max < t3_min):263                    raise Test2Fail264            assert t3_min265            t.stamp = t3_min266            267        for x in t.deps:268            X = Q and x.tid not in tid_watch269            X or e('\tpid %s has I-commit dep on %s (d-commit:%s i-commit:%s)',270                   pid, x.tid, x.dcommit, x.icommit)271            try:272                assert not x.icommit273            except AssertionError:274                errlog('pid %s has stale I-commit dep on %s', pid, x.tid)275                raise276            ds.add(x)277            ds.update(d for d in x.depstar if not d.icommit)278            279        if ds:280            Q or e('\tdepstar for %d at D-commit: %s', pid,281                   ' '.join(map(str, (d.tid for d in ds))))282        283        t.dcommit = tracker.on_precommit(pid) # writes now visible to non-dependents284        if not t.stamp:285            t.stamp = t.dcommit286            287        t.depstar = ds288        # /// END CRITICAL SECTION ///289            290        for x in t.war:291            assert not x.icommit292            assert x.tid in in_flight293        for x in t.deps:294            assert not x.icommit295            assert x.tid in in_flight296        if not t.war and not t.deps:297            i_commit(t)298            299        yield from tracker.on_finish(pid, True)300        yield from sys_sleep(random.randint(5*ONE_TICK, 10*ONE_TICK))301        yield from sys_busy(random.randint(ONE_TICK, 2*ONE_TICK), color='orange')302        303    def tx_abort(pid):304        Q = q and pid not in tid_watch305        t = in_flight[pid]306        Q or e('Abort %d', pid)307        commits = set()308        for rid,dep in t.clobbers.items():309            R = Q and rid not in rid_watch310            # I created this version, delete it311            R or e('\tRoll back update of rid=%d', rid)312            313        for rid,dep in t.reads.items():314            R = Q and rid not in rid_watch315            R or e('\tRemove %d from read set of rid=%d', t.tid, rid)316            v_readers[rid,dep].remove(t)317            x,_ = tracker.get_overwriter(rid,dep,False)318            x = in_flight.get(x,None)319            if x and x.war:320                X = Q and x.tid not in tid_watch321                x.war.discard(t)322                X or e('\tpid %d I-commit no longer blocked on WAR %d', x.tid, pid)323                X or e('\t=> remaining WAR deps: %s', 324                       ' '.join(map(str, (d.tid for d in x.war))))325                if not x.war and not x.deps and x.dcommit and not x.icommit:326                    assert not x.icommit327                    assert x not in commits328                    commits.add(x)329                else:330                    q or e('\trid=%s still has readers waiting to I-commit: %s',331                           rid, ' '.join(map(str, (d.tid for d in x.war))))332            elif x:333                q or e('\tskipping pid=%d with empty WAR', x.tid)334        t.dcommit = t.icommit = t.durable = yield from tracker.on_finish(pid, False)335        del in_flight[t.tid]336        for x in commits:337            i_commit(x)338                                  339        yield from sys_busy(random.randint(ONE_TICK, 2*ONE_TICK), color='red')340    def fini():341        live = [t for t in in_flight.values() if not t.icommit]342        if live:343            errlog('\nFound %d live transactions at exit (oldest from tick %.2f):',344                   len(live), min(t.begin for t in live)/float(ONE_TICK))345            if not q:346                for t in live:347                    errlog('\tpid=%s deps={%s}', t.tid, ' '.join(map(str, (x.tid for x in t.war))))348        print_general_stats(stats)349        print_failure_causes(stats)350    return NamedTuple(nrec=nrec, tx_begin=tx_create,351                      tx_read=tx_read, tx_write=tx_write,352                      tx_commit=tx_commit, tx_abort=tx_abort,353                      fini=fini,354                      begin_tracking=tracker.begin_tracking,355                      end_tracking=tracker.end_tracking)356        357                358def test_nbl_db():359    R,U,X = 1,2,3360    def test_fini(db):361        done = False362        def callback():363            nonlocal done364            done = True365        def nop():366            pid = yield from sys_getpid()367            yield from db.tx_begin(pid)368            yield from db.tx_abort(pid)369        yield from sys_sleep(1000*ONE_TICK)370        db.end_tracking(callback)371        yield from sys_spawn(nop())372        yield from sys_sleep(1000*ONE_TICK)373        yield from sys_spawn(nop())374        yield from sys_sleep(10000*ONE_TICK)375        assert done    376        db.fini()377        yield from sys_exit()378            379    def access(db, pid, rid, mode, delay):380        yield from sys_sleep(delay*ONE_TICK)381        yield from db.tx_write(pid, rid) if mode == X else db.tx_read(pid, rid, mode == U)382    def commit(db, pid, delay):383        if not isinstance(delay, int):384            errlog('bad delay: %s', delay)385            386        yield from sys_sleep(delay*ONE_TICK)387        yield from db.tx_commit(pid)388        389    def tx_one(db, rid, mode, delay1, delay2):390        def thunk():391            pid = yield from sys_getpid()392            try:393                yield from db.tx_begin(pid)394                yield from access(db, pid, rid, mode, delay1)395                yield from commit(db, pid, delay2)396            except AbortTransaction:397                yield from db.tx_abort(pid)398                399        return (yield from sys_spawn(thunk()))400        401    def tx_two(db, rid1, mode1, rid2=None, mode2=None, delay1=0, delay2=0, delay3=0):402        def thunk():403            pid = yield from sys_getpid()404            try:405                yield from db.tx_begin(pid)406                yield from access(db, pid, rid1, mode1, delay1)407                yield from access(db, pid, rid2 or rid1, mode2 or mode1, delay2)408                yield from commit(db, pid, delay3)409            except AbortTransaction:410                yield from db.tx_abort(pid)411                412        return (yield from sys_spawn(thunk()))413    def tx_n(db, commit_delay, *args):414        # accept (rid,mode,delay) triples415        def thunk():416            pid = yield from sys_getpid()417            try:418                yield from db.tx_begin(pid)419                for rid,mode,delay in args:420                    yield from access(db, pid, rid, mode, delay)421                yield from commit(db, pid, commit_delay)422            except AbortTransaction:423                yield from db.tx_abort(pid)424                425        return (yield from sys_spawn(thunk()))426        427    def test1(db):428        '''reads coexist peacefully'''429        # R-lock at t=0, S-locks at t=2 and t=2430        yield from tx_one(db, 1, R, 0, 10)431        yield from tx_one(db, 1, R, 1, 8)432        yield from tx_one(db, 1, R, 2, 6)433        yield from test_fini(db)434    def test2(db):435        '''incompatible requests block until the holder leaves'''436        yield from tx_one(db, 1, R, 0, 10)437        yield from tx_one(db, 1, X, 2, 0)438        yield from test_fini(db)439            440            441    def test3(db):442        '''reads can coexist with one upgrade lock, but a second upgrader443        blocks until the first leaves. Also make sure that the first444        upgrader can still upgrade when the lock mode is W.445        '''446        yield from tx_one(db, 1, R, 0, 10)447        yield from tx_two(db, rid1=1, mode1=U, mode2=X, delay1=2, delay2=8)448        yield from tx_two(db, rid1=1, mode1=U, mode2=X, delay1=3, delay2=6)449        yield from test_fini(db)450            451    def test4(db):452        '''new readers can coexist with an upgrade lock but not with an453        in-progress upgrade.454        '''455        yield from tx_one(db, 1, R, 0, 4)456        yield from tx_two(db, rid1=1, mode1=U, mode2=X, delay1=1, delay2=1)457        yield from tx_one(db, 1, R, 2, 4)458        yield from tx_one(db, 1, R, 4, 0)459        yield from test_fini(db)460    def test5(db):461        '''reader cannot upgrade if an upgrade lock has been granted'''462        yield from tx_two(db, rid1=1, mode1=R, mode2=X, delay2=5, delay3=1)463        yield from tx_one(db, 1, U, 1, 10)464        yield from test_fini(db)465    def test6(db):466        '''simple two-party deadlock detected'''467        yield from tx_two(db, rid1=1, mode1=R, rid2=2, mode2=X, delay2=5)468        yield from tx_two(db, rid1=2, mode1=R, rid2=1, mode2=X, delay1=1)469        yield from test_fini(db)470    def test7(db):471        '''three-party deadlock detected'''472        yield from tx_two(db, rid1=1, mode1=R, rid2=2, mode2=X, delay2=5)473        yield from tx_two(db, rid1=2, mode1=R, rid2=3, mode2=X, delay1=1)474        yield from tx_two(db, rid1=3, mode1=R, rid2=1, mode2=X, delay1=2)475        yield from test_fini(db)476    def test8(db):477        '''SI read-only anomaly detected:478	T1		T2		T3479			read A480			read B481					write B' (T2)482					D-commit (T2)483	read A484	read B' (T3)485	read C486	D-commit (T2 T3)487			write C' (T1)488			!! abort !!489					I-commit490	I-commit491        '''492        yield from tx_n(db, 0, (0, R, 200), (1, R, 0), (2, R, 0))493        yield from tx_n(db, 0, (0, R, 0), (1, R, 0), (2, X, 300))494        yield from tx_n(db, 0, (1, X, 100))495        yield from test_fini(db)496    A,B,C,D = 0,1,2,3497    def test9(db):498        '''Complex scenario that SI and 2PL both reject:499	T1		T2		T3		T4		T5500	read A501							write A' (T1)502							D-commit (T1)503					read B504    			write B' (T3)505			write C'506			D-commit (T3)507									write C'' (T2)508									D-commit (T2, T3)    509    	read C'' (T5)510					read C (**)511    					write D'512					DI-commit513			I-commit514									I-commit515	write D''516	DI-commit517							I-commit    518        519        '''520        yield from tx_n(db, 0, (A, R, 0), (C, R, 500), (D, X, 300))521        yield from tx_n(db, 0, (B, X, 300), (C, X, 0))522        yield from tx_n(db, 0, (B, R, 200), (C, R, 400), (D, X, 0))523        yield from tx_n(db, 0, (A, X, 100))524        yield from tx_n(db, 0, (C, X, 400))525        yield from test_fini(db)526    def test10(db):527        '''Schedule found during a measurement run. The Dangerous Structure528        has not yet formed when T2 commits first; when529        T3 commits second it sees T1 rw(A) T2 rw(B) T3:530        531	T1		T2		T3		532	                read B533	                		write B'534	                write A'535	read A          				536	                D-commit537	                                write C'538	                                D-commit539	read C'                         		540	D-commit	                                541        '''542        yield from tx_n(db, 100, (A, R, 0), (C, X, 200))543        yield from tx_n(db, 0,   (B, X, 100), (A, X, 400))544        yield from tx_n(db, 0,   (C, R, 300), (B, R, 300))545        yield from test_fini(db)546    def test11(db):547        '''Schedule found during a measurement run. The Dangerous Structure548        has not yet formed when T2 commits first; when549        T3 commits second it sees T1 rw(A) T2 rw(B) T3:550        551	T1		T2		T3552			read B553			write A'554					write C'555	read A556			commit557					write B'558	                                commit559	write C''560	commit561        '''562        yield from tx_n(db, 0, (A, R, 200), (C, X, 300))563        yield from tx_n(db, 300,   (B, R, 0), (A, X, 0))564        yield from tx_n(db, 0,   (C, X, 200), (B, X, 300))565        yield from test_fini(db)566    def test12(db):567        '''Schedule found during a measurement run. A dependency cycle forms568        without manifesting any Dangerous Structures:569        570	T1		T2		T3		T4571	                				read A572	write A'573	write B'574			read C575	D-commit576					write C'577	                                write D'578	                                D-commit579	                                		read D'580			read B'581	                				D-commit582			D-commit583        Adversarial commit ordering can make it even nastier. Here, T1584        is arguably the problem transaction, since its read of D' is585        impossible under SI, but it commits before T3 even reads586        C'. That means it's not enough for T1 to detect that it has587        gone awry; T3 must know to abort itself even though it has588        done nothing "wrong."589        590	T1		T2		T3		T4		591	read A                          				592	                write A'593	                write C'594	                D-commit595	                		read B596	                				write B'597	                                                write D'598	                                                D-commit599	read D'                                         		600	D-commit                        				601	                		read C'602					D-commit603        Note that there's still a shadow of the Dangerous Structure604        here. Every cycle must include (at least) two RW dependencies:605        one leaks information out of an uncommitted transaction and606        the other circumvents the commit-time dependency tracking that607        normally prevents the leakage from causing isolation608        failures. The RW deps just don't have to be adjacent any more609        after we give up SI.610        Thought: perhaps we can check D-commit times: it's a bad sign611        if I read a version that was clobbered before 1+ of the612        transactions in my dependency set. In the above example, T3613        depends on {T1 T2}, and gives a RW dep to T4; T4 D-committed614        before T1, and so T1 could potentially be poisoned by615        T4. There could be a lot of false positives, though: if T4 did616        not write D' at all (writing only B', for example), then there617        is no cycle. This looks suspiciously similar to wound-wait,618        which has a painfully high false positive rate.619        '''620        yield from tx_n(db, 300, (A, X, 0), (B, X, 0))621        yield from tx_n(db, 200, (C, R, 100), (B, R, 600))622        yield from tx_n(db, 0,   (C, X, 500), (D, X, 0))623        yield from tx_n(db, 200, (A, R, 0), (D, R, 600))624        yield from test_fini(db)625    def test13(db):626        '''One of the simplest possible serialization anomalies627        T1		T2628        read A629        		write A'630        		write B'631        		D-commit632        read B633        D-commit634        '''635        yield from tx_n(db, 0, (A, R, 0), (B, R, 300))636        yield from tx_n(db, 0, (A, X, 100), (B, X, 0))637        yield from test_fini(db)638    def test14(db):639        '''Scenario suggested by Alan Fekete:640	T1		T2		T3641					Write A'642			Read B643					Write B'644	Write C'645					Commit646	Read A'647	Commit648			Write C''649			Commit650        T2 should abort.651        '''652        yield from tx_n(db, 0, (C, X, 400), (A, R, 200))653        yield from tx_n(db, 0, (B, R, 100), (C, X, 600))654        yield from tx_n(db, 200, (A, X, 0), (B, X, 200))655        yield from test_fini(db)656    def test15(db):657        '''This scenario found during simulation run:658        T1		T2		T3		T4659        write A'660        write B'661        		read B662        D-commit663        				read C664        		write C'665        		D-commit666        						read D667        				write D'668        						read A'669        				D-commit670        						D-commit671        Produces the cycle T1 -- T4 rw T3 rw T2 rw T1 672        '''673        yield from tx_n(db, 200, (A, X, 0), (B, X, 0))674        yield from tx_n(db, 0, (B, R, 200), (C, X, 300))675        yield from tx_n(db, 200, (C, R, 400), (D, X, 400))676        yield from tx_n(db, 200, (D, R, 700), (A, R, 200))677        yield from test_fini(db)678    failures = 0679    for test in (test1,test2,test3,test4,test5,test6,test7,test8,test9,680                 test10,test11,test12,test13,test14,test15):681        errlog('\n%s\nRunning test:\n\n\t%s\n', '='*72, test.__doc__)682        stats = NamedTuple()683        tracker = dependency_tracker(stats)684        db = make_nbl_db(stats=stats, tracker=tracker,nrec=10, ww_blocks=True, verbose=True)685        db.begin_tracking()686        try:687            simulator(test(db), log=log_svg_task())688        except:689            traceback.print_exc()690            failures +=1691        692    if failures:693        exit(1)694    695    696if __name__ == '__main__':697    try:698        seed = sys.argv[1]699    except:700        seed = make_seed()701    errlog('Using RNG seed: %s', seed)702    random.seed(seed)703    test_nbl_db()...

test_abs_project.py

Source:test_abs_project.py

1import unittest2import time3from selenium import webdriver4class TestAbs(unittest.TestCase):5    def test_registration1(self):6        link = "http://suninjuly.github.io/registration1.html"7        #link = "http://suninjuly.github.io/registration2.html"8        browser = webdriver.Chrome('/Users/d.lang/chromedriver')9        browser.get(link)10        # ÐÐ°Ñ ÐºÐ¾Ð´, ÐºÐ¾ÑÐ¾ÑÑÐ¹ Ð·Ð°Ð¿Ð¾Ð»Ð½ÑÐµÑ Ð¾Ð±ÑÐ·Ð°ÑÐµÐ»ÑÐ½ÑÐµ Ð¿Ð¾Ð»Ñ11        input1 = browser.find_element_by_tag_name("input[placeholder='Input your first name']")12        input1.send_keys("Ivan")13        input2 = browser.find_element_by_tag_name("input[placeholder='Input your last name']")14        input2.send_keys("Petrov")15        input3 = browser.find_element_by_tag_name("input[placeholder='Input your email']")16        input3.send_keys("@mail")17        input4 = browser.find_element_by_tag_name("input[placeholder='Input your phone:']")18        input4.send_keys("899999")19        input5 = browser.find_element_by_tag_name("input[placeholder='Input your address:']")20        input5.send_keys("Msc")21        # ÐÑÐ¿ÑÐ°Ð²Ð»ÑÐµÐ¼ Ð·Ð°Ð¿Ð¾Ð»Ð½ÐµÐ½Ð½ÑÑ ÑÐ¾ÑÐ¼Ñ22        button = browser.find_element_by_css_selector("button.btn")23        button.click()24        # ÐÑÐ¾Ð²ÐµÑÑÐµÐ¼, ÑÑÐ¾ ÑÐ¼Ð¾Ð³Ð»Ð¸ Ð·Ð°ÑÐµÐ³Ð¸ÑÑÑÐ¸ÑÐ¾Ð²Ð°ÑÑÑÑ25        # Ð¶Ð´ÐµÐ¼ Ð·Ð°Ð³ÑÑÐ·ÐºÐ¸ ÑÑÑÐ°Ð½Ð¸ÑÑ26        time.sleep(1)27        # Ð½Ð°ÑÐ¾Ð´Ð¸Ð¼ ÑÐ»ÐµÐ¼ÐµÐ½Ñ, ÑÐ¾Ð´ÐµÑÐ¶Ð°ÑÐ¸Ð¹ ÑÐµÐºÑÑ28        welcome_text_elt = browser.find_element_by_tag_name("h1")29        # Ð·Ð°Ð¿Ð¸ÑÑÐ²Ð°ÐµÐ¼ Ð² Ð¿ÐµÑÐµÐ¼ÐµÐ½Ð½ÑÑ welcome_text ÑÐµÐºÑÑ Ð¸Ð· ÑÐ»ÐµÐ¼ÐµÐ½ÑÐ° welcome_text_elt30        welcome_text = welcome_text_elt.text31        # Ñ Ð¿Ð¾Ð¼Ð¾ÑÑÑ assert Ð¿ÑÐ¾Ð²ÐµÑÑÐµÐ¼, ÑÑÐ¾ Ð¾Ð¶Ð¸Ð´Ð°ÐµÐ¼ÑÐ¹ ÑÐµÐºÑÑ ÑÐ¾Ð²Ð¿Ð°Ð´Ð°ÐµÑ Ñ ÑÐµÐºÑÑÐ¾Ð¼ Ð½Ð° ÑÑÑÐ°Ð½Ð¸ÑÐµ ÑÐ°Ð¹ÑÐ°32        self.assertEqual("Congratulations! You have successfully registered!", welcome_text, "Test1fail")33        #assert "Congratulations! You have successfully registered!" == welcome_text34    def test_registration2(self):35        #link = "http://suninjuly.github.io/registration1.html"36        link = "http://suninjuly.github.io/registration2.html"37        browser = webdriver.Chrome('/Users/d.lang/chromedriver')38        browser.get(link)39        # ÐÐ°Ñ ÐºÐ¾Ð´, ÐºÐ¾ÑÐ¾ÑÑÐ¹ Ð·Ð°Ð¿Ð¾Ð»Ð½ÑÐµÑ Ð¾Ð±ÑÐ·Ð°ÑÐµÐ»ÑÐ½ÑÐµ Ð¿Ð¾Ð»Ñ40        input1 = browser.find_element_by_tag_name("input[placeholder='Input your first name']")41        input1.send_keys("Ivan")42        input2 = browser.find_element_by_tag_name("input[placeholder='Input your last name']")43        input2.send_keys("Petrov")44        input3 = browser.find_element_by_tag_name("input[placeholder='Input your email']")45        input3.send_keys("@mail")46        input4 = browser.find_element_by_tag_name("input[placeholder='Input your phone:']")47        input4.send_keys("899999")48        input5 = browser.find_element_by_tag_name("input[placeholder='Input your address:']")49        input5.send_keys("Msc")50        # ÐÑÐ¿ÑÐ°Ð²Ð»ÑÐµÐ¼ Ð·Ð°Ð¿Ð¾Ð»Ð½ÐµÐ½Ð½ÑÑ ÑÐ¾ÑÐ¼Ñ51        button = browser.find_element_by_css_selector("button.btn")52        button.click()53        # ÐÑÐ¾Ð²ÐµÑÑÐµÐ¼, ÑÑÐ¾ ÑÐ¼Ð¾Ð³Ð»Ð¸ Ð·Ð°ÑÐµÐ³Ð¸ÑÑÑÐ¸ÑÐ¾Ð²Ð°ÑÑÑÑ54        # Ð¶Ð´ÐµÐ¼ Ð·Ð°Ð³ÑÑÐ·ÐºÐ¸ ÑÑÑÐ°Ð½Ð¸ÑÑ55        time.sleep(1)56        # Ð½Ð°ÑÐ¾Ð´Ð¸Ð¼ ÑÐ»ÐµÐ¼ÐµÐ½Ñ, ÑÐ¾Ð´ÐµÑÐ¶Ð°ÑÐ¸Ð¹ ÑÐµÐºÑÑ57        welcome_text_elt = browser.find_element_by_tag_name("h1")58        # Ð·Ð°Ð¿Ð¸ÑÑÐ²Ð°ÐµÐ¼ Ð² Ð¿ÐµÑÐµÐ¼ÐµÐ½Ð½ÑÑ welcome_text ÑÐµÐºÑÑ Ð¸Ð· ÑÐ»ÐµÐ¼ÐµÐ½ÑÐ° welcome_text_elt59        welcome_text = welcome_text_elt.text60        # Ñ Ð¿Ð¾Ð¼Ð¾ÑÑÑ assert Ð¿ÑÐ¾Ð²ÐµÑÑÐµÐ¼, ÑÑÐ¾ Ð¾Ð¶Ð¸Ð´Ð°ÐµÐ¼ÑÐ¹ ÑÐµÐºÑÑ ÑÐ¾Ð²Ð¿Ð°Ð´Ð°ÐµÑ Ñ ÑÐµÐºÑÑÐ¾Ð¼ Ð½Ð° ÑÑÑÐ°Ð½Ð¸ÑÐµ ÑÐ°Ð¹ÑÐ°61        self.assertEqual("Congratulations! You have successfully registered!", welcome_text, "Test2fail")62if __name__ == "__main__":...

examples.py

Source:examples.py

...7        """8        This test will print output to stdout, and then pass.9        """10        print("Sunshine and daisies")11    def test1Fail(self):12        """13        This test will print output to stderr, and then fail an assertion.14        """15        sys.stderr.write("Doom and gloom.\n")16        self.assertTrue(False)17    def test2Error(self):18        """19        An Exception will be raised (and not caught) while running this test.20        """21        raise Exception22    @unittest.skip("This is the 'reason' portion of the skipped test.")23    def test3Skip(self):24        """25        This test will be skipped....

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.