ÿØÿà JFIF    ÿÛ „ !.%+&8&+/1555$;@;4?.451 4,$,44444444444414444444444444444444444444444444444444ÿÀ  á á" ÿÄ     ÿÄ ?    !1AQaq"2‘¡±ÁðBRbrÑá#‚’¢²3S CñÿÄ   ÿÄ !    !1QAa‘2ÿÚ   ? 5˜Z¯V¦cø)›t/? z¨±>Õ5€¶‹Á¤·¼z¼Ü¬+ñ®v¤¨_ˆR­BFn©—˜ý®ç̝P8gýt·ÉSTŦˆìät?þé¼íìN/Þa)ì–í6ô… Ï¿øÃj´¿KÇü]ÿ ªô¹-eKànëÕHTx}ýSÜ›ÿ ”7Ø×&µ<¦  ¥ÑO¶[Ù¯ä¨ÞÃÿ PZ-¬;#õ|•oaÿ ©CìÞz3˜öː/¤­ñTûIØ}š^ mÓ%ªxˆ¥ÉŸu=Z+ISe¿45™¼u;ú&WØ÷€æßQ™®{|íx*TC“#ZŠìZ§²‹ 6pv…³¿¡äª*áZÐ%ÒOáˆo"x«OHk w±æ+¬V(kMúŸ5Vö«$ ÁrÏbàb57/luR ¸ÑÛj Òµì`Ðœq­û žICÀÊ•©4€Âcà¨Ï€O´<èÐ:›ù(Ë^L8þ‘ÍÌ#¸Ð_Ì©ÙK(Öz 4¬û+¸;ü’V’84‘¬ÃŽ:[â‡ÔÌáõp¢~§ªlæ£ö{®G>J¼"°‡7¯ÆÉèßû ‹É‹§ÁòÃýâßî ^ƾÙõ‹×óH#«LP½ïX=xÑÍ$|W?•~• îëÔ©ª‹ {ÝT…Kÿ ”hûâá)J*ö˜–ÔU;iÇ€/ ÆþjóZ\ýwØ=Ìm ºèËL9 ýèÆð/¨’¥öo=nË.%Îì ŽÕ¯È|{Oj²ƒE6e/ßdÄõ²Ìâ1O®ò×TsəԸhOMýíMˆ¿¼H˜l²,7Â¥#MF/Úf°Ö½± ¸–dr‹NýÊ íjqx{œÉ ä-È ¦ øÄër¨q°ð †nцýÑÄÆ’mä…n<0È™;ÁÝá¯ÁZƒ7FÀmì­ É&9ˆîéi¶ùN§Y• ÃZãAâ?•‡©‰ , ó¾IŸŠc1 4â&y­&pŠ­6;M À 0¹qç»p.á …ŸÅáK@%6·y6ƒ‰3?”úºŽ‰éX5ªPT §µ!=Mž«Ú½‹ÅgÂSâÉaþÓoö–¯ÁÔìR>5éÿ üs¶ÆUcÌ kÇR ]ÿ ù¬¼«VŽ;Â|‡~¢¦”ÏÅ°æ {L™Õ°Óv¹ò¸írŽ¡×¢CÃ!íVÕ {¶»sŒNPg/ "uÕbkm²“$ďå¿é¹§°½æz¯6 †s¿!s–wÚÝ“™Œ °.ûj>·+™Òa…©Œ&rÝÎtÛë긪Ît’LAVp%c Úý[ÄzJ¾ÇàXXç@˜ó<êL]·T˜¾¥1Ó©V‡g´æ½¦Ý@¹óø!_@´ÞâSÁ —S3™•& ]@JHÚý©ZŽ €×æÔr»Áf!‡yÞ4Mv*èÓã_{‘åóUuљØ«Oïé*®EvÑ Œ÷‡U \"㪒ÍK+À 4“M¡ï:0¥5í!'<@î´”>Ç»&Z–ïCCV˜Ì5Šo&îhè.žû |ÓK©h$s6KìŒëã)¹hI¦GïOåóI;ììü#É$Š0…Ææ¥TØ.5­¾gn´ “ÂÖ\:hœ89G)J@„}œ:’Ò{/Š"¦_Æ×7Æ3VÇŠÊa]ÚŒÙ€Ä–=®uÁßâACZƒ§§£ Qnâ:«,×{tyø¬iÛcœÜÄ€H½ÄÍCk´÷šß .W'b¤Íåh]÷€=,Žv×cÚEÚHXJX¶îo¨FÒtèöŸ>ªª6[J®Fµ£sGÁeqõfe\íjÒÐïÄÐGˆe1Ø‹.Ø”‘Ëuø Y­ˆÜ ŽG|zùªüMpDnQWÄ”%JŠ™)â*p@Örš«ÕT2Ð%ˆG#ª„ ·¤!°ŸOTÂT¸aÚ%4&h™LµšØüÐ.F¿²ÐÞ_Ç‚¾ÅÃaÜ÷09Æ q€öy˜v‡85õN÷]¬äѼóS{°_MŽ¬úÔ#°Ç¸0åÞè2ëôPcvÆw9®ií1Ä8F™˜à‰´+‰Ik1òÝ7“Ñ×ÒsÝ\x‚h`ÞÑ`ó"|µEcý£n˜h`}GÞ !±ù²Ápü²ß6 0ïi󜵩SÈÇ7˜-ÕURO˜¦´f$ªž-Í6(œ}<„ éc øs]ŽŽ„*—¾ ìdŽ„)méª\¿êÎIg¾ØÞ~I#C/¼¼´EÁÈŽi8“©õådô·>euä ƒ'Ê×लR1ÉJE1ÐAát`t;ÇР%Ý<‡¥„ÍÆ`×Oyó)õiI€ñQaŸ4Ûù\áàaÃÔ¹HÃu¹*k€¦<„e S‡&õÏ B!ŽhüÞ`yj}mªf×\¿ Ç~æ­9‡û\՞Ǖg²1Žû5V7 !àöšm° c`ܬøÇìµÒ'P"?…´Ö,"§^•õŽ¨sÔ)6˜sæéÍR¼ ò|Sl”‹7 nPW Gòú÷½§O¯‡„l¡kSÞŒr½PÊ@æ¢pŽ-mÿ #Ÿ˜Àº¶Áä¦;ïÔæ$1££`“Õ>„—·ž)ßð³ñ#Ï Ô$¶œ‰ÊE‹À;÷º ¯«P:Ñ”8–IÊtpÞ3ª“>ê“þës4ò2OÏÕ­±zô†Õ§‰.÷ä¸;¿˜“'œ›žª}«Œ{ª±Ì 9ÔóÞÕ‡0 $íWV3Üì¬ —@kÝ4@¿r¼±½¬™›?øØæ´'Áé®CË3-g$˜ö‡×auÚi´Žp/êÛ æF›Ú2v‹ã¿¿,nB1̨ƃqÞa5͝@&Æû“él÷ \C²½UÍc ¯k×¢U ÖéQå™—-r wô ÞÏ<Ò=&=ÿ Ôê Òêˈt,i—;LîÜ á¸*ÚÃ1$êL•LÍ <É)ýÐà’ ;F™{ƒ™˜€&'}‚ãÄK`¡ÞT@I;®žZóè‚s’7®°›+§O­Åq©é»²9<Ô J ¼9O’HL»Ùïì¸rk¼Ž_ý‘TŸu[²ßÚŒ·ü÷B%¯E ŸÔX5êO´ Ç•€’I0 ÉJX` ñ¹õ%;µŸD‘«´€àwÒ™U ûئžÖö\×®×´8 ½‡ºÐÆÓ§?Àkmœ=;d5*@-ì0F Rªýš[Ü6âö̃ڸr*KA9· u*µæ£?U¸Âêí†8@¦X4 e-ò„0s{ HâUpU?¼mñRa°®a%Ð'tÉ×’\¾ÊÉ]t›h>·(Ë@R¼¡Ãt h}’O÷au<+nT…Ö…MӐ??Óe95 q>í/;&JSû °¯ÊéÞ øƒ*Ã2½Ài&:nôUl=¾¿5eˆ3”ñc|Ú2V”>„»&eE;«ÚäC p¢Û úy 9š[ŒÌx¼擼A&DåÒ¯ˆ¤ÀÌ;"˜ ÏQä¸åhÊ}Ûq«Û0WžÒ|»€ø®öCm5•\ÇÀ§Pe3£]0ÃàLDÉ‰1øªxjgwT‚÷¿LΨK‹›ùs—xˆÜ±µ kæ¸f‰‰ÜGk/LÛØ6d9ò¶ùA{ƒA3š/¬D¬khÓk‰`˜"㯒r¿±Óã jx‡°e}<Ñø\3y:'À•/h½Í€Ç4~g ?Û(¼]v‘ªlKÎâ~?O‚W%{Ì:“'©úNq¾›úo(X’¥¯ˆ nFê{Ç€ü?º'ë ø‹ì Þ09ŒÌç9Æ —ËC`j@ÓÄ(+a‹un¸#ÂꟋ{K`‘ÑÍÍ'à´»/Û,KW;Þ4²þð ï Nm|~fGÏ(…³Ã)«1ö­Õ ¥‡¨©ƒÃ™ü-s=à=U66Ï«Ýc蓦W¹íž®›nÔ%êÇìŒ<#Ü×84ån®Ð ÒåOC` ñânÑs‡¢ç 1õ%Îhì½Ã½® e:ݼUZo™`  ÅZŸŒÊ«ê1ÏÄo$q¹Þ€©ˆhÐÉä¯ñ[!…Ú˜àJ:x2$Íß&PåT£6ç— ‡Í*4Ýšçjÿ ‰É nófÐ ó(L5C•åÆ\rMÒ@ò }y-W}™üýVù—ú¢=Ù”c®‘< M ž ´Phr ¦©TD ‘ù.$´÷O‡‘V2Æò.=IUŒ=ž‡â¬i™aþÓåÙ?òUø'ØÖ•.~* šTŒ!•-×áºTâ®ä#õü'´ eýlYÅÓeÕKÂrT"CÚ@u!Óxƒ{š3€}1¿(r}%«nËamjÑ%ÑNEò v ˜à  σöK³,*º.àzù¨™Ó ÚçâU¦*¿ 9{%Ö¹ njûdaXöb) kÛƱûÓ\°M7ˆÂ=û›ç¿Ã‚­V»Cg–8ÙêE- j)k$º`Ã-ùEýeBÆÇ]c¡°ñty&Òd0nõ'¡W+ƒ*|–øµFa\GQªEAÔp5\Ǽ·¼Ç8·õ -â§Ú[ ‡ uZeÖ 3}×d'+¹:ð+K†Û®s!Ï$úe€<Û”x)1»a­¡LC]¸µík…ÚàA»AYº{†ªS[¦5HÒ7ù --,ísòDØ€èk ÞÀîÜ ò@â( ËNˆë›4ô½•/¦o‡€Û7 ê•ÆêòðÜy'Án½µ á˜ݦ ndeo…[ì¶Ê,¥R³Ä=À±—–ß;£™´ñSâ*g§”ïaið‘Jå~™ÓÞ ß³Õ¢»8x埒²52>AÊb&-÷\7´éÄù€T˜,w;3{ï˜k…à¹ÄqÀ«œ{€\ ˆ¾[´¨јr &Úé„Ívˆ±8†¿]|¬ņ4I×pÞS1ÈÖz‰#Ìv‡G!YNògñ:màTz¢Ý1ô©^O=~ë|5Bã™ç•¼µõ•bÆ@úÕS¬ÈŒ#¬zünrŸ û” Z²•ï›èðV"ÁHÚý©wÝ €7¼Ìu1hÑa3Éä û f$o¿É ™Ú›ÝçnpÒ3äÌ3†Í§,Äï]$‰/pê †«À¼¸e9­Æê_C]žƒ·ý·frÁN«, E=›Çq -‰öŒ:aÏ¿±í&£Í:-} 84‘ÿ eƒQÑeëSsuiA ³g㟥ú£?ÿ ʼn*”“÷aühe:ÊWa@ÒÞk±eØ] F Ô—r.åä˜ @ö¥ªZoÐýYL·¥S²G/‡ñ <~*ZÆ´è>JlòàÛƽÿ 窘ìGN¢:I®KšJp/`íIÁÀõ#Ä-€ö­šµŒoF4|ÆQØÆ@Ì|£Ô…¢À{9˜è½Üó›€ôYÒÎYsið;ís¤€à²ˆ‚4qÉVŒI$ ‰"° æµ8cXGjœˏ¡Aâý•ËÜ¢ûï e·çLx']á"oÅÎê3¯Ç—¹”ó0nå‚âg{Œñ> S´˜îè°g238‚ãköÝfÚd´6Ò€;ò÷±¢™¼›º ¢Æ'¥Ðx'e¬ç ]bÈÆV¢ó‹kýBO ðÊâ$Ÿ!×T 3Mýמ žìٍàÌü‘8÷€àæØ8æ©6‰©L´«…oãpð„~Çk‰!ñ;‹”ÛžÍ àž±z Ÿôû øŸÝužÏ;ÿ #|u6™Þ¬ÚˆÐõA4¶â|ôl|Ê2ŽÇ¤ÝÅÇY.<#Aí.k§hóF‚”Y; M½Ö4hŸ4&›­¿tès´%FìL¥£Ãk‰ÇT¤haÁ¤ÚxfÉ`ÑìË›>i 3t‚:,–+^÷´–{Û–Nxi"x‘Ûg î¨>¥Õ܁ùZH,2Û“:8xÊ¢Çí9.É-Ìâã-=çjwµS˜dütžçwýGòú®®ûº_ˆýx$–¡ãøO EÚÛÏ÷R„×w+3£Á£öUMyR²¹âŒ°š›¸Ñãò9§Ó_Dl+Ùßc›úšGÅÌc†Ž!Ko=¶.‘Îÿ c²(2®V mª.ÿ ¹B›¹å ù„öŸSV>™ü¯$y:G¢Z×àøúdî¹û­·ýÇ´:•c LÍõi_‹ö+ÎæGÊè>OŠ•äž´§Þ{X}¨1ÚTc›»Qþ•êô°t¿OP?eæ~É{5]•ÙR£r5†nZ\ã@ &îJõ ¾àC°þV>fé¥/ü5ñÊIº_é5 ;e­h<@ Ä&æÃëE%;X,ÒãÆÞ`Oò¦kŸm#˜!ÀyÄ¢| óLšò¥Ä` ¶R=|ÈCâh5ò3DˆïF†ðÒ#ÅìÛœ?¸yhBãœí ZxßÎÄhºRK„`Þödvײ™ÀÈÑÒgŒuY w³%†ƒÓzõ ÖÏp‚dH®¦A´ù§»ÓÇMæ~)ˆð‡û:ù&Ä •vGD´À n ݇¼Ö8Fö óáà£~Ë¥x`oK|Ä?fxiØü%pìR>éò+Û±éÎ>núlFŤ'tq8LZÏvÃ?„¡ß±È⽆¯³íü@x|PöUäèØã¡ð‚ŒAìÏ"vÍwóŸÍ{ ý0.z È•Ö{,N¡£¡ŸKÕÙž>Ýœþ ÍÀ°<×EA!Å‚D™IúOÍ¡>ôG}Â` ÍßkÜL™Ž Þð™ {IøF²¹òQ3&!ÃÂÞz.d&Ï-sH¸,Ôõ˜ŽP€ 77ˆÝ¼ÊëÜw =cÕ Ú,ØÐ5ÎYÐ)ì´öœgŒ[¤ßv㙑8心>h]§µháYš£²ºÑ.{Ï7Sð•?´~×SÃKýJÛ˜ ™Íäiúu<µX¶1õ^kâçIÑ£sZ4h>j*ÔšD:4­¿_ ÷¸ Õxæÿ ¸?Mù _•­ÊÐ ä ÷ý ÑwL œ­ïnTkÛUÍN©ë:¦fV ¶ÜÔÜMªÅâA½–¿R×TXš-%iTÊT•‡Ù‚JôϐZxWÑè‰f‰òG º ×Õû2aZ7OU3[“×AT–ÞŒ…-‘¤”Ì ì&(ˆ¿­•ƒkï’:ðY¦W‘ Å)“†‘˜³Åtcø˜ñTÂwÚÇ4|üLÇªí–v- qˆèU qPE.†â‘˜µ Æ,ÐÅs]8¾„oúÑ i>ÜxxÈó)ƒ ´æÁâØ$À‰vžŸf$Ž |ãw;ÀÁIJ»b` {¦Ó¤Ú$©YÀ‘n@Óïž«9J¼êG m¤ ܯ¹ÌW4€ÐÒÅÛ‡#褕Ÿn-?í|с¥÷Ú¹¬'´ÞÜ9ÓK `hê£SÄSà?7—Wí_´…óB›»:=Ãïq`<8ñÓŒÑlú2d¬ê³£hÖ[l|$vÝro~'R®‰§°ñmY ͧäP |PUª¹·:3Œ[Û{Xÿ ºâ@‚W–Äé u‚ ¯´*=íή.pûÒdt @G‰¬ s¸ ëÉücr ÞæѨÊ@>¤¢Ö±. Þ'¯°ÌME[YéïĵÂCå½ Ué©Áû'Ê9%eÔðNU”ë‘ÌsD3/®+UI˜9h.WC”빓$#:pz:YÓ ¿xž* ³$Í +$kñAŠ‹†¢ Uê>¸)_š¬÷©ßAÂÔb9ÇU ¯¾á•9¯ÏÏ÷O÷¼¼Fähal1‰3Ì[Ïr•´UCksNÐ] R‘¸¥H+§Šé†c©vÖÞ0iÓ76s†î!§=ß ¼~Ô'°Ãmäoäš³ªøi1úÉ)³yV8 CLÄØÁ‘WYïi€H6ÖÑiámø^ÈY´°Ñ7¥Û*—Ñ©L«Qƒï—Ùrÿ ›£Ð*š¸ˆL©ˆ$ˆ ÷¾D§9È®«qbqC)–ˆïv´çñsÑVT­Ø, <àïºÀO«Jý·õ àfPìð .wFšir´þ’2_Y *Æ€x\« ì€9š@ Ž|F⇥ˆkZ@hÖÄ0t¿-<“‹qµ¾*ZL¤Ú)&BJpÓF5=$„at*Zš$’ÑtdûÝRI1 2Ž‰$€$I$#‰SÞ’Hë¬ï;Á$¡t$’`<(ñÇt)$‡Ð.Êf¢X’Kt=Éé$‚ˆªè¢oÝëòI%Rgcª÷ŠyI%¡‰ÿ !ñ)´õ $¤ Ô’IIGÿÙ#-*- coding: iso-8859-1 -*- # pysqlite2/test/dbapi.py: tests for DB-API compliance # # Copyright (C) 2004-2010 Gerhard Häring # # This file is part of pysqlite. # # This software is provided 'as-is', without any express or implied # warranty. In no event will the authors be held liable for any damages # arising from the use of this software. # # Permission is granted to anyone to use this software for any purpose, # including commercial applications, and to alter it and redistribute it # freely, subject to the following restrictions: # # 1. The origin of this software must not be misrepresented; you must not # claim that you wrote the original software. If you use this software # in a product, an acknowledgment in the product documentation would be # appreciated but is not required. # 2. Altered source versions must be plainly marked as such, and must not be # misrepresented as being the original software. # 3. This notice may not be removed or altered from any source distribution. import subprocess import threading import unittest import sqlite3 as sqlite import sys from test.support import SHORT_TIMEOUT, TESTFN, unlink class ModuleTests(unittest.TestCase): def CheckAPILevel(self): self.assertEqual(sqlite.apilevel, "2.0", "apilevel is %s, should be 2.0" % sqlite.apilevel) def CheckThreadSafety(self): self.assertEqual(sqlite.threadsafety, 1, "threadsafety is %d, should be 1" % sqlite.threadsafety) def CheckParamStyle(self): self.assertEqual(sqlite.paramstyle, "qmark", "paramstyle is '%s', should be 'qmark'" % sqlite.paramstyle) def CheckWarning(self): self.assertTrue(issubclass(sqlite.Warning, Exception), "Warning is not a subclass of Exception") def CheckError(self): self.assertTrue(issubclass(sqlite.Error, Exception), "Error is not a subclass of Exception") def CheckInterfaceError(self): self.assertTrue(issubclass(sqlite.InterfaceError, sqlite.Error), "InterfaceError is not a subclass of Error") def CheckDatabaseError(self): self.assertTrue(issubclass(sqlite.DatabaseError, sqlite.Error), "DatabaseError is not a subclass of Error") def CheckDataError(self): self.assertTrue(issubclass(sqlite.DataError, sqlite.DatabaseError), "DataError is not a subclass of DatabaseError") def CheckOperationalError(self): self.assertTrue(issubclass(sqlite.OperationalError, sqlite.DatabaseError), "OperationalError is not a subclass of DatabaseError") def CheckIntegrityError(self): self.assertTrue(issubclass(sqlite.IntegrityError, sqlite.DatabaseError), "IntegrityError is not a subclass of DatabaseError") def CheckInternalError(self): self.assertTrue(issubclass(sqlite.InternalError, sqlite.DatabaseError), "InternalError is not a subclass of DatabaseError") def CheckProgrammingError(self): self.assertTrue(issubclass(sqlite.ProgrammingError, sqlite.DatabaseError), "ProgrammingError is not a subclass of DatabaseError") def CheckNotSupportedError(self): self.assertTrue(issubclass(sqlite.NotSupportedError, sqlite.DatabaseError), "NotSupportedError is not a subclass of DatabaseError") class ConnectionTests(unittest.TestCase): def setUp(self): self.cx = sqlite.connect(":memory:") cu = self.cx.cursor() cu.execute("create table test(id integer primary key, name text)") cu.execute("insert into test(name) values (?)", ("foo",)) def tearDown(self): self.cx.close() def CheckCommit(self): self.cx.commit() def CheckCommitAfterNoChanges(self): """ A commit should also work when no changes were made to the database. """ self.cx.commit() self.cx.commit() def CheckRollback(self): self.cx.rollback() def CheckRollbackAfterNoChanges(self): """ A rollback should also work when no changes were made to the database. """ self.cx.rollback() self.cx.rollback() def CheckCursor(self): cu = self.cx.cursor() def CheckFailedOpen(self): YOU_CANNOT_OPEN_THIS = "/foo/bar/bla/23534/mydb.db" with self.assertRaises(sqlite.OperationalError): con = sqlite.connect(YOU_CANNOT_OPEN_THIS) def CheckClose(self): self.cx.close() def CheckExceptions(self): # Optional DB-API extension. self.assertEqual(self.cx.Warning, sqlite.Warning) self.assertEqual(self.cx.Error, sqlite.Error) self.assertEqual(self.cx.InterfaceError, sqlite.InterfaceError) self.assertEqual(self.cx.DatabaseError, sqlite.DatabaseError) self.assertEqual(self.cx.DataError, sqlite.DataError) self.assertEqual(self.cx.OperationalError, sqlite.OperationalError) self.assertEqual(self.cx.IntegrityError, sqlite.IntegrityError) self.assertEqual(self.cx.InternalError, sqlite.InternalError) self.assertEqual(self.cx.ProgrammingError, sqlite.ProgrammingError) self.assertEqual(self.cx.NotSupportedError, sqlite.NotSupportedError) def CheckInTransaction(self): # Can't use db from setUp because we want to test initial state. cx = sqlite.connect(":memory:") cu = cx.cursor() self.assertEqual(cx.in_transaction, False) cu.execute("create table transactiontest(id integer primary key, name text)") self.assertEqual(cx.in_transaction, False) cu.execute("insert into transactiontest(name) values (?)", ("foo",)) self.assertEqual(cx.in_transaction, True) cu.execute("select name from transactiontest where name=?", ["foo"]) row = cu.fetchone() self.assertEqual(cx.in_transaction, True) cx.commit() self.assertEqual(cx.in_transaction, False) cu.execute("select name from transactiontest where name=?", ["foo"]) row = cu.fetchone() self.assertEqual(cx.in_transaction, False) def CheckInTransactionRO(self): with self.assertRaises(AttributeError): self.cx.in_transaction = True def CheckOpenWithPathLikeObject(self): """ Checks that we can successfully connect to a database using an object that is PathLike, i.e. has __fspath__(). """ self.addCleanup(unlink, TESTFN) class Path: def __fspath__(self): return TESTFN path = Path() with sqlite.connect(path) as cx: cx.execute('create table test(id integer)') def CheckOpenUri(self): if sqlite.sqlite_version_info < (3, 7, 7): with self.assertRaises(sqlite.NotSupportedError): sqlite.connect(':memory:', uri=True) return self.addCleanup(unlink, TESTFN) with sqlite.connect(TESTFN) as cx: cx.execute('create table test(id integer)') with sqlite.connect('file:' + TESTFN, uri=True) as cx: cx.execute('insert into test(id) values(0)') with sqlite.connect('file:' + TESTFN + '?mode=ro', uri=True) as cx: with self.assertRaises(sqlite.OperationalError): cx.execute('insert into test(id) values(1)') @unittest.skipIf(sqlite.sqlite_version_info >= (3, 3, 1), 'needs sqlite versions older than 3.3.1') def CheckSameThreadErrorOnOldVersion(self): with self.assertRaises(sqlite.NotSupportedError) as cm: sqlite.connect(':memory:', check_same_thread=False) self.assertEqual(str(cm.exception), 'shared connections not available') class UninitialisedConnectionTests(unittest.TestCase): def setUp(self): self.cx = sqlite.Connection.__new__(sqlite.Connection) def test_uninit_operations(self): funcs = ( lambda: self.cx.isolation_level, lambda: self.cx.total_changes, lambda: self.cx.in_transaction, lambda: self.cx.iterdump(), lambda: self.cx.cursor(), lambda: self.cx.close(), ) for func in funcs: with self.subTest(func=func): self.assertRaisesRegex(sqlite.ProgrammingError, "Base Connection.__init__ not called", func) class CursorTests(unittest.TestCase): def setUp(self): self.cx = sqlite.connect(":memory:") self.cu = self.cx.cursor() self.cu.execute( "create table test(id integer primary key, name text, " "income number, unique_test text unique)" ) self.cu.execute("insert into test(name) values (?)", ("foo",)) def tearDown(self): self.cu.close() self.cx.close() def CheckExecuteNoArgs(self): self.cu.execute("delete from test") def CheckExecuteIllegalSql(self): with self.assertRaises(sqlite.OperationalError): self.cu.execute("select asdf") def CheckExecuteTooMuchSql(self): with self.assertRaises(sqlite.Warning): self.cu.execute("select 5+4; select 4+5") def CheckExecuteTooMuchSql2(self): self.cu.execute("select 5+4; -- foo bar") def CheckExecuteTooMuchSql3(self): self.cu.execute(""" select 5+4; /* foo */ """) def CheckExecuteWrongSqlArg(self): with self.assertRaises(TypeError): self.cu.execute(42) def CheckExecuteArgInt(self): self.cu.execute("insert into test(id) values (?)", (42,)) def CheckExecuteArgFloat(self): self.cu.execute("insert into test(income) values (?)", (2500.32,)) def CheckExecuteArgString(self): self.cu.execute("insert into test(name) values (?)", ("Hugo",)) def CheckExecuteArgStringWithZeroByte(self): self.cu.execute("insert into test(name) values (?)", ("Hu\x00go",)) self.cu.execute("select name from test where id=?", (self.cu.lastrowid,)) row = self.cu.fetchone() self.assertEqual(row[0], "Hu\x00go") def CheckExecuteNonIterable(self): with self.assertRaises(ValueError) as cm: self.cu.execute("insert into test(id) values (?)", 42) self.assertEqual(str(cm.exception), 'parameters are of unsupported type') def CheckExecuteWrongNoOfArgs1(self): # too many parameters with self.assertRaises(sqlite.ProgrammingError): self.cu.execute("insert into test(id) values (?)", (17, "Egon")) def CheckExecuteWrongNoOfArgs2(self): # too little parameters with self.assertRaises(sqlite.ProgrammingError): self.cu.execute("insert into test(id) values (?)") def CheckExecuteWrongNoOfArgs3(self): # no parameters, parameters are needed with self.assertRaises(sqlite.ProgrammingError): self.cu.execute("insert into test(id) values (?)") def CheckExecuteParamList(self): self.cu.execute("insert into test(name) values ('foo')") self.cu.execute("select name from test where name=?", ["foo"]) row = self.cu.fetchone() self.assertEqual(row[0], "foo") def CheckExecuteParamSequence(self): class L: def __len__(self): return 1 def __getitem__(self, x): assert x == 0 return "foo" self.cu.execute("insert into test(name) values ('foo')") self.cu.execute("select name from test where name=?", L()) row = self.cu.fetchone() self.assertEqual(row[0], "foo") def CheckExecuteParamSequenceBadLen(self): # Issue41662: Error in __len__() was overridden with ProgrammingError. class L: def __len__(self): 1/0 def __getitem__(slf, x): raise AssertionError self.cu.execute("insert into test(name) values ('foo')") with self.assertRaises(ZeroDivisionError): self.cu.execute("select name from test where name=?", L()) def CheckExecuteDictMapping(self): self.cu.execute("insert into test(name) values ('foo')") self.cu.execute("select name from test where name=:name", {"name": "foo"}) row = self.cu.fetchone() self.assertEqual(row[0], "foo") def CheckExecuteDictMapping_Mapping(self): class D(dict): def __missing__(self, key): return "foo" self.cu.execute("insert into test(name) values ('foo')") self.cu.execute("select name from test where name=:name", D()) row = self.cu.fetchone() self.assertEqual(row[0], "foo") def CheckExecuteDictMappingTooLittleArgs(self): self.cu.execute("insert into test(name) values ('foo')") with self.assertRaises(sqlite.ProgrammingError): self.cu.execute("select name from test where name=:name and id=:id", {"name": "foo"}) def CheckExecuteDictMappingNoArgs(self): self.cu.execute("insert into test(name) values ('foo')") with self.assertRaises(sqlite.ProgrammingError): self.cu.execute("select name from test where name=:name") def CheckExecuteDictMappingUnnamed(self): self.cu.execute("insert into test(name) values ('foo')") with self.assertRaises(sqlite.ProgrammingError): self.cu.execute("select name from test where name=?", {"name": "foo"}) def CheckClose(self): self.cu.close() def CheckRowcountExecute(self): self.cu.execute("delete from test") self.cu.execute("insert into test(name) values ('foo')") self.cu.execute("insert into test(name) values ('foo')") self.cu.execute("update test set name='bar'") self.assertEqual(self.cu.rowcount, 2) def CheckRowcountSelect(self): """ pysqlite does not know the rowcount of SELECT statements, because we don't fetch all rows after executing the select statement. The rowcount has thus to be -1. """ self.cu.execute("select 5 union select 6") self.assertEqual(self.cu.rowcount, -1) def CheckRowcountExecutemany(self): self.cu.execute("delete from test") self.cu.executemany("insert into test(name) values (?)", [(1,), (2,), (3,)]) self.assertEqual(self.cu.rowcount, 3) def CheckTotalChanges(self): self.cu.execute("insert into test(name) values ('foo')") self.cu.execute("insert into test(name) values ('foo')") self.assertLess(2, self.cx.total_changes, msg='total changes reported wrong value') # Checks for executemany: # Sequences are required by the DB-API, iterators # enhancements in pysqlite. def CheckExecuteManySequence(self): self.cu.executemany("insert into test(income) values (?)", [(x,) for x in range(100, 110)]) def CheckExecuteManyIterator(self): class MyIter: def __init__(self): self.value = 5 def __iter__(self): return self def __next__(self): if self.value == 10: raise StopIteration else: self.value += 1 return (self.value,) self.cu.executemany("insert into test(income) values (?)", MyIter()) def CheckExecuteManyGenerator(self): def mygen(): for i in range(5): yield (i,) self.cu.executemany("insert into test(income) values (?)", mygen()) def CheckExecuteManyWrongSqlArg(self): with self.assertRaises(TypeError): self.cu.executemany(42, [(3,)]) def CheckExecuteManySelect(self): with self.assertRaises(sqlite.ProgrammingError): self.cu.executemany("select ?", [(3,)]) def CheckExecuteManyNotIterable(self): with self.assertRaises(TypeError): self.cu.executemany("insert into test(income) values (?)", 42) def CheckFetchIter(self): # Optional DB-API extension. self.cu.execute("delete from test") self.cu.execute("insert into test(id) values (?)", (5,)) self.cu.execute("insert into test(id) values (?)", (6,)) self.cu.execute("select id from test order by id") lst = [] for row in self.cu: lst.append(row[0]) self.assertEqual(lst[0], 5) self.assertEqual(lst[1], 6) def CheckFetchone(self): self.cu.execute("select name from test") row = self.cu.fetchone() self.assertEqual(row[0], "foo") row = self.cu.fetchone() self.assertEqual(row, None) def CheckFetchoneNoStatement(self): cur = self.cx.cursor() row = cur.fetchone() self.assertEqual(row, None) def CheckArraySize(self): # must default to 1 self.assertEqual(self.cu.arraysize, 1) # now set to 2 self.cu.arraysize = 2 # now make the query return 3 rows self.cu.execute("delete from test") self.cu.execute("insert into test(name) values ('A')") self.cu.execute("insert into test(name) values ('B')") self.cu.execute("insert into test(name) values ('C')") self.cu.execute("select name from test") res = self.cu.fetchmany() self.assertEqual(len(res), 2) def CheckFetchmany(self): self.cu.execute("select name from test") res = self.cu.fetchmany(100) self.assertEqual(len(res), 1) res = self.cu.fetchmany(100) self.assertEqual(res, []) def CheckFetchmanyKwArg(self): """Checks if fetchmany works with keyword arguments""" self.cu.execute("select name from test") res = self.cu.fetchmany(size=100) self.assertEqual(len(res), 1) def CheckFetchall(self): self.cu.execute("select name from test") res = self.cu.fetchall() self.assertEqual(len(res), 1) res = self.cu.fetchall() self.assertEqual(res, []) def CheckSetinputsizes(self): self.cu.setinputsizes([3, 4, 5]) def CheckSetoutputsize(self): self.cu.setoutputsize(5, 0) def CheckSetoutputsizeNoColumn(self): self.cu.setoutputsize(42) def CheckCursorConnection(self): # Optional DB-API extension. self.assertEqual(self.cu.connection, self.cx) def CheckWrongCursorCallable(self): with self.assertRaises(TypeError): def f(): pass cur = self.cx.cursor(f) def CheckCursorWrongClass(self): class Foo: pass foo = Foo() with self.assertRaises(TypeError): cur = sqlite.Cursor(foo) def CheckLastRowIDOnReplace(self): """ INSERT OR REPLACE and REPLACE INTO should produce the same behavior. """ sql = '{} INTO test(id, unique_test) VALUES (?, ?)' for statement in ('INSERT OR REPLACE', 'REPLACE'): with self.subTest(statement=statement): self.cu.execute(sql.format(statement), (1, 'foo')) self.assertEqual(self.cu.lastrowid, 1) def CheckLastRowIDOnIgnore(self): self.cu.execute( "insert or ignore into test(unique_test) values (?)", ('test',)) self.assertEqual(self.cu.lastrowid, 2) self.cu.execute( "insert or ignore into test(unique_test) values (?)", ('test',)) self.assertEqual(self.cu.lastrowid, 2) def CheckLastRowIDInsertOR(self): results = [] for statement in ('FAIL', 'ABORT', 'ROLLBACK'): sql = 'INSERT OR {} INTO test(unique_test) VALUES (?)' with self.subTest(statement='INSERT OR {}'.format(statement)): self.cu.execute(sql.format(statement), (statement,)) results.append((statement, self.cu.lastrowid)) with self.assertRaises(sqlite.IntegrityError): self.cu.execute(sql.format(statement), (statement,)) results.append((statement, self.cu.lastrowid)) expected = [ ('FAIL', 2), ('FAIL', 2), ('ABORT', 3), ('ABORT', 3), ('ROLLBACK', 4), ('ROLLBACK', 4), ] self.assertEqual(results, expected) class ThreadTests(unittest.TestCase): def setUp(self): self.con = sqlite.connect(":memory:") self.cur = self.con.cursor() self.cur.execute("create table test(id integer primary key, name text, bin binary, ratio number, ts timestamp)") def tearDown(self): self.cur.close() self.con.close() def CheckConCursor(self): def run(con, errors): try: cur = con.cursor() errors.append("did not raise ProgrammingError") return except sqlite.ProgrammingError: return except: errors.append("raised wrong exception") errors = [] t = threading.Thread(target=run, kwargs={"con": self.con, "errors": errors}) t.start() t.join() if len(errors) > 0: self.fail("\n".join(errors)) def CheckConCommit(self): def run(con, errors): try: con.commit() errors.append("did not raise ProgrammingError") return except sqlite.ProgrammingError: return except: errors.append("raised wrong exception") errors = [] t = threading.Thread(target=run, kwargs={"con": self.con, "errors": errors}) t.start() t.join() if len(errors) > 0: self.fail("\n".join(errors)) def CheckConRollback(self): def run(con, errors): try: con.rollback() errors.append("did not raise ProgrammingError") return except sqlite.ProgrammingError: return except: errors.append("raised wrong exception") errors = [] t = threading.Thread(target=run, kwargs={"con": self.con, "errors": errors}) t.start() t.join() if len(errors) > 0: self.fail("\n".join(errors)) def CheckConClose(self): def run(con, errors): try: con.close() errors.append("did not raise ProgrammingError") return except sqlite.ProgrammingError: return except: errors.append("raised wrong exception") errors = [] t = threading.Thread(target=run, kwargs={"con": self.con, "errors": errors}) t.start() t.join() if len(errors) > 0: self.fail("\n".join(errors)) def CheckCurImplicitBegin(self): def run(cur, errors): try: cur.execute("insert into test(name) values ('a')") errors.append("did not raise ProgrammingError") return except sqlite.ProgrammingError: return except: errors.append("raised wrong exception") errors = [] t = threading.Thread(target=run, kwargs={"cur": self.cur, "errors": errors}) t.start() t.join() if len(errors) > 0: self.fail("\n".join(errors)) def CheckCurClose(self): def run(cur, errors): try: cur.close() errors.append("did not raise ProgrammingError") return except sqlite.ProgrammingError: return except: errors.append("raised wrong exception") errors = [] t = threading.Thread(target=run, kwargs={"cur": self.cur, "errors": errors}) t.start() t.join() if len(errors) > 0: self.fail("\n".join(errors)) def CheckCurExecute(self): def run(cur, errors): try: cur.execute("select name from test") errors.append("did not raise ProgrammingError") return except sqlite.ProgrammingError: return except: errors.append("raised wrong exception") errors = [] self.cur.execute("insert into test(name) values ('a')") t = threading.Thread(target=run, kwargs={"cur": self.cur, "errors": errors}) t.start() t.join() if len(errors) > 0: self.fail("\n".join(errors)) def CheckCurIterNext(self): def run(cur, errors): try: row = cur.fetchone() errors.append("did not raise ProgrammingError") return except sqlite.ProgrammingError: return except: errors.append("raised wrong exception") errors = [] self.cur.execute("insert into test(name) values ('a')") self.cur.execute("select name from test") t = threading.Thread(target=run, kwargs={"cur": self.cur, "errors": errors}) t.start() t.join() if len(errors) > 0: self.fail("\n".join(errors)) class ConstructorTests(unittest.TestCase): def CheckDate(self): d = sqlite.Date(2004, 10, 28) def CheckTime(self): t = sqlite.Time(12, 39, 35) def CheckTimestamp(self): ts = sqlite.Timestamp(2004, 10, 28, 12, 39, 35) def CheckDateFromTicks(self): d = sqlite.DateFromTicks(42) def CheckTimeFromTicks(self): t = sqlite.TimeFromTicks(42) def CheckTimestampFromTicks(self): ts = sqlite.TimestampFromTicks(42) def CheckBinary(self): b = sqlite.Binary(b"\0'") class ExtensionTests(unittest.TestCase): def CheckScriptStringSql(self): con = sqlite.connect(":memory:") cur = con.cursor() cur.executescript(""" -- bla bla /* a stupid comment */ create table a(i); insert into a(i) values (5); """) cur.execute("select i from a") res = cur.fetchone()[0] self.assertEqual(res, 5) def CheckScriptSyntaxError(self): con = sqlite.connect(":memory:") cur = con.cursor() with self.assertRaises(sqlite.OperationalError): cur.executescript("create table test(x); asdf; create table test2(x)") def CheckScriptErrorNormal(self): con = sqlite.connect(":memory:") cur = con.cursor() with self.assertRaises(sqlite.OperationalError): cur.executescript("create table test(sadfsadfdsa); select foo from hurz;") def CheckCursorExecutescriptAsBytes(self): con = sqlite.connect(":memory:") cur = con.cursor() with self.assertRaises(ValueError) as cm: cur.executescript(b"create table test(foo); insert into test(foo) values (5);") self.assertEqual(str(cm.exception), 'script argument must be unicode.') def CheckConnectionExecute(self): con = sqlite.connect(":memory:") result = con.execute("select 5").fetchone()[0] self.assertEqual(result, 5, "Basic test of Connection.execute") def CheckConnectionExecutemany(self): con = sqlite.connect(":memory:") con.execute("create table test(foo)") con.executemany("insert into test(foo) values (?)", [(3,), (4,)]) result = con.execute("select foo from test order by foo").fetchall() self.assertEqual(result[0][0], 3, "Basic test of Connection.executemany") self.assertEqual(result[1][0], 4, "Basic test of Connection.executemany") def CheckConnectionExecutescript(self): con = sqlite.connect(":memory:") con.executescript("create table test(foo); insert into test(foo) values (5);") result = con.execute("select foo from test").fetchone()[0] self.assertEqual(result, 5, "Basic test of Connection.executescript") class ClosedConTests(unittest.TestCase): def CheckClosedConCursor(self): con = sqlite.connect(":memory:") con.close() with self.assertRaises(sqlite.ProgrammingError): cur = con.cursor() def CheckClosedConCommit(self): con = sqlite.connect(":memory:") con.close() with self.assertRaises(sqlite.ProgrammingError): con.commit() def CheckClosedConRollback(self): con = sqlite.connect(":memory:") con.close() with self.assertRaises(sqlite.ProgrammingError): con.rollback() def CheckClosedCurExecute(self): con = sqlite.connect(":memory:") cur = con.cursor() con.close() with self.assertRaises(sqlite.ProgrammingError): cur.execute("select 4") def CheckClosedCreateFunction(self): con = sqlite.connect(":memory:") con.close() def f(x): return 17 with self.assertRaises(sqlite.ProgrammingError): con.create_function("foo", 1, f) def CheckClosedCreateAggregate(self): con = sqlite.connect(":memory:") con.close() class Agg: def __init__(self): pass def step(self, x): pass def finalize(self): return 17 with self.assertRaises(sqlite.ProgrammingError): con.create_aggregate("foo", 1, Agg) def CheckClosedSetAuthorizer(self): con = sqlite.connect(":memory:") con.close() def authorizer(*args): return sqlite.DENY with self.assertRaises(sqlite.ProgrammingError): con.set_authorizer(authorizer) def CheckClosedSetProgressCallback(self): con = sqlite.connect(":memory:") con.close() def progress(): pass with self.assertRaises(sqlite.ProgrammingError): con.set_progress_handler(progress, 100) def CheckClosedCall(self): con = sqlite.connect(":memory:") con.close() with self.assertRaises(sqlite.ProgrammingError): con() class ClosedCurTests(unittest.TestCase): def CheckClosed(self): con = sqlite.connect(":memory:") cur = con.cursor() cur.close() for method_name in ("execute", "executemany", "executescript", "fetchall", "fetchmany", "fetchone"): if method_name in ("execute", "executescript"): params = ("select 4 union select 5",) elif method_name == "executemany": params = ("insert into foo(bar) values (?)", [(3,), (4,)]) else: params = [] with self.assertRaises(sqlite.ProgrammingError): method = getattr(cur, method_name) method(*params) class SqliteOnConflictTests(unittest.TestCase): """ Tests for SQLite's "insert on conflict" feature. See https://www.sqlite.org/lang_conflict.html for details. """ def setUp(self): self.cx = sqlite.connect(":memory:") self.cu = self.cx.cursor() self.cu.execute(""" CREATE TABLE test( id INTEGER PRIMARY KEY, name TEXT, unique_name TEXT UNIQUE ); """) def tearDown(self): self.cu.close() self.cx.close() def CheckOnConflictRollbackWithExplicitTransaction(self): self.cx.isolation_level = None # autocommit mode self.cu = self.cx.cursor() # Start an explicit transaction. self.cu.execute("BEGIN") self.cu.execute("INSERT INTO test(name) VALUES ('abort_test')") self.cu.execute("INSERT OR ROLLBACK INTO test(unique_name) VALUES ('foo')") with self.assertRaises(sqlite.IntegrityError): self.cu.execute("INSERT OR ROLLBACK INTO test(unique_name) VALUES ('foo')") # Use connection to commit. self.cx.commit() self.cu.execute("SELECT name, unique_name from test") # Transaction should have rolled back and nothing should be in table. self.assertEqual(self.cu.fetchall(), []) def CheckOnConflictAbortRaisesWithExplicitTransactions(self): # Abort cancels the current sql statement but doesn't change anything # about the current transaction. self.cx.isolation_level = None # autocommit mode self.cu = self.cx.cursor() # Start an explicit transaction. self.cu.execute("BEGIN") self.cu.execute("INSERT INTO test(name) VALUES ('abort_test')") self.cu.execute("INSERT OR ABORT INTO test(unique_name) VALUES ('foo')") with self.assertRaises(sqlite.IntegrityError): self.cu.execute("INSERT OR ABORT INTO test(unique_name) VALUES ('foo')") self.cx.commit() self.cu.execute("SELECT name, unique_name FROM test") # Expect the first two inserts to work, third to do nothing. self.assertEqual(self.cu.fetchall(), [('abort_test', None), (None, 'foo',)]) def CheckOnConflictRollbackWithoutTransaction(self): # Start of implicit transaction self.cu.execute("INSERT INTO test(name) VALUES ('abort_test')") self.cu.execute("INSERT OR ROLLBACK INTO test(unique_name) VALUES ('foo')") with self.assertRaises(sqlite.IntegrityError): self.cu.execute("INSERT OR ROLLBACK INTO test(unique_name) VALUES ('foo')") self.cu.execute("SELECT name, unique_name FROM test") # Implicit transaction is rolled back on error. self.assertEqual(self.cu.fetchall(), []) def CheckOnConflictAbortRaisesWithoutTransactions(self): # Abort cancels the current sql statement but doesn't change anything # about the current transaction. self.cu.execute("INSERT INTO test(name) VALUES ('abort_test')") self.cu.execute("INSERT OR ABORT INTO test(unique_name) VALUES ('foo')") with self.assertRaises(sqlite.IntegrityError): self.cu.execute("INSERT OR ABORT INTO test(unique_name) VALUES ('foo')") # Make sure all other values were inserted. self.cu.execute("SELECT name, unique_name FROM test") self.assertEqual(self.cu.fetchall(), [('abort_test', None), (None, 'foo',)]) def CheckOnConflictFail(self): self.cu.execute("INSERT OR FAIL INTO test(unique_name) VALUES ('foo')") with self.assertRaises(sqlite.IntegrityError): self.cu.execute("INSERT OR FAIL INTO test(unique_name) VALUES ('foo')") self.assertEqual(self.cu.fetchall(), []) def CheckOnConflictIgnore(self): self.cu.execute("INSERT OR IGNORE INTO test(unique_name) VALUES ('foo')") # Nothing should happen. self.cu.execute("INSERT OR IGNORE INTO test(unique_name) VALUES ('foo')") self.cu.execute("SELECT unique_name FROM test") self.assertEqual(self.cu.fetchall(), [('foo',)]) def CheckOnConflictReplace(self): self.cu.execute("INSERT OR REPLACE INTO test(name, unique_name) VALUES ('Data!', 'foo')") # There shouldn't be an IntegrityError exception. self.cu.execute("INSERT OR REPLACE INTO test(name, unique_name) VALUES ('Very different data!', 'foo')") self.cu.execute("SELECT name, unique_name FROM test") self.assertEqual(self.cu.fetchall(), [('Very different data!', 'foo')]) class MultiprocessTests(unittest.TestCase): CONNECTION_TIMEOUT = SHORT_TIMEOUT / 1000. # Defaults to 30 ms def tearDown(self): unlink(TESTFN) def test_ctx_mgr_rollback_if_commit_failed(self): # bpo-27334: ctx manager does not rollback if commit fails SCRIPT = f"""if 1: import sqlite3 def wait(): print("started") assert "database is locked" in input() cx = sqlite3.connect("{TESTFN}", timeout={self.CONNECTION_TIMEOUT}) cx.create_function("wait", 0, wait) with cx: cx.execute("create table t(t)") try: # execute two transactions; both will try to lock the db cx.executescript(''' -- start a transaction and wait for parent begin transaction; select * from t; select wait(); rollback; -- start a new transaction; would fail if parent holds lock begin transaction; select * from t; rollback; ''') finally: cx.close() """ # spawn child process proc = subprocess.Popen( [sys.executable, "-c", SCRIPT], encoding="utf-8", bufsize=0, stdin=subprocess.PIPE, stdout=subprocess.PIPE, ) self.addCleanup(proc.communicate) # wait for child process to start self.assertEqual("started", proc.stdout.readline().strip()) cx = sqlite.connect(TESTFN, timeout=self.CONNECTION_TIMEOUT) try: # context manager should correctly release the db lock with cx: cx.execute("insert into t values('test')") except sqlite.OperationalError as exc: proc.stdin.write(str(exc)) else: proc.stdin.write("no error") finally: cx.close() # terminate child process self.assertIsNone(proc.returncode) try: proc.communicate(input="end", timeout=SHORT_TIMEOUT) except subprocess.TimeoutExpired: proc.kill() proc.communicate() raise self.assertEqual(proc.returncode, 0) def suite(): module_suite = unittest.makeSuite(ModuleTests, "Check") connection_suite = unittest.makeSuite(ConnectionTests, "Check") cursor_suite = unittest.makeSuite(CursorTests, "Check") thread_suite = unittest.makeSuite(ThreadTests, "Check") constructor_suite = unittest.makeSuite(ConstructorTests, "Check") ext_suite = unittest.makeSuite(ExtensionTests, "Check") closed_con_suite = unittest.makeSuite(ClosedConTests, "Check") closed_cur_suite = unittest.makeSuite(ClosedCurTests, "Check") on_conflict_suite = unittest.makeSuite(SqliteOnConflictTests, "Check") uninit_con_suite = unittest.makeSuite(UninitialisedConnectionTests) multiproc_con_suite = unittest.makeSuite(MultiprocessTests) return unittest.TestSuite(( module_suite, connection_suite, cursor_suite, thread_suite, constructor_suite, ext_suite, closed_con_suite, closed_cur_suite, on_conflict_suite, uninit_con_suite, multiproc_con_suite, )) def test(): runner = unittest.TextTestRunner() runner.run(suite()) if __name__ == "__main__": test()