ÿØÿà JFIF    ÿÛ „  ( %"1!%)+...383,7(-.+  -+++--++++---+-+-----+---------------+---+-++7-----ÿÀ  ß â" ÿÄ     ÿÄ H    !1AQaq"‘¡2B±ÁÑð#R“Ò Tbr‚²á3csƒ’ÂñDS¢³$CÿÄ   ÿÄ %  !1AQa"23‘ÿÚ   ? ôÿ ¨pŸªáÿ —åYõõ\?àÒü©ŠÄï¨pŸªáÿ —åYõõ\?àÓü©ŠÄá 0Ÿªáÿ Ÿå[úƒ ú®ði~TÁbqÐ8OÕpÿ ƒOò¤Oè`–RÂáœá™êi€ßÉ< FtŸI“öÌ8úDf´°å}“¾œ6  öFá°y¥jñÇh†ˆ¢ã/ÃÐ:ªcÈ "Y¡ðÑl>ÿ ”ÏËte:qž\oäŠe÷󲍷˜HT4&ÿ ÓÐü6ö®¿øþßèô Ÿ•7Ñi’•j|“ñì>b…þS?*Óôÿ ÓÐü*h¥£ír¶ü UãSç‚Ÿ[AÐaè[ûª•õ&õj?†Éö+EzP—WeÒírJFt ‘BŒ†Ï‡%#tE Øz ¥OÛ«!1›üä±Í™%ºÍãö]°î(–:@<‹ŒÊö×òÆt¦ãº+‡¦%ÌÁ²h´OƒJŒtMÜ>ÀÜÊw3Y´•牋4ǍýʏTì>œú=Íwhyë,¾Ôò×õ¿ßÊa»«þˆѪQ|%6ž™A õ%:øj<>É—ÿ Å_ˆCbõ¥š±ý¯Ýƒï…¶|RëócÍf溪“t.СøTÿ *Ä¿-{†çàczůŽ_–^XþŒ±miB[X±d 1,é”zEù»& î9gœf™9Ð'.;—™i}!ôšåîqêÛ٤ёý£½ÆA–àôe"A$˝Úsäÿ ÷Û #°xŸëí(l »ý3—¥5m! rt`†0~'j2(]S¦¦kv,ÚÇ l¦øJA£Šƒ J3E8ÙiŽ:cÉžúeZ°€¯\®kÖ(79«Ž:¯X”¾³Š&¡* ….‰Ž(ÜíŸ2¥ª‡×Hi²TF¤ò[¨íÈRëÉ䢍mgÑ.Ÿ<öäS0í„ǹÁU´f#Vß;Õ–…P@3ío<ä-±»Ž.L|kªÀê›fÂ6@»eu‚|ÓaÞÆŸ…¨ááå>åŠ?cKü6ùTÍÆ”†sĤÚ;H2RÚ†õ\Ö·Ÿn'¾ ñ#ºI¤Å´%çÁ­‚â7›‹qT3Iï¨ÖÚ5I7Ë!ÅOóŸ¶øÝñØôת¦$Tcö‘[«Ö³šÒ';Aþ ¸èíg A2Z"i¸vdÄ÷.iõ®§)¿]¤À†–‡É&ä{V¶iŽ”.Ó×Õÿ û?h¬Mt–íª[ÿ Ñÿ ÌV(í}=ibÔ¡›¥¢±b Lô¥‡piη_Z<‡z§èŒ)iÖwiÇ 2hÙ3·=’d÷8éŽ1¦¸c¤µ€7›7Ø ð\á)} ¹fËí›pAÃL%âc2 í§æQz¿;T8sæ°qø)QFMð‰XŒÂ±N¢aF¨…8¯!U  Z©RÊ ÖPVÄÀÍin™Ì-GˆªÅËŠ›•zË}º±ŽÍFò¹}Uw×#ä5B¤{î}Ð<ÙD é©¤&‡ïDbàÁôMÁ." ¤‡ú*õ'VŽ|¼´Úgllº¼klz[Æüï÷Aób‡Eÿ dÑ»Xx9ÃÜ£ÁT/`¼¸vI±Ýµ·Ë‚“G³þ*Ÿû´r|*}<¨îºœ @¦mÄ’M¹”.œ«Y–|6ÏU¤jç¥ÕÞqO ˜kDÆÁ¨5ÿ š;Њ¦¦€GÙk \ –Þ=â¼=SͧµªS°ÚÍpÜãQűÀõ¬?ÃÁ1Ñ•õZà?hóœ€ L¦l{Y*K˜Ù›zc˜–ˆâ ø+¾ ­-Ök¥%ùEÜA'}ˆ><ÊIè“bpÍ/qÞâvoX€w,\úªò6Z[XdÒæ­@Ö—€$òJí#é>'°Ú ôª˜<)4ryÙ£|óAÅn5žêŸyÒäMÝ2{"}‰–¤l÷ûWX\l¾Á¸góÉOÔ /óñB¤f¸çñ[.P˜ZsÊË*ßT܈§QN¢’¡¨§V¼(Üù*eÕ“”5T¨‹Âê¥FŒã½Dü[8'Ò¥a…Ú¶k7a *•›¼'Ò·\8¨ª\@\õ¢¦íq+DÙrmÎ…_ªæ»ŠÓœ¡¯’Ré9MÅ×D™lælffc+ŒÑ,ý™ÿ ¯þǤ=Å’Á7µ÷ÚÛ/“Ü€ñýã¼àí¾ÕÑ+ƒ,uµMâÀÄbm:ÒÎPæ{˜Gz[ƒ¯«® KHà`ßØŠéí¯P8Aq.C‰ à€kòpj´kN¶qô€…Õ,ÜNŠª-­{Zö’æû44‰sŽè‰îVíRœÕm" 6?³D9¡ÇTíÅꋇ`4«¸ÝÁô ï’ýorqКÇZ«x4Žâéþuïf¹µö[P ,Q£éaX±`PÉÍZ ¸äYúg üAx ’6Lê‚xÝÓ*äQ  Ï’¨hÍ =²,6ï#rÃ<¯–£»ƒ‹,–ê•€ aÛsñ'%Æ"®ÛüìBᝠHÚ3ß°©$“XnœÖ’î2ËTeûìxîß ¦å¿çÉ ðK§þ{‘t‚Ϋ¬jéîZ[ ”š7L¥4VÚCE×]m¤Øy”ä4-dz£œ§¸x.*ãÊÊ b÷•h:©‡¦s`BTÁRû¾g⻩‹jø sF¢àJøFl‘È•Xᓁà~*j¯ +(ÚÕ6-£¯÷GŠØy‚<Ç’.F‹Hœw(+)ÜÜâÈzÄäT§FߘãÏ;DmVœ3Àu@mÚüXÝü•3B¨òÌÁÛ<·ÃÜ z,Ì@õÅ·d2]ü8s÷IôÞ¯^Ç9¢u„~ëAŸï4«M? K]­ÅàPl@s_ p:°¬ZR”´›JC[CS.h‹ƒïËœ«Æ]–÷ó‚wR×k7X‰k›‘´ù¦=¡«‰¨¨Â')—71ó’c‡Ðúµ `é.{§p¹ój\Ž{1h{o±Ý=áUÊïGÖŒõ–-BÄm+AZX¶¡ ïHðæ¥JmÙ;…ä¡Ÿˆ¦ ° äšiÉg«$üMk5¤L“’çÊvïâï ,=f“"íἊ5ô¬x6{ɏžID0e¸vçmi'︧ºð9$ò¹÷*£’9ÿ ²TÔ…×>JV¥}Œ}$p[bÔ®*[jzS*8 ”·T›Í–ñUîƒwo$áè=LT™ç—~ô·¤ÈÚ$榍q‰„+´kFm)ž‹©i–ËqÞŠ‰à¶ü( ‚•§ •°ò·‡#5ª•µÊ﯅¡X¨šÁ*F#TXJÊ ušJVÍ&=iÄs1‚3•'fý§5Ñ<=[íÞ­ PÚ;ѱÌ_~Ä££8rÞ ²w;’hDT°>ÈG¬8Á²ÚzŽ®ò®qZcqJêäÞ-ö[ܘbň±çb“Š¶31²n×iƒðÕ;1¶þÉ ªX‰,ßqÏ$>•î íZ¥Z 1{ç൵+ƒÕµ¥°T$§K]á»Ûï*·¤tMI’ÂZbŽÕiÒ˜}bÓ0£ª5›¨ [5Ž^ÝœWøÂÝh° ¢OWun£¤5 a2Z.G2³YL]jåtì”ä ÁÓ‘%"©<ÔúÊ°sº UZvä‡ÄiÆÒM .÷V·™ø#kèýiíÌ–ª)µT[)BˆõÑ xB¾B€ÖT¨.¥~ð@VĶr#¸ü*åZNDŽH;âi ],©£öØpù(šºãö¼T.uCê•4@ÿ GÕÛ)Cx›®0ø#:ÏðFÒbR\(€€Ä®fã4Þ‰Fä¯HXƒÅ,†öEÑÔÜ]Öv²?tLÃvBY£ú6Êu5ÅAQ³1‘’¬x–HŒÐ‡ ^ ¸KwJôÖŽ5×CÚ¨vÜ«/B0$×k°=ðbÇ(Ï)w±A†Á† 11Í=èQšµ626ŒÜ/`G«µ<}—-Ö7KEHÈÉðóȤmݱû±·ø«Snmá=“ä«šmݱŸ¡¶~ó·“äUóJæúòB|E LêŽy´jDÔ$G¢þÐñ7óR8ýÒ…Ç› WVe#·Ÿ p·Fx~•ݤF÷0Èÿ K¯æS<6’¡WШ; ´ÿ ¥Êø\Òuî†åÝ–VNœkÒ7oòX¨Á­Ø÷FÎÑä±g÷ÿ M~Çî=p,X´ ÝÌÚÅ‹’ÃjÖ.ØöÏñ qïQ¤ÓZE†° =6·]܈ s¸>v•Ž^Ý\wq9r‰Î\¸¡kURÒ$­*‹Nq?Þª*!sŠÆ:TU_u±T+øX¡ ®¹¡,ÄâÃBTsÜ$Ø›4m椴zÜK]’’›Pƒ @€#â˜`é¹=I‡fiV•Ôî“nRm+µFPOhÍ0B£ €+¬5c v•:P'ÒyÎ ‰V~‚Ó†ÖuókDoh$å\*ö%Ю=£«…aȼ½÷Û.-½VŒŠ¼'lyî±1¬3ó#ÞE¿ÔS¤gV£m›=§\û"—WU¤ÚǼÿ ÂnÁGŒÃ ‚õN D³õNÚíŒÕ;HôyÄÈ©P¹Ä{:?R‘Ô¨âF÷ø£bÅó® JS|‚R÷ivýáâ€Æé¡è³´IئÑT!§˜•Øª‚¬â@q€wnïCWÄ@JU€ê¯m6]Ï:£âx'+ÒðXvÓ¦Úm=–´7œ $ì“B£~p%ÕŸUþ« N@¼üï~w˜ñø5®—'Ôe»¤5ã//€ž~‰Tþ›Å7•#¤× Íö pÄ$ùeåì*«ÓŠEØWEÈsßg ¦ûvžSsLpºÊW–âµEWöˬH; ™!CYõZ ÃÄf æ#1W. \uWâ\,\Çf j’<qTbên›Î[vxx£ë 'ö¨1›˜ÀM¼Pÿ H)ƒêêŒA7s,|F“ 꺸k³9Ìö*ç®;Ö!Ö$Eiž•¹ÒÚ†ýóéÝû¾ÕS®ó$’NÝäŸz¤5r¦ãÄÃD÷Üø!°ø‡Ô&@m™Ì^Ãä­d q5Lnÿ N;.6½·N|#ä"1Nƒx“ã<3('&ñßt  ~ªu”1Tb㫨9ê–›–bìd$ߣ=#ÕãÒmU¯eí$EFù5ýYô櫨æ왍Ǘ±ssM]·á¿0ÕåJRÓªîiƒ+O58ÖñªŠÒx" \µâá¨i’¤i —Ö ” M+M¤ë9‚‰A¦°Qõ¾ßøK~¼Ã‘g…Ö´~÷Ï[3GUœÒ½#…kàÔ®Ò”‰³·dWV‰IP‰Ú8u¹”E ÖqLj¾êÕCBš{A^Âß;–¨`¯¬ìö ˼ ×tìø.tƐm*n¨y4o&Àx¥n¦×î‡aupáÛj8¿m›è¶ã!o½;ß0y^ý×^EÑ¿ÒjzŒ­)vÚÑnÄL …^ªô× ‡—‚3k Îý­hï]içå–îÏ*÷ñþ»Ô CÒjøjÍznˆ´ ¹#b'Fô‹ ‰v¥'’à'T´ƒHýÍ%M‰ ƒ&ÆÇŒï1 ‘ –Þ ‰i¬s žR-Ÿ kŠ¬á¬7:þ 0ŒÅÒÕ/aÙ¬ÃÝ#Úøœ ©aiVc‰. ¹¦ãµ” ›Yg¦›ÆÎýº°f³7ƒhá·¸­}&D9¡ÂsÉÙÞèŠõØàC™¨ñbFC|´Ü(ŸƒÚÒ-%»'a Ì¿)ËÇn¿úÿ ÞŽX…4ÊÅH^ôΑí@ù¹Eh¶“L8Çjù ¼ÎåVªóR©Ï5uà V4lZß®=€xÖŸ–ÑÈ ÷”¨°¾__yM1tÉ?uÆþIkÄgæ@þ[¢†°XÃJ£j·:nkÅ¢u ‘}âGzö­/IµèŠ¬¼48q¦F°ŽR¼=ûì{´¯RýicS ÕÛ íNtÍÙï£,w4rêì®»~x(©Uñ§#Ñ&œÕ¤>ÎåÍÓ9’Ö{9eV­[Öjâ²ãu]˜å2›qÑšÕJç0€sÄ|Êëè0튔bÁ>“{×_F`Ø©ºê:µä,v¤ðfc1±"«ÔÍän1#=· Âøv~H½ÐßA¾¿Ü€Óš]Õ; I¾÷ç‚Qi†î¹9ywÔKG˜áñ zQY—§ÃÕZ07§X‚ Áh;ÁM)iÌCH-¯T‘ë|A0{Ò½LÚ–TâÖkÜ’dÀ“rmm»”جPF³ÖcbE§T€ÒxKºû’Ó®7±²(\4ŽÃ¸Uu@j™yĵ;³µ!Á¢b.W¤=mõ´êµK k ¸K^ÜÛ#p*Ü14qkZç5ïë †°5Ï%ÍÛ<Õ¤×Ô¥ê†C Õ´¼ú$ƒÖ“”]Ù¬qÞÚ[4©ý!ûÏ—Áb쳐XµA¬â~`›Çr¸8ìùÝ䫦<>ä÷«?xs´ÇÑ /á;¹øüÊÈÙà{"@Žïzâ¬[âß‚ U_<ÇŸ½4èN˜ú61®qŠu ¦þF£»äJ_ˆÙÎ~ ÞAã–Ý„Ï—rŠD;xTž‘ô`É«…suãO`?³à™ô Lý#Íc5öoæØ‚y´´÷«ZR§<&JÇ+éâô´€i!Àˆ0æAoàðLèÖ-2ŸõW.’t^–(KÁmHµV@xÜÇy®Ñø­â^:Ú3w· 7½¹°ñ¸â¹®:',«Mœ—n­Á+Ãbš LÈ‘ÄnRÓÅœ%¦²‰¨ùQ:¤f‚ "PÕtô¸…cæl…&˜Ú˜Ôkv‹ž+vŠ,=¢v­6—Xy*¥t£«<™:“aîϲ=¦6rO]XI¿Œ÷¤zÚ­›¶ 6÷”w\d ü~v®ˆÌk«^m<ÿ ¢‰Õ\)ùºŽ;… lîÙÅEŠ®cѾ@vnMÏ,¼“ñ•ŽBxðÃzãÇç%3ˆ"}Ù•Åî> BÉú;Ò]V+P˜F_´ßé> Øše|ï‡ÄOmFæÇ ãqÞ$/xÐx­z`ï9"œÜij‚!7.\Td…9M‡•iŽ‹¾‘50ÞŽn¥ß4ÉôO ¹*í^QêËÜÇÌ8=Ž§s‰'ÂëÙ«á%Pú[O †ÅP¯VsŽ°.‰,kc¶ ¬A9n˜XÎ-ÞšN["¹QÕ‰ƒMýÁߺXJæÍaLj¾×Ãmã¾ãÚ uñÒþåQô¦¥ /ÄUx:‚ÍÜ’ Đ©ØÝ3V¨‰ÕnÐ6ó*óúK­«…c ¯U òhsý­jóÔj#,ímŒRµ«lbïUTŒÑ8†Ä0œÏr`ð¡¬É Ї ë"À² ™ 6¥ f¶ ¢ÚoܱԷ-<Àî)†a¶ž'Ú»¨TXqØæ¶÷YÄHy˜9ÈIW­YÀuMFë ºÏ’AqÌ4·/Ú †ô'i$øä­=Ä Ý|öK×40è|È6p‘0§)o¥ctî§H+CA-“ xØ|ÐXŠç l8íºð3Ø:³¤¬KX¯UÿÙ import sys import unittest from supervisor import xmlrpc from supervisor.compat import StringIO from supervisor.compat import xmlrpclib from supervisor.supervisorctl import LSBInitExitStatuses, LSBStatusExitStatuses from supervisor.tests.base import DummyRPCServer class fgthread_Tests(unittest.TestCase): def _getTargetClass(self): from supervisor.supervisorctl import fgthread return fgthread def _makeOne(self, program, ctl): return self._getTargetClass()(program, ctl) def test_ctor(self): options = DummyClientOptions() ctl = DummyController(options) inst = self._makeOne(None, ctl) self.assertEqual(inst.killed, False) def test_globaltrace_call(self): options = DummyClientOptions() ctl = DummyController(options) inst = self._makeOne(None, ctl) result = inst.globaltrace(None, 'call', None) self.assertEqual(result, inst.localtrace) def test_globaltrace_noncall(self): options = DummyClientOptions() ctl = DummyController(options) inst = self._makeOne(None, ctl) result = inst.globaltrace(None, None, None) self.assertEqual(result, None) def test_localtrace_killed_whyline(self): options = DummyClientOptions() ctl = DummyController(options) inst = self._makeOne(None, ctl) inst.killed = True try: inst.localtrace(None, 'line', None) except SystemExit as e: self.assertEqual(e.code, None) else: self.fail("No exception thrown. Excepted SystemExit") def test_localtrace_killed_not_whyline(self): options = DummyClientOptions() ctl = DummyController(options) inst = self._makeOne(None, ctl) inst.killed = True result = inst.localtrace(None, None, None) self.assertEqual(result, inst.localtrace) def test_kill(self): options = DummyClientOptions() ctl = DummyController(options) inst = self._makeOne(None, ctl) inst.killed = True class DummyCloseable(object): def close(self): self.closed = True inst.output_handler = DummyCloseable() inst.error_handler = DummyCloseable() inst.kill() self.assertTrue(inst.killed) self.assertTrue(inst.output_handler.closed) self.assertTrue(inst.error_handler.closed) class ControllerTests(unittest.TestCase): def _getTargetClass(self): from supervisor.supervisorctl import Controller return Controller def _makeOne(self, options): return self._getTargetClass()(options) def test_ctor(self): options = DummyClientOptions() controller = self._makeOne(options) self.assertEqual(controller.prompt, options.prompt + '> ') def test__upcheck(self): options = DummyClientOptions() controller = self._makeOne(options) result = controller.upcheck() self.assertEqual(result, True) def test__upcheck_wrong_server_version(self): options = DummyClientOptions() options._server.supervisor.getVersion = lambda *x: '1.0' controller = self._makeOne(options) controller.stdout = StringIO() result = controller.upcheck() self.assertEqual(result, False) self.assertEqual(controller.stdout.getvalue(), 'Sorry, this version of supervisorctl expects' ' to talk to a server with API version 3.0, but' ' the remote version is 1.0.\n') def test__upcheck_unknown_method(self): options = DummyClientOptions() from supervisor.xmlrpc import Faults def getVersion(): raise xmlrpclib.Fault(Faults.UNKNOWN_METHOD, 'duh') options._server.supervisor.getVersion = getVersion controller = self._makeOne(options) controller.stdout = StringIO() result = controller.upcheck() self.assertEqual(result, False) self.assertEqual(controller.stdout.getvalue(), 'Sorry, supervisord responded but did not recognize' ' the supervisor namespace commands that' ' supervisorctl uses to control it. Please check' ' that the [rpcinterface:supervisor] section is' ' enabled in the configuration file' ' (see sample.conf).\n') def test__upcheck_reraises_other_xmlrpc_faults(self): options = DummyClientOptions() from supervisor.xmlrpc import Faults def f(*arg, **kw): raise xmlrpclib.Fault(Faults.FAILED, '') options._server.supervisor.getVersion = f controller = self._makeOne(options) controller.stdout = StringIO() self.assertRaises(xmlrpclib.Fault, controller.upcheck) self.assertEqual(controller.exitstatus, LSBInitExitStatuses.GENERIC) def test__upcheck_catches_socket_error_ECONNREFUSED(self): options = DummyClientOptions() import socket import errno def raise_fault(*arg, **kw): raise socket.error(errno.ECONNREFUSED, 'nobody home') options._server.supervisor.getVersion = raise_fault controller = self._makeOne(options) controller.stdout = StringIO() result = controller.upcheck() self.assertEqual(result, False) output = controller.stdout.getvalue() self.assertTrue('refused connection' in output) self.assertEqual(controller.exitstatus, LSBInitExitStatuses.INSUFFICIENT_PRIVILEGES) def test__upcheck_catches_socket_error_ENOENT(self): options = DummyClientOptions() import socket import errno def raise_fault(*arg, **kw): raise socket.error(errno.ENOENT, 'nobody home') options._server.supervisor.getVersion = raise_fault controller = self._makeOne(options) controller.stdout = StringIO() result = controller.upcheck() self.assertEqual(result, False) output = controller.stdout.getvalue() self.assertTrue('no such file' in output) self.assertEqual(controller.exitstatus, LSBInitExitStatuses.NOT_RUNNING) def test__upcheck_reraises_other_socket_faults(self): options = DummyClientOptions() import socket import errno def f(*arg, **kw): raise socket.error(errno.EBADF, '') options._server.supervisor.getVersion = f controller = self._makeOne(options) controller.stdout = StringIO() self.assertRaises(socket.error, controller.upcheck) def test_onecmd(self): options = DummyClientOptions() controller = self._makeOne(options) controller.stdout = StringIO() plugin = DummyPlugin() controller.options.plugins = (plugin,) result = controller.onecmd('help') self.assertEqual(result, None) self.assertEqual(plugin.helped, True) def test_onecmd_empty_does_not_repeat_previous_cmd(self): options = DummyClientOptions() controller = self._makeOne(options) controller.stdout = StringIO() plugin = DummyPlugin() controller.options.plugins = (plugin,) plugin.helped = False controller.onecmd('help') self.assertTrue(plugin.helped) plugin.helped = False controller.onecmd('') self.assertFalse(plugin.helped) def test_onecmd_clears_completion_cache(self): options = DummyClientOptions() controller = self._makeOne(options) controller.stdout = StringIO() controller._complete_info = {} controller.onecmd('help') self.assertEqual(controller._complete_info, None) def test_onecmd_bad_command_error(self): options = DummyClientOptions() controller = self._makeOne(options) controller.stdout = StringIO() controller.onecmd("badcmd") self.assertEqual(controller.stdout.getvalue(), "*** Unknown syntax: badcmd\n") self.assertEqual(controller.exitstatus, LSBInitExitStatuses.GENERIC) def test_complete_action_empty(self): options = DummyClientOptions() controller = self._makeOne(options) controller.stdout=StringIO() controller.vocab = ['help'] result = controller.complete('', 0, line='') self.assertEqual(result, 'help ') result = controller.complete('', 1, line='') self.assertEqual(result, None) def test_complete_action_partial(self): options = DummyClientOptions() controller = self._makeOne(options) controller.stdout=StringIO() controller.vocab = ['help'] result = controller.complete('h', 0, line='h') self.assertEqual(result, 'help ') result = controller.complete('h', 1, line='h') self.assertEqual(result, None) def test_complete_action_whole(self): options = DummyClientOptions() controller = self._makeOne(options) controller.stdout=StringIO() controller.vocab = ['help'] result = controller.complete('help', 0, line='help') self.assertEqual(result, 'help ') def test_complete_unknown_action_uncompletable(self): options = DummyClientOptions() controller = self._makeOne(options) controller.stdout=StringIO() result = controller.complete('bad', 0, line='bad') self.assertEqual(result, None) def test_complete_unknown_action_arg_uncompletable(self): options = DummyClientOptions() controller = self._makeOne(options) controller.stdout=StringIO() controller.vocab = ['help', 'add'] result = controller.complete('', 1, line='bad ') self.assertEqual(result, None) def test_complete_help_empty(self): options = DummyClientOptions() controller = self._makeOne(options) controller.stdout=StringIO() controller.vocab = ['help', 'start'] result = controller.complete('', 0, line='help ') self.assertEqual(result, 'help ') result = controller.complete('', 1, line='help ') self.assertEqual(result, 'start ') result = controller.complete('', 2, line='help ') self.assertEqual(result, None) def test_complete_help_action(self): options = DummyClientOptions() controller = self._makeOne(options) controller.stdout=StringIO() controller.vocab = ['help', 'start'] result = controller.complete('he', 0, line='help he') self.assertEqual(result, 'help ') result = controller.complete('he', 1, line='help he') self.assertEqual(result, None) def test_complete_start_empty(self): options = DummyClientOptions() controller = self._makeOne(options) controller.stdout=StringIO() controller.vocab = ['help', 'start'] result = controller.complete('', 0, line='start ') self.assertEqual(result, 'foo ') result = controller.complete('', 1, line='start ') self.assertEqual(result, 'bar ') result = controller.complete('', 2, line='start ') self.assertEqual(result, 'baz:baz_01 ') result = controller.complete('', 3, line='start ') self.assertEqual(result, 'baz:* ') result = controller.complete('', 4, line='start ') self.assertEqual(result, None) def test_complete_start_no_colon(self): options = DummyClientOptions() controller = self._makeOne(options) controller.stdout=StringIO() controller.vocab = ['help', 'start'] result = controller.complete('f', 0, line='start f') self.assertEqual(result, 'foo ') result = controller.complete('f', 1, line='start f') self.assertEqual(result, None) def test_complete_start_with_colon(self): options = DummyClientOptions() controller = self._makeOne(options) controller.stdout=StringIO() controller.vocab = ['help', 'start'] result = controller.complete('foo:', 0, line='start foo:') self.assertEqual(result, 'foo:foo ') result = controller.complete('foo:', 1, line='start foo:') self.assertEqual(result, 'foo:* ') result = controller.complete('foo:', 2, line='start foo:') self.assertEqual(result, None) def test_complete_start_uncompletable(self): options = DummyClientOptions() controller = self._makeOne(options) controller.stdout=StringIO() controller.vocab = ['help', 'start'] result = controller.complete('bad', 0, line='start bad') self.assertEqual(result, None) def test_complete_caches_process_info(self): options = DummyClientOptions() controller = self._makeOne(options) controller.stdout=StringIO() controller.vocab = ['help', 'start'] result = controller.complete('', 0, line='start ') self.assertNotEqual(result, None) def f(*arg, **kw): raise Exception("should not have called getAllProcessInfo") controller.options._server.supervisor.getAllProcessInfo = f controller.complete('', 1, line='start ') def test_complete_add_empty(self): options = DummyClientOptions() controller = self._makeOne(options) controller.stdout=StringIO() controller.vocab = ['help', 'add'] result = controller.complete('', 0, line='add ') self.assertEqual(result, 'foo ') result = controller.complete('', 1, line='add ') self.assertEqual(result, 'bar ') result = controller.complete('', 2, line='add ') self.assertEqual(result, 'baz ') result = controller.complete('', 3, line='add ') self.assertEqual(result, None) def test_complete_add_uncompletable(self): options = DummyClientOptions() controller = self._makeOne(options) controller.stdout=StringIO() controller.vocab = ['help', 'add'] result = controller.complete('bad', 0, line='add bad') self.assertEqual(result, None) def test_complete_add_group(self): options = DummyClientOptions() controller = self._makeOne(options) controller.stdout=StringIO() controller.vocab = ['help', 'add'] result = controller.complete('f', 0, line='add f') self.assertEqual(result, 'foo ') result = controller.complete('f', 1, line='add f') self.assertEqual(result, None) def test_complete_reload_arg_uncompletable(self): options = DummyClientOptions() controller = self._makeOne(options) controller.stdout=StringIO() controller.vocab = ['help', 'reload'] result = controller.complete('', 1, line='reload ') self.assertEqual(result, None) def test_nohelp(self): options = DummyClientOptions() controller = self._makeOne(options) self.assertEqual(controller.nohelp, '*** No help on %s') def test_do_help(self): options = DummyClientOptions() controller = self._makeOne(options) controller.stdout = StringIO() results = controller.do_help('') helpval = controller.stdout.getvalue() self.assertEqual(results, None) self.assertEqual(helpval, 'foo helped') def test_do_help_for_help(self): options = DummyClientOptions() controller = self._makeOne(options) controller.stdout = StringIO() results = controller.do_help("help") self.assertEqual(results, None) helpval = controller.stdout.getvalue() self.assertTrue("help\t\tPrint a list" in helpval) def test_get_supervisor_returns_serverproxy_supervisor_namespace(self): options = DummyClientOptions() controller = self._makeOne(options) proxy = controller.get_supervisor() expected = options.getServerProxy().supervisor self.assertEqual(proxy, expected) def test_get_server_proxy_with_no_args_returns_serverproxy(self): options = DummyClientOptions() controller = self._makeOne(options) proxy = controller.get_server_proxy() expected = options.getServerProxy() self.assertEqual(proxy, expected) def test_get_server_proxy_with_namespace_returns_that_namespace(self): options = DummyClientOptions() controller = self._makeOne(options) proxy = controller.get_server_proxy('system') expected = options.getServerProxy().system self.assertEqual(proxy, expected) def test_real_controller_initialization(self): from supervisor.options import ClientOptions args = [] # simulating starting without parameters options = ClientOptions() # No default config file search in case they would exist self.assertTrue(len(options.searchpaths) > 0) options.searchpaths = [] options.realize(args, doc=__doc__) self._makeOne(options) # should not raise class TestControllerPluginBase(unittest.TestCase): def _getTargetClass(self): from supervisor.supervisorctl import ControllerPluginBase return ControllerPluginBase def _makeOne(self, *arg, **kw): klass = self._getTargetClass() options = DummyClientOptions() ctl = DummyController(options) plugin = klass(ctl, *arg, **kw) return plugin def test_do_help_noarg(self): plugin = self._makeOne() result = plugin.do_help(None) self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), '\n') self.assertEqual(len(plugin.ctl.topics_printed), 1) topics = plugin.ctl.topics_printed[0] self.assertEqual(topics[0], 'unnamed commands (type help ):') self.assertEqual(topics[1], []) self.assertEqual(topics[2], 15) self.assertEqual(topics[3], 80) def test_do_help_witharg(self): plugin = self._makeOne() result = plugin.do_help('foo') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'no help on foo\n') self.assertEqual(len(plugin.ctl.topics_printed), 0) class TestDefaultControllerPlugin(unittest.TestCase): def _getTargetClass(self): from supervisor.supervisorctl import DefaultControllerPlugin return DefaultControllerPlugin def _makeOne(self, *arg, **kw): klass = self._getTargetClass() options = DummyClientOptions() ctl = DummyController(options) plugin = klass(ctl, *arg, **kw) return plugin def test_tail_toofewargs(self): plugin = self._makeOne() result = plugin.do_tail('') self.assertEqual(result, None) lines = plugin.ctl.stdout.getvalue().split('\n') self.assertEqual(lines[0], 'Error: too few arguments') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_tail_toomanyargs(self): plugin = self._makeOne() result = plugin.do_tail('one two three four') self.assertEqual(result, None) lines = plugin.ctl.stdout.getvalue().split('\n') self.assertEqual(lines[0], 'Error: too many arguments') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_tail_f_noprocname(self): plugin = self._makeOne() result = plugin.do_tail('-f') self.assertEqual(result, None) lines = plugin.ctl.stdout.getvalue().split('\n') self.assertEqual(lines[0], 'Error: tail requires process name') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_tail_bad_modifier(self): plugin = self._makeOne() result = plugin.do_tail('-z foo') self.assertEqual(result, None) lines = plugin.ctl.stdout.getvalue().split('\n') self.assertEqual(lines[0], 'Error: bad argument -z') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_tail_defaults(self): plugin = self._makeOne() result = plugin.do_tail('foo') self.assertEqual(result, None) lines = plugin.ctl.stdout.getvalue().split('\n') self.assertEqual(len(lines), 12) self.assertEqual(lines[0], 'stdout line') def test_tail_no_file(self): plugin = self._makeOne() result = plugin.do_tail('NO_FILE') self.assertEqual(result, None) lines = plugin.ctl.stdout.getvalue().split('\n') self.assertEqual(len(lines), 2) self.assertEqual(lines[0], 'NO_FILE: ERROR (no log file)') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_tail_failed(self): plugin = self._makeOne() result = plugin.do_tail('FAILED') self.assertEqual(result, None) lines = plugin.ctl.stdout.getvalue().split('\n') self.assertEqual(len(lines), 2) self.assertEqual(lines[0], 'FAILED: ERROR (unknown error reading log)') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_tail_bad_name(self): plugin = self._makeOne() result = plugin.do_tail('BAD_NAME') self.assertEqual(result, None) lines = plugin.ctl.stdout.getvalue().split('\n') self.assertEqual(len(lines), 2) self.assertEqual(lines[0], 'BAD_NAME: ERROR (no such process name)') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_tail_bytesmodifier(self): plugin = self._makeOne() result = plugin.do_tail('-10 foo') self.assertEqual(result, None) lines = plugin.ctl.stdout.getvalue().split('\n') self.assertEqual(len(lines), 3) self.assertEqual(lines[0], 'dout line') def test_tail_explicit_channel_stdout_nomodifier(self): plugin = self._makeOne() result = plugin.do_tail('foo stdout') self.assertEqual(result, None) lines = plugin.ctl.stdout.getvalue().split('\n') self.assertEqual(len(lines), 12) self.assertEqual(lines[0], 'stdout line') def test_tail_explicit_channel_stderr_nomodifier(self): plugin = self._makeOne() result = plugin.do_tail('foo stderr') self.assertEqual(result, None) lines = plugin.ctl.stdout.getvalue().split('\n') self.assertEqual(len(lines), 12) self.assertEqual(lines[0], 'stderr line') def test_tail_explicit_channel_unrecognized(self): plugin = self._makeOne() result = plugin.do_tail('foo fudge') self.assertEqual(result, None) value = plugin.ctl.stdout.getvalue().strip() self.assertEqual(value, "Error: bad channel 'fudge'") self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_tail_upcheck_failed(self): plugin = self._makeOne() plugin.ctl.upcheck = lambda: False called = [] def f(*arg, **kw): called.append(True) plugin.ctl.options._server.supervisor.readProcessStdoutLog = f plugin.do_tail('foo') self.assertEqual(called, []) def test_status_help(self): plugin = self._makeOne() plugin.help_status() out = plugin.ctl.stdout.getvalue() self.assertTrue("status " in out) def test_status_upcheck_failed(self): plugin = self._makeOne() plugin.ctl.upcheck = lambda: False called = [] def f(*arg, **kw): called.append(True) plugin.ctl.options._server.supervisor.getAllProcessInfo = f plugin.do_status('') self.assertEqual(called, []) def test_status_table_process_column_min_width(self): plugin = self._makeOne() result = plugin.do_status('') self.assertEqual(result, None) lines = plugin.ctl.stdout.getvalue().split("\n") self.assertEqual(lines[0].index("RUNNING"), 33) def test_status_table_process_column_expands(self): plugin = self._makeOne() options = plugin.ctl.options def f(*arg, **kw): from supervisor.states import ProcessStates return [{'name': 'foo'*50, # long name 'group':'foo', 'pid': 11, 'state': ProcessStates.RUNNING, 'statename': 'RUNNING', 'start': 0, 'stop': 0, 'spawnerr': '', 'now': 0, 'description':'foo description'}, { 'name': 'bar', # short name 'group': 'bar', 'pid': 12, 'state': ProcessStates.FATAL, 'statename': 'RUNNING', 'start': 0, 'stop': 0, 'spawnerr': '', 'now': 0, 'description': 'bar description', }] options._server.supervisor.getAllProcessInfo = f self.assertEqual(plugin.do_status(''), None) lines = plugin.ctl.stdout.getvalue().split("\n") self.assertEqual(lines[0].index("RUNNING"), 157) self.assertEqual(lines[1].index("RUNNING"), 157) def test_status_all_processes_no_arg(self): plugin = self._makeOne() result = plugin.do_status('') self.assertEqual(result, None) value = plugin.ctl.stdout.getvalue().split('\n') self.assertEqual(value[0].split(None, 2), ['foo', 'RUNNING', 'foo description']) self.assertEqual(value[1].split(None, 2), ['bar', 'FATAL', 'bar description']) self.assertEqual(value[2].split(None, 2), ['baz:baz_01', 'STOPPED', 'baz description']) self.assertEqual(plugin.ctl.exitstatus, LSBStatusExitStatuses.NOT_RUNNING) def test_status_success(self): plugin = self._makeOne() result = plugin.do_status('foo') self.assertEqual(result, None) self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.SUCCESS) value = plugin.ctl.stdout.getvalue().split('\n') self.assertEqual(value[0].split(None, 2), ['foo', 'RUNNING', 'foo description']) def test_status_unknown_process(self): plugin = self._makeOne() result = plugin.do_status('unknownprogram') self.assertEqual(result, None) value = plugin.ctl.stdout.getvalue() self.assertEqual("unknownprogram: ERROR (no such process)\n", value) self.assertEqual(plugin.ctl.exitstatus, LSBStatusExitStatuses.UNKNOWN) def test_status_all_processes_all_arg(self): plugin = self._makeOne() result = plugin.do_status('all') self.assertEqual(result, None) value = plugin.ctl.stdout.getvalue().split('\n') self.assertEqual(value[0].split(None, 2), ['foo', 'RUNNING', 'foo description']) self.assertEqual(value[1].split(None, 2), ['bar', 'FATAL', 'bar description']) self.assertEqual(value[2].split(None, 2), ['baz:baz_01', 'STOPPED', 'baz description']) self.assertEqual(plugin.ctl.exitstatus, LSBStatusExitStatuses.NOT_RUNNING) def test_status_process_name(self): plugin = self._makeOne() result = plugin.do_status('foo') self.assertEqual(result, None) value = plugin.ctl.stdout.getvalue().strip() self.assertEqual(value.split(None, 2), ['foo', 'RUNNING', 'foo description']) self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.SUCCESS) def test_status_group_name(self): plugin = self._makeOne() result = plugin.do_status('baz:*') self.assertEqual(result, None) value = plugin.ctl.stdout.getvalue().split('\n') self.assertEqual(value[0].split(None, 2), ['baz:baz_01', 'STOPPED', 'baz description']) self.assertEqual(plugin.ctl.exitstatus, LSBStatusExitStatuses.NOT_RUNNING) def test_status_mixed_names(self): plugin = self._makeOne() result = plugin.do_status('foo baz:*') self.assertEqual(result, None) value = plugin.ctl.stdout.getvalue().split('\n') self.assertEqual(value[0].split(None, 2), ['foo', 'RUNNING', 'foo description']) self.assertEqual(value[1].split(None, 2), ['baz:baz_01', 'STOPPED', 'baz description']) self.assertEqual(plugin.ctl.exitstatus, LSBStatusExitStatuses.NOT_RUNNING) def test_status_bad_group_name(self): plugin = self._makeOne() result = plugin.do_status('badgroup:*') self.assertEqual(result, None) value = plugin.ctl.stdout.getvalue().split('\n') self.assertEqual(value[0], "badgroup: ERROR (no such group)") self.assertEqual(plugin.ctl.exitstatus, LSBStatusExitStatuses.UNKNOWN) def test_status_bad_process_name(self): plugin = self._makeOne() result = plugin.do_status('badprocess') self.assertEqual(result, None) value = plugin.ctl.stdout.getvalue().split('\n') self.assertEqual(value[0], "badprocess: ERROR (no such process)") self.assertEqual(plugin.ctl.exitstatus, LSBStatusExitStatuses.UNKNOWN) def test_status_bad_process_name_with_group(self): plugin = self._makeOne() result = plugin.do_status('badgroup:badprocess') self.assertEqual(result, None) value = plugin.ctl.stdout.getvalue().split('\n') self.assertEqual(value[0], "badgroup:badprocess: " "ERROR (no such process)") self.assertEqual(plugin.ctl.exitstatus, LSBStatusExitStatuses.UNKNOWN) def test_start_help(self): plugin = self._makeOne() plugin.help_start() out = plugin.ctl.stdout.getvalue() self.assertTrue("start " in out) def test_start_fail(self): plugin = self._makeOne() result = plugin.do_start('') self.assertEqual(result, None) expected = "Error: start requires a process name" self.assertEqual(plugin.ctl.stdout.getvalue().split('\n')[0], expected) self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.INVALID_ARGS) def test_start_badname(self): plugin = self._makeOne() result = plugin.do_start('BAD_NAME') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'BAD_NAME: ERROR (no such process)\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_start_no_file(self): plugin = self._makeOne() result = plugin.do_start('NO_FILE') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'NO_FILE: ERROR (no such file)\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_start_not_executable(self): plugin = self._makeOne() result = plugin.do_start('NOT_EXECUTABLE') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'NOT_EXECUTABLE: ERROR (file is not executable)\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_start_alreadystarted(self): plugin = self._makeOne() result = plugin.do_start('ALREADY_STARTED') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'ALREADY_STARTED: ERROR (already started)\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.SUCCESS) def test_start_spawnerror(self): plugin = self._makeOne() result = plugin.do_start('SPAWN_ERROR') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'SPAWN_ERROR: ERROR (spawn error)\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.NOT_RUNNING) def test_start_abnormaltermination(self): plugin = self._makeOne() result = plugin.do_start('ABNORMAL_TERMINATION') self.assertEqual(result, None) expected = 'ABNORMAL_TERMINATION: ERROR (abnormal termination)\n' self.assertEqual(plugin.ctl.stdout.getvalue(), expected) self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.NOT_RUNNING) def test_start_one_success(self): plugin = self._makeOne() result = plugin.do_start('foo') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'foo: started\n') def test_start_one_with_group_name_success(self): plugin = self._makeOne() result = plugin.do_start('foo:foo') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'foo: started\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.SUCCESS) def test_start_many(self): plugin = self._makeOne() result = plugin.do_start('foo bar') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'foo: started\nbar: started\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.SUCCESS) def test_start_group(self): plugin = self._makeOne() result = plugin.do_start('foo:') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'foo:foo_00: started\n' 'foo:foo_01: started\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.SUCCESS) def test_start_group_bad_name(self): plugin = self._makeOne() result = plugin.do_start('BAD_NAME:') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'BAD_NAME: ERROR (no such group)\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.INVALID_ARGS) def test_start_all(self): plugin = self._makeOne() result = plugin.do_start('all') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'foo: started\n' 'foo2: started\n' 'failed_group:failed: ERROR (spawn error)\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.NOT_RUNNING) def test_start_upcheck_failed(self): plugin = self._makeOne() plugin.ctl.upcheck = lambda: False called = [] def f(*arg, **kw): called.append(True) supervisor = plugin.ctl.options._server.supervisor supervisor.startAllProcesses = f supervisor.startProcessGroup = f plugin.do_start('foo') self.assertEqual(called, []) def test_stop_help(self): plugin = self._makeOne() plugin.help_stop() out = plugin.ctl.stdout.getvalue() self.assertTrue("stop " in out) def test_stop_fail(self): plugin = self._makeOne() result = plugin.do_stop('') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue().split('\n')[0], "Error: stop requires a process name") self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_stop_badname(self): plugin = self._makeOne() result = plugin.do_stop('BAD_NAME') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'BAD_NAME: ERROR (no such process)\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_stop_notrunning(self): plugin = self._makeOne() result = plugin.do_stop('NOT_RUNNING') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'NOT_RUNNING: ERROR (not running)\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.SUCCESS) def test_stop_failed(self): plugin = self._makeOne() result = plugin.do_stop('FAILED') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'FAILED\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_stop_one_success(self): plugin = self._makeOne() result = plugin.do_stop('foo') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'foo: stopped\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.SUCCESS) def test_stop_one_with_group_name_success(self): plugin = self._makeOne() result = plugin.do_stop('foo:foo') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'foo: stopped\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.SUCCESS) def test_stop_many(self): plugin = self._makeOne() result = plugin.do_stop('foo bar') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'foo: stopped\n' 'bar: stopped\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.SUCCESS) def test_stop_group(self): plugin = self._makeOne() result = plugin.do_stop('foo:') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'foo:foo_00: stopped\n' 'foo:foo_01: stopped\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.SUCCESS) def test_stop_group_bad_name(self): plugin = self._makeOne() result = plugin.do_stop('BAD_NAME:') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'BAD_NAME: ERROR (no such group)\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_stop_all(self): plugin = self._makeOne() result = plugin.do_stop('all') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'foo: stopped\n' 'foo2: stopped\n' 'failed_group:failed: ERROR (no such process)\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_stop_upcheck_failed(self): plugin = self._makeOne() plugin.ctl.upcheck = lambda: False called = [] def f(*arg, **kw): called.append(True) supervisor = plugin.ctl.options._server.supervisor supervisor.stopAllProcesses = f supervisor.stopProcessGroup = f plugin.do_stop('foo') self.assertEqual(called, []) def test_signal_help(self): plugin = self._makeOne() plugin.help_signal() out = plugin.ctl.stdout.getvalue() self.assertTrue("signal " in out) def test_signal_fail_no_arg(self): plugin = self._makeOne() result = plugin.do_signal('') self.assertEqual(result, None) msg = 'Error: signal requires a signal name and a process name' self.assertEqual(plugin.ctl.stdout.getvalue().split('\n')[0], msg) self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_signal_fail_one_arg(self): plugin = self._makeOne() result = plugin.do_signal('hup') self.assertEqual(result, None) msg = 'Error: signal requires a signal name and a process name' self.assertEqual(plugin.ctl.stdout.getvalue().split('\n')[0], msg) self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_signal_bad_signal(self): plugin = self._makeOne() result = plugin.do_signal('BAD_SIGNAL foo') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'foo: ERROR (bad signal name)\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_signal_bad_name(self): plugin = self._makeOne() result = plugin.do_signal('HUP BAD_NAME') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'BAD_NAME: ERROR (no such process)\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_signal_bad_group(self): plugin = self._makeOne() result = plugin.do_signal('HUP BAD_NAME:') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'BAD_NAME: ERROR (no such group)\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_signal_not_running(self): plugin = self._makeOne() result = plugin.do_signal('HUP NOT_RUNNING') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'NOT_RUNNING: ERROR (not running)\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.NOT_RUNNING) def test_signal_failed(self): plugin = self._makeOne() result = plugin.do_signal('HUP FAILED') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'FAILED\n') self.assertEqual(plugin.ctl.exitstatus, 1) def test_signal_one_success(self): plugin = self._makeOne() result = plugin.do_signal('HUP foo') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'foo: signalled\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.SUCCESS) def test_signal_many(self): plugin = self._makeOne() result = plugin.do_signal('HUP foo bar') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'foo: signalled\n' 'bar: signalled\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.SUCCESS) def test_signal_group(self): plugin = self._makeOne() result = plugin.do_signal('HUP foo:') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'foo:foo_00: signalled\n' 'foo:foo_01: signalled\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.SUCCESS) def test_signal_all(self): plugin = self._makeOne() result = plugin.do_signal('HUP all') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'foo: signalled\n' 'foo2: signalled\n' 'failed_group:failed: ERROR (no such process)\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_signal_upcheck_failed(self): plugin = self._makeOne() plugin.ctl.upcheck = lambda: False called = [] def f(*arg, **kw): called.append(True) supervisor = plugin.ctl.options._server.supervisor supervisor.signalAllProcesses = f supervisor.signalProcessGroup = f plugin.do_signal('term foo') self.assertEqual(called, []) def test_restart_help(self): plugin = self._makeOne() plugin.help_restart() out = plugin.ctl.stdout.getvalue() self.assertTrue("restart " in out) def test_restart_fail(self): plugin = self._makeOne() result = plugin.do_restart('') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue().split('\n')[0], 'Error: restart requires a process name') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_restart_one(self): plugin = self._makeOne() result = plugin.do_restart('foo') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'foo: stopped\nfoo: started\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.SUCCESS) def test_restart_all(self): plugin = self._makeOne() result = plugin.do_restart('all') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'foo: stopped\nfoo2: stopped\n' 'failed_group:failed: ERROR (no such process)\n' 'foo: started\nfoo2: started\n' 'failed_group:failed: ERROR (spawn error)\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.NOT_RUNNING) def test_restart_upcheck_failed(self): plugin = self._makeOne() plugin.ctl.upcheck = lambda: False called = [] def f(*arg, **kw): called.append(True) supervisor = plugin.ctl.options._server.supervisor supervisor.stopAllProcesses = f supervisor.stopProcessGroup = f plugin.do_restart('foo') self.assertEqual(called, []) def test_clear_help(self): plugin = self._makeOne() plugin.help_clear() out = plugin.ctl.stdout.getvalue() self.assertTrue("clear " in out) def test_clear_fail(self): plugin = self._makeOne() result = plugin.do_clear('') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue().split('\n')[0], "Error: clear requires a process name") self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_clear_badname(self): plugin = self._makeOne() result = plugin.do_clear('BAD_NAME') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'BAD_NAME: ERROR (no such process)\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_clear_one_success(self): plugin = self._makeOne() result = plugin.do_clear('foo') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'foo: cleared\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.SUCCESS) def test_clear_one_with_group_success(self): plugin = self._makeOne() result = plugin.do_clear('foo:foo') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'foo: cleared\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.SUCCESS) def test_clear_many(self): plugin = self._makeOne() result = plugin.do_clear('foo bar') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'foo: cleared\nbar: cleared\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.SUCCESS) def test_clear_all(self): plugin = self._makeOne() result = plugin.do_clear('all') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'foo: cleared\n' 'foo2: cleared\n' 'failed_group:failed: ERROR (failed)\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_clear_upcheck_failed(self): plugin = self._makeOne() plugin.ctl.upcheck = lambda: False called = [] def f(*arg, **kw): called.append(True) supervisor = plugin.ctl.options._server.supervisor supervisor.clearAllProcessLogs = f supervisor.clearProcessLogs = f plugin.do_clear('foo') self.assertEqual(called, []) def test_open_help(self): plugin = self._makeOne() plugin.help_open() out = plugin.ctl.stdout.getvalue() self.assertTrue("open " in out) def test_open_fail(self): plugin = self._makeOne() result = plugin.do_open('badname') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'ERROR: url must be http:// or unix://\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_open_succeed(self): plugin = self._makeOne() result = plugin.do_open('http://localhost:9002') self.assertEqual(result, None) value = plugin.ctl.stdout.getvalue().split('\n') self.assertEqual(value[0].split(None, 2), ['foo', 'RUNNING', 'foo description']) self.assertEqual(value[1].split(None, 2), ['bar', 'FATAL', 'bar description']) self.assertEqual(value[2].split(None, 2), ['baz:baz_01', 'STOPPED', 'baz description']) self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.SUCCESS) def test_version_help(self): plugin = self._makeOne() plugin.help_version() out = plugin.ctl.stdout.getvalue() self.assertTrue("Show the version of the remote supervisord" in out) def test_version(self): plugin = self._makeOne() plugin.do_version(None) self.assertEqual(plugin.ctl.stdout.getvalue(), '3000\n') def test_version_arg(self): plugin = self._makeOne() result = plugin.do_version('bad') self.assertEqual(result, None) val = plugin.ctl.stdout.getvalue() self.assertTrue(val.startswith('Error: version accepts no arguments'), val) self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_version_upcheck_failed(self): plugin = self._makeOne() plugin.ctl.upcheck = lambda: False called = [] def f(*arg, **kw): called.append(True) plugin.ctl.options._server.supervisor.getSupervisorVersion = f plugin.do_version('') self.assertEqual(called, []) def test_reload_help(self): plugin = self._makeOne() plugin.help_reload() out = plugin.ctl.stdout.getvalue() self.assertTrue("Restart the remote supervisord" in out) def test_reload_fail(self): plugin = self._makeOne() options = plugin.ctl.options options._server.supervisor._restartable = False result = plugin.do_reload('') self.assertEqual(result, None) self.assertEqual(options._server.supervisor._restarted, False) def test_reload(self): plugin = self._makeOne() options = plugin.ctl.options result = plugin.do_reload('') self.assertEqual(result, None) self.assertEqual(options._server.supervisor._restarted, True) def test_reload_arg(self): plugin = self._makeOne() result = plugin.do_reload('bad') self.assertEqual(result, None) val = plugin.ctl.stdout.getvalue() self.assertTrue(val.startswith('Error: reload accepts no arguments'), val) self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_shutdown_help(self): plugin = self._makeOne() plugin.help_shutdown() out = plugin.ctl.stdout.getvalue() self.assertTrue("Shut the remote supervisord down" in out) def test_shutdown(self): plugin = self._makeOne() options = plugin.ctl.options result = plugin.do_shutdown('') self.assertEqual(result, None) self.assertEqual(options._server.supervisor._shutdown, True) def test_shutdown_catches_xmlrpc_fault_shutdown_state(self): plugin = self._makeOne() from supervisor import xmlrpc def raise_fault(*arg, **kw): raise xmlrpclib.Fault(xmlrpc.Faults.SHUTDOWN_STATE, 'bye') plugin.ctl.options._server.supervisor.shutdown = raise_fault result = plugin.do_shutdown('') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'ERROR: already shutting down\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.SUCCESS) def test_shutdown_reraises_other_xmlrpc_faults(self): plugin = self._makeOne() from supervisor import xmlrpc def raise_fault(*arg, **kw): raise xmlrpclib.Fault(xmlrpc.Faults.CANT_REREAD, 'ouch') plugin.ctl.options._server.supervisor.shutdown = raise_fault self.assertRaises(xmlrpclib.Fault, plugin.do_shutdown, '') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_shutdown_catches_socket_error_ECONNREFUSED(self): plugin = self._makeOne() import socket import errno def raise_fault(*arg, **kw): raise socket.error(errno.ECONNREFUSED, 'nobody home') plugin.ctl.options._server.supervisor.shutdown = raise_fault result = plugin.do_shutdown('') self.assertEqual(result, None) output = plugin.ctl.stdout.getvalue() self.assertTrue('refused connection (already shut down?)' in output) self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_shutdown_catches_socket_error_ENOENT(self): plugin = self._makeOne() import socket import errno def raise_fault(*arg, **kw): raise socket.error(errno.ENOENT, 'no file') plugin.ctl.options._server.supervisor.shutdown = raise_fault result = plugin.do_shutdown('') self.assertEqual(result, None) output = plugin.ctl.stdout.getvalue() self.assertTrue('no such file (already shut down?)' in output) self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_shutdown_reraises_other_socket_errors(self): plugin = self._makeOne() import socket import errno def raise_fault(*arg, **kw): raise socket.error(errno.EPERM, 'denied') plugin.ctl.options._server.supervisor.shutdown = raise_fault self.assertRaises(socket.error, plugin.do_shutdown, '') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test__formatChanges(self): plugin = self._makeOne() # Don't explode, plz plugin._formatChanges([['added'], ['changed'], ['removed']]) plugin._formatChanges([[], [], []]) def test_reread_help(self): plugin = self._makeOne() plugin.help_reread() out = plugin.ctl.stdout.getvalue() self.assertTrue("Reload the daemon's configuration files" in out) self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.SUCCESS) def test_reread(self): plugin = self._makeOne() calls = [] plugin._formatChanges = lambda x: calls.append(x) result = plugin.do_reread(None) self.assertEqual(result, None) self.assertEqual(calls[0], [['added'], ['changed'], ['removed']]) self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.SUCCESS) def test_reread_arg(self): plugin = self._makeOne() result = plugin.do_reread('bad') self.assertEqual(result, None) val = plugin.ctl.stdout.getvalue() self.assertTrue(val.startswith('Error: reread accepts no arguments'), val) self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_reread_cant_reread(self): plugin = self._makeOne() from supervisor import xmlrpc def reloadConfig(*arg, **kw): raise xmlrpclib.Fault(xmlrpc.Faults.CANT_REREAD, 'cant') plugin.ctl.options._server.supervisor.reloadConfig = reloadConfig plugin.do_reread(None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'ERROR: cant\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_reread_shutdown_state(self): plugin = self._makeOne() from supervisor import xmlrpc def reloadConfig(*arg, **kw): raise xmlrpclib.Fault(xmlrpc.Faults.SHUTDOWN_STATE, '') plugin.ctl.options._server.supervisor.reloadConfig = reloadConfig plugin.do_reread(None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'ERROR: supervisor shutting down\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_reread_reraises_other_faults(self): plugin = self._makeOne() from supervisor import xmlrpc def reloadConfig(*arg, **kw): raise xmlrpclib.Fault(xmlrpc.Faults.FAILED, '') plugin.ctl.options._server.supervisor.reloadConfig = reloadConfig self.assertRaises(xmlrpclib.Fault, plugin.do_reread, '') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test__formatConfigInfo(self): info = { 'group': 'group1', 'name': 'process1', 'inuse': True, 'autostart': True, 'process_prio': 999, 'group_prio': 999 } plugin = self._makeOne() result = plugin._formatConfigInfo(info) self.assertTrue('in use' in result) info = { 'group': 'group1', 'name': 'process1', 'inuse': False, 'autostart': False, 'process_prio': 999, 'group_prio': 999 } result = plugin._formatConfigInfo(info) self.assertTrue('avail' in result) def test_avail_help(self): plugin = self._makeOne() plugin.help_avail() out = plugin.ctl.stdout.getvalue() self.assertTrue("Display all configured" in out) def test_avail(self): calls = [] plugin = self._makeOne() class FakeSupervisor(object): def getAllConfigInfo(self): return [{ 'group': 'group1', 'name': 'process1', 'inuse': False, 'autostart': False, 'process_prio': 999, 'group_prio': 999 }] plugin.ctl.get_supervisor = lambda : FakeSupervisor() plugin.ctl.output = calls.append result = plugin.do_avail('') self.assertEqual(result, None) def test_avail_arg(self): plugin = self._makeOne() result = plugin.do_avail('bad') self.assertEqual(result, None) val = plugin.ctl.stdout.getvalue() self.assertTrue(val.startswith('Error: avail accepts no arguments'), val) self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_avail_shutdown_state(self): plugin = self._makeOne() supervisor = plugin.ctl.options._server.supervisor def getAllConfigInfo(): from supervisor import xmlrpc raise xmlrpclib.Fault(xmlrpc.Faults.SHUTDOWN_STATE, '') supervisor.getAllConfigInfo = getAllConfigInfo result = plugin.do_avail('') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'ERROR: supervisor shutting down\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_avail_reraises_other_faults(self): plugin = self._makeOne() supervisor = plugin.ctl.options._server.supervisor def getAllConfigInfo(): from supervisor import xmlrpc raise xmlrpclib.Fault(xmlrpc.Faults.FAILED, '') supervisor.getAllConfigInfo = getAllConfigInfo self.assertRaises(xmlrpclib.Fault, plugin.do_avail, '') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_add_help(self): plugin = self._makeOne() plugin.help_add() out = plugin.ctl.stdout.getvalue() self.assertTrue("add " in out) def test_add(self): plugin = self._makeOne() result = plugin.do_add('foo') self.assertEqual(result, None) supervisor = plugin.ctl.options._server.supervisor self.assertEqual(supervisor.processes, ['foo']) self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.SUCCESS) def test_add_already_added(self): plugin = self._makeOne() result = plugin.do_add('ALREADY_ADDED') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'ERROR: process group already active\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.SUCCESS) def test_add_bad_name(self): plugin = self._makeOne() result = plugin.do_add('BAD_NAME') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'ERROR: no such process/group: BAD_NAME\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_add_shutdown_state(self): plugin = self._makeOne() result = plugin.do_add('SHUTDOWN_STATE') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'ERROR: shutting down\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_add_reraises_other_faults(self): plugin = self._makeOne() self.assertRaises(xmlrpclib.Fault, plugin.do_add, 'FAILED') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_remove_help(self): plugin = self._makeOne() plugin.help_remove() out = plugin.ctl.stdout.getvalue() self.assertTrue("remove " in out) self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.SUCCESS) def test_remove(self): plugin = self._makeOne() supervisor = plugin.ctl.options._server.supervisor supervisor.processes = ['foo'] result = plugin.do_remove('foo') self.assertEqual(result, None) self.assertEqual(supervisor.processes, []) def test_remove_bad_name(self): plugin = self._makeOne() supervisor = plugin.ctl.options._server.supervisor supervisor.processes = ['foo'] result = plugin.do_remove('BAD_NAME') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'ERROR: no such process/group: BAD_NAME\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_remove_still_running(self): plugin = self._makeOne() supervisor = plugin.ctl.options._server.supervisor supervisor.processes = ['foo'] result = plugin.do_remove('STILL_RUNNING') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'ERROR: process/group still running: STILL_RUNNING\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_remove_reraises_other_faults(self): plugin = self._makeOne() self.assertRaises(xmlrpclib.Fault, plugin.do_remove, 'FAILED') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_update_help(self): plugin = self._makeOne() plugin.help_update() out = plugin.ctl.stdout.getvalue() self.assertTrue("Reload config and add/remove" in out) def test_update_not_on_shutdown(self): plugin = self._makeOne() supervisor = plugin.ctl.options._server.supervisor def reloadConfig(): from supervisor import xmlrpc raise xmlrpclib.Fault(xmlrpc.Faults.SHUTDOWN_STATE, 'blah') supervisor.reloadConfig = reloadConfig supervisor.processes = ['removed'] plugin.do_update('') self.assertEqual(supervisor.processes, ['removed']) def test_update_added_procs(self): plugin = self._makeOne() supervisor = plugin.ctl.options._server.supervisor def reloadConfig(): return [[['new_proc'], [], []]] supervisor.reloadConfig = reloadConfig result = plugin.do_update('') self.assertEqual(result, None) self.assertEqual(supervisor.processes, ['new_proc']) def test_update_with_gname(self): plugin = self._makeOne() supervisor = plugin.ctl.options._server.supervisor def reloadConfig(): return [[['added1', 'added2'], ['changed'], ['removed']]] supervisor.reloadConfig = reloadConfig supervisor.processes = ['changed', 'removed'] plugin.do_update('changed') self.assertEqual(sorted(supervisor.processes), sorted(['changed', 'removed'])) plugin.do_update('added1 added2') self.assertEqual(sorted(supervisor.processes), sorted(['changed', 'removed', 'added1', 'added2'])) plugin.do_update('removed') self.assertEqual(sorted(supervisor.processes), sorted(['changed', 'added1', 'added2'])) supervisor.processes = ['changed', 'removed'] plugin.do_update('removed added1') self.assertEqual(sorted(supervisor.processes), sorted(['changed', 'added1'])) supervisor.processes = ['changed', 'removed'] plugin.do_update('all') self.assertEqual(sorted(supervisor.processes), sorted(['changed', 'added1', 'added2'])) def test_update_changed_procs(self): from supervisor import xmlrpc plugin = self._makeOne() supervisor = plugin.ctl.options._server.supervisor calls = [] def reloadConfig(): return [[[], ['changed_group'], []]] supervisor.reloadConfig = reloadConfig supervisor.startProcess = lambda x: calls.append(('start', x)) supervisor.addProcessGroup('changed_group') # fake existence results = [{'name': 'changed_process', 'group': 'changed_group', 'status': xmlrpc.Faults.SUCCESS, 'description': 'blah'}] def stopProcessGroup(name): calls.append(('stop', name)) return results supervisor.stopProcessGroup = stopProcessGroup plugin.do_update('') self.assertEqual(calls, [('stop', 'changed_group')]) supervisor.addProcessGroup('changed_group') # fake existence calls[:] = [] results[:] = [{'name': 'changed_process1', 'group': 'changed_group', 'status': xmlrpc.Faults.NOT_RUNNING, 'description': 'blah'}, {'name': 'changed_process2', 'group': 'changed_group', 'status': xmlrpc.Faults.FAILED, 'description': 'blah'}] plugin.do_update('') self.assertEqual(calls, [('stop', 'changed_group')]) supervisor.addProcessGroup('changed_group') # fake existence calls[:] = [] results[:] = [{'name': 'changed_process1', 'group': 'changed_group', 'status': xmlrpc.Faults.FAILED, 'description': 'blah'}, {'name': 'changed_process2', 'group': 'changed_group', 'status': xmlrpc.Faults.SUCCESS, 'description': 'blah'}] plugin.do_update('') self.assertEqual(calls, [('stop', 'changed_group')]) def test_update_removed_procs(self): from supervisor import xmlrpc plugin = self._makeOne() supervisor = plugin.ctl.options._server.supervisor def reloadConfig(): return [[[], [], ['removed_group']]] supervisor.reloadConfig = reloadConfig results = [{'name': 'removed_process', 'group': 'removed_group', 'status': xmlrpc.Faults.SUCCESS, 'description': 'blah'}] supervisor.processes = ['removed_group'] def stopProcessGroup(name): return results supervisor.stopProcessGroup = stopProcessGroup plugin.do_update('') self.assertEqual(supervisor.processes, []) results[:] = [{'name': 'removed_process', 'group': 'removed_group', 'status': xmlrpc.Faults.NOT_RUNNING, 'description': 'blah'}] supervisor.processes = ['removed_group'] plugin.do_update('') self.assertEqual(supervisor.processes, []) results[:] = [{'name': 'removed_process', 'group': 'removed_group', 'status': xmlrpc.Faults.FAILED, 'description': 'blah'}] supervisor.processes = ['removed_group'] plugin.do_update('') self.assertEqual(supervisor.processes, ['removed_group']) def test_update_reraises_other_faults(self): plugin = self._makeOne() supervisor = plugin.ctl.options._server.supervisor def reloadConfig(): from supervisor import xmlrpc raise xmlrpclib.Fault(xmlrpc.Faults.FAILED, 'FAILED') supervisor.reloadConfig = reloadConfig self.assertRaises(xmlrpclib.Fault, plugin.do_update, '') self.assertEqual(plugin.ctl.exitstatus, 1) def test_pid_help(self): plugin = self._makeOne() plugin.help_pid() out = plugin.ctl.stdout.getvalue() self.assertTrue("pid " in out) def test_pid_supervisord(self): plugin = self._makeOne() result = plugin.do_pid('') self.assertEqual(result, None) options = plugin.ctl.options lines = plugin.ctl.stdout.getvalue().split('\n') self.assertEqual(len(lines), 2) self.assertEqual(lines[0], str(options._server.supervisor.getPID())) def test_pid_allprocesses(self): plugin = self._makeOne() result = plugin.do_pid('all') self.assertEqual(result, None) value = plugin.ctl.stdout.getvalue().strip() self.assertEqual(value.split(), ['11', '12', '13']) self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.SUCCESS) def test_pid_badname(self): plugin = self._makeOne() result = plugin.do_pid('BAD_NAME') self.assertEqual(result, None) value = plugin.ctl.stdout.getvalue().strip() self.assertEqual(value, 'No such process BAD_NAME') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_pid_oneprocess(self): plugin = self._makeOne() result = plugin.do_pid('foo') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue().strip(), '11') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.SUCCESS) def test_pid_oneprocess_not_running(self): plugin = self._makeOne() options = plugin.ctl.options def f(*arg, **kw): from supervisor.states import ProcessStates return {'name': 'foo', 'group':'foo', 'pid': 0, 'state': ProcessStates.STOPPED, 'statename': 'STOPPED', 'start': 0, 'stop': 0, 'spawnerr': '', 'now': 0, 'description':'foo description' } options._server.supervisor.getProcessInfo = f result = plugin.do_pid('foo') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue().strip(), '0') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.NOT_RUNNING) def test_pid_upcheck_failed(self): plugin = self._makeOne() plugin.ctl.upcheck = lambda: False called = [] def f(*arg, **kw): called.append(True) plugin.ctl.options._server.supervisor.getPID = f plugin.do_pid('') self.assertEqual(called, []) def test_maintail_help(self): plugin = self._makeOne() plugin.help_maintail() out = plugin.ctl.stdout.getvalue() self.assertTrue("tail of supervisor main log file" in out) def test_maintail_toomanyargs(self): plugin = self._makeOne() result = plugin.do_maintail('foo bar') self.assertEqual(result, None) val = plugin.ctl.stdout.getvalue() self.assertTrue(val.startswith('Error: too many'), val) self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_maintail_minus_string_fails(self): plugin = self._makeOne() result = plugin.do_maintail('-wrong') self.assertEqual(result, None) val = plugin.ctl.stdout.getvalue() self.assertTrue(val.startswith('Error: bad argument -wrong'), val) self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_maintail_wrong(self): plugin = self._makeOne() result = plugin.do_maintail('wrong') self.assertEqual(result, None) val = plugin.ctl.stdout.getvalue() self.assertTrue(val.startswith('Error: bad argument wrong'), val) self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def _dont_test_maintail_dashf(self): # https://github.com/Supervisor/supervisor/issues/285 # TODO: Refactor so we can test more of maintail -f than just a # connect error, and fix this test so it passes on FreeBSD. plugin = self._makeOne() plugin.listener = DummyListener() result = plugin.do_maintail('-f') self.assertEqual(result, None) errors = plugin.listener.errors self.assertEqual(len(errors), 1) error = errors[0] self.assertEqual(plugin.listener.closed, 'http://localhost:65532/mainlogtail') self.assertEqual(error[0], 'http://localhost:65532/mainlogtail') self.assertTrue('Cannot connect' in error[1]) def test_maintail_bad_modifier(self): plugin = self._makeOne() result = plugin.do_maintail('-z') self.assertEqual(result, None) lines = plugin.ctl.stdout.getvalue().split('\n') self.assertEqual(lines[0], 'Error: bad argument -z') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_maintail_nobytes(self): plugin = self._makeOne() result = plugin.do_maintail('') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'mainlogdata\n') def test_maintail_dashbytes(self): plugin = self._makeOne() result = plugin.do_maintail('-100') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'mainlogdata\n') def test_maintail_readlog_error_nofile(self): plugin = self._makeOne() supervisor_rpc = plugin.ctl.get_supervisor() from supervisor import xmlrpc supervisor_rpc._readlog_error = xmlrpc.Faults.NO_FILE result = plugin.do_maintail('-100') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'supervisord: ERROR (no log file)\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_maintail_readlog_error_failed(self): plugin = self._makeOne() supervisor_rpc = plugin.ctl.get_supervisor() from supervisor import xmlrpc supervisor_rpc._readlog_error = xmlrpc.Faults.FAILED result = plugin.do_maintail('-100') self.assertEqual(result, None) self.assertEqual(plugin.ctl.stdout.getvalue(), 'supervisord: ERROR (unknown error reading log)\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_maintail_upcheck_failed(self): plugin = self._makeOne() plugin.ctl.upcheck = lambda: False called = [] def f(*arg, **kw): called.append(True) plugin.ctl.options._server.supervisor.readLog = f plugin.do_maintail('') self.assertEqual(called, []) def test_fg_help(self): plugin = self._makeOne() plugin.help_fg() out = plugin.ctl.stdout.getvalue() self.assertTrue("fg " in out) def test_fg_too_few_args(self): plugin = self._makeOne() result = plugin.do_fg('') self.assertEqual(result, None) lines = plugin.ctl.stdout.getvalue().split('\n') self.assertEqual(lines[0], 'ERROR: no process name supplied') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_fg_too_many_args(self): plugin = self._makeOne() result = plugin.do_fg('foo bar') self.assertEqual(result, None) line = plugin.ctl.stdout.getvalue() self.assertEqual(line, 'ERROR: too many process names supplied\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_fg_badprocname(self): plugin = self._makeOne() result = plugin.do_fg('BAD_NAME') self.assertEqual(result, None) line = plugin.ctl.stdout.getvalue() self.assertEqual(line, 'ERROR: bad process name supplied\n') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_fg_procnotrunning(self): plugin = self._makeOne() result = plugin.do_fg('bar') self.assertEqual(result, None) line = plugin.ctl.stdout.getvalue() self.assertEqual(line, 'ERROR: process not running\n') result = plugin.do_fg('baz_01') lines = plugin.ctl.stdout.getvalue().split('\n') self.assertEqual(result, None) self.assertEqual(lines[-2], 'ERROR: process not running') self.assertEqual(plugin.ctl.exitstatus, LSBInitExitStatuses.GENERIC) def test_fg_upcheck_failed(self): plugin = self._makeOne() plugin.ctl.upcheck = lambda: False called = [] def f(*arg, **kw): called.append(True) plugin.ctl.options._server.supervisor.getProcessInfo = f plugin.do_fg('foo') self.assertEqual(called, []) def test_exit_help(self): plugin = self._makeOne() plugin.help_exit() out = plugin.ctl.stdout.getvalue() self.assertTrue("Exit the supervisor shell" in out) def test_quit_help(self): plugin = self._makeOne() plugin.help_quit() out = plugin.ctl.stdout.getvalue() self.assertTrue("Exit the supervisor shell" in out) class DummyListener: def __init__(self): self.errors = [] def error(self, url, msg): self.errors.append((url, msg)) def close(self, url): self.closed = url class DummyPluginFactory: def __init__(self, ctl, **kw): self.ctl = ctl def do_help(self, arg): self.ctl.stdout.write('foo helped') class DummyClientOptions: def __init__(self): self.prompt = 'supervisor' self.serverurl = 'http://localhost:65532' self.username = 'chrism' self.password = '123' self.history_file = None self.plugins = () self._server = DummyRPCServer() self.interactive = False self.plugin_factories = [('dummy', DummyPluginFactory, {})] def getServerProxy(self): return self._server class DummyController: nohelp = 'no help on %s' def __init__(self, options): self.options = options self.topics_printed = [] self.stdout = StringIO() self.exitstatus = LSBInitExitStatuses.SUCCESS def upcheck(self): return True def get_supervisor(self): return self.get_server_proxy('supervisor') def get_server_proxy(self, namespace=None): proxy = self.options.getServerProxy() if namespace is None: return proxy else: return getattr(proxy, namespace) def output(self, data): self.stdout.write(data + '\n') def print_topics(self, doc_headers, cmds_doc, rows, cols): self.topics_printed.append((doc_headers, cmds_doc, rows, cols)) def set_exitstatus_from_xmlrpc_fault(self, faultcode, ignored_faultcode=None): from supervisor.supervisorctl import DEAD_PROGRAM_FAULTS if faultcode in (ignored_faultcode, xmlrpc.Faults.SUCCESS): pass elif faultcode in DEAD_PROGRAM_FAULTS: self.exitstatus = LSBInitExitStatuses.NOT_RUNNING else: self.exitstatus = LSBInitExitStatuses.GENERIC class DummyPlugin: def __init__(self, controller=None): self.ctl = controller def do_help(self, arg): self.helped = True def test_suite(): return unittest.findTestCases(sys.modules[__name__]) if __name__ == '__main__': unittest.main(defaultTest='test_suite')