KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use set_db_version method in autotest
Best Python code snippet using autotest_python
pycs_db.py
Source:pycs_db.py 
...94 return self._execute(sql, args).fetchone()95 def get_db_version(self):96 self.db_id, = self.fetchone("SELECT db_id FROM pycs_meta")97 print "Database is at version %d" % self.db_id98 def set_db_version(self, newver):99 self.execute("UPDATE pycs_meta SET db_id=%d", (newver,))100 self.execute("COMMIT")101 self.get_db_version()102 def update_schema(self):103 storFn = pycs_paths.DATADIR + "/settings.dat"104 print "Connecting to MetaKit database in %s" % storFn105 import metakit106 mkdb = metakit.storage(storFn, 1)107 108 print "Updating PostgreSQL database schema"109 # See if the database is completely new110 try:111 self.fetchone("SELECT db_id FROM pycs_meta")112 except DBE, e:113 if e.args[0].find("does not exist") != -1:114 print "Creating pycs_meta table"115 self.execute("""CREATE TABLE pycs_meta (db_id INT)""")116 self.execute("INSERT INTO pycs_meta (db_id) VALUES (0)")117 else:118 raise119 self.get_db_version()120 # if version 0, need to create the referrers table121 if self.db_id < 1:122 # grab MK table123 referrers_table = mkdb.getas("referrers[time:S,usernum:S,group:S,referrer:S,count:I]").ordered(2)124 # set up PG125 print "Creating pycs_referrers table"126 self.execute("""CREATE TABLE pycs_referrers (id INT PRIMARY KEY, hit_time TIMESTAMP, usernum INT, usergroup VARCHAR(32), referrer VARCHAR(2048), hit_count INT, is_search_engine BOOLEAN, search_engine VARCHAR(32), search_term VARCHAR(1024))""")127 self.execute("""CREATE SEQUENCE pycs_referrers_id_seq""")128 self.execute("""CREATE INDEX pycs_referrers_user_index ON pycs_referrers (usernum, usergroup)""")129 # copy data over130 print "Copying referrer data (%d rows) into pycs_referrers table" % len(referrers_table)131 count = 0132 for row in referrers_table:133 if not (count % 1000):134 print "\r%s" % count,135 try:136 matched, term = search_engines.checkUrlForSearchEngine(row.referrer)137 if term: term = toutf8(term)138 self.execute("INSERT INTO pycs_referrers (id, hit_time, usernum, usergroup, referrer, hit_count, is_search_engine, search_engine, search_term) VALUES (NEXTVAL('pycs_referrers_id_seq'), %s, %d, %s, %s, %d, %s, %s, %s)", (pyto8601(row.time), int(row.usernum), row.group, toutf8(row.referrer), row.count, (matched and term) and 't' or 'f', matched, term))139 except DBE, e:140 print e141 print (row.time, row.usernum, row.group, row.referrer, row.count)142 raise143 count += 1144 print145 self.set_db_version(1)146 if self.db_id < 2:147 comments_table = mkdb.getas('comments[user:S,paragraph:S,link:S,notes[name:S,email:S,url:S,comment:S,date:S]]').ordered( 2 )148 149 print "Creating pycs_comments table"150 #self.execute("""DROP TABLE pycs_comments""")151 #self.execute("""DROP SEQUENCE pycs_comments_id_seq""")152 self.execute("""CREATE TABLE pycs_comments (id INT PRIMARY KEY, usernum INT, postid VARCHAR(255), postlink VARCHAR(2048), commentdate TIMESTAMP, postername VARCHAR(255), posteremail VARCHAR(255), posterurl VARCHAR(2048), commenttext TEXT)""")153 self.execute("""CREATE SEQUENCE pycs_comments_id_seq""")154 self.execute("""CREATE INDEX pycs_comments_post_index ON pycs_comments (usernum, postid, id)""")155 print "Copying comments (for %d users) into pycs_comments table" % len(comments_table)156 for user_row in comments_table:157 usernum, postid, postlink = (user_row.user, user_row.paragraph, user_row.link)158 if not usernum:159 usernum = 0160 else:161 usernum = int(usernum)162 print "user %s post %s postlink %s: %d comments" % (`usernum`, `postid`, `postlink`, len(user_row.notes))163 for cmt_row in user_row.notes:164 name, email, url, comment, date = (cmt_row.name, cmt_row.email, cmt_row.url, cmt_row.comment, cmt_row.date)165 #print "name %s email %s url %s cmt %s... date %s" % (`name`, `email`, `url`, `comment[:20]`, date)166 if date:167 date = pyto8601(date, has_am=0)168 else:169 date = None170 try:171 self.execute("""INSERT INTO pycs_comments (id, usernum, postid, postlink, commentdate, postername, posteremail, posterurl, commenttext) VALUES (NEXTVAL('pycs_comments_id_seq'), %d, %s, %s, %s, %s, %s, %s, %s)""", (usernum, postid, postlink, date, toutf8(name), toutf8(email), toutf8(url), toutf8(comment)))172 except:173 print (usernum, postid, postlink, date, name, email, url, comment)174 raise175 self.set_db_version(2)176 if self.db_id < 3:177 self.execute("""ALTER TABLE pycs_comments ADD COLUMN is_spam INT""")178 self.execute("""UPDATE pycs_comments SET is_spam=0""")179 self.set_db_version(3)180 if self.db_id < 4:181 self.execute("""DROP INDEX pycs_comments_post_index""")182 self.execute("""CREATE INDEX pycs_comments_post_index ON pycs_comments (is_spam, usernum, postid, id)""")183 self.set_db_version(4)184 if self.db_id < 5:185 # repair damage by comment bug that inserted NULL instead of 0 for is_spam on comment posting186 self.execute("""UPDATE pycs_comments SET is_spam=0 WHERE is_spam IS NULL""")187 self.set_db_version(5)188 if self.db_id < 6:189 print "Creating pycs_updates table for weblog updates"190 self.execute("""CREATE TABLE pycs_updates (update_time TIMESTAMP, url VARCHAR(1024), title VARCHAR(1024))""")191 self.execute("""CREATE INDEX pycs_updates_by_time ON pycs_updates (update_time)""")192 self.execute("""CREATE INDEX pycs_updates_url_title ON pycs_updates (url, title)""")193 # fill it with data194 print "Copying weblog update data over to pycs_updates table"195 updates_table = mkdb.getas('blogUpdates[updateTime:I,blogUrl:S,blogName:S]')196 for row in updates_table:197 self.execute("INSERT INTO pycs_updates (update_time, url, title) VALUES (%s, %s, %s)",198 (timeto8601(time.localtime(row.updateTime)),199 row.blogUrl,200 row.blogName,201 ))202 self.set_db_version(6)203 if self.db_id < 7:204 print "Creating pycs_spam_commenters table"205 self.execute("""CREATE TABLE pycs_spam_commenters (name VARCHAR(1024))""")206 self.execute("""CREATE INDEX pycs_spam_commenters_name ON pycs_spam_commenters (name)""")207 self.set_db_version(7)208 if self.db_id < 8:209 self.execute("""CREATE TABLE pycs_good_commenters (name VARCHAR(1024))""")210 self.execute("""CREATE INDEX pycs_good_commenters_name ON pycs_good_commenters (name)""")211 self.set_db_version(8)212 if self.db_id < 9:213 print "Indexing poster names in comments table"214 self.execute("""CREATE INDEX pycs_comments_postername ON pycs_comments (postername)""")215 self.set_db_version(9)216 if self.db_id < 10:217 print "Moving meta table into postgres"218 meta_table = mkdb.getas("meta[nextUsernum:I]")219 if len(meta_table) == 0:220 nextUsernum = 1221 else:222 nextUsernum = meta_table[0].nextUsernum223 224 self.execute("""CREATE SEQUENCE pycs_users_id_seq START %d""",225 (nextUsernum,))226 self.set_db_version(10)227 if self.db_id < 11:228 print "Creating pycs_users table"229 users_table = mkdb.getas(230 "users[usernum:S,email:S,password:S,name:S,weblogTitle:S,serialNumber:S,organization:S," +231 "flBehindFirewall:I,hitstoday:I,hitsyesterday:I,hitsalltime:I," +232 "membersince:S,lastping:S,pings:I,lastupstream:S,upstreams:I,lastdelete:S,deletes:I,bytesupstreamed:I," +233 "signons:I,signedon:I,lastsignon:S,lastsignoff:S,clientPort:I,disabled:I,alias:S,flManila:I,bytesused:I,stylesheet:S,commentsdisabled:I]"234 ).ordered()235 236 if self.fetchone("SELECT COLUMN_NAME FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_NAME='pycs_users'"):237 self.execute("DROP TABLE pycs_users")238 self.execute("""CREATE TABLE pycs_users (usernum INT NOT NULL PRIMARY KEY, email VARCHAR(1024) NOT NULL, password CHAR(32) NOT NULL, name VARCHAR(1024) NOT NULL, weblogTitle VARCHAR(1024) NOT NULL DEFAULT '', serialNumber VARCHAR(128) NOT NULL DEFAULT '', organization VARCHAR(1024) NOT NULL DEFAULT '', flBehindFirewall INT NOT NULL DEFAULT 0, hitstoday INT NOT NULL DEFAULT 0, hitsyesterday INT NOT NULL DEFAULT 0, hitsalltime INT NOT NULL DEFAULT 0, membersince TIMESTAMP, lastping TIMESTAMP, pings INT NOT NULL DEFAULT 0, lastupstream TIMESTAMP, upstreams INT NOT NULL DEFAULT 0, lastdelete TIMESTAMP, deletes INT NOT NULL DEFAULT 0, bytesupstreamed INT NOT NULL DEFAULT 0, signons INT NOT NULL DEFAULT 0, signedon INT NOT NULL DEFAULT 0, lastsignon TIMESTAMP, lastsignoff TIMESTAMP, clientPort INT NOT NULL DEFAULT 0, disabled INT NOT NULL DEFAULT 0, alias VARCHAR(256), flManila INT NOT NULL DEFAULT 0, bytesused INT NOT NULL DEFAULT 0, stylesheet VARCHAR(2048), commentsdisabled INT)""")239 for user in users_table:240 try:241 usernum = int(user.usernum)242 except:243 print "can't decode usernum %s for user %s" % (`user.usernum`, `user`)244 continue245 print "Adding usernum %d (%s) to the database" % (usernum, user.name)246 self.execute("INSERT INTO pycs_users (usernum, email, password, name, weblogTitle, serialNumber, organization, flBehindFirewall, hitstoday, hitsyesterday, hitsalltime, membersince, lastping, pings, lastupstream, upstreams, lastdelete, deletes, bytesupstreamed, signons, signedon, lastsignon, lastsignoff, clientPort, disabled, alias, flManila, bytesused, stylesheet, commentsdisabled) VALUES (%d, %s, %s, %s, %s, %s, %s, %d, %d, %d, %d, %s, %s, %d, %s, %d, %s, %d, %d, %d, %d, %s, %s, %d, %d, %s, %d, %d, %s, %d)", (247 usernum,248 toutf8(user.email),249 user.password,250 toutf8(user.name),251 toutf8(user.weblogTitle),252 toutf8(user.serialNumber),253 toutf8(user.organization),254 user.flBehindFirewall,255 user.hitstoday,256 user.hitsyesterday,257 user.hitsalltime,258 pyto8601(user.membersince, has_am=1, allow_empty=1),259 pyto8601(user.lastping, has_am=1, allow_empty=1),260 user.pings,261 pyto8601(user.lastupstream, has_am=1, allow_empty=1),262 user.upstreams,263 pyto8601(user.lastdelete, has_am=1, allow_empty=1),264 user.deletes,265 user.bytesupstreamed,266 user.signons,267 user.signedon,268 pyto8601(user.lastsignon, has_am=1, allow_empty=1),269 pyto8601(user.lastsignoff, has_am=1, allow_empty=1),270 user.clientPort,271 user.disabled,272 toutf8(user.alias),273 user.flManila,274 user.bytesused,275 toutf8(user.stylesheet),276 user.commentsdisabled,277 ))278 279 self.set_db_version(11)280 if self.db_id < 12:281 print "Adding pycs_mirroredposts table"282 if self.fetchone("SELECT COLUMN_NAME FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_NAME='pycs_mirrored_posts'"):283 self.execute("DROP TABLE pycs_mirrored_posts")284 285 self.execute("""CREATE TABLE pycs_mirrored_posts (usernum INT, postdate TIMESTAMP, postid VARCHAR(1024), postguid VARCHAR(1024), posturl VARCHAR(1024), posttitle VARCHAR(1024), postcontent TEXT)""")286 mirrored_posts = mkdb.getas("mirroredPosts[usernum:S,posts[date:S,postid:S,guid:S,url:S,title:S,description:S]]")287 for user_posts in mirrored_posts:288 usernum = int(user_posts.usernum)289 print "Adding posts for usernum %d" % usernum290 for post in user_posts.posts:291 print "\t%s" % post.title292 self.execute("""INSERT INTO pycs_mirrored_posts (usernum, postdate, postid, postguid, posturl, posttitle, postcontent) VALUES (%d, %s, %s, %s, %s, %s, %s)""", (293 int(usernum),294 fix8601(post.date), # already iso-8601 so we don't need to decode it295 toutf8(post.postid),296 toutf8(post.guid),297 toutf8(post.url),298 toutf8(post.title),299 toutf8(post.description),300 ))301 self.set_db_version(12)302 # first hack at getting access restrictions in there. it's303 # all messed up though - done without thinking enough. i'm304 # checking this in to cvs anyway just in case someone else305 # wants to fix it up for me :-)306 if 0 and self.db_id < 13: # if you want to fix this, give it a new id - higher than the current max, whatever that is307 print "moving access restrictions into postgres..."308 self.execute("CREATE TABLE pycs_access_locations (blogid INT, locname VARCHAR(1024), regexp VARCHAR(1024), groupname VARCHAR(1024))")309 self.execute("CREATE TABLE pycs_access_groups (blogid INT, groupname VARCHAR(1024), username VARCHAR(1024))")310 self.execute("CREATE TABLE pycs_access_users (blogid INT, username VARCHAR(1024), password VARCHAR(1024))")311 ar_locations = mkdb.getas(312 "arlocations[blogid:S,locname:S,regexp:S,group[name:S]]"313 ).ordered(2)314 for loc in ar_locations:315 for group in loc.group:316 self.execute("INSERT INTO pycs_access_locations (blogid, locname, regexp, groupname) VALUES (%d, %s, %s, %s)",317 (int(loc.blogid),318 toutf8(loc.locname),319 toutf8(loc.regexp),320 toutf8(group.name),321 ))322 ar_groups = mkdb.getas(323 "argroups[blogid:S,name:S,user[name:S]]"324 ).ordered(2)325 for group in ar_groups:326 for user in group.user:327 self.execute("INSERT INTO pycs_access_groups (blogid, groupname, username) VALUES (%d, %s, %s)",328 (int(group.blogid),329 toutf8(group.name),330 toutf8(user.name),331 ))332 ar_users = mkdb.getas(333 "arusers[blogid:S,name:S,password:S]"334 ).ordered(2)335 for user in ar_users:336 self.execute("INSERT INTO pycs_access_users (blogid, username, password) VALUES (%d, %s, %s)",337 (int(user.blogid),338 toutf8(user.name),339 toutf8(user.password),340 ))341 self.set_db_version(13)342 if self.db_id < 14:343 print "Moving trackbacks over to Postgres"344 trackbacks = mkdb.getas(345 'trackbacks[user:S,paragraph:S,notes[name:S,title:S,url:S,excerpt:S,date:S]]'346 ).ordered( 2 )347 self.execute("CREATE TABLE pycs_trackbacks (id INT PRIMARY KEY, usernum INT, postid VARCHAR(255), postlink VARCHAR(2048), tbname VARCHAR(2048), tbtitle VARCHAR(2048), tburl VARCHAR(2048), tbexcerpt TEXT, tbdate TIMESTAMP)")348 self.execute("CREATE SEQUENCE pycs_trackbacks_id_seq")349 self.execute("CREATE INDEX pycs_trackbacks_post_index ON pycs_trackbacks (usernum, postid, id)")350 self.execute("DELETE FROM pycs_trackbacks") # in case something broke during an earlier import351 352 for user in trackbacks:353 usernum = int(user.user)354 postid = user.paragraph355 print "\tusernum %d postid %s" % (usernum, postid)356 for tb in user.notes:357 self.execute("INSERT INTO pycs_trackbacks (id, usernum, postid, tbname, tbtitle, tburl, tbexcerpt, tbdate) VALUES (NEXTVAL('pycs_trackbacks_id_seq'), %d, %s, %s, %s, %s, %s, %s)", (358 usernum,359 toutf8(postid),360 toutf8(tb.name),361 toutf8(tb.title),362 toutf8(tb.url),363 toutf8(tb.excerpt),364 pyto8601(tb.date, has_am=0),365 ))366 self.set_db_version(14)367 # next db id: 15368# not doing this because we don't get the IP address from the front-end proxy anyway369# if self.db_id < XXX:370# print "Adding IP address field to comments table"371# self.execute("""ALTER TABLE pycs_comments ADD COLUMN posterip VARCHAR(15)""")372 ...
__init__.py
Source:__init__.py
...84 result['new_version'] = new_version85 result['message'] = 'Evolved %s to %s' % (86 pkg_name, new_version)87 elif mark_all_current:88 manager._set_db_version(sw_version)89 transaction.commit()90 result['new_version'] = sw_version91 result['message'] = 'Evolved %s to %s' % (92 pkg_name, sw_version)93 else:94 result['new_version'] = db_version95 result['message'] = 'Not evolving (latest not specified)'96 else:97 if set_db_version == db_version:98 result['new_version'] = db_version99 result['message'] = 'Nothing to do'100 else:101 manager._set_db_version(set_db_version)102 transaction.commit()103 result['new_version'] = set_db_version104 result['message'] = (105 'Database version set to %s' % set_db_version106 )107 results.append(result)108 return results109def includeme(config): #pragma: no cover110 config.add_directive('add_evolution_package', add_evolution_package)...
evolve.py
Source:evolve.py
1""" Run database evolution steps """2import getopt3import sys4from pyramid.paster import (5 setup_logging,6 bootstrap,7 )8from substanced.evolution import evolve_packages9def usage(e=None):10 if e is not None:11 print e12 print ''13 print "sd_evolve [--latest] [--set-db-version=num] [--package=name] config_uri"14 print " Evolves new database with changes from scripts in evolve packages"15 print " - with no arguments, evolve just displays versions"16 print " - with the --latest argument, evolve runs scripts as necessary"17 print " - if --package is specified, only operate against the specified"18 print " package name."19 print " - with the --set-db-version argument, evolve runs no scripts"20 print " but just sets the database 'version number' for a package "21 print " to an arbitrary integer number. Requires --package."22 print23 print "e.g. sd_evolve --latest etc/development.ini"24 sys.exit(2)25def main(argv=sys.argv):26 name, argv = argv[0], argv[1:]27 latest = False28 set_db_version = None29 package = None30 try:31 opts, args = getopt.getopt(argv, 'l?hs:p:',32 ['latest',33 'package=',34 'set-db-version=',35 'help',36 ])37 except getopt.GetoptError, e:38 usage(e)39 if args:40 config_uri = args[0]41 else:42 usage('Requires a config_uri as an argument')43 for k, v in opts:44 if k in ('-l', '--latest'):45 latest = True46 if k in ('-h', '-?', '--help'):47 usage()48 if k in ('-p,' '--package'):49 package = v50 if k in ('-s', '--set-db-version'):51 try:52 set_db_version = int(v)53 if set_db_version < 0:54 raise Exception55 except:56 usage('Bad version number %s' % v)57 setup_logging(config_uri)58 env = bootstrap(config_uri)59 root = env['root']60 registry = env['registry']61 try:62 results = evolve_packages(63 registry,64 root,65 package=package,66 set_db_version=set_db_version,67 latest=latest,68 )69 except Exception as e:70 usage(repr(e))71 for result in results:72 print 'Package %(package)s' % result73 print 'Code at software version %(sw_version)s' % result74 print 'Database at version %(db_version)s' % result75 print result['message']76 print ''77if __name__ == '__main__':...
Automation Testing Tutorials
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run 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?
