ÿØÿà 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ÿÙ """Test suite for supervisor.datatypes""" import os import signal import socket import sys import tempfile import unittest from mock import Mock, patch, sentinel from supervisor import datatypes class ProcessOrGroupName(unittest.TestCase): def _callFUT(self, arg): return datatypes.process_or_group_name(arg) def test_strips_surrounding_whitespace(self): name = " foo\t" self.assertEqual(self._callFUT(name), "foo") def test_disallows_inner_spaces(self): name = "foo bar" self.assertRaises(ValueError, self._callFUT, name) def test_disallows_colons(self): name = "foo:bar" self.assertRaises(ValueError, self._callFUT, name) class IntegerTests(unittest.TestCase): def _callFUT(self, arg): return datatypes.integer(arg) def test_converts_numeric(self): self.assertEqual(self._callFUT('1'), 1) def test_converts_numeric_overflowing_int(self): self.assertEqual(self._callFUT(str(sys.maxint+1)), sys.maxint+1) def test_raises_for_non_numeric(self): self.assertRaises(ValueError, self._callFUT, 'abc') class BooleanTests(unittest.TestCase): def _callFUT(self, arg): return datatypes.boolean(arg) def test_returns_true_for_truthy_values(self): for s in datatypes.TRUTHY_STRINGS: self.assertEqual(self._callFUT(s), True) def test_returns_true_for_upper_truthy_values(self): for s in map(str.upper, datatypes.TRUTHY_STRINGS): self.assertEqual(self._callFUT(s), True) def test_returns_false_for_falsy_values(self): for s in datatypes.FALSY_STRINGS: self.assertEqual(self._callFUT(s), False) def test_returns_false_for_upper_falsy_values(self): for s in map(str.upper, datatypes.FALSY_STRINGS): self.assertEqual(self._callFUT(s), False) def test_braises_value_error_for_bad_value(self): self.assertRaises(ValueError, self._callFUT, 'not-a-value') class ListOfStringsTests(unittest.TestCase): def _callFUT(self, arg): return datatypes.list_of_strings(arg) def test_returns_empty_list_for_empty_string(self): self.assertEqual(self._callFUT(''), []) def test_returns_list_of_strings_by_comma_split(self): self.assertEqual(self._callFUT('foo,bar'), ['foo', 'bar']) def test_returns_strings_with_whitespace_stripped(self): self.assertEqual(self._callFUT(' foo , bar '), ['foo', 'bar']) def test_raises_value_error_when_comma_split_fails(self): self.assertRaises(ValueError, self._callFUT, 42) class ListOfIntsTests(unittest.TestCase): def _callFUT(self, arg): return datatypes.list_of_ints(arg) def test_returns_empty_list_for_empty_string(self): self.assertEqual(self._callFUT(''), []) def test_returns_list_of_ints_by_comma_split(self): self.assertEqual(self._callFUT('1,42'), [1,42]) def test_returns_ints_even_if_whitespace_in_string(self): self.assertEqual(self._callFUT(' 1 , 42 '), [1,42]) def test_raises_value_error_when_comma_split_fails(self): self.assertRaises(ValueError, self._callFUT, 42) def test_raises_value_error_when_one_value_is_bad(self): self.assertRaises(ValueError, self._callFUT, '1, bad, 42') class ListOfExitcodesTests(unittest.TestCase): def _callFUT(self, arg): return datatypes.list_of_exitcodes(arg) def test_returns_list_of_ints_from_csv(self): self.assertEqual(self._callFUT('1,2,3'), [1,2,3]) def test_returns_list_of_ints_from_one(self): self.assertEqual(self._callFUT('1'), [1]) def test_raises_for_invalid_exitcode_values(self): self.assertRaises(ValueError, self._callFUT, 'a,b,c') self.assertRaises(ValueError, self._callFUT, '1024') self.assertRaises(ValueError, self._callFUT, '-1,1') class DictOfKeyValuePairsTests(unittest.TestCase): def _callFUT(self, arg): return datatypes.dict_of_key_value_pairs(arg) def test_returns_empty_dict_for_empty_str(self): actual = self._callFUT('') self.assertEqual({}, actual) def test_returns_dict_from_single_pair_str(self): actual = self._callFUT('foo=bar') expected = {'foo': 'bar'} self.assertEqual(actual, expected) def test_returns_dict_from_multi_pair_str(self): actual = self._callFUT('foo=bar,baz=qux') expected = {'foo': 'bar', 'baz': 'qux'} self.assertEqual(actual, expected) def test_returns_dict_even_if_whitespace(self): actual = self._callFUT(' foo = bar , baz = qux ') expected = {'foo': 'bar', 'baz': 'qux'} self.assertEqual(actual, expected) def test_returns_dict_even_if_newlines(self): actual = self._callFUT('foo\n=\nbar\n,\nbaz\n=\nqux') expected = {'foo': 'bar', 'baz': 'qux'} self.assertEqual(actual, expected) def test_handles_commas_inside_apostrophes(self): actual = self._callFUT("foo='bar,baz',baz='q,ux'") expected = {'foo': 'bar,baz', 'baz': 'q,ux'} self.assertEqual(actual, expected) def test_handles_commas_inside_quotes(self): actual = self._callFUT('foo="bar,baz",baz="q,ux"') expected = {'foo': 'bar,baz', 'baz': 'q,ux'} self.assertEqual(actual, expected) def test_handles_unquoted_non_alphanum(self): actual = self._callFUT( 'HOME=/home/auser,FOO=/.foo+(1.2)-_/,' 'SUPERVISOR_SERVER_URL=http://127.0.0.1:9001') expected = {'HOME': '/home/auser', 'FOO': '/.foo+(1.2)-_/', 'SUPERVISOR_SERVER_URL': 'http://127.0.0.1:9001'} self.assertEqual(actual, expected) def test_allows_trailing_comma(self): actual = self._callFUT('foo=bar,') expected = {'foo': 'bar'} self.assertEqual(actual, expected) def test_raises_value_error_on_too_short(self): self.assertRaises(ValueError, self._callFUT, 'foo') self.assertRaises(ValueError, self._callFUT, 'foo=') self.assertRaises(ValueError, self._callFUT, 'foo=bar,baz') self.assertRaises(ValueError, self._callFUT, 'foo=bar,baz=') def test_raises_when_comma_is_missing(self): kvp = 'KEY1=no-comma KEY2=ends-with-comma,' self.assertRaises(ValueError, self._callFUT, kvp) class LogfileNameTests(unittest.TestCase): def _callFUT(self, arg): return datatypes.logfile_name(arg) def test_returns_none_for_none_values(self): for thing in datatypes.LOGFILE_NONES: actual = self._callFUT(thing) self.assertEqual(actual, None) def test_returns_none_for_uppered_none_values(self): for thing in datatypes.LOGFILE_NONES: if hasattr(thing, 'upper'): thing = thing.upper() actual = self._callFUT(thing) self.assertEqual(actual, None) def test_returns_automatic_for_auto_values(self): for thing in datatypes.LOGFILE_AUTOS: actual = self._callFUT(thing) self.assertEqual(actual, datatypes.Automatic) def test_returns_automatic_for_uppered_auto_values(self): for thing in datatypes.LOGFILE_AUTOS: if hasattr(thing, 'upper'): thing = thing.upper() actual = self._callFUT(thing) self.assertEqual(actual, datatypes.Automatic) def test_returns_existing_dirpath_for_other_values(self): func = datatypes.existing_dirpath datatypes.existing_dirpath = lambda path: path try: path = '/path/to/logfile/With/Case/Preserved' actual = self._callFUT(path) self.assertEqual(actual, path) finally: datatypes.existing_dirpath = func class RangeCheckedConversionTests(unittest.TestCase): def _getTargetClass(self): return datatypes.RangeCheckedConversion def _makeOne(self, conversion, min=None, max=None): return self._getTargetClass()(conversion, min, max) def test_below_lower_bound(self): conversion = self._makeOne(lambda *arg: -1, 0) self.assertRaises(ValueError, conversion, None) def test_above_upper_lower_bound(self): conversion = self._makeOne(lambda *arg: 1, 0, 0) self.assertRaises(ValueError, conversion, None) def test_passes(self): conversion = self._makeOne(lambda *arg: 0, 0, 0) self.assertEqual(conversion(0), 0) class NameToGidTests(unittest.TestCase): def _callFUT(self, arg): return datatypes.name_to_gid(arg) @patch("grp.getgrnam", Mock(return_value=[0,0,42])) def test_gets_gid_from_group_name(self): gid = self._callFUT("foo") self.assertEqual(gid, 42) @patch("grp.getgrgid", Mock(return_value=[0,0,42])) def test_gets_gid_from_group_id(self): gid = self._callFUT("42") self.assertEqual(gid, 42) @patch("grp.getgrnam", Mock(side_effect=KeyError("bad group name"))) def test_raises_for_bad_group_name(self): self.assertRaises(ValueError, self._callFUT, "foo") @patch("grp.getgrgid", Mock(side_effect=KeyError("bad group id"))) def test_raises_for_bad_group_id(self): self.assertRaises(ValueError, self._callFUT, "42") class NameToUidTests(unittest.TestCase): def _callFUT(self, arg): return datatypes.name_to_uid(arg) @patch("pwd.getpwnam", Mock(return_value=[0,0,42])) def test_gets_uid_from_username(self): uid = self._callFUT("foo") self.assertEqual(uid, 42) @patch("pwd.getpwuid", Mock(return_value=[0,0,42])) def test_gets_uid_from_user_id(self): uid = self._callFUT("42") self.assertEqual(uid, 42) @patch("pwd.getpwnam", Mock(side_effect=KeyError("bad username"))) def test_raises_for_bad_username(self): self.assertRaises(ValueError, self._callFUT, "foo") @patch("pwd.getpwuid", Mock(side_effect=KeyError("bad user id"))) def test_raises_for_bad_user_id(self): self.assertRaises(ValueError, self._callFUT, "42") class OctalTypeTests(unittest.TestCase): def _callFUT(self, arg): return datatypes.octal_type(arg) def test_success(self): self.assertEqual(self._callFUT('10'), 8) def test_raises_for_non_numeric(self): try: self._callFUT('bad') self.fail() except ValueError, e: expected = 'bad can not be converted to an octal type' self.assertEqual(e.args[0], expected) def test_raises_for_unconvertable_numeric(self): try: self._callFUT('1.2') self.fail() except ValueError, e: expected = '1.2 can not be converted to an octal type' self.assertEqual(e.args[0], expected) class ExistingDirectoryTests(unittest.TestCase): def _callFUT(self, arg): return datatypes.existing_directory(arg) def test_dir_exists(self): path = os.path.dirname(__file__) self.assertEqual(path, self._callFUT(path)) def test_dir_does_not_exist(self): path = os.path.join(os.path.dirname(__file__), 'nonexistent') try: self._callFUT(path) self.fail() except ValueError, e: expected = "%s is not an existing directory" % path self.assertEqual(e.args[0], expected) def test_not_a_directory(self): path = __file__ try: self._callFUT(path) self.fail() except ValueError, e: expected = "%s is not an existing directory" % path self.assertEqual(e.args[0], expected) def test_expands_home(self): home = os.path.expanduser('~') if os.path.exists(home): path = self._callFUT('~') self.assertEqual(home, path) class ExistingDirpathTests(unittest.TestCase): def _callFUT(self, arg): return datatypes.existing_dirpath(arg) def test_returns_existing_dirpath(self): self.assertEqual(self._callFUT(__file__), __file__) def test_returns_dirpath_if_relative(self): self.assertEqual(self._callFUT('foo'), 'foo') def test_raises_if_dir_does_not_exist(self): path = os.path.join(os.path.dirname(__file__), 'nonexistent', 'foo') try: self._callFUT(path) self.fail() except ValueError, e: expected = ('The directory named as part of the path %s ' 'does not exist.' % path) self.assertEqual(e.args[0], expected) def test_raises_if_exists_but_not_a_dir(self): path = os.path.join(os.path.dirname(__file__), os.path.basename(__file__), 'foo') try: self._callFUT(path) self.fail() except ValueError, e: expected = ('The directory named as part of the path %s ' 'does not exist.' % path) self.assertEqual(e.args[0], expected) def test_expands_home(self): home = os.path.expanduser('~') if os.path.exists(home): path = self._callFUT('~/foo') self.assertEqual(os.path.join(home, 'foo'), path) class LoggingLevelTests(unittest.TestCase): def _callFUT(self, arg): return datatypes.logging_level(arg) def test_returns_level_from_name_case_insensitive(self): from supervisor.loggers import LevelsByName self.assertEqual(self._callFUT("wArN"), LevelsByName.WARN) def test_raises_for_bad_level_name(self): self.assertRaises(ValueError, self._callFUT, "foo") class UrlTests(unittest.TestCase): def _callFUT(self, arg): return datatypes.url(arg) def test_accepts_urlparse_recognized_scheme_with_netloc(self): good_url = 'http://localhost:9001' self.assertEqual(self._callFUT(good_url), good_url) def test_rejects_urlparse_recognized_scheme_but_no_netloc(self): bad_url = 'http://' self.assertRaises(ValueError, self._callFUT, bad_url) def test_accepts_unix_scheme_with_path(self): good_url = "unix://somepath" self.assertEqual(good_url, self._callFUT(good_url)) def test_rejects_unix_scheme_with_no_slashes_or_path(self): bad_url = "unix:" self.assertRaises(ValueError, self._callFUT, bad_url) def test_rejects_unix_scheme_with_slashes_but_no_path(self): bad_url = "unix://" self.assertRaises(ValueError, self._callFUT, bad_url) class InetStreamSocketConfigTests(unittest.TestCase): def _getTargetClass(self): return datatypes.InetStreamSocketConfig def _makeOne(self, *args, **kw): return self._getTargetClass()(*args, **kw) def test_url(self): conf = self._makeOne('127.0.0.1', 8675) self.assertEqual(conf.url, 'tcp://127.0.0.1:8675') def test___str__(self): cfg = self._makeOne('localhost', 65531) self.assertEqual(str(cfg), 'tcp://localhost:65531') def test_repr(self): conf = self._makeOne('127.0.0.1', 8675) s = repr(conf) self.assertTrue(s.startswith( ''), s) def test_addr(self): conf = self._makeOne('127.0.0.1', 8675) addr = conf.addr() self.assertEqual(addr, ('127.0.0.1', 8675)) def test_port_as_string(self): conf = self._makeOne('localhost', '5001') addr = conf.addr() self.assertEqual(addr, ('localhost', 5001)) def test_create_and_bind(self): conf = self._makeOne('127.0.0.1', 8675) sock = conf.create_and_bind() reuse = sock.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR) self.assertTrue(reuse) self.assertEqual(conf.addr(), sock.getsockname()) #verifies that bind was called sock.close() def test_same_urls_are_equal(self): conf1 = self._makeOne('localhost', 5001) conf2 = self._makeOne('localhost', 5001) self.assertTrue(conf1 == conf2) self.assertFalse(conf1 != conf2) def test_diff_urls_are_not_equal(self): conf1 = self._makeOne('localhost', 5001) conf2 = self._makeOne('localhost', 5002) self.assertTrue(conf1 != conf2) self.assertFalse(conf1 == conf2) def test_diff_objs_are_not_equal(self): conf1 = self._makeOne('localhost', 5001) conf2 = 'blah' self.assertTrue(conf1 != conf2) self.assertFalse(conf1 == conf2) class UnixStreamSocketConfigTests(unittest.TestCase): def _getTargetClass(self): return datatypes.UnixStreamSocketConfig def _makeOne(self, *args, **kw): return self._getTargetClass()(*args, **kw) def test_url(self): conf = self._makeOne('/tmp/foo.sock') self.assertEqual(conf.url, 'unix:///tmp/foo.sock') def test___str__(self): cfg = self._makeOne('foo/bar') self.assertEqual(str(cfg), 'unix://foo/bar') def test_repr(self): conf = self._makeOne('/tmp/foo.sock') s = repr(conf) self.assertTrue(s.startswith( ''), s) def test_get_addr(self): conf = self._makeOne('/tmp/foo.sock') addr = conf.addr() self.assertEqual(addr, '/tmp/foo.sock') def test_create_and_bind(self): (tf_fd, tf_name) = tempfile.mkstemp() owner = (sentinel.uid, sentinel.gid) mode = sentinel.mode conf = self._makeOne(tf_name, owner=owner, mode=mode) # Patch os.chmod and os.chown functions with mocks # objects so that the test does not depend on # any specific system users or permissions chown_mock = Mock() chmod_mock = Mock() @patch('os.chown', chown_mock) @patch('os.chmod', chmod_mock) def call_create_and_bind(conf): return conf.create_and_bind() sock = call_create_and_bind(conf) self.assertTrue(os.path.exists(tf_name)) # verifies that bind was called self.assertEqual(conf.addr(), sock.getsockname()) sock.close() self.assertTrue(os.path.exists(tf_name)) os.unlink(tf_name) # Verify that os.chown was called with correct args self.assertEqual(1, chown_mock.call_count) path_arg = chown_mock.call_args[0][0] uid_arg = chown_mock.call_args[0][1] gid_arg = chown_mock.call_args[0][2] self.assertEqual(tf_name, path_arg) self.assertEqual(owner[0], uid_arg) self.assertEqual(owner[1], gid_arg) # Verify that os.chmod was called with correct args self.assertEqual(1, chmod_mock.call_count) path_arg = chmod_mock.call_args[0][0] mode_arg = chmod_mock.call_args[0][1] self.assertEqual(tf_name, path_arg) self.assertEqual(mode, mode_arg) def test_create_and_bind_when_chown_fails(self): (tf_fd, tf_name) = tempfile.mkstemp() owner = (sentinel.uid, sentinel.gid) mode = sentinel.mode conf = self._makeOne(tf_name, owner=owner, mode=mode) @patch('os.chown', Mock(side_effect=OSError("msg"))) @patch('os.chmod', Mock()) def call_create_and_bind(conf): return conf.create_and_bind() try: call_create_and_bind(conf) self.fail() except ValueError, e: expected = "Could not change ownership of socket file: msg" self.assertEqual(e.args[0], expected) self.assertFalse(os.path.exists(tf_name)) def test_create_and_bind_when_chmod_fails(self): (tf_fd, tf_name) = tempfile.mkstemp() owner = (sentinel.uid, sentinel.gid) mode = sentinel.mode conf = self._makeOne(tf_name, owner=owner, mode=mode) @patch('os.chown', Mock()) @patch('os.chmod', Mock(side_effect=OSError("msg"))) def call_create_and_bind(conf): return conf.create_and_bind() try: call_create_and_bind(conf) self.fail() except ValueError, e: expected = "Could not change permissions of socket file: msg" self.assertEqual(e.args[0], expected) self.assertFalse(os.path.exists(tf_name)) def test_same_paths_are_equal(self): conf1 = self._makeOne('/tmp/foo.sock') conf2 = self._makeOne('/tmp/foo.sock') self.assertTrue(conf1 == conf2) self.assertFalse(conf1 != conf2) def test_diff_paths_are_not_equal(self): conf1 = self._makeOne('/tmp/foo.sock') conf2 = self._makeOne('/tmp/bar.sock') self.assertTrue(conf1 != conf2) self.assertFalse(conf1 == conf2) def test_diff_objs_are_not_equal(self): conf1 = self._makeOne('/tmp/foo.sock') conf2 = 'blah' self.assertTrue(conf1 != conf2) self.assertFalse(conf1 == conf2) class InetAddressTests(unittest.TestCase): def _callFUT(self, s): return datatypes.inet_address(s) def test_no_port_number(self): self.assertRaises(ValueError, self._callFUT, 'a:') def test_bad_port_number(self): self.assertRaises(ValueError, self._callFUT, 'a') def test_default_host(self): host, port = self._callFUT('*:8080') self.assertEqual(host, '') self.assertEqual(port, 8080) def test_boring(self): host, port = self._callFUT('localhost:80') self.assertEqual(host, 'localhost') self.assertEqual(port, 80) class SocketAddressTests(unittest.TestCase): def _getTargetClass(self): return datatypes.SocketAddress def _makeOne(self, s): return self._getTargetClass()(s) def test_unix_socket(self): import socket addr = self._makeOne('/foo/bar') self.assertEqual(addr.family, socket.AF_UNIX) self.assertEqual(addr.address, '/foo/bar') def test_inet_socket(self): import socket addr = self._makeOne('localhost:8080') self.assertEqual(addr.family, socket.AF_INET) self.assertEqual(addr.address, ('localhost', 8080)) class ColonSeparatedUserGroupTests(unittest.TestCase): def _callFUT(self, arg): return datatypes.colon_separated_user_group(arg) def test_ok_username(self): self.assertEqual(self._callFUT('root')[0], 0) def test_missinguser_username(self): self.assertRaises(ValueError, self._callFUT, 'godihopethisuserdoesntexist') def test_missinguser_username_and_groupname(self): self.assertRaises(ValueError, self._callFUT, 'godihopethisuserdoesntexist:foo') def test_separated_user_group_returns_both(self): name_to_uid = Mock(return_value=12) name_to_gid = Mock(return_value=34) @patch("supervisor.datatypes.name_to_uid", name_to_uid) @patch("supervisor.datatypes.name_to_gid", name_to_gid) def colon_separated(value): return self._callFUT(value) uid, gid = colon_separated("foo:bar") name_to_uid.assert_called_with("foo") self.assertEqual(12, uid) name_to_gid.assert_called_with("bar") self.assertEqual(34, gid) def test_separated_user_group_returns_user_only(self): name_to_uid = Mock(return_value=42) @patch("supervisor.datatypes.name_to_uid", name_to_uid) def colon_separated(value): return self._callFUT(value) uid, gid = colon_separated("foo") name_to_uid.assert_called_with("foo") self.assertEqual(42, uid) self.assertEqual(-1, gid) class SignalNumberTests(unittest.TestCase): def _callFUT(self, arg): return datatypes.signal_number(arg) def test_converts_number(self): self.assertEqual(self._callFUT(signal.SIGTERM), signal.SIGTERM) def test_converts_name(self): self.assertEqual(self._callFUT(' term '), signal.SIGTERM) def test_converts_signame(self): self.assertEqual(self._callFUT('SIGTERM'), signal.SIGTERM) def test_raises_for_bad_number(self): try: self._callFUT('12345678') self.fail() except ValueError, e: expected = "value '12345678' is not a valid signal number" self.assertEqual(e.args[0], expected) def test_raises_for_bad_name(self): try: self._callFUT('BADSIG') self.fail() except ValueError, e: expected = "value 'BADSIG' is not a valid signal name" self.assertEqual(e.args[0], expected) class AutoRestartTests(unittest.TestCase): def _callFUT(self, arg): return datatypes.auto_restart(arg) def test_converts_truthy(self): for s in datatypes.TRUTHY_STRINGS: result = self._callFUT(s) self.assertEqual(result, datatypes.RestartUnconditionally) def test_converts_falsy(self): for s in datatypes.FALSY_STRINGS: self.assertFalse(self._callFUT(s)) def test_converts_unexpected(self): for s in ('unexpected', 'UNEXPECTED'): result = self._callFUT(s) self.assertEqual(result, datatypes.RestartWhenExitUnexpected) def test_raises_for_bad_value(self): try: self._callFUT('bad') self.fail() except ValueError, e: self.assertEqual(e.args[0], "invalid 'autorestart' value 'bad'") class ProfileOptionsTests(unittest.TestCase): def _callFUT(self, arg): return datatypes.profile_options(arg) def test_empty(self): sort_options, callers = self._callFUT('') self.assertEqual([], sort_options) self.assertFalse(callers) def test_without_callers(self): sort_options, callers = self._callFUT('CUMULATIVE,calls') self.assertEqual(['cumulative', 'calls'], sort_options) self.assertFalse(callers) def test_with_callers(self): sort_options, callers = self._callFUT('cumulative, callers') self.assertEqual(['cumulative'], sort_options) self.assertTrue(callers)