����JFIFXX�����    $.' ",#(7),01444'9=82<.342  2!!22222222222222222222222222222222222222222222222222����"��4�� ���,�PG"Z_�4�˷����kjز�Z�,F+��_z�,�© �����zh6�٨�ic�fu���#ډb���_�N�?��wQ���5-�~�I���8����TK<5o�Iv-�����k�_U_�����~b�M��d����Ӝ�U�Hh��?]��E�w��Q���k�{��_}qFW7HTՑ��Y��F�?_�'ϔ��_�Ջt��=||I ��6�έ"�����D���/[�k�9���Y�8ds|\���Ҿp6�Ҵ���]��.����6�z<�v��@]�i%��$j��~�g��J>��no����pM[me�i$[����s�o�ᘨ�˸ nɜG-�ĨU�ycP�3.DB�li�;��hj���x7Z^�N�h������N3u{�:j�x�힞��#M&��jL P@_���� P��&��o8������9�����@Sz6�t7#O�ߋ �s}Yf�T���lmr����Z)'N��k�۞p����w\�Tȯ?�8`�O��i{wﭹW�[�r�� ��Q4F�׊���3m&L�=��h3����z~��#�\�l :�F,j@�� ʱ�wQT����8�"kJO���6�֚l����}���R�>ډK���]��y����&����p�}b��;N�1�m�r$�|��7�>e�@B�TM*-iH��g�D�)� E�m�|�ؘbҗ�a��Ҿ����t4���o���G��*oCN�rP���Q��@z,|?W[0�����:�n,jWiE��W��$~/�hp\��?��{(�0���+�Y8rΟ�+����>S-S����VN;�}�s?.����� w�9��˟<���Mq4�Wv'��{)0�1mB��V����W[�����8�/<� �%���wT^�5���b��)iM� pg�N�&ݝ��VO~�q���u���9� ����!��J27����$O-���! �:�%H��� ـ����y�ΠM=t{!S�� oK8������t<����è:a������[�����ա�H���~��w��Qz`�po�^ ����Q��n� �,uu�C�$ ^���,������8�#��:�6��e�|~���!�3�3.�\0��q��o�4`.|� ����y�Q�`~;�d�ׯ,��O�Zw�������`73�v�܋�<���Ȏ�� ـ4k��5�K�a�u�=9Yd��$>x�A�&�� j0� ���vF��� Y�|�y��� ~�6�@c��1vOp�Ig����4��l�OD���L����� R���c���j�_�uX6��3?nk��Wy�f;^*B� ��@�~a�`��Eu������+���6�L��.ü>��}y���}_�O�6�͐�:�YrG�X��kG�����l^w���~㒶sy��Iu�!� W ��X��N�7BV��O��!X�2����wvG�R�f�T#�����t�/?���%8�^�W�aT��G�cL�M���I��(J����1~�8�?aT ���]����AS�E��(��*E}� 2��#I/�׍qz��^t�̔���b�Yz4x���t�){ OH��+(E��A&�N�������XT��o��"�XC��'���)}�J�z�p� ��~5�}�^����+�6����w��c��Q�|Lp�d�H��}�(�.|����k��c4^�"�����Z?ȕ ��a<�L�!039C� �Eu�C�F�Ew�ç ;�n?�*o���B�8�bʝ���'#Rqf���M}7����]����s2tcS{�\icTx;�\��7K���P���ʇ Z O-��~��c>"��?�������P��E��O�8��@�8��G��Q�g�a�Վ���󁶠�䧘��_%#r�>�1�z�a��eb��qcPѵ��n���#L��� =��׀t� L�7�`��V���A{�C:�g���e@�w1 Xp3�c3�ġ����p��M"'-�@n4���fG��B3�DJ�8[Jo�ߐ���gK)ƛ��$���� ���8�3�����+���� �����6�ʻ���� ���S�kI�*KZlT _`���?��K����QK�d����B`�s}�>���`��*�>��,*@J�d�oF*����弝��O}�k��s��]��y�ߘ��c1G�V���<=�7��7����6�q�PT��tXԀ�!9*4�4Tހ3XΛex�46���Y��D ����� �BdemDa����\�_l,��G�/���֌7���Y�](�xTt^%�GE�����4�}bT���ڹ�����;Y)���B�Q��u��>J/J �⮶.�XԄ��j�ݳ�+E��d ��r�5�_D�1 ��o�� �B�x�΢�#���<��W�����8���R6�@g�M�.��� dr�D��>(otU��@x=��~v���2� ӣ�d�oBd��3�eO�6�㣷�����ݜ6��6Y��Qz`��S��{���\P�~z m5{J/L��1������<�e�ͅPu�b�]�ϔ���'������f�b� Zpw��c`"��i���BD@:)ִ�:�]��hv�E�w���T�l��P���"Ju�}��وV J��G6��. J/�Qgl߭�e�����@�z�Zev2u�)]կ�����7x���s�M�-<ɯ�c��r�v�����@��$�ޮ}lk���a���'����>x��O\�ZFu>�����ck#��&:��`�$�ai�>2Δ����l���oF[h��lE�ܺ�Πk:)���`�� $[6�����9�����kOw�\|���8}������ބ:��񶐕��I�A1/�=�2[�,�!��.}gN#�u����b��� ~��݊��}34q����d�E��Lc��$��"�[q�U�硬g^��%B �z���r�pJ�ru%v\h1Y�ne`ǥ:g���pQM~�^�Xi� ��`S�:V29.�P���V�?B�k�� AEvw%�_�9C�Q����wKekPؠ�\�;Io d�{ ߞo�c1eP����\� `����E=���@K<�Y���eڼ�J���w����{av�F�'�M�@/J��+9p���|]�����Iw &`��8���&M�hg��[�{��Xj��%��Ӓ�$��(����ʹN���<>�I���RY���K2�NPlL�ɀ)��&e����B+ь����( � �JTx���_?EZ� }@ 6�U���뙢ط�z��dWI�n` D����噥�[��uV��"�G&Ú����2g�}&m��?ċ�"����Om#��������� ��{�ON��"S�X��Ne��ysQ���@Fn��Vg���dX�~nj�]J�<�K]:��FW��b�������62�=��5f����JKw��bf�X�55��~J �%^����:�-�QIE��P��v�nZum� z � ~ə ���� ���ة����;�f��\v���g�8�1��f24;�V���ǔ�)����9���1\��c��v�/'Ƞ�w�������$�4�R-��t���� e�6�/�ġ �̕Ecy�J���u�B���<�W�ַ~�w[B1L۲�-JS΂�{���΃������A��20�c#��@ 0!1@AP"#2Q`$3V�%45a6�FRUq��� ����^7ׅ,$n�������+��F�`��2X'��0vM��p�L=������5��8������u�p~���.�`r�����\���O��,ư�0oS ��_�M�����l���4�kv\JSd���x���SW�<��Ae�IX����������$I���w�:S���y���›R��9�Q[���,�5�;�@]�%���u�@ *ro�lbI �� ��+���%m:�͇ZV�����u�̉����θau<�fc�.����{�4Ա� �Q����*�Sm��8\ujqs]{kN���)qO�y�_*dJ�b�7���yQqI&9�ԌK!�M}�R�;������S�T���1���i[U�ɵz�]��U)V�S6���3$K{�ߊ<�(� E]Զ[ǼENg�����'�\?#)Dkf��J���o��v���'�%ƞ�&K�u�!��b�35LX�Ϸ��63$K�a�;�9>,R��W��3�3� d�JeTYE.Mϧ��-�o�j3+y��y^�c�������VO�9NV\nd�1 ��!͕_)a�v;����թ�M�lWR1��)El��P;��yوÏ�u 3�k�5Pr6<�⒲l�!˞*��u־�n�!�l:����UNW ��%��Chx8vL'��X�@��*��)���̮��ˍ��� ���D-M�+J�U�kvK����+�x8��cY������?�Ԡ��~3mo��|�u@[XeY�C�\Kp�x8�oC�C�&����N�~3-H���� ��MX�s�u<`���~"WL��$8ξ��3���a�)|:@�m�\���^�`�@ҷ)�5p+��6���p�%i)P M���ngc�����#0Aruz���RL+xSS?���ʮ}()#�t��mˇ!��0}}y����<�e� �-ή�Ԩ��X������ MF���ԙ~l L.3���}�V뽺�v�����멬��Nl�)�2����^�Iq��a��M��qG��T�����c3#������3U�Ǎ���}��לS�|qa��ڃ�+���-��2�f����/��bz��ڐ�� �ݼ[2�ç����k�X�2�* �Z�d���J�G����M*9W���s{��w���T��x��y,�in�O�v��]���n����P�$�JB@=4�OTI�n��e�22a\����q�d���%�$��(���:���: /*�K[PR�fr\nڙdN���F�n�$�4�[�� U�zƶ����� �mʋ���,�ao�u 3�z� �x��Kn����\[��VFmbE;�_U��&V�Gg�]L�۪&#n%�$ɯ�dG���D�TI=�%+AB�Ru#��b4�1�»x�cs�YzڙJG��f��Il��d�eF'T� iA��T���uC�$����Y��H?����[!G`}���ͪ� �纤Hv\������j�Ex�K���!���OiƸ�Yj�+u-<���'q����uN�*�r\��+�]���<�wOZ.fp�ێ��,-*)V?j-kÊ#�`�r��dV����(�ݽBk�����G�ƛk�QmUڗe��Z���f}|����8�8��a���i��3'J�����~G_�^���d�8w������ R�`(�~�.��u���l�s+g�bv���W���lGc}��u���afE~1�Ue������Z�0�8�=e�� f@/�jqEKQQ�J��oN��J���W5~M>$6�Lt�;$ʳ{���^��6�{����v6���ķܰg�V�cnn �~z�x�«�,2�u�?cE+Ș�H؎�%�Za�)���X>uW�Tz�Nyo����s���FQƤ��$��*�&�LLXL)�1�" L��eO��ɟ�9=���:t��Z���c��Ž���Y?�ӭV�wv�~,Y��r�ۗ�|�y��GaF�����C�����.�+� ���v1���fήJ�����]�S��T��B��n5sW}y�$��~z�'�c ��8 ��� ,! �p��VN�S��N�N�q��y8z˱�A��4��*��'������2n<�s���^ǧ˭P�Jޮɏ�U�G�L�J�*#��<�V��t7�8����TĜ>��i}K%,���)[��z�21z ?�N�i�n1?T�I�R#��m-�����������������1����lA�`��fT5+��ܐ�c�q՝��ʐ��,���3�f2U�եmab��#ŠdQ�y>\��)�SLY����w#��.���ʑ�f��� ,"+�w�~�N�'�c�O�3F�������N<���)j��&��,-� �љ���֊�_�zS���TǦ����w�>��?�������n��U仆�V���e�����0���$�C�d���rP �m�׈e�Xm�Vu� �L��.�bֹ��� �[Դaզ���*��\y�8�Է:�Ez\�0�Kq�C b��̘��cө���Q��=0Y��s�N��S.���3.���O�o:���#���v7�[#߫ ��5�܎�L���Er4���9n��COWlG�^��0k�%<���ZB���aB_���������'=��{i�v�l�$�uC���mƎҝ{�c㱼�y]���W�i ��ߧc��m�H� m�"�"�����;Y�ߝ�Z�Ǔ�����:S#��|}�y�,/k�Ld� TA�(�AI$+I3��;Y*���Z��}|��ӧO��d�v��..#:n��f>�>���ȶI�TX��� 8��y����"d�R�|�)0���=���n4��6ⲑ�+��r<�O�܂~zh�z����7ܓ�HH�Ga롏���nCo�>������a ���~]���R���̲c?�6(�q�;5%� |�uj�~z8R=X��I�V=�|{v�Gj\gc��q����z�؋%M�ߍ����1y��#��@f^���^�>N�����#x#۹��6�Y~�?�dfPO��{��P�4��V��u1E1J �*|���%���JN��`eWu�zk M6���q t[�� ��g�G���v��WIG��u_ft����5�j�"�Y�:T��ɐ���*�;� e5���4����q$C��2d�}���� _S�L#m�Yp��O�.�C�;��c����Hi#֩%+) �Ӎ��ƲV���SYź��g |���tj��3�8���r|���V��1#;.SQ�A[���S������#���`n�+���$��$I �P\[�@�s��(�ED�z���P��])8�G#��0B��[ى��X�II�q<��9�~[Z멜�Z�⊔IWU&A>�P~�#��dp<�?����7���c��'~���5 ��+$���lx@�M�dm��n<=e�dyX��?{�|Aef ,|n3�<~z�ƃ�uۧ�����P��Y,�ӥQ�*g�#먙R�\���;T��i,��[9Qi歉����c>]9�� ��"�c��P�� �Md?٥��If�ت�u��k��/����F��9�c*9��Ǎ:�ØF���z�n*�@|I�ށ9����N3{'��[�'ͬ�Ҳ4��#}��!�V� Fu��,�,mTIk���v C�7v���B�6k�T9��1�*l� '~��ƞF��lU��'�M ����][ΩũJ_�{�i�I�n��$���L�� j��O�dx�����kza۪��#�E��Cl����x˘�o�����V���ɞ�ljr��)�/,�߬h�L��#��^��L�ф�,íMƁe�̩�NB�L�����iL����q�}��(��q��6IçJ$�W�E$��:������=#����(�K�B����zђ <��K(�N�۫K�w��^O{!����)�H���>x�������lx�?>Պ�+�>�W���,Ly!_�D���Ō�l���Q�!�[ �S����J��1��Ɛ�Y}��b,+�Lo�x�ɓ)����=�y�oh�@�꥟/��I��ѭ=��P�y9��� �ۍYӘ�e+�p�Jnϱ?V\SO%�(�t� ���=?MR�[Ș�����d�/ ��n�l��B�7j� ��!�;ӥ�/�[-���A�>�dN�sLj ��,ɪv��=1c�.SQ�O3�U���ƀ�ܽ�E����������̻��9G�ϷD�7(�}��Ävӌ\�y�_0[w ���<΍>����a_��[0+�L��F.�޺��f�>oN�T����q;���y\��bՃ��y�jH�<|q-eɏ�_?_9+P���Hp$�����[ux�K w�Mw��N�ی'$Y2�=��q���KB��P��~������Yul:�[<����F1�2�O���5=d����]Y�sw:���Ϯ���E��j,_Q��X��z`H1,#II ��d�wr��P˂@�ZJV����y$�\y�{}��^~���[:N����ߌ�U�������O��d�����ؾe��${p>G��3c���Ė�lʌ�� ת��[��`ϱ�-W����dg�I��ig2��� ��}s ��ؤ(%#sS@���~���3�X�nRG�~\jc3�v��ӍL��M[JB�T��s3}��j�Nʖ��W����;7��ç?=X�F=-�=����q�ߚ���#���='�c��7���ڑW�I(O+=:uxq�������������e2�zi+�kuG�R��������0�&e�n���iT^J����~\jy���p'dtG��s����O��3����9* �b#Ɋ�� p������[Bws�T�>d4�ۧs���nv�n���U���_�~,�v����ƜJ1��s�� �QIz��)�(lv8M���U=�;����56��G���s#�K���MP�=��LvyGd��}�VwWBF�'�à �?MH�U�g2�� ����!�p�7Q��j��ڴ����=��j�u��� Jn�A s���uM������e��Ɔ�Ҕ�!)'��8Ϣ�ٔ��ޝ(��Vp���צ֖d=�IC�J�Ǡ{q������kԭ�߸���i��@K����u�|�p=..�*+����x�����z[Aqġ#s2a�Ɗ���RR�)*HRsi�~�a &f��M��P����-K�L@��Z��Xy�'x�{}��Zm+���:�)�) IJ�-i�u���� ���ܒH��'�L(7�y�GӜq���� j��� 6ߌg1�g�o���,kر���tY�?W,���p���e���f�OQS��!K�۟cҒA�|ս�j�>��=⬒��˧L[�� �߿2JaB~R��u�:��Q�] �0H~���]�7��Ƽ�I���(}��cq '�ήET���q�?f�ab���ӥvr� �)o��-Q��_'����ᴎo��K������;��V���o��%���~OK ����*��b�f:���-ťIR��`B�5!RB@���ï�� �u �̯e\�_U�_������� g�ES��3�������QT��a����x����U<~�c?�*�#]�MW,[8O�a�x��]�1bC|踤�P��lw5V%�)�{t�<��d��5���0i�XSU��m:��Z�┵�i�"��1�^B�-��P�hJ��&)O��*�D��c�W��vM��)����}���P��ܗ-q����\mmζZ-l@�}��a��E�6��F�@��&Sg@���ݚ�M����� ȹ 4����#p�\H����dYDo�H���"��\��..R�B�H�z_�/5˘����6��KhJR��P�mƶi�m���3�,#c�co��q�a)*Pt����R�m�k�7x�D�E�\Y�閣_X�<���~�)���c[[�BP����6�Yq���S��0����%_����;��Àv�~�| VS؇ ��'O0��F0��\���U�-�d@�����7�SJ*z��3n��y��P����O���������m�~�P�3|Y��ʉr#�C�<�G~�.,! ���bqx���h~0=��!ǫ�jy����l�O,�[B��~��|9��ٱ����Xly�#�i�B��g%�S��������tˋ���e���ې��\[d�t)��.+u�|1 ������#�~Oj����hS�%��i.�~X���I�H�m��0n���c�1uE�q��cF�RF�o���7� �O�ꮧ� ���ۛ{��ʛi5�rw?׌#Qn�TW��~?y$��m\�\o����%W� ?=>S�N@�� �Ʈ���R����N�)�r"C�:��:����� �����#��qb��Y�. �6[��2K����2u�Ǧ�HYR��Q�MV��� �G�$��Q+.>�����nNH��q�^��� ����q��mM��V��D�+�-�#*�U�̒ ���p욳��u:�������IB���m���PV@O���r[b= �� ��1U�E��_Nm�yKbN�O���U�}�the�`�|6֮P>�\2�P�V���I�D�i�P�O;�9�r�mAHG�W�S]��J*�_�G��+kP�2����Ka�Z���H�'K�x�W�MZ%�O�YD�Rc+o��?�q��Ghm��d�S�oh�\�D�|:W������UA�Qc yT�q������~^�H��/��#p�CZ���T�I�1�ӏT����4��"�ČZ�����}��`w�#�*,ʹ�� ��0�i��課�Om�*�da��^gJ݅{���l�e9uF#T�ֲ��̲�ٞC"�q���ߍ ոޑ�o#�XZTp����@ o�8��(jd��xw�]�,f���`~�|,s��^����f�1���t��|��m�򸄭/ctr��5s��7�9Q�4�H1꠲BB@l9@���C�����+�wp�xu�£Yc�9��?`@#�o�mH�s2��)�=��2�.�l����jg�9$�Y�S�%*L������R�Y������7Z���,*=�䷘$�������arm�o�ϰ���UW.|�r�uf����IGw�t����Zwo��~5 ��YյhO+=8fF�)�W�7�L9lM�̘·Y���֘YLf�큹�pRF���99.A �"wz��=E\Z���'a� 2��Ǚ�#;�'}�G���*��l��^"q��+2FQ� hj��kŦ��${���ޮ-�T�٭cf�|�3#~�RJ����t��$b�(R��(����r���dx� >U b�&9,>���%E\� Ά�e�$��'�q't��*�א���ެ�b��-|d���SB�O�O��$�R+�H�)�܎�K��1m`;�J�2�Y~9��O�g8=vqD`K[�F)k�[���1m޼c��n���]s�k�z$@��)!I �x՝"v��9=�ZA=`Ɠi �:�E��)`7��vI��}d�YI�_ �o�:ob���o ���3Q��&D&�2=�� �Ά��;>�h����y.*ⅥS������Ӭ�+q&����j|UƧ����}���J0��WW< ۋS�)jQR�j���Ư��rN)�Gű�4Ѷ(�S)Ǣ�8��i��W52���No˓� ۍ%�5brOn�L�;�n��\G����=�^U�dI���8$�&���h��'���+�(������cȁ߫k�l��S^���cƗjԌE�ꭔ��gF���Ȓ��@���}O���*;e�v�WV���YJ\�]X'5��ղ�k�F��b 6R�o՜m��i N�i����>J����?��lPm�U��}>_Z&�KK��q�r��I�D�Չ~�q�3fL�:S�e>���E���-G���{L�6p�e,8��������QI��h��a�Xa��U�A'���ʂ���s�+טIjP�-��y�8ۈZ?J$��W�P� ��R�s�]��|�l(�ԓ��sƊi��o(��S0��Y� 8�T97.�����WiL��c�~�dxc�E|�2!�X�K�Ƙਫ਼�$((�6�~|d9u+�qd�^3�89��Y�6L�.I�����?���iI�q���9�)O/뚅����O���X��X�V��ZF[�یgQ�L��K1���RҖr@v�#��X�l��F���Нy�S�8�7�kF!A��sM���^rkp�jP�DyS$N���q��nxҍ!U�f�!eh�i�2�m���`�Y�I�9r�6� �TF���C}/�y�^���Η���5d�'��9A-��J��>{�_l+�`��A���[�'��յ�ϛ#w:݅�%��X�}�&�PSt�Q�"�-��\縵�/����$Ɨh�Xb�*�y��BS����;W�ջ_mc�����vt?2}1�;qS�d�d~u:2k5�2�R�~�z+|HE!)�Ǟl��7`��0�<�,�2*���Hl-��x�^����'_TV�gZA�'j� ^�2Ϊ��N7t�����?w�� �x1��f��Iz�C-Ȗ��K�^q�;���-W�DvT�7��8�Z�������� hK�(P:��Q- �8�n�Z���܃e貾�<�1�YT<�,�����"�6{/ �?�͟��|1�:�#g��W�>$����d��J��d�B��=��jf[��%rE^��il:��B���x���Sּ�1հ��,�=��*�7 fcG��#q� �eh?��2�7�����,�!7x��6�n�LC�4x��},Geǝ�tC.��vS �F�43��zz\��;QYC,6����~;RYS/6���|2���5���v��T��i����������mlv��������&� �nRh^ejR�LG�f���? �ۉҬܦƩ��|��Ȱ����>3����!v��i�ʯ�>�v��オ�X3e���_1z�Kȗ\<������!�8���V��]��?b�k41�Re��T�q��mz��TiOʦ�Z��Xq���L������q"+���2ۨ��8}�&N7XU7Ap�d�X��~�׿��&4e�o�F��� �H����O���č�c�� 懴�6���͉��+)��v;j��ݷ�� �UV�� i��� j���Y9GdÒJ1��詞�����V?h��l����l�cGs�ځ�������y�Ac�����\V3�? �� ܙg�>qH�S,�E�W�[�㺨�uch�⍸�O�}���a��>�q�6�n6����N6�q������N ! 1AQaq�0@����"2BRb�#Pr���3C`��Scst���$4D���%Td�� ?���N����a��3��m���C���w��������xA�m�q�m���m������$����4n淿t'��C"w��zU=D�\R+w�p+Y�T�&�պ@��ƃ��3ޯ?�Aﶂ��aŘ���@-�����Q�=���9D��ռ�ѻ@��M�V��P��܅�G5�f�Y<�u=,EC)�<�Fy'�"�&�չ�X~f��l�KԆV��?�� �W�N����=(� �;���{�r����ٌ�Y���h{�١������jW����P���Tc�����X�K�r��}���w�R��%��?���E��m�� �Y�q|����\lEE4���r���}�lsI�Y������f�$�=�d�yO����p�����yBj8jU�o�/�S��?�U��*������ˍ�0������u�q�m [�?f����a�� )Q�>����6#������� ?����0UQ����,IX���(6ڵ[�DI�MNލ�c&���υ�j\��X�R|,4��� j������T�hA�e��^���d���b<����n�� �즇�=!���3�^�`j�h�ȓr��jẕ�c�,ٞX����-����a�ﶔ���#�$��]w�O��Ӫ�1y%��L�Y<�wg#�ǝ�̗`�x�xa�t�w��»1���o7o5��>�m뭛C���Uƃߜ}�C���y1Xνm�F8�jI���]����H���ۺиE@I�i;r�8ӭ����V�F�Շ| ��&?�3|x�B�MuS�Ge�=Ӕ�#BE5G�����Y!z��_e��q�р/W>|-�Ci߇�t�1ޯќd�R3�u��g�=0 5��[?�#͏��q�cf���H��{ ?u�=?�?ǯ���}Z��z���hmΔ�BFTW�����<�q�(v� ��!��z���iW]*�J�V�z��gX֧A�q�&��/w���u�gYӘa���; �i=����g:��?2�dž6�ى�k�4�>�Pxs����}������G�9��3 ���)gG�R<>r h�$��'nc�h�P��Bj��J�ҧH� -��N1���N��?��~��}-q!=��_2hc�M��l�vY%UE�@|�v����M2�.Y[|y�"Eï��K�ZF,�ɯ?,q�?v�M 80jx�"�;�9vk�����+ ֧�� �ȺU��?�%�vcV��mA�6��Qg^M����A}�3�nl� QRN�l8�kkn�'�����(��M�7m9و�q���%ޟ���*h$Zk"��$�9��: �?U8�Sl��,,|ɒ��xH(ѷ����Gn�/Q�4�P��G�%��Ա8�N��!� �&�7�;���eKM7�4��9R/%����l�c>�x;������>��C�:�����t��h?aKX�bhe�ᜋ^�$�Iհ �hr7%F$�E��Fd���t��5���+�(M6�t����Ü�UU|zW�=a�Ts�Tg������dqP�Q����b'�m���1{|Y����X�N��b �P~��F^F:����k6�"�j!�� �I�r�`��1&�-$�Bevk:y���#yw��I0��x��=D�4��tU���P�ZH��ڠ底taP��6����b>�xa����Q�#� WeF��ŮNj�p�J* mQ�N����*I�-*�ȩ�F�g�3 �5��V�ʊ�ɮ�a��5F���O@{���NX��?����H�]3��1�Ri_u��������ѕ�� ����0��� F��~��:60�p�͈�S��qX#a�5>���`�o&+�<2�D����: �������ڝ�$�nP���*)�N�|y�Ej�F�5ټ�e���ihy�Z �>���k�bH�a�v��h�-#���!�Po=@k̆IEN��@��}Ll?j�O������߭�ʞ���Q|A07x���wt!xf���I2?Z��<ץ�T���cU�j��]��陎Ltl �}5�ϓ��$�,��O�mˊ�;�@O��jE��j(�ا,��LX���LO���Ц�90�O �.����a��nA���7������j4 ��W��_ٓ���zW�jcB������y՗+EM�)d���N�g6�y1_x��p�$Lv:��9�"z��p���ʙ$��^��JԼ*�ϭ����o���=x�Lj�6�J��u82�A�H�3$�ٕ@�=Vv�]�'�qEz�;I˼��)��=��ɯ���x �/�W(V���p�����$ �m�������u�����񶤑Oqˎ�T����r��㠚x�sr�GC��byp�G��1ߠ�w e�8�$⿄����/�M{*}��W�]˷.�CK\�ުx���/$�WPw���r� |i���&�}�{�X� �>��$-��l���?-z���g����lΆ���(F���h�vS*���b���߲ڡn,|)mrH[���a�3�ר�[1��3o_�U�3�TC�$��(�=�)0�kgP���� ��u�^=��4 �WYCҸ:��vQ�ר�X�à��tk�m,�t*��^�,�}D*� �"(�I��9R����>`�`��[~Q]�#af��i6l��8���6�:,s�s�N6�j"�A4���IuQ��6E,�GnH��zS�HO�uk�5$�I�4��ؤ�Q9�@��C����wp�BGv[]�u�Ov���0I4���\��y�����Q�Ѹ��~>Z��8�T��a��q�ޣ;z��a���/��S��I:�ܫ_�|������>=Z����8:�S��U�I�J��"IY���8%b8���H��:�QO�6�;7�I�S��J��ҌAά3��>c���E+&jf$eC+�z�;��V����� �r���ʺ������my�e���aQ�f&��6�ND��.:��NT�vm�<- u���ǝ\MvZY�N�NT��-A�>jr!S��n�O 1�3�Ns�%�3D@���`������ܟ 1�^c<���� �a�ɽ�̲�Xë#�w�|y�cW�=�9I*H8�p�^(4���՗�k��arOcW�tO�\�ƍR��8����'�K���I�Q�����?5�>[�}��yU�ײ -h��=��% q�ThG�2�)���"ו3]�!kB��*p�FDl�A���,�eEi�H�f�Ps�����5�H:�Փ~�H�0Dت�D�I����h�F3�������c��2���E��9�H��5�zԑ�ʚ�i�X�=:m�xg�hd(�v����׊�9iS��O��d@0ڽ���:�p�5�h-��t�&���X�q�ӕ,��ie�|���7A�2���O%P��E��htj��Y1��w�Ѓ!����  ���� ࢽ��My�7�\�a�@�ţ�J �4�Ȼ�F�@o�̒?4�wx��)��]�P��~�����u�����5�����7X ��9��^ܩ�U;Iꭆ 5 �������eK2�7(�{|��Y׎ �V��\"���Z�1� Z�����}��(�Ǝ"�1S���_�vE30>���p;� ΝD��%x�W�?W?v����o�^V�i�d��r[��/&>�~`�9Wh��y�;���R��� ;;ɮT��?����r$�g1�K����A��C��c��K��l:�'��3 c�ﳯ*"t8�~l��)���m��+U,z��`(�>yJ�?����h>��]��v��ЍG*�{`��;y]��I�T� ;c��NU�fo¾h���/$���|NS���1�S�"�H��V���T���4��uhǜ�]�v;���5�͠x��'C\�SBpl���h}�N����� A�Bx���%��ޭ�l��/����T��w�ʽ]D�=����K���ž�r㻠l4�S�O?=�k �M:� ��c�C�a�#ha���)�ѐxc�s���gP�iG��{+���x���Q���I= �� z��ԫ+ �8"�k�ñ�j=|����c ��y��CF��/��*9ж�h{ �?4�o� ��k�m�Q�N�x��;�Y��4膚�a�w?�6�>e]�����Q�r�:����g�,i"�����ԩA�*M�<�G��b�if��l^M��5� �Ҩ�{����6J��ZJ�����P�*�����Y���ݛu�_4�9�I8�7���������,^ToR���m4�H��?�N�S�ѕw��/S��甍�@�9H�S�T��t�ƻ���ʒU��*{Xs�@����f�����֒Li�K{H�w^���������Ϥm�tq���s� ���ք��f:��o~s��g�r��ט� �S�ѱC�e]�x���a��) ���(b-$(�j>�7q�B?ӕ�F��hV25r[7 Y� }L�R��}����*sg+��x�r�2�U=�*'WS��ZDW]�WǞ�<��叓���{�$�9Ou4��y�90-�1�'*D`�c�^o?(�9��u���ݐ��'PI&� f�Jݮ�������:wS����jfP1F:X �H�9dԯ���˝[�_54 �}*;@�ܨ�� ð�yn�T���?�ןd�#���4rG�ͨ��H�1�|-#���Mr�S3��G�3�����)�.᧏3v�z֑��r����$G"�`j �1t��x0<Ɔ�Wh6�y�6��,œ�Ga��gA����y��b��)��h�D��ß�_�m��ü �gG;��e�v��ݝ�nQ� ��C����-�*��o���y�a��M��I�>�<���]obD��"�:���G�A��-\%LT�8���c�)��+y76���o�Q�#*{�(F�⽕�y����=���rW�\p���۩�c���A���^e6��K������ʐ�cVf5$�'->���ՉN"���F�"�UQ@�f��Gb~��#�&�M=��8�ט�JNu9��D��[̤�s�o�~������ G��9T�tW^g5y$b��Y'��س�Ǵ�=��U-2 #�MC�t(�i� �lj�@Q 5�̣i�*�O����s�x�K�f��}\��M{E�V�{�υ��Ƈ�����);�H����I��fe�Lȣr�2��>��W�I�Ȃ6������i��k�� �5�YOxȺ����>��Y�f5'��|��H+��98pj�n�.O�y�������jY��~��i�w'������l�;�s�2��Y��:'lg�ꥴ)o#'Sa�a�K��Z� �m��}�`169�n���"���x��I ��*+� }F<��cГ���F�P�������ֹ*�PqX�x۩��,� ��N�� �4<-����%����:��7����W���u�`����� $�?�I��&����o��o��`v�>��P��"��l���4��5'�Z�gE���8���?��[�X�7(��.Q�-��*���ތL@̲����v��.5���[��=�t\+�CNܛ��,g�SQnH����}*F�G16���&:�t��4ُ"A��̣��$�b �|����#rs��a�����T�� ]�<�j��BS�('$�ɻ� �wP;�/�n��?�ݜ��x�F��yUn�~mL*-�������Xf�wd^�a�}��f�,=t�׵i�.2/wpN�Ep8�OР���•��R�FJ� 55TZ��T �ɭ�<��]��/�0�r�@�f��V��V����Nz�G��^���7hZi����k��3�,kN�e|�vg�1{9]_i��X5y7� 8e]�U����'�-2,���e"����]ot�I��Y_��n�(JҼ��1�O ]bXc���Nu�No��pS���Q_���_�?i�~�x h5d'�(qw52] ��'ޤ�q��o1�R!���`ywy�A4u���h<קy���\[~�4�\ X�Wt/� 6�����n�F�a8��f���z �3$�t(���q��q�x��^�XWeN'p<-v�!�{�(>ӽDP7��ո0�y)�e$ٕv�Ih'Q�EA�m*�H��RI��=:��� ���4牢) �%_iN�ݧ�l]� �Nt���G��H�L��� ɱ�g<���1V�,�J~�ٹ�"K��Q�� 9�HS�9�?@��k����r�;we݁�]I�!{ �@�G�[�"��`���J:�n]�{�cA�E����V��ʆ���#��U9�6����j�#Y�m\��q�e4h�B�7��C�������d<�?J����1g:ٳ���=Y���D�p�ц� ׈ǔ��1�]26؜oS�'��9�V�FVu�P�h�9�xc�oq�X��p�o�5��Ա5$�9W�V(�[Ak�aY錎qf;�'�[�|���b�6�Ck��)��#a#a˙��8���=äh�4��2��C��4tm^ �n'c���]GQ$[Wҿ��i���vN�{Fu ��1�gx��1┷���N�m��{j-,��x�� Ūm�ЧS�[�s���Gna���䑴�� x�p 8<������97�Q���ϴ�v�aϚG��Rt�Һ׈�f^\r��WH�JU�7Z���y)�vg=����n��4�_)y��D'y�6�]�c�5̪�\� �PF�k����&�c;��cq�$~T�7j ���nç]�<�g ":�to�t}�159�<�/�8������m�b�K#g'I'.W�����6��I/��>v��\�MN��g���m�A�yQL�4u�Lj�j9��#44�t��l^�}L����n��R��!��t��±]��r��h6ٍ>�yҏ�N��fU�� ���� Fm@�8}�/u��jb9������he:A�y�ծw��GpΧh�5����l}�3p468��)U��d��c����;Us/�֔�YX�1�O2��uq�s��`hwg�r~�{ R��mhN��؎*q 42�*th��>�#���E����#��Hv�O����q�}�����6�e��\�,Wk�#���X��b>��p}�դ��3���T5��†��6��[��@�P�y*n��|'f�֧>�lư΂�̺����SU�'*�q�p�_S�����M�� '��c�6�����m�� ySʨ;M��r���Ƌ�m�Kxo,���Gm�P��A�G�:��i��w�9�}M(�^�V��$ǒ�ѽ�9���|���� �a����J�SQ�a���r�B;����}���ٻ֢�2�%U���c�#�g���N�a�ݕ�'�v�[�OY'��3L�3�;,p�]@�S��{ls��X�'���c�jw�k'a�.��}�}&�� �dP�*�bK=ɍ!����;3n�gΊU�ߴmt�'*{,=SzfD� A��ko~�G�aoq�_mi}#�m�������P�Xhύ����mxǍ�΂���巿zf��Q���c���|kc�����?���W��Y�$���_Lv����l߶��c���`?����l�j�ݲˏ!V��6����U�Ђ(A���4y)H���p�Z_�x��>���e��R��$�/�`^'3qˏ�-&Q�=?��CFVR �D�fV�9��{�8g�������n�h�(P"��6�[�D���< E�����~0<@�`�G�6����Hг�cc�� �c�K.5��D��d�B���`?�XQ��2��ٿyqo&+�1^� DW�0�ꊩ���G�#��Q�nL3��c���������/��x ��1�1[y�x�პCW��C�c�UĨ80�m�e�4.{�m��u���I=��f�����0QRls9���f���������9���~f�����Ǩ��a�"@�8���ȁ�Q����#c�ic������G��$���G���r/$W�(��W���V�"��m�7�[m�A�m����bo��D� j����۳� l���^�k�h׽����� ��#� iXn�v��eT�k�a�^Y�4�BN��ĕ��0 !01@Q"2AaPq3BR������?���@4�Q�����T3,���㺠�W�[=JK�Ϟ���2�r^7��vc�:�9 �E�ߴ�w�S#d���Ix��u��:��Hp��9E!�� V 2;73|F��9Y���*ʬ�F��D����u&���y؟��^EA��A��(ɩ���^��GV:ݜDy�`��Jr29ܾ�㝉��[���E;Fzx��YG��U�e�Y�C���� ����v-tx����I�sם�Ę�q��Eb�+P\ :>�i�C'�;�����k|z�رn�y]�#ǿb��Q��������w�����(�r|ӹs��[�D��2v-%��@;�8<a���[\o[ϧw��I!��*0�krs)�[�J9^��ʜ��p1)� "��/_>��o��<1����A�E�y^�C��`�x1'ܣn�p��s`l���fQ��):�l����b>�Me�jH^?�kl3(�z:���1ŠK&?Q�~�{�ٺ�h�y���/�[��V�|6��}�KbX����mn[-��7�5q�94�������dm���c^���h� X��5��<�eޘ>G���-�}�دB�ޟ� ��|�rt�M��V+�]�c?�-#ڛ��^ǂ}���Lkr���O��u�>�-D�ry� D?:ޞ�U��ǜ�7�V��?瓮�"�#���r��չģVR;�n���/_� ؉v�ݶe5d�b9��/O��009�G���5n�W����JpA�*�r9�>�1��.[t���s�F���nQ� V 77R�]�ɫ8����_0<՜�IF�u(v��4��F�k�3��E)��N:��yڮe��P�`�1}�$WS��J�SQ�N�j�ٺ��޵�#l���ј(�5=��5�lǏmoW�v-�1����v,W�mn��߀$x�<����v�j(����c]��@#��1������Ǔ���o'��u+����;G�#�޸��v-lη��/(`i⣍Pm^���ԯ̾9Z��F��������n��1��� ��]�[��)�'������:�֪�W��FC����� �B9،!?���]��V��A�Վ�M��b�w��G F>_DȬ0¤�#�QR�[V��kz���m�w�"��9ZG�7'[��=�Q����j8R?�zf�\a�=��O�U����*oB�A�|G���2�54 �p��.w7� �� ��&������ξxGHp� B%��$g�����t�Џ򤵍z���HN�u�Я�-�'4��0��;_��3 !01"@AQa2Pq#3BR������?��ʩca��en��^��8���<�u#��m*08r��y�N"�<�Ѳ0��@\�p��� �����Kv�D��J8�Fҽ� �f�Y��-m�ybX�NP����}�!*8t(�OqѢ��Q�wW�K��ZD��Δ^e��!� ��B�K��p~�����e*l}z#9ң�k���q#�Ft�o��S�R����-�w�!�S���Ӥß|M�l޶V��!eˈ�8Y���c�ЮM2��tk���� ������J�fS����Ö*i/2�����n]�k�\���|4yX�8��U�P.���Ы[���l��@"�t�<������5�lF���vU�����W��W��;�b�cД^6[#7@vU�xgZv��F�6��Q,K�v��� �+Ъ��n��Ǣ��Ft���8��0��c�@�!�Zq s�v�t�;#](B��-�nῃ~���3g������5�J�%���O������n�kB�ĺ�.r��+���#�N$?�q�/�s�6��p��a����a��J/��M�8��6�ܰ"�*������ɗud"\w���aT(����[��F��U՛����RT�b���n�*��6���O��SJ�.�ij<�v�MT��R\c��5l�sZB>F��<7�;EA��{��E���Ö��1U/�#��d1�a�n.1ě����0�ʾR�h��|�R��Ao�3�m3 ��%�� ���28Q� ��y��φ���H�To�7�lW>����#i`�q���c����a��� �m,B�-j����݋�'mR1Ήt�>��V��p���s�0IbI�C.���1R�ea�����]H�6����������4B>��o��](��$B���m�����a�!=��?�B� K�Ǿ+�Ծ"�n���K��*��+��[T#�{E�J�S����Q�����s�5�:�U�\wĐ�f�3����܆&�)����I���Ԇw��E T�lrTf6Q|R�h:��[K�� �z��c֧�G�C��%\��_�a�84��HcO�bi��ؖV��7H �)*ģK~Xhչ0��4?�0��� �E<���}3���#���u�?�� ��|g�S�6ꊤ�|�I#Hڛ� �ա��w�X��9��7���Ŀ%�SL��y6č��|�F�a 8���b��$�sק�h���b9RAu7�˨p�Č�_\*w��묦��F ����4D~�f����|(�"m���NK��i�S�>�$d7SlA��/�²����SL��|6N�}���S�˯���g��]6��; �#�.��<���q'Q�1|KQ$�����񛩶"�$r�b:���N8�w@��8$�� �AjfG|~�9F ���Y��ʺ��Bwؒ������M:I岎�G��`s�YV5����6��A �b:�W���G�q%l�����F��H���7�������Fsv7��k�� 403WebShell
403Webshell
Server IP : 213.165.242.4  /  Your IP : 216.73.216.78
Web Server : Apache
System : Linux amsngx344.inmotionhosting.com 4.18.0-553.40.1.lve.el8.x86_64 #1 SMP Wed Feb 12 18:54:57 UTC 2025 x86_64
User : aquafi9 ( 1305)
PHP Version : 7.4.33
Disable Function : NONE
MySQL : OFF  |  cURL : ON  |  WGET : ON  |  Perl : ON  |  Python : ON  |  Sudo : ON  |  Pkexec : ON
Directory :  /opt/imh-python/lib/python3.9/site-packages/tornado/test/

Upload File :
current_dir [ Writeable ] document_root [ Writeable ]

 

Command :

[ Back ]     

Current File : /opt/imh-python/lib/python3.9/site-packages/tornado/test/web_test.py
from tornado.concurrent import Future
from tornado import gen
from tornado.escape import (
    json_decode,
    utf8,
    to_unicode,
    recursive_unicode,
    native_str,
    to_basestring,
)
from tornado.httpclient import HTTPClientError
from tornado.httputil import format_timestamp
from tornado.iostream import IOStream
from tornado import locale
from tornado.locks import Event
from tornado.log import app_log, gen_log
from tornado.simple_httpclient import SimpleAsyncHTTPClient
from tornado.template import DictLoader
from tornado.testing import AsyncHTTPTestCase, AsyncTestCase, ExpectLog, gen_test
from tornado.util import ObjectDict, unicode_type
from tornado.web import (
    Application,
    RequestHandler,
    StaticFileHandler,
    RedirectHandler as WebRedirectHandler,
    HTTPError,
    MissingArgumentError,
    ErrorHandler,
    authenticated,
    url,
    _create_signature_v1,
    create_signed_value,
    decode_signed_value,
    get_signature_key_version,
    UIModule,
    Finish,
    stream_request_body,
    removeslash,
    addslash,
    GZipContentEncoding,
)

import binascii
import contextlib
import copy
import datetime
import email.utils
import gzip
from io import BytesIO
import itertools
import logging
import os
import re
import socket
import typing  # noqa: F401
import unittest
import urllib.parse


def relpath(*a):
    return os.path.join(os.path.dirname(__file__), *a)


class WebTestCase(AsyncHTTPTestCase):
    """Base class for web tests that also supports WSGI mode.

    Override get_handlers and get_app_kwargs instead of get_app.
    This class is deprecated since WSGI mode is no longer supported.
    """

    def get_app(self):
        self.app = Application(self.get_handlers(), **self.get_app_kwargs())
        return self.app

    def get_handlers(self):
        raise NotImplementedError()

    def get_app_kwargs(self):
        return {}


class SimpleHandlerTestCase(WebTestCase):
    """Simplified base class for tests that work with a single handler class.

    To use, define a nested class named ``Handler``.
    """

    def get_handlers(self):
        return [("/", self.Handler)]


class HelloHandler(RequestHandler):
    def get(self):
        self.write("hello")


class CookieTestRequestHandler(RequestHandler):
    # stub out enough methods to make the secure_cookie functions work
    def __init__(self, cookie_secret="0123456789", key_version=None):
        # don't call super.__init__
        self._cookies = {}  # type: typing.Dict[str, bytes]
        if key_version is None:
            self.application = ObjectDict(settings=dict(cookie_secret=cookie_secret))
        else:
            self.application = ObjectDict(
                settings=dict(cookie_secret=cookie_secret, key_version=key_version)
            )

    def get_cookie(self, name):
        return self._cookies.get(name)

    def set_cookie(self, name, value, expires_days=None):
        self._cookies[name] = value


# See SignedValueTest below for more.
class SecureCookieV1Test(unittest.TestCase):
    def test_round_trip(self):
        handler = CookieTestRequestHandler()
        handler.set_secure_cookie("foo", b"bar", version=1)
        self.assertEqual(handler.get_secure_cookie("foo", min_version=1), b"bar")

    def test_cookie_tampering_future_timestamp(self):
        handler = CookieTestRequestHandler()
        # this string base64-encodes to '12345678'
        handler.set_secure_cookie("foo", binascii.a2b_hex(b"d76df8e7aefc"), version=1)
        cookie = handler._cookies["foo"]
        match = re.match(br"12345678\|([0-9]+)\|([0-9a-f]+)", cookie)
        assert match is not None
        timestamp = match.group(1)
        sig = match.group(2)
        self.assertEqual(
            _create_signature_v1(
                handler.application.settings["cookie_secret"],
                "foo",
                "12345678",
                timestamp,
            ),
            sig,
        )
        # shifting digits from payload to timestamp doesn't alter signature
        # (this is not desirable behavior, just confirming that that's how it
        # works)
        self.assertEqual(
            _create_signature_v1(
                handler.application.settings["cookie_secret"],
                "foo",
                "1234",
                b"5678" + timestamp,
            ),
            sig,
        )
        # tamper with the cookie
        handler._cookies["foo"] = utf8(
            "1234|5678%s|%s" % (to_basestring(timestamp), to_basestring(sig))
        )
        # it gets rejected
        with ExpectLog(gen_log, "Cookie timestamp in future"):
            self.assertTrue(handler.get_secure_cookie("foo", min_version=1) is None)

    def test_arbitrary_bytes(self):
        # Secure cookies accept arbitrary data (which is base64 encoded).
        # Note that normal cookies accept only a subset of ascii.
        handler = CookieTestRequestHandler()
        handler.set_secure_cookie("foo", b"\xe9", version=1)
        self.assertEqual(handler.get_secure_cookie("foo", min_version=1), b"\xe9")


# See SignedValueTest below for more.
class SecureCookieV2Test(unittest.TestCase):
    KEY_VERSIONS = {0: "ajklasdf0ojaisdf", 1: "aslkjasaolwkjsdf"}

    def test_round_trip(self):
        handler = CookieTestRequestHandler()
        handler.set_secure_cookie("foo", b"bar", version=2)
        self.assertEqual(handler.get_secure_cookie("foo", min_version=2), b"bar")

    def test_key_version_roundtrip(self):
        handler = CookieTestRequestHandler(
            cookie_secret=self.KEY_VERSIONS, key_version=0
        )
        handler.set_secure_cookie("foo", b"bar")
        self.assertEqual(handler.get_secure_cookie("foo"), b"bar")

    def test_key_version_roundtrip_differing_version(self):
        handler = CookieTestRequestHandler(
            cookie_secret=self.KEY_VERSIONS, key_version=1
        )
        handler.set_secure_cookie("foo", b"bar")
        self.assertEqual(handler.get_secure_cookie("foo"), b"bar")

    def test_key_version_increment_version(self):
        handler = CookieTestRequestHandler(
            cookie_secret=self.KEY_VERSIONS, key_version=0
        )
        handler.set_secure_cookie("foo", b"bar")
        new_handler = CookieTestRequestHandler(
            cookie_secret=self.KEY_VERSIONS, key_version=1
        )
        new_handler._cookies = handler._cookies
        self.assertEqual(new_handler.get_secure_cookie("foo"), b"bar")

    def test_key_version_invalidate_version(self):
        handler = CookieTestRequestHandler(
            cookie_secret=self.KEY_VERSIONS, key_version=0
        )
        handler.set_secure_cookie("foo", b"bar")
        new_key_versions = self.KEY_VERSIONS.copy()
        new_key_versions.pop(0)
        new_handler = CookieTestRequestHandler(
            cookie_secret=new_key_versions, key_version=1
        )
        new_handler._cookies = handler._cookies
        self.assertEqual(new_handler.get_secure_cookie("foo"), None)


class FinalReturnTest(WebTestCase):
    def get_handlers(self):
        test = self

        class FinishHandler(RequestHandler):
            @gen.coroutine
            def get(self):
                test.final_return = self.finish()
                yield test.final_return

            @gen.coroutine
            def post(self):
                self.write("hello,")
                yield self.flush()
                test.final_return = self.finish("world")
                yield test.final_return

        class RenderHandler(RequestHandler):
            def create_template_loader(self, path):
                return DictLoader({"foo.html": "hi"})

            @gen.coroutine
            def get(self):
                test.final_return = self.render("foo.html")

        return [("/finish", FinishHandler), ("/render", RenderHandler)]

    def get_app_kwargs(self):
        return dict(template_path="FinalReturnTest")

    def test_finish_method_return_future(self):
        response = self.fetch(self.get_url("/finish"))
        self.assertEqual(response.code, 200)
        self.assertIsInstance(self.final_return, Future)
        self.assertTrue(self.final_return.done())

        response = self.fetch(self.get_url("/finish"), method="POST", body=b"")
        self.assertEqual(response.code, 200)
        self.assertIsInstance(self.final_return, Future)
        self.assertTrue(self.final_return.done())

    def test_render_method_return_future(self):
        response = self.fetch(self.get_url("/render"))
        self.assertEqual(response.code, 200)
        self.assertIsInstance(self.final_return, Future)


class CookieTest(WebTestCase):
    def get_handlers(self):
        class SetCookieHandler(RequestHandler):
            def get(self):
                # Try setting cookies with different argument types
                # to ensure that everything gets encoded correctly
                self.set_cookie("str", "asdf")
                self.set_cookie("unicode", u"qwer")
                self.set_cookie("bytes", b"zxcv")

        class GetCookieHandler(RequestHandler):
            def get(self):
                self.write(self.get_cookie("foo", "default"))

        class SetCookieDomainHandler(RequestHandler):
            def get(self):
                # unicode domain and path arguments shouldn't break things
                # either (see bug #285)
                self.set_cookie("unicode_args", "blah", domain=u"foo.com", path=u"/foo")

        class SetCookieSpecialCharHandler(RequestHandler):
            def get(self):
                self.set_cookie("equals", "a=b")
                self.set_cookie("semicolon", "a;b")
                self.set_cookie("quote", 'a"b')

        class SetCookieOverwriteHandler(RequestHandler):
            def get(self):
                self.set_cookie("a", "b", domain="example.com")
                self.set_cookie("c", "d", domain="example.com")
                # A second call with the same name clobbers the first.
                # Attributes from the first call are not carried over.
                self.set_cookie("a", "e")

        class SetCookieMaxAgeHandler(RequestHandler):
            def get(self):
                self.set_cookie("foo", "bar", max_age=10)

        class SetCookieExpiresDaysHandler(RequestHandler):
            def get(self):
                self.set_cookie("foo", "bar", expires_days=10)

        class SetCookieFalsyFlags(RequestHandler):
            def get(self):
                self.set_cookie("a", "1", secure=True)
                self.set_cookie("b", "1", secure=False)
                self.set_cookie("c", "1", httponly=True)
                self.set_cookie("d", "1", httponly=False)

        return [
            ("/set", SetCookieHandler),
            ("/get", GetCookieHandler),
            ("/set_domain", SetCookieDomainHandler),
            ("/special_char", SetCookieSpecialCharHandler),
            ("/set_overwrite", SetCookieOverwriteHandler),
            ("/set_max_age", SetCookieMaxAgeHandler),
            ("/set_expires_days", SetCookieExpiresDaysHandler),
            ("/set_falsy_flags", SetCookieFalsyFlags),
        ]

    def test_set_cookie(self):
        response = self.fetch("/set")
        self.assertEqual(
            sorted(response.headers.get_list("Set-Cookie")),
            ["bytes=zxcv; Path=/", "str=asdf; Path=/", "unicode=qwer; Path=/"],
        )

    def test_get_cookie(self):
        response = self.fetch("/get", headers={"Cookie": "foo=bar"})
        self.assertEqual(response.body, b"bar")

        response = self.fetch("/get", headers={"Cookie": 'foo="bar"'})
        self.assertEqual(response.body, b"bar")

        response = self.fetch("/get", headers={"Cookie": "/=exception;"})
        self.assertEqual(response.body, b"default")

    def test_set_cookie_domain(self):
        response = self.fetch("/set_domain")
        self.assertEqual(
            response.headers.get_list("Set-Cookie"),
            ["unicode_args=blah; Domain=foo.com; Path=/foo"],
        )

    def test_cookie_special_char(self):
        response = self.fetch("/special_char")
        headers = sorted(response.headers.get_list("Set-Cookie"))
        self.assertEqual(len(headers), 3)
        self.assertEqual(headers[0], 'equals="a=b"; Path=/')
        self.assertEqual(headers[1], 'quote="a\\"b"; Path=/')
        # python 2.7 octal-escapes the semicolon; older versions leave it alone
        self.assertTrue(
            headers[2] in ('semicolon="a;b"; Path=/', 'semicolon="a\\073b"; Path=/'),
            headers[2],
        )

        data = [
            ("foo=a=b", "a=b"),
            ('foo="a=b"', "a=b"),
            ('foo="a;b"', '"a'),  # even quoted, ";" is a delimiter
            ("foo=a\\073b", "a\\073b"),  # escapes only decoded in quotes
            ('foo="a\\073b"', "a;b"),
            ('foo="a\\"b"', 'a"b'),
        ]
        for header, expected in data:
            logging.debug("trying %r", header)
            response = self.fetch("/get", headers={"Cookie": header})
            self.assertEqual(response.body, utf8(expected))

    def test_set_cookie_overwrite(self):
        response = self.fetch("/set_overwrite")
        headers = response.headers.get_list("Set-Cookie")
        self.assertEqual(
            sorted(headers), ["a=e; Path=/", "c=d; Domain=example.com; Path=/"]
        )

    def test_set_cookie_max_age(self):
        response = self.fetch("/set_max_age")
        headers = response.headers.get_list("Set-Cookie")
        self.assertEqual(sorted(headers), ["foo=bar; Max-Age=10; Path=/"])

    def test_set_cookie_expires_days(self):
        response = self.fetch("/set_expires_days")
        header = response.headers.get("Set-Cookie")
        match = re.match("foo=bar; expires=(?P<expires>.+); Path=/", header)
        assert match is not None

        expires = datetime.datetime.utcnow() + datetime.timedelta(days=10)
        parsed = email.utils.parsedate(match.groupdict()["expires"])
        assert parsed is not None
        header_expires = datetime.datetime(*parsed[:6])
        self.assertTrue(abs((expires - header_expires).total_seconds()) < 10)

    def test_set_cookie_false_flags(self):
        response = self.fetch("/set_falsy_flags")
        headers = sorted(response.headers.get_list("Set-Cookie"))
        # The secure and httponly headers are capitalized in py35 and
        # lowercase in older versions.
        self.assertEqual(headers[0].lower(), "a=1; path=/; secure")
        self.assertEqual(headers[1].lower(), "b=1; path=/")
        self.assertEqual(headers[2].lower(), "c=1; httponly; path=/")
        self.assertEqual(headers[3].lower(), "d=1; path=/")


class AuthRedirectRequestHandler(RequestHandler):
    def initialize(self, login_url):
        self.login_url = login_url

    def get_login_url(self):
        return self.login_url

    @authenticated
    def get(self):
        # we'll never actually get here because the test doesn't follow redirects
        self.send_error(500)


class AuthRedirectTest(WebTestCase):
    def get_handlers(self):
        return [
            ("/relative", AuthRedirectRequestHandler, dict(login_url="/login")),
            (
                "/absolute",
                AuthRedirectRequestHandler,
                dict(login_url="http://example.com/login"),
            ),
        ]

    def test_relative_auth_redirect(self):
        response = self.fetch(self.get_url("/relative"), follow_redirects=False)
        self.assertEqual(response.code, 302)
        self.assertEqual(response.headers["Location"], "/login?next=%2Frelative")

    def test_absolute_auth_redirect(self):
        response = self.fetch(self.get_url("/absolute"), follow_redirects=False)
        self.assertEqual(response.code, 302)
        self.assertTrue(
            re.match(
                r"http://example.com/login\?next=http%3A%2F%2F127.0.0.1%3A[0-9]+%2Fabsolute",
                response.headers["Location"],
            ),
            response.headers["Location"],
        )


class ConnectionCloseHandler(RequestHandler):
    def initialize(self, test):
        self.test = test

    @gen.coroutine
    def get(self):
        self.test.on_handler_waiting()
        yield self.test.cleanup_event.wait()

    def on_connection_close(self):
        self.test.on_connection_close()


class ConnectionCloseTest(WebTestCase):
    def get_handlers(self):
        self.cleanup_event = Event()
        return [("/", ConnectionCloseHandler, dict(test=self))]

    def test_connection_close(self):
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
        s.connect(("127.0.0.1", self.get_http_port()))
        self.stream = IOStream(s)
        self.stream.write(b"GET / HTTP/1.0\r\n\r\n")
        self.wait()
        # Let the hanging coroutine clean up after itself
        self.cleanup_event.set()
        self.io_loop.run_sync(lambda: gen.sleep(0))

    def on_handler_waiting(self):
        logging.debug("handler waiting")
        self.stream.close()

    def on_connection_close(self):
        logging.debug("connection closed")
        self.stop()


class EchoHandler(RequestHandler):
    def get(self, *path_args):
        # Type checks: web.py interfaces convert argument values to
        # unicode strings (by default, but see also decode_argument).
        # In httpserver.py (i.e. self.request.arguments), they're left
        # as bytes.  Keys are always native strings.
        for key in self.request.arguments:
            if type(key) != str:
                raise Exception("incorrect type for key: %r" % type(key))
            for value in self.request.arguments[key]:
                if type(value) != bytes:
                    raise Exception("incorrect type for value: %r" % type(value))
            for value in self.get_arguments(key):
                if type(value) != unicode_type:
                    raise Exception("incorrect type for value: %r" % type(value))
        for arg in path_args:
            if type(arg) != unicode_type:
                raise Exception("incorrect type for path arg: %r" % type(arg))
        self.write(
            dict(
                path=self.request.path,
                path_args=path_args,
                args=recursive_unicode(self.request.arguments),
            )
        )


class RequestEncodingTest(WebTestCase):
    def get_handlers(self):
        return [("/group/(.*)", EchoHandler), ("/slashes/([^/]*)/([^/]*)", EchoHandler)]

    def fetch_json(self, path):
        return json_decode(self.fetch(path).body)

    def test_group_question_mark(self):
        # Ensure that url-encoded question marks are handled properly
        self.assertEqual(
            self.fetch_json("/group/%3F"),
            dict(path="/group/%3F", path_args=["?"], args={}),
        )
        self.assertEqual(
            self.fetch_json("/group/%3F?%3F=%3F"),
            dict(path="/group/%3F", path_args=["?"], args={"?": ["?"]}),
        )

    def test_group_encoding(self):
        # Path components and query arguments should be decoded the same way
        self.assertEqual(
            self.fetch_json("/group/%C3%A9?arg=%C3%A9"),
            {
                u"path": u"/group/%C3%A9",
                u"path_args": [u"\u00e9"],
                u"args": {u"arg": [u"\u00e9"]},
            },
        )

    def test_slashes(self):
        # Slashes may be escaped to appear as a single "directory" in the path,
        # but they are then unescaped when passed to the get() method.
        self.assertEqual(
            self.fetch_json("/slashes/foo/bar"),
            dict(path="/slashes/foo/bar", path_args=["foo", "bar"], args={}),
        )
        self.assertEqual(
            self.fetch_json("/slashes/a%2Fb/c%2Fd"),
            dict(path="/slashes/a%2Fb/c%2Fd", path_args=["a/b", "c/d"], args={}),
        )

    def test_error(self):
        # Percent signs (encoded as %25) should not mess up printf-style
        # messages in logs
        with ExpectLog(gen_log, ".*Invalid unicode"):
            self.fetch("/group/?arg=%25%e9")


class TypeCheckHandler(RequestHandler):
    def prepare(self):
        self.errors = {}  # type: typing.Dict[str, str]

        self.check_type("status", self.get_status(), int)

        # get_argument is an exception from the general rule of using
        # type str for non-body data mainly for historical reasons.
        self.check_type("argument", self.get_argument("foo"), unicode_type)
        self.check_type("cookie_key", list(self.cookies.keys())[0], str)
        self.check_type("cookie_value", list(self.cookies.values())[0].value, str)

        # Secure cookies return bytes because they can contain arbitrary
        # data, but regular cookies are native strings.
        if list(self.cookies.keys()) != ["asdf"]:
            raise Exception(
                "unexpected values for cookie keys: %r" % self.cookies.keys()
            )
        self.check_type("get_secure_cookie", self.get_secure_cookie("asdf"), bytes)
        self.check_type("get_cookie", self.get_cookie("asdf"), str)

        self.check_type("xsrf_token", self.xsrf_token, bytes)
        self.check_type("xsrf_form_html", self.xsrf_form_html(), str)

        self.check_type("reverse_url", self.reverse_url("typecheck", "foo"), str)

        self.check_type("request_summary", self._request_summary(), str)

    def get(self, path_component):
        # path_component uses type unicode instead of str for consistency
        # with get_argument()
        self.check_type("path_component", path_component, unicode_type)
        self.write(self.errors)

    def post(self, path_component):
        self.check_type("path_component", path_component, unicode_type)
        self.write(self.errors)

    def check_type(self, name, obj, expected_type):
        actual_type = type(obj)
        if expected_type != actual_type:
            self.errors[name] = "expected %s, got %s" % (expected_type, actual_type)


class DecodeArgHandler(RequestHandler):
    def decode_argument(self, value, name=None):
        if type(value) != bytes:
            raise Exception("unexpected type for value: %r" % type(value))
        # use self.request.arguments directly to avoid recursion
        if "encoding" in self.request.arguments:
            return value.decode(to_unicode(self.request.arguments["encoding"][0]))
        else:
            return value

    def get(self, arg):
        def describe(s):
            if type(s) == bytes:
                return ["bytes", native_str(binascii.b2a_hex(s))]
            elif type(s) == unicode_type:
                return ["unicode", s]
            raise Exception("unknown type")

        self.write({"path": describe(arg), "query": describe(self.get_argument("foo"))})


class LinkifyHandler(RequestHandler):
    def get(self):
        self.render("linkify.html", message="http://example.com")


class UIModuleResourceHandler(RequestHandler):
    def get(self):
        self.render("page.html", entries=[1, 2])


class OptionalPathHandler(RequestHandler):
    def get(self, path):
        self.write({"path": path})


class MultiHeaderHandler(RequestHandler):
    def get(self):
        self.set_header("x-overwrite", "1")
        self.set_header("X-Overwrite", 2)
        self.add_header("x-multi", 3)
        self.add_header("X-Multi", "4")


class RedirectHandler(RequestHandler):
    def get(self):
        if self.get_argument("permanent", None) is not None:
            self.redirect("/", permanent=int(self.get_argument("permanent")))
        elif self.get_argument("status", None) is not None:
            self.redirect("/", status=int(self.get_argument("status")))
        else:
            raise Exception("didn't get permanent or status arguments")


class EmptyFlushCallbackHandler(RequestHandler):
    @gen.coroutine
    def get(self):
        # Ensure that the flush callback is run whether or not there
        # was any output.  The gen.Task and direct yield forms are
        # equivalent.
        yield self.flush()  # "empty" flush, but writes headers
        yield self.flush()  # empty flush
        self.write("o")
        yield self.flush()  # flushes the "o"
        yield self.flush()  # empty flush
        self.finish("k")


class HeaderInjectionHandler(RequestHandler):
    def get(self):
        try:
            self.set_header("X-Foo", "foo\r\nX-Bar: baz")
            raise Exception("Didn't get expected exception")
        except ValueError as e:
            if "Unsafe header value" in str(e):
                self.finish(b"ok")
            else:
                raise


class GetArgumentHandler(RequestHandler):
    def prepare(self):
        if self.get_argument("source", None) == "query":
            method = self.get_query_argument
        elif self.get_argument("source", None) == "body":
            method = self.get_body_argument
        else:
            method = self.get_argument
        self.finish(method("foo", "default"))


class GetArgumentsHandler(RequestHandler):
    def prepare(self):
        self.finish(
            dict(
                default=self.get_arguments("foo"),
                query=self.get_query_arguments("foo"),
                body=self.get_body_arguments("foo"),
            )
        )


# This test was shared with wsgi_test.py; now the name is meaningless.
class WSGISafeWebTest(WebTestCase):
    COOKIE_SECRET = "WebTest.COOKIE_SECRET"

    def get_app_kwargs(self):
        loader = DictLoader(
            {
                "linkify.html": "{% module linkify(message) %}",
                "page.html": """\
<html><head></head><body>
{% for e in entries %}
{% module Template("entry.html", entry=e) %}
{% end %}
</body></html>""",
                "entry.html": """\
{{ set_resources(embedded_css=".entry { margin-bottom: 1em; }",
                 embedded_javascript="js_embed()",
                 css_files=["/base.css", "/foo.css"],
                 javascript_files="/common.js",
                 html_head="<meta>",
                 html_body='<script src="/analytics.js"/>') }}
<div class="entry">...</div>""",
            }
        )
        return dict(
            template_loader=loader,
            autoescape="xhtml_escape",
            cookie_secret=self.COOKIE_SECRET,
        )

    def tearDown(self):
        super(WSGISafeWebTest, self).tearDown()
        RequestHandler._template_loaders.clear()

    def get_handlers(self):
        urls = [
            url("/typecheck/(.*)", TypeCheckHandler, name="typecheck"),
            url("/decode_arg/(.*)", DecodeArgHandler, name="decode_arg"),
            url("/decode_arg_kw/(?P<arg>.*)", DecodeArgHandler),
            url("/linkify", LinkifyHandler),
            url("/uimodule_resources", UIModuleResourceHandler),
            url("/optional_path/(.+)?", OptionalPathHandler),
            url("/multi_header", MultiHeaderHandler),
            url("/redirect", RedirectHandler),
            url(
                "/web_redirect_permanent",
                WebRedirectHandler,
                {"url": "/web_redirect_newpath"},
            ),
            url(
                "/web_redirect",
                WebRedirectHandler,
                {"url": "/web_redirect_newpath", "permanent": False},
            ),
            url(
                "//web_redirect_double_slash",
                WebRedirectHandler,
                {"url": "/web_redirect_newpath"},
            ),
            url("/header_injection", HeaderInjectionHandler),
            url("/get_argument", GetArgumentHandler),
            url("/get_arguments", GetArgumentsHandler),
        ]
        return urls

    def fetch_json(self, *args, **kwargs):
        response = self.fetch(*args, **kwargs)
        response.rethrow()
        return json_decode(response.body)

    def test_types(self):
        cookie_value = to_unicode(
            create_signed_value(self.COOKIE_SECRET, "asdf", "qwer")
        )
        response = self.fetch(
            "/typecheck/asdf?foo=bar", headers={"Cookie": "asdf=" + cookie_value}
        )
        data = json_decode(response.body)
        self.assertEqual(data, {})

        response = self.fetch(
            "/typecheck/asdf?foo=bar",
            method="POST",
            headers={"Cookie": "asdf=" + cookie_value},
            body="foo=bar",
        )

    def test_decode_argument(self):
        # These urls all decode to the same thing
        urls = [
            "/decode_arg/%C3%A9?foo=%C3%A9&encoding=utf-8",
            "/decode_arg/%E9?foo=%E9&encoding=latin1",
            "/decode_arg_kw/%E9?foo=%E9&encoding=latin1",
        ]
        for req_url in urls:
            response = self.fetch(req_url)
            response.rethrow()
            data = json_decode(response.body)
            self.assertEqual(
                data,
                {u"path": [u"unicode", u"\u00e9"], u"query": [u"unicode", u"\u00e9"]},
            )

        response = self.fetch("/decode_arg/%C3%A9?foo=%C3%A9")
        response.rethrow()
        data = json_decode(response.body)
        self.assertEqual(
            data, {u"path": [u"bytes", u"c3a9"], u"query": [u"bytes", u"c3a9"]}
        )

    def test_decode_argument_invalid_unicode(self):
        # test that invalid unicode in URLs causes 400, not 500
        with ExpectLog(gen_log, ".*Invalid unicode.*"):
            response = self.fetch("/typecheck/invalid%FF")
            self.assertEqual(response.code, 400)
            response = self.fetch("/typecheck/invalid?foo=%FF")
            self.assertEqual(response.code, 400)

    def test_decode_argument_plus(self):
        # These urls are all equivalent.
        urls = [
            "/decode_arg/1%20%2B%201?foo=1%20%2B%201&encoding=utf-8",
            "/decode_arg/1%20+%201?foo=1+%2B+1&encoding=utf-8",
        ]
        for req_url in urls:
            response = self.fetch(req_url)
            response.rethrow()
            data = json_decode(response.body)
            self.assertEqual(
                data,
                {u"path": [u"unicode", u"1 + 1"], u"query": [u"unicode", u"1 + 1"]},
            )

    def test_reverse_url(self):
        self.assertEqual(self.app.reverse_url("decode_arg", "foo"), "/decode_arg/foo")
        self.assertEqual(self.app.reverse_url("decode_arg", 42), "/decode_arg/42")
        self.assertEqual(self.app.reverse_url("decode_arg", b"\xe9"), "/decode_arg/%E9")
        self.assertEqual(
            self.app.reverse_url("decode_arg", u"\u00e9"), "/decode_arg/%C3%A9"
        )
        self.assertEqual(
            self.app.reverse_url("decode_arg", "1 + 1"), "/decode_arg/1%20%2B%201"
        )

    def test_uimodule_unescaped(self):
        response = self.fetch("/linkify")
        self.assertEqual(
            response.body, b'<a href="http://example.com">http://example.com</a>'
        )

    def test_uimodule_resources(self):
        response = self.fetch("/uimodule_resources")
        self.assertEqual(
            response.body,
            b"""\
<html><head><link href="/base.css" type="text/css" rel="stylesheet"/><link href="/foo.css" type="text/css" rel="stylesheet"/>
<style type="text/css">
.entry { margin-bottom: 1em; }
</style>
<meta>
</head><body>


<div class="entry">...</div>


<div class="entry">...</div>

<script src="/common.js" type="text/javascript"></script>
<script type="text/javascript">
//<![CDATA[
js_embed()
//]]>
</script>
<script src="/analytics.js"/>
</body></html>""",  # noqa: E501
        )

    def test_optional_path(self):
        self.assertEqual(self.fetch_json("/optional_path/foo"), {u"path": u"foo"})
        self.assertEqual(self.fetch_json("/optional_path/"), {u"path": None})

    def test_multi_header(self):
        response = self.fetch("/multi_header")
        self.assertEqual(response.headers["x-overwrite"], "2")
        self.assertEqual(response.headers.get_list("x-multi"), ["3", "4"])

    def test_redirect(self):
        response = self.fetch("/redirect?permanent=1", follow_redirects=False)
        self.assertEqual(response.code, 301)
        response = self.fetch("/redirect?permanent=0", follow_redirects=False)
        self.assertEqual(response.code, 302)
        response = self.fetch("/redirect?status=307", follow_redirects=False)
        self.assertEqual(response.code, 307)

    def test_web_redirect(self):
        response = self.fetch("/web_redirect_permanent", follow_redirects=False)
        self.assertEqual(response.code, 301)
        self.assertEqual(response.headers["Location"], "/web_redirect_newpath")
        response = self.fetch("/web_redirect", follow_redirects=False)
        self.assertEqual(response.code, 302)
        self.assertEqual(response.headers["Location"], "/web_redirect_newpath")

    def test_web_redirect_double_slash(self):
        response = self.fetch("//web_redirect_double_slash", follow_redirects=False)
        self.assertEqual(response.code, 301)
        self.assertEqual(response.headers["Location"], "/web_redirect_newpath")

    def test_header_injection(self):
        response = self.fetch("/header_injection")
        self.assertEqual(response.body, b"ok")

    def test_get_argument(self):
        response = self.fetch("/get_argument?foo=bar")
        self.assertEqual(response.body, b"bar")
        response = self.fetch("/get_argument?foo=")
        self.assertEqual(response.body, b"")
        response = self.fetch("/get_argument")
        self.assertEqual(response.body, b"default")

        # Test merging of query and body arguments.
        # In singular form, body arguments take precedence over query arguments.
        body = urllib.parse.urlencode(dict(foo="hello"))
        response = self.fetch("/get_argument?foo=bar", method="POST", body=body)
        self.assertEqual(response.body, b"hello")
        # In plural methods they are merged.
        response = self.fetch("/get_arguments?foo=bar", method="POST", body=body)
        self.assertEqual(
            json_decode(response.body),
            dict(default=["bar", "hello"], query=["bar"], body=["hello"]),
        )

    def test_get_query_arguments(self):
        # send as a post so we can ensure the separation between query
        # string and body arguments.
        body = urllib.parse.urlencode(dict(foo="hello"))
        response = self.fetch(
            "/get_argument?source=query&foo=bar", method="POST", body=body
        )
        self.assertEqual(response.body, b"bar")
        response = self.fetch(
            "/get_argument?source=query&foo=", method="POST", body=body
        )
        self.assertEqual(response.body, b"")
        response = self.fetch("/get_argument?source=query", method="POST", body=body)
        self.assertEqual(response.body, b"default")

    def test_get_body_arguments(self):
        body = urllib.parse.urlencode(dict(foo="bar"))
        response = self.fetch(
            "/get_argument?source=body&foo=hello", method="POST", body=body
        )
        self.assertEqual(response.body, b"bar")

        body = urllib.parse.urlencode(dict(foo=""))
        response = self.fetch(
            "/get_argument?source=body&foo=hello", method="POST", body=body
        )
        self.assertEqual(response.body, b"")

        body = urllib.parse.urlencode(dict())
        response = self.fetch(
            "/get_argument?source=body&foo=hello", method="POST", body=body
        )
        self.assertEqual(response.body, b"default")

    def test_no_gzip(self):
        response = self.fetch("/get_argument")
        self.assertNotIn("Accept-Encoding", response.headers.get("Vary", ""))
        self.assertNotIn("gzip", response.headers.get("Content-Encoding", ""))


class NonWSGIWebTests(WebTestCase):
    def get_handlers(self):
        return [("/empty_flush", EmptyFlushCallbackHandler)]

    def test_empty_flush(self):
        response = self.fetch("/empty_flush")
        self.assertEqual(response.body, b"ok")


class ErrorResponseTest(WebTestCase):
    def get_handlers(self):
        class DefaultHandler(RequestHandler):
            def get(self):
                if self.get_argument("status", None):
                    raise HTTPError(int(self.get_argument("status")))
                1 / 0

        class WriteErrorHandler(RequestHandler):
            def get(self):
                if self.get_argument("status", None):
                    self.send_error(int(self.get_argument("status")))
                else:
                    1 / 0

            def write_error(self, status_code, **kwargs):
                self.set_header("Content-Type", "text/plain")
                if "exc_info" in kwargs:
                    self.write("Exception: %s" % kwargs["exc_info"][0].__name__)
                else:
                    self.write("Status: %d" % status_code)

        class FailedWriteErrorHandler(RequestHandler):
            def get(self):
                1 / 0

            def write_error(self, status_code, **kwargs):
                raise Exception("exception in write_error")

        return [
            url("/default", DefaultHandler),
            url("/write_error", WriteErrorHandler),
            url("/failed_write_error", FailedWriteErrorHandler),
        ]

    def test_default(self):
        with ExpectLog(app_log, "Uncaught exception"):
            response = self.fetch("/default")
            self.assertEqual(response.code, 500)
            self.assertTrue(b"500: Internal Server Error" in response.body)

            response = self.fetch("/default?status=503")
            self.assertEqual(response.code, 503)
            self.assertTrue(b"503: Service Unavailable" in response.body)

            response = self.fetch("/default?status=435")
            self.assertEqual(response.code, 435)
            self.assertTrue(b"435: Unknown" in response.body)

    def test_write_error(self):
        with ExpectLog(app_log, "Uncaught exception"):
            response = self.fetch("/write_error")
            self.assertEqual(response.code, 500)
            self.assertEqual(b"Exception: ZeroDivisionError", response.body)

            response = self.fetch("/write_error?status=503")
            self.assertEqual(response.code, 503)
            self.assertEqual(b"Status: 503", response.body)

    def test_failed_write_error(self):
        with ExpectLog(app_log, "Uncaught exception"):
            response = self.fetch("/failed_write_error")
            self.assertEqual(response.code, 500)
            self.assertEqual(b"", response.body)


class StaticFileTest(WebTestCase):
    # The expected MD5 hash of robots.txt, used in tests that call
    # StaticFileHandler.get_version
    robots_txt_hash = b"f71d20196d4caf35b6a670db8c70b03d"
    static_dir = os.path.join(os.path.dirname(__file__), "static")

    def get_handlers(self):
        class StaticUrlHandler(RequestHandler):
            def get(self, path):
                with_v = int(self.get_argument("include_version", 1))
                self.write(self.static_url(path, include_version=with_v))

        class AbsoluteStaticUrlHandler(StaticUrlHandler):
            include_host = True

        class OverrideStaticUrlHandler(RequestHandler):
            def get(self, path):
                do_include = bool(self.get_argument("include_host"))
                self.include_host = not do_include

                regular_url = self.static_url(path)
                override_url = self.static_url(path, include_host=do_include)
                if override_url == regular_url:
                    return self.write(str(False))

                protocol = self.request.protocol + "://"
                protocol_length = len(protocol)
                check_regular = regular_url.find(protocol, 0, protocol_length)
                check_override = override_url.find(protocol, 0, protocol_length)

                if do_include:
                    result = check_override == 0 and check_regular == -1
                else:
                    result = check_override == -1 and check_regular == 0
                self.write(str(result))

        return [
            ("/static_url/(.*)", StaticUrlHandler),
            ("/abs_static_url/(.*)", AbsoluteStaticUrlHandler),
            ("/override_static_url/(.*)", OverrideStaticUrlHandler),
            ("/root_static/(.*)", StaticFileHandler, dict(path="/")),
        ]

    def get_app_kwargs(self):
        return dict(static_path=relpath("static"))

    def test_static_files(self):
        response = self.fetch("/robots.txt")
        self.assertTrue(b"Disallow: /" in response.body)

        response = self.fetch("/static/robots.txt")
        self.assertTrue(b"Disallow: /" in response.body)
        self.assertEqual(response.headers.get("Content-Type"), "text/plain")

    def test_static_compressed_files(self):
        response = self.fetch("/static/sample.xml.gz")
        self.assertEqual(response.headers.get("Content-Type"), "application/gzip")
        response = self.fetch("/static/sample.xml.bz2")
        self.assertEqual(
            response.headers.get("Content-Type"), "application/octet-stream"
        )
        # make sure the uncompressed file still has the correct type
        response = self.fetch("/static/sample.xml")
        self.assertTrue(
            response.headers.get("Content-Type") in set(("text/xml", "application/xml"))
        )

    def test_static_url(self):
        response = self.fetch("/static_url/robots.txt")
        self.assertEqual(response.body, b"/static/robots.txt?v=" + self.robots_txt_hash)

    def test_absolute_static_url(self):
        response = self.fetch("/abs_static_url/robots.txt")
        self.assertEqual(
            response.body,
            (utf8(self.get_url("/")) + b"static/robots.txt?v=" + self.robots_txt_hash),
        )

    def test_relative_version_exclusion(self):
        response = self.fetch("/static_url/robots.txt?include_version=0")
        self.assertEqual(response.body, b"/static/robots.txt")

    def test_absolute_version_exclusion(self):
        response = self.fetch("/abs_static_url/robots.txt?include_version=0")
        self.assertEqual(response.body, utf8(self.get_url("/") + "static/robots.txt"))

    def test_include_host_override(self):
        self._trigger_include_host_check(False)
        self._trigger_include_host_check(True)

    def _trigger_include_host_check(self, include_host):
        path = "/override_static_url/robots.txt?include_host=%s"
        response = self.fetch(path % int(include_host))
        self.assertEqual(response.body, utf8(str(True)))

    def get_and_head(self, *args, **kwargs):
        """Performs a GET and HEAD request and returns the GET response.

        Fails if any ``Content-*`` headers returned by the two requests
        differ.
        """
        head_response = self.fetch(*args, method="HEAD", **kwargs)
        get_response = self.fetch(*args, method="GET", **kwargs)
        content_headers = set()
        for h in itertools.chain(head_response.headers, get_response.headers):
            if h.startswith("Content-"):
                content_headers.add(h)
        for h in content_headers:
            self.assertEqual(
                head_response.headers.get(h),
                get_response.headers.get(h),
                "%s differs between GET (%s) and HEAD (%s)"
                % (h, head_response.headers.get(h), get_response.headers.get(h)),
            )
        return get_response

    def test_static_304_if_modified_since(self):
        response1 = self.get_and_head("/static/robots.txt")
        response2 = self.get_and_head(
            "/static/robots.txt",
            headers={"If-Modified-Since": response1.headers["Last-Modified"]},
        )
        self.assertEqual(response2.code, 304)
        self.assertTrue("Content-Length" not in response2.headers)
        self.assertTrue("Last-Modified" not in response2.headers)

    def test_static_304_if_none_match(self):
        response1 = self.get_and_head("/static/robots.txt")
        response2 = self.get_and_head(
            "/static/robots.txt", headers={"If-None-Match": response1.headers["Etag"]}
        )
        self.assertEqual(response2.code, 304)

    def test_static_304_etag_modified_bug(self):
        response1 = self.get_and_head("/static/robots.txt")
        response2 = self.get_and_head(
            "/static/robots.txt",
            headers={
                "If-None-Match": '"MISMATCH"',
                "If-Modified-Since": response1.headers["Last-Modified"],
            },
        )
        self.assertEqual(response2.code, 200)

    def test_static_if_modified_since_pre_epoch(self):
        # On windows, the functions that work with time_t do not accept
        # negative values, and at least one client (processing.js) seems
        # to use if-modified-since 1/1/1960 as a cache-busting technique.
        response = self.get_and_head(
            "/static/robots.txt",
            headers={"If-Modified-Since": "Fri, 01 Jan 1960 00:00:00 GMT"},
        )
        self.assertEqual(response.code, 200)

    def test_static_if_modified_since_time_zone(self):
        # Instead of the value from Last-Modified, make requests with times
        # chosen just before and after the known modification time
        # of the file to ensure that the right time zone is being used
        # when parsing If-Modified-Since.
        stat = os.stat(relpath("static/robots.txt"))

        response = self.get_and_head(
            "/static/robots.txt",
            headers={"If-Modified-Since": format_timestamp(stat.st_mtime - 1)},
        )
        self.assertEqual(response.code, 200)
        response = self.get_and_head(
            "/static/robots.txt",
            headers={"If-Modified-Since": format_timestamp(stat.st_mtime + 1)},
        )
        self.assertEqual(response.code, 304)

    def test_static_etag(self):
        response = self.get_and_head("/static/robots.txt")
        self.assertEqual(
            utf8(response.headers.get("Etag")), b'"' + self.robots_txt_hash + b'"'
        )

    def test_static_with_range(self):
        response = self.get_and_head(
            "/static/robots.txt", headers={"Range": "bytes=0-9"}
        )
        self.assertEqual(response.code, 206)
        self.assertEqual(response.body, b"User-agent")
        self.assertEqual(
            utf8(response.headers.get("Etag")), b'"' + self.robots_txt_hash + b'"'
        )
        self.assertEqual(response.headers.get("Content-Length"), "10")
        self.assertEqual(response.headers.get("Content-Range"), "bytes 0-9/26")

    def test_static_with_range_full_file(self):
        response = self.get_and_head(
            "/static/robots.txt", headers={"Range": "bytes=0-"}
        )
        # Note: Chrome refuses to play audio if it gets an HTTP 206 in response
        # to ``Range: bytes=0-`` :(
        self.assertEqual(response.code, 200)
        robots_file_path = os.path.join(self.static_dir, "robots.txt")
        with open(robots_file_path) as f:
            self.assertEqual(response.body, utf8(f.read()))
        self.assertEqual(response.headers.get("Content-Length"), "26")
        self.assertEqual(response.headers.get("Content-Range"), None)

    def test_static_with_range_full_past_end(self):
        response = self.get_and_head(
            "/static/robots.txt", headers={"Range": "bytes=0-10000000"}
        )
        self.assertEqual(response.code, 200)
        robots_file_path = os.path.join(self.static_dir, "robots.txt")
        with open(robots_file_path) as f:
            self.assertEqual(response.body, utf8(f.read()))
        self.assertEqual(response.headers.get("Content-Length"), "26")
        self.assertEqual(response.headers.get("Content-Range"), None)

    def test_static_with_range_partial_past_end(self):
        response = self.get_and_head(
            "/static/robots.txt", headers={"Range": "bytes=1-10000000"}
        )
        self.assertEqual(response.code, 206)
        robots_file_path = os.path.join(self.static_dir, "robots.txt")
        with open(robots_file_path) as f:
            self.assertEqual(response.body, utf8(f.read()[1:]))
        self.assertEqual(response.headers.get("Content-Length"), "25")
        self.assertEqual(response.headers.get("Content-Range"), "bytes 1-25/26")

    def test_static_with_range_end_edge(self):
        response = self.get_and_head(
            "/static/robots.txt", headers={"Range": "bytes=22-"}
        )
        self.assertEqual(response.body, b": /\n")
        self.assertEqual(response.headers.get("Content-Length"), "4")
        self.assertEqual(response.headers.get("Content-Range"), "bytes 22-25/26")

    def test_static_with_range_neg_end(self):
        response = self.get_and_head(
            "/static/robots.txt", headers={"Range": "bytes=-4"}
        )
        self.assertEqual(response.body, b": /\n")
        self.assertEqual(response.headers.get("Content-Length"), "4")
        self.assertEqual(response.headers.get("Content-Range"), "bytes 22-25/26")

    def test_static_with_range_neg_past_start(self):
        response = self.get_and_head(
            "/static/robots.txt", headers={"Range": "bytes=-1000000"}
        )
        self.assertEqual(response.code, 200)
        robots_file_path = os.path.join(self.static_dir, "robots.txt")
        with open(robots_file_path) as f:
            self.assertEqual(response.body, utf8(f.read()))
        self.assertEqual(response.headers.get("Content-Length"), "26")
        self.assertEqual(response.headers.get("Content-Range"), None)

    def test_static_invalid_range(self):
        response = self.get_and_head("/static/robots.txt", headers={"Range": "asdf"})
        self.assertEqual(response.code, 200)

    def test_static_unsatisfiable_range_zero_suffix(self):
        response = self.get_and_head(
            "/static/robots.txt", headers={"Range": "bytes=-0"}
        )
        self.assertEqual(response.headers.get("Content-Range"), "bytes */26")
        self.assertEqual(response.code, 416)

    def test_static_unsatisfiable_range_invalid_start(self):
        response = self.get_and_head(
            "/static/robots.txt", headers={"Range": "bytes=26"}
        )
        self.assertEqual(response.code, 416)
        self.assertEqual(response.headers.get("Content-Range"), "bytes */26")

    def test_static_unsatisfiable_range_end_less_than_start(self):
        response = self.get_and_head(
            "/static/robots.txt", headers={"Range": "bytes=10-3"}
        )
        self.assertEqual(response.code, 416)
        self.assertEqual(response.headers.get("Content-Range"), "bytes */26")

    def test_static_head(self):
        response = self.fetch("/static/robots.txt", method="HEAD")
        self.assertEqual(response.code, 200)
        # No body was returned, but we did get the right content length.
        self.assertEqual(response.body, b"")
        self.assertEqual(response.headers["Content-Length"], "26")
        self.assertEqual(
            utf8(response.headers["Etag"]), b'"' + self.robots_txt_hash + b'"'
        )

    def test_static_head_range(self):
        response = self.fetch(
            "/static/robots.txt", method="HEAD", headers={"Range": "bytes=1-4"}
        )
        self.assertEqual(response.code, 206)
        self.assertEqual(response.body, b"")
        self.assertEqual(response.headers["Content-Length"], "4")
        self.assertEqual(
            utf8(response.headers["Etag"]), b'"' + self.robots_txt_hash + b'"'
        )

    def test_static_range_if_none_match(self):
        response = self.get_and_head(
            "/static/robots.txt",
            headers={
                "Range": "bytes=1-4",
                "If-None-Match": b'"' + self.robots_txt_hash + b'"',
            },
        )
        self.assertEqual(response.code, 304)
        self.assertEqual(response.body, b"")
        self.assertTrue("Content-Length" not in response.headers)
        self.assertEqual(
            utf8(response.headers["Etag"]), b'"' + self.robots_txt_hash + b'"'
        )

    def test_static_404(self):
        response = self.get_and_head("/static/blarg")
        self.assertEqual(response.code, 404)

    def test_path_traversal_protection(self):
        # curl_httpclient processes ".." on the client side, so we
        # must test this with simple_httpclient.
        self.http_client.close()
        self.http_client = SimpleAsyncHTTPClient()
        with ExpectLog(gen_log, ".*not in root static directory"):
            response = self.get_and_head("/static/../static_foo.txt")
        # Attempted path traversal should result in 403, not 200
        # (which means the check failed and the file was served)
        # or 404 (which means that the file didn't exist and
        # is probably a packaging error).
        self.assertEqual(response.code, 403)

    @unittest.skipIf(os.name != "posix", "non-posix OS")
    def test_root_static_path(self):
        # Sometimes people set the StaticFileHandler's path to '/'
        # to disable Tornado's path validation (in conjunction with
        # their own validation in get_absolute_path). Make sure
        # that the stricter validation in 4.2.1 doesn't break them.
        path = os.path.join(
            os.path.dirname(os.path.abspath(__file__)), "static/robots.txt"
        )
        response = self.get_and_head("/root_static" + urllib.parse.quote(path))
        self.assertEqual(response.code, 200)


class StaticDefaultFilenameTest(WebTestCase):
    def get_app_kwargs(self):
        return dict(
            static_path=relpath("static"),
            static_handler_args=dict(default_filename="index.html"),
        )

    def get_handlers(self):
        return []

    def test_static_default_filename(self):
        response = self.fetch("/static/dir/", follow_redirects=False)
        self.assertEqual(response.code, 200)
        self.assertEqual(b"this is the index\n", response.body)

    def test_static_default_redirect(self):
        response = self.fetch("/static/dir", follow_redirects=False)
        self.assertEqual(response.code, 301)
        self.assertTrue(response.headers["Location"].endswith("/static/dir/"))


class StaticFileWithPathTest(WebTestCase):
    def get_app_kwargs(self):
        return dict(
            static_path=relpath("static"),
            static_handler_args=dict(default_filename="index.html"),
        )

    def get_handlers(self):
        return [("/foo/(.*)", StaticFileHandler, {"path": relpath("templates/")})]

    def test_serve(self):
        response = self.fetch("/foo/utf8.html")
        self.assertEqual(response.body, b"H\xc3\xa9llo\n")


class CustomStaticFileTest(WebTestCase):
    def get_handlers(self):
        class MyStaticFileHandler(StaticFileHandler):
            @classmethod
            def make_static_url(cls, settings, path):
                version_hash = cls.get_version(settings, path)
                extension_index = path.rindex(".")
                before_version = path[:extension_index]
                after_version = path[(extension_index + 1) :]
                return "/static/%s.%s.%s" % (
                    before_version,
                    version_hash,
                    after_version,
                )

            def parse_url_path(self, url_path):
                extension_index = url_path.rindex(".")
                version_index = url_path.rindex(".", 0, extension_index)
                return "%s%s" % (url_path[:version_index], url_path[extension_index:])

            @classmethod
            def get_absolute_path(cls, settings, path):
                return "CustomStaticFileTest:" + path

            def validate_absolute_path(self, root, absolute_path):
                return absolute_path

            @classmethod
            def get_content(self, path, start=None, end=None):
                assert start is None and end is None
                if path == "CustomStaticFileTest:foo.txt":
                    return b"bar"
                raise Exception("unexpected path %r" % path)

            def get_content_size(self):
                if self.absolute_path == "CustomStaticFileTest:foo.txt":
                    return 3
                raise Exception("unexpected path %r" % self.absolute_path)

            def get_modified_time(self):
                return None

            @classmethod
            def get_version(cls, settings, path):
                return "42"

        class StaticUrlHandler(RequestHandler):
            def get(self, path):
                self.write(self.static_url(path))

        self.static_handler_class = MyStaticFileHandler

        return [("/static_url/(.*)", StaticUrlHandler)]

    def get_app_kwargs(self):
        return dict(static_path="dummy", static_handler_class=self.static_handler_class)

    def test_serve(self):
        response = self.fetch("/static/foo.42.txt")
        self.assertEqual(response.body, b"bar")

    def test_static_url(self):
        with ExpectLog(gen_log, "Could not open static file", required=False):
            response = self.fetch("/static_url/foo.txt")
            self.assertEqual(response.body, b"/static/foo.42.txt")


class HostMatchingTest(WebTestCase):
    class Handler(RequestHandler):
        def initialize(self, reply):
            self.reply = reply

        def get(self):
            self.write(self.reply)

    def get_handlers(self):
        return [("/foo", HostMatchingTest.Handler, {"reply": "wildcard"})]

    def test_host_matching(self):
        self.app.add_handlers(
            "www.example.com", [("/foo", HostMatchingTest.Handler, {"reply": "[0]"})]
        )
        self.app.add_handlers(
            r"www\.example\.com", [("/bar", HostMatchingTest.Handler, {"reply": "[1]"})]
        )
        self.app.add_handlers(
            "www.example.com", [("/baz", HostMatchingTest.Handler, {"reply": "[2]"})]
        )
        self.app.add_handlers(
            "www.e.*e.com", [("/baz", HostMatchingTest.Handler, {"reply": "[3]"})]
        )

        response = self.fetch("/foo")
        self.assertEqual(response.body, b"wildcard")
        response = self.fetch("/bar")
        self.assertEqual(response.code, 404)
        response = self.fetch("/baz")
        self.assertEqual(response.code, 404)

        response = self.fetch("/foo", headers={"Host": "www.example.com"})
        self.assertEqual(response.body, b"[0]")
        response = self.fetch("/bar", headers={"Host": "www.example.com"})
        self.assertEqual(response.body, b"[1]")
        response = self.fetch("/baz", headers={"Host": "www.example.com"})
        self.assertEqual(response.body, b"[2]")
        response = self.fetch("/baz", headers={"Host": "www.exe.com"})
        self.assertEqual(response.body, b"[3]")


class DefaultHostMatchingTest(WebTestCase):
    def get_handlers(self):
        return []

    def get_app_kwargs(self):
        return {"default_host": "www.example.com"}

    def test_default_host_matching(self):
        self.app.add_handlers(
            "www.example.com", [("/foo", HostMatchingTest.Handler, {"reply": "[0]"})]
        )
        self.app.add_handlers(
            r"www\.example\.com", [("/bar", HostMatchingTest.Handler, {"reply": "[1]"})]
        )
        self.app.add_handlers(
            "www.test.com", [("/baz", HostMatchingTest.Handler, {"reply": "[2]"})]
        )

        response = self.fetch("/foo")
        self.assertEqual(response.body, b"[0]")
        response = self.fetch("/bar")
        self.assertEqual(response.body, b"[1]")
        response = self.fetch("/baz")
        self.assertEqual(response.code, 404)

        response = self.fetch("/foo", headers={"X-Real-Ip": "127.0.0.1"})
        self.assertEqual(response.code, 404)

        self.app.default_host = "www.test.com"

        response = self.fetch("/baz")
        self.assertEqual(response.body, b"[2]")


class NamedURLSpecGroupsTest(WebTestCase):
    def get_handlers(self):
        class EchoHandler(RequestHandler):
            def get(self, path):
                self.write(path)

        return [
            ("/str/(?P<path>.*)", EchoHandler),
            (u"/unicode/(?P<path>.*)", EchoHandler),
        ]

    def test_named_urlspec_groups(self):
        response = self.fetch("/str/foo")
        self.assertEqual(response.body, b"foo")

        response = self.fetch("/unicode/bar")
        self.assertEqual(response.body, b"bar")


class ClearHeaderTest(SimpleHandlerTestCase):
    class Handler(RequestHandler):
        def get(self):
            self.set_header("h1", "foo")
            self.set_header("h2", "bar")
            self.clear_header("h1")
            self.clear_header("nonexistent")

    def test_clear_header(self):
        response = self.fetch("/")
        self.assertTrue("h1" not in response.headers)
        self.assertEqual(response.headers["h2"], "bar")


class Header204Test(SimpleHandlerTestCase):
    class Handler(RequestHandler):
        def get(self):
            self.set_status(204)
            self.finish()

    def test_204_headers(self):
        response = self.fetch("/")
        self.assertEqual(response.code, 204)
        self.assertNotIn("Content-Length", response.headers)
        self.assertNotIn("Transfer-Encoding", response.headers)


class Header304Test(SimpleHandlerTestCase):
    class Handler(RequestHandler):
        def get(self):
            self.set_header("Content-Language", "en_US")
            self.write("hello")

    def test_304_headers(self):
        response1 = self.fetch("/")
        self.assertEqual(response1.headers["Content-Length"], "5")
        self.assertEqual(response1.headers["Content-Language"], "en_US")

        response2 = self.fetch(
            "/", headers={"If-None-Match": response1.headers["Etag"]}
        )
        self.assertEqual(response2.code, 304)
        self.assertTrue("Content-Length" not in response2.headers)
        self.assertTrue("Content-Language" not in response2.headers)
        # Not an entity header, but should not be added to 304s by chunking
        self.assertTrue("Transfer-Encoding" not in response2.headers)


class StatusReasonTest(SimpleHandlerTestCase):
    class Handler(RequestHandler):
        def get(self):
            reason = self.request.arguments.get("reason", [])
            self.set_status(
                int(self.get_argument("code")), reason=reason[0] if reason else None
            )

    def get_http_client(self):
        # simple_httpclient only: curl doesn't expose the reason string
        return SimpleAsyncHTTPClient()

    def test_status(self):
        response = self.fetch("/?code=304")
        self.assertEqual(response.code, 304)
        self.assertEqual(response.reason, "Not Modified")
        response = self.fetch("/?code=304&reason=Foo")
        self.assertEqual(response.code, 304)
        self.assertEqual(response.reason, "Foo")
        response = self.fetch("/?code=682&reason=Bar")
        self.assertEqual(response.code, 682)
        self.assertEqual(response.reason, "Bar")
        response = self.fetch("/?code=682")
        self.assertEqual(response.code, 682)
        self.assertEqual(response.reason, "Unknown")


class DateHeaderTest(SimpleHandlerTestCase):
    class Handler(RequestHandler):
        def get(self):
            self.write("hello")

    def test_date_header(self):
        response = self.fetch("/")
        parsed = email.utils.parsedate(response.headers["Date"])
        assert parsed is not None
        header_date = datetime.datetime(*parsed[:6])
        self.assertTrue(
            header_date - datetime.datetime.utcnow() < datetime.timedelta(seconds=2)
        )


class RaiseWithReasonTest(SimpleHandlerTestCase):
    class Handler(RequestHandler):
        def get(self):
            raise HTTPError(682, reason="Foo")

    def get_http_client(self):
        # simple_httpclient only: curl doesn't expose the reason string
        return SimpleAsyncHTTPClient()

    def test_raise_with_reason(self):
        response = self.fetch("/")
        self.assertEqual(response.code, 682)
        self.assertEqual(response.reason, "Foo")
        self.assertIn(b"682: Foo", response.body)

    def test_httperror_str(self):
        self.assertEqual(str(HTTPError(682, reason="Foo")), "HTTP 682: Foo")

    def test_httperror_str_from_httputil(self):
        self.assertEqual(str(HTTPError(682)), "HTTP 682: Unknown")


class ErrorHandlerXSRFTest(WebTestCase):
    def get_handlers(self):
        # note that if the handlers list is empty we get the default_host
        # redirect fallback instead of a 404, so test with both an
        # explicitly defined error handler and an implicit 404.
        return [("/error", ErrorHandler, dict(status_code=417))]

    def get_app_kwargs(self):
        return dict(xsrf_cookies=True)

    def test_error_xsrf(self):
        response = self.fetch("/error", method="POST", body="")
        self.assertEqual(response.code, 417)

    def test_404_xsrf(self):
        response = self.fetch("/404", method="POST", body="")
        self.assertEqual(response.code, 404)


class GzipTestCase(SimpleHandlerTestCase):
    class Handler(RequestHandler):
        def get(self):
            for v in self.get_arguments("vary"):
                self.add_header("Vary", v)
            # Must write at least MIN_LENGTH bytes to activate compression.
            self.write("hello world" + ("!" * GZipContentEncoding.MIN_LENGTH))

    def get_app_kwargs(self):
        return dict(
            gzip=True, static_path=os.path.join(os.path.dirname(__file__), "static")
        )

    def assert_compressed(self, response):
        # simple_httpclient renames the content-encoding header;
        # curl_httpclient doesn't.
        self.assertEqual(
            response.headers.get(
                "Content-Encoding", response.headers.get("X-Consumed-Content-Encoding")
            ),
            "gzip",
        )

    def test_gzip(self):
        response = self.fetch("/")
        self.assert_compressed(response)
        self.assertEqual(response.headers["Vary"], "Accept-Encoding")

    def test_gzip_static(self):
        # The streaming responses in StaticFileHandler have subtle
        # interactions with the gzip output so test this case separately.
        response = self.fetch("/robots.txt")
        self.assert_compressed(response)
        self.assertEqual(response.headers["Vary"], "Accept-Encoding")

    def test_gzip_not_requested(self):
        response = self.fetch("/", use_gzip=False)
        self.assertNotIn("Content-Encoding", response.headers)
        self.assertEqual(response.headers["Vary"], "Accept-Encoding")

    def test_vary_already_present(self):
        response = self.fetch("/?vary=Accept-Language")
        self.assert_compressed(response)
        self.assertEqual(
            [s.strip() for s in response.headers["Vary"].split(",")],
            ["Accept-Language", "Accept-Encoding"],
        )

    def test_vary_already_present_multiple(self):
        # Regression test for https://github.com/tornadoweb/tornado/issues/1670
        response = self.fetch("/?vary=Accept-Language&vary=Cookie")
        self.assert_compressed(response)
        self.assertEqual(
            [s.strip() for s in response.headers["Vary"].split(",")],
            ["Accept-Language", "Cookie", "Accept-Encoding"],
        )


class PathArgsInPrepareTest(WebTestCase):
    class Handler(RequestHandler):
        def prepare(self):
            self.write(dict(args=self.path_args, kwargs=self.path_kwargs))

        def get(self, path):
            assert path == "foo"
            self.finish()

    def get_handlers(self):
        return [("/pos/(.*)", self.Handler), ("/kw/(?P<path>.*)", self.Handler)]

    def test_pos(self):
        response = self.fetch("/pos/foo")
        response.rethrow()
        data = json_decode(response.body)
        self.assertEqual(data, {"args": ["foo"], "kwargs": {}})

    def test_kw(self):
        response = self.fetch("/kw/foo")
        response.rethrow()
        data = json_decode(response.body)
        self.assertEqual(data, {"args": [], "kwargs": {"path": "foo"}})


class ClearAllCookiesTest(SimpleHandlerTestCase):
    class Handler(RequestHandler):
        def get(self):
            self.clear_all_cookies()
            self.write("ok")

    def test_clear_all_cookies(self):
        response = self.fetch("/", headers={"Cookie": "foo=bar; baz=xyzzy"})
        set_cookies = sorted(response.headers.get_list("Set-Cookie"))
        # Python 3.5 sends 'baz="";'; older versions use 'baz=;'
        self.assertTrue(
            set_cookies[0].startswith("baz=;") or set_cookies[0].startswith('baz="";')
        )
        self.assertTrue(
            set_cookies[1].startswith("foo=;") or set_cookies[1].startswith('foo="";')
        )


class PermissionError(Exception):
    pass


class ExceptionHandlerTest(SimpleHandlerTestCase):
    class Handler(RequestHandler):
        def get(self):
            exc = self.get_argument("exc")
            if exc == "http":
                raise HTTPError(410, "no longer here")
            elif exc == "zero":
                1 / 0
            elif exc == "permission":
                raise PermissionError("not allowed")

        def write_error(self, status_code, **kwargs):
            if "exc_info" in kwargs:
                typ, value, tb = kwargs["exc_info"]
                if isinstance(value, PermissionError):
                    self.set_status(403)
                    self.write("PermissionError")
                    return
            RequestHandler.write_error(self, status_code, **kwargs)

        def log_exception(self, typ, value, tb):
            if isinstance(value, PermissionError):
                app_log.warning("custom logging for PermissionError: %s", value.args[0])
            else:
                RequestHandler.log_exception(self, typ, value, tb)

    def test_http_error(self):
        # HTTPErrors are logged as warnings with no stack trace.
        # TODO: extend ExpectLog to test this more precisely
        with ExpectLog(gen_log, ".*no longer here"):
            response = self.fetch("/?exc=http")
            self.assertEqual(response.code, 410)

    def test_unknown_error(self):
        # Unknown errors are logged as errors with a stack trace.
        with ExpectLog(app_log, "Uncaught exception"):
            response = self.fetch("/?exc=zero")
            self.assertEqual(response.code, 500)

    def test_known_error(self):
        # log_exception can override logging behavior, and write_error
        # can override the response.
        with ExpectLog(app_log, "custom logging for PermissionError: not allowed"):
            response = self.fetch("/?exc=permission")
            self.assertEqual(response.code, 403)


class BuggyLoggingTest(SimpleHandlerTestCase):
    class Handler(RequestHandler):
        def get(self):
            1 / 0

        def log_exception(self, typ, value, tb):
            1 / 0

    def test_buggy_log_exception(self):
        # Something gets logged even though the application's
        # logger is broken.
        with ExpectLog(app_log, ".*"):
            self.fetch("/")


class UIMethodUIModuleTest(SimpleHandlerTestCase):
    """Test that UI methods and modules are created correctly and
    associated with the handler.
    """

    class Handler(RequestHandler):
        def get(self):
            self.render("foo.html")

        def value(self):
            return self.get_argument("value")

    def get_app_kwargs(self):
        def my_ui_method(handler, x):
            return "In my_ui_method(%s) with handler value %s." % (x, handler.value())

        class MyModule(UIModule):
            def render(self, x):
                return "In MyModule(%s) with handler value %s." % (
                    x,
                    self.handler.value(),
                )

        loader = DictLoader(
            {"foo.html": "{{ my_ui_method(42) }} {% module MyModule(123) %}"}
        )
        return dict(
            template_loader=loader,
            ui_methods={"my_ui_method": my_ui_method},
            ui_modules={"MyModule": MyModule},
        )

    def tearDown(self):
        super(UIMethodUIModuleTest, self).tearDown()
        # TODO: fix template loader caching so this isn't necessary.
        RequestHandler._template_loaders.clear()

    def test_ui_method(self):
        response = self.fetch("/?value=asdf")
        self.assertEqual(
            response.body,
            b"In my_ui_method(42) with handler value asdf. "
            b"In MyModule(123) with handler value asdf.",
        )


class GetArgumentErrorTest(SimpleHandlerTestCase):
    class Handler(RequestHandler):
        def get(self):
            try:
                self.get_argument("foo")
                self.write({})
            except MissingArgumentError as e:
                self.write({"arg_name": e.arg_name, "log_message": e.log_message})

    def test_catch_error(self):
        response = self.fetch("/")
        self.assertEqual(
            json_decode(response.body),
            {"arg_name": "foo", "log_message": "Missing argument foo"},
        )


class SetLazyPropertiesTest(SimpleHandlerTestCase):
    class Handler(RequestHandler):
        def prepare(self):
            self.current_user = "Ben"
            self.locale = locale.get("en_US")

        def get_user_locale(self):
            raise NotImplementedError()

        def get_current_user(self):
            raise NotImplementedError()

        def get(self):
            self.write("Hello %s (%s)" % (self.current_user, self.locale.code))

    def test_set_properties(self):
        # Ensure that current_user can be assigned to normally for apps
        # that want to forgo the lazy get_current_user property
        response = self.fetch("/")
        self.assertEqual(response.body, b"Hello Ben (en_US)")


class GetCurrentUserTest(WebTestCase):
    def get_app_kwargs(self):
        class WithoutUserModule(UIModule):
            def render(self):
                return ""

        class WithUserModule(UIModule):
            def render(self):
                return str(self.current_user)

        loader = DictLoader(
            {
                "without_user.html": "",
                "with_user.html": "{{ current_user }}",
                "without_user_module.html": "{% module WithoutUserModule() %}",
                "with_user_module.html": "{% module WithUserModule() %}",
            }
        )
        return dict(
            template_loader=loader,
            ui_modules={
                "WithUserModule": WithUserModule,
                "WithoutUserModule": WithoutUserModule,
            },
        )

    def tearDown(self):
        super(GetCurrentUserTest, self).tearDown()
        RequestHandler._template_loaders.clear()

    def get_handlers(self):
        class CurrentUserHandler(RequestHandler):
            def prepare(self):
                self.has_loaded_current_user = False

            def get_current_user(self):
                self.has_loaded_current_user = True
                return ""

        class WithoutUserHandler(CurrentUserHandler):
            def get(self):
                self.render_string("without_user.html")
                self.finish(str(self.has_loaded_current_user))

        class WithUserHandler(CurrentUserHandler):
            def get(self):
                self.render_string("with_user.html")
                self.finish(str(self.has_loaded_current_user))

        class CurrentUserModuleHandler(CurrentUserHandler):
            def get_template_namespace(self):
                # If RequestHandler.get_template_namespace is called, then
                # get_current_user is evaluated. Until #820 is fixed, this
                # is a small hack to circumvent the issue.
                return self.ui

        class WithoutUserModuleHandler(CurrentUserModuleHandler):
            def get(self):
                self.render_string("without_user_module.html")
                self.finish(str(self.has_loaded_current_user))

        class WithUserModuleHandler(CurrentUserModuleHandler):
            def get(self):
                self.render_string("with_user_module.html")
                self.finish(str(self.has_loaded_current_user))

        return [
            ("/without_user", WithoutUserHandler),
            ("/with_user", WithUserHandler),
            ("/without_user_module", WithoutUserModuleHandler),
            ("/with_user_module", WithUserModuleHandler),
        ]

    @unittest.skip("needs fix")
    def test_get_current_user_is_lazy(self):
        # TODO: Make this test pass. See #820.
        response = self.fetch("/without_user")
        self.assertEqual(response.body, b"False")

    def test_get_current_user_works(self):
        response = self.fetch("/with_user")
        self.assertEqual(response.body, b"True")

    def test_get_current_user_from_ui_module_is_lazy(self):
        response = self.fetch("/without_user_module")
        self.assertEqual(response.body, b"False")

    def test_get_current_user_from_ui_module_works(self):
        response = self.fetch("/with_user_module")
        self.assertEqual(response.body, b"True")


class UnimplementedHTTPMethodsTest(SimpleHandlerTestCase):
    class Handler(RequestHandler):
        pass

    def test_unimplemented_standard_methods(self):
        for method in ["HEAD", "GET", "DELETE", "OPTIONS"]:
            response = self.fetch("/", method=method)
            self.assertEqual(response.code, 405)
        for method in ["POST", "PUT"]:
            response = self.fetch("/", method=method, body=b"")
            self.assertEqual(response.code, 405)


class UnimplementedNonStandardMethodsTest(SimpleHandlerTestCase):
    class Handler(RequestHandler):
        def other(self):
            # Even though this method exists, it won't get called automatically
            # because it is not in SUPPORTED_METHODS.
            self.write("other")

    def test_unimplemented_patch(self):
        # PATCH is recently standardized; Tornado supports it by default
        # but wsgiref.validate doesn't like it.
        response = self.fetch("/", method="PATCH", body=b"")
        self.assertEqual(response.code, 405)

    def test_unimplemented_other(self):
        response = self.fetch("/", method="OTHER", allow_nonstandard_methods=True)
        self.assertEqual(response.code, 405)


class AllHTTPMethodsTest(SimpleHandlerTestCase):
    class Handler(RequestHandler):
        def method(self):
            self.write(self.request.method)

        get = delete = options = post = put = method  # type: ignore

    def test_standard_methods(self):
        response = self.fetch("/", method="HEAD")
        self.assertEqual(response.body, b"")
        for method in ["GET", "DELETE", "OPTIONS"]:
            response = self.fetch("/", method=method)
            self.assertEqual(response.body, utf8(method))
        for method in ["POST", "PUT"]:
            response = self.fetch("/", method=method, body=b"")
            self.assertEqual(response.body, utf8(method))


class PatchMethodTest(SimpleHandlerTestCase):
    class Handler(RequestHandler):
        SUPPORTED_METHODS = RequestHandler.SUPPORTED_METHODS + (  # type: ignore
            "OTHER",
        )

        def patch(self):
            self.write("patch")

        def other(self):
            self.write("other")

    def test_patch(self):
        response = self.fetch("/", method="PATCH", body=b"")
        self.assertEqual(response.body, b"patch")

    def test_other(self):
        response = self.fetch("/", method="OTHER", allow_nonstandard_methods=True)
        self.assertEqual(response.body, b"other")


class FinishInPrepareTest(SimpleHandlerTestCase):
    class Handler(RequestHandler):
        def prepare(self):
            self.finish("done")

        def get(self):
            # It's difficult to assert for certain that a method did not
            # or will not be called in an asynchronous context, but this
            # will be logged noisily if it is reached.
            raise Exception("should not reach this method")

    def test_finish_in_prepare(self):
        response = self.fetch("/")
        self.assertEqual(response.body, b"done")


class Default404Test(WebTestCase):
    def get_handlers(self):
        # If there are no handlers at all a default redirect handler gets added.
        return [("/foo", RequestHandler)]

    def test_404(self):
        response = self.fetch("/")
        self.assertEqual(response.code, 404)
        self.assertEqual(
            response.body,
            b"<html><title>404: Not Found</title>"
            b"<body>404: Not Found</body></html>",
        )


class Custom404Test(WebTestCase):
    def get_handlers(self):
        return [("/foo", RequestHandler)]

    def get_app_kwargs(self):
        class Custom404Handler(RequestHandler):
            def get(self):
                self.set_status(404)
                self.write("custom 404 response")

        return dict(default_handler_class=Custom404Handler)

    def test_404(self):
        response = self.fetch("/")
        self.assertEqual(response.code, 404)
        self.assertEqual(response.body, b"custom 404 response")


class DefaultHandlerArgumentsTest(WebTestCase):
    def get_handlers(self):
        return [("/foo", RequestHandler)]

    def get_app_kwargs(self):
        return dict(
            default_handler_class=ErrorHandler,
            default_handler_args=dict(status_code=403),
        )

    def test_403(self):
        response = self.fetch("/")
        self.assertEqual(response.code, 403)


class HandlerByNameTest(WebTestCase):
    def get_handlers(self):
        # All three are equivalent.
        return [
            ("/hello1", HelloHandler),
            ("/hello2", "tornado.test.web_test.HelloHandler"),
            url("/hello3", "tornado.test.web_test.HelloHandler"),
        ]

    def test_handler_by_name(self):
        resp = self.fetch("/hello1")
        self.assertEqual(resp.body, b"hello")
        resp = self.fetch("/hello2")
        self.assertEqual(resp.body, b"hello")
        resp = self.fetch("/hello3")
        self.assertEqual(resp.body, b"hello")


class StreamingRequestBodyTest(WebTestCase):
    def get_handlers(self):
        @stream_request_body
        class StreamingBodyHandler(RequestHandler):
            def initialize(self, test):
                self.test = test

            def prepare(self):
                self.test.prepared.set_result(None)

            def data_received(self, data):
                self.test.data.set_result(data)

            def get(self):
                self.test.finished.set_result(None)
                self.write({})

        @stream_request_body
        class EarlyReturnHandler(RequestHandler):
            def prepare(self):
                # If we finish the response in prepare, it won't continue to
                # the (non-existent) data_received.
                raise HTTPError(401)

        @stream_request_body
        class CloseDetectionHandler(RequestHandler):
            def initialize(self, test):
                self.test = test

            def on_connection_close(self):
                super(CloseDetectionHandler, self).on_connection_close()
                self.test.close_future.set_result(None)

        return [
            ("/stream_body", StreamingBodyHandler, dict(test=self)),
            ("/early_return", EarlyReturnHandler),
            ("/close_detection", CloseDetectionHandler, dict(test=self)),
        ]

    def connect(self, url, connection_close):
        # Use a raw connection so we can control the sending of data.
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
        s.connect(("127.0.0.1", self.get_http_port()))
        stream = IOStream(s)
        stream.write(b"GET " + url + b" HTTP/1.1\r\n")
        if connection_close:
            stream.write(b"Connection: close\r\n")
        stream.write(b"Transfer-Encoding: chunked\r\n\r\n")
        return stream

    @gen_test
    def test_streaming_body(self):
        self.prepared = Future()  # type: Future[None]
        self.data = Future()  # type: Future[bytes]
        self.finished = Future()  # type: Future[None]

        stream = self.connect(b"/stream_body", connection_close=True)
        yield self.prepared
        stream.write(b"4\r\nasdf\r\n")
        # Ensure the first chunk is received before we send the second.
        data = yield self.data
        self.assertEqual(data, b"asdf")
        self.data = Future()
        stream.write(b"4\r\nqwer\r\n")
        data = yield self.data
        self.assertEquals(data, b"qwer")
        stream.write(b"0\r\n\r\n")
        yield self.finished
        data = yield stream.read_until_close()
        # This would ideally use an HTTP1Connection to read the response.
        self.assertTrue(data.endswith(b"{}"))
        stream.close()

    @gen_test
    def test_early_return(self):
        stream = self.connect(b"/early_return", connection_close=False)
        data = yield stream.read_until_close()
        self.assertTrue(data.startswith(b"HTTP/1.1 401"))

    @gen_test
    def test_early_return_with_data(self):
        stream = self.connect(b"/early_return", connection_close=False)
        stream.write(b"4\r\nasdf\r\n")
        data = yield stream.read_until_close()
        self.assertTrue(data.startswith(b"HTTP/1.1 401"))

    @gen_test
    def test_close_during_upload(self):
        self.close_future = Future()  # type: Future[None]
        stream = self.connect(b"/close_detection", connection_close=False)
        stream.close()
        yield self.close_future


# Each method in this handler returns a yieldable object and yields to the
# IOLoop so the future is not immediately ready.  Ensure that the
# yieldables are respected and no method is called before the previous
# one has completed.
@stream_request_body
class BaseFlowControlHandler(RequestHandler):
    def initialize(self, test):
        self.test = test
        self.method = None
        self.methods = []  # type: typing.List[str]

    @contextlib.contextmanager
    def in_method(self, method):
        if self.method is not None:
            self.test.fail("entered method %s while in %s" % (method, self.method))
        self.method = method
        self.methods.append(method)
        try:
            yield
        finally:
            self.method = None

    @gen.coroutine
    def prepare(self):
        # Note that asynchronous prepare() does not block data_received,
        # so we don't use in_method here.
        self.methods.append("prepare")
        yield gen.moment

    @gen.coroutine
    def post(self):
        with self.in_method("post"):
            yield gen.moment
        self.write(dict(methods=self.methods))


class BaseStreamingRequestFlowControlTest(object):
    def get_httpserver_options(self):
        # Use a small chunk size so flow control is relevant even though
        # all the data arrives at once.
        return dict(chunk_size=10, decompress_request=True)

    def get_http_client(self):
        # simple_httpclient only: curl doesn't support body_producer.
        return SimpleAsyncHTTPClient()

    # Test all the slightly different code paths for fixed, chunked, etc bodies.
    def test_flow_control_fixed_body(self):
        response = self.fetch("/", body="abcdefghijklmnopqrstuvwxyz", method="POST")
        response.rethrow()
        self.assertEqual(
            json_decode(response.body),
            dict(
                methods=[
                    "prepare",
                    "data_received",
                    "data_received",
                    "data_received",
                    "post",
                ]
            ),
        )

    def test_flow_control_chunked_body(self):
        chunks = [b"abcd", b"efgh", b"ijkl"]

        @gen.coroutine
        def body_producer(write):
            for i in chunks:
                yield write(i)

        response = self.fetch("/", body_producer=body_producer, method="POST")
        response.rethrow()
        self.assertEqual(
            json_decode(response.body),
            dict(
                methods=[
                    "prepare",
                    "data_received",
                    "data_received",
                    "data_received",
                    "post",
                ]
            ),
        )

    def test_flow_control_compressed_body(self):
        bytesio = BytesIO()
        gzip_file = gzip.GzipFile(mode="w", fileobj=bytesio)
        gzip_file.write(b"abcdefghijklmnopqrstuvwxyz")
        gzip_file.close()
        compressed_body = bytesio.getvalue()
        response = self.fetch(
            "/",
            body=compressed_body,
            method="POST",
            headers={"Content-Encoding": "gzip"},
        )
        response.rethrow()
        self.assertEqual(
            json_decode(response.body),
            dict(
                methods=[
                    "prepare",
                    "data_received",
                    "data_received",
                    "data_received",
                    "post",
                ]
            ),
        )


class DecoratedStreamingRequestFlowControlTest(
    BaseStreamingRequestFlowControlTest, WebTestCase
):
    def get_handlers(self):
        class DecoratedFlowControlHandler(BaseFlowControlHandler):
            @gen.coroutine
            def data_received(self, data):
                with self.in_method("data_received"):
                    yield gen.moment

        return [("/", DecoratedFlowControlHandler, dict(test=self))]


class NativeStreamingRequestFlowControlTest(
    BaseStreamingRequestFlowControlTest, WebTestCase
):
    def get_handlers(self):
        class NativeFlowControlHandler(BaseFlowControlHandler):
            async def data_received(self, data):
                with self.in_method("data_received"):
                    import asyncio

                    await asyncio.sleep(0)

        return [("/", NativeFlowControlHandler, dict(test=self))]


class IncorrectContentLengthTest(SimpleHandlerTestCase):
    def get_handlers(self):
        test = self
        self.server_error = None

        # Manually set a content-length that doesn't match the actual content.
        class TooHigh(RequestHandler):
            def get(self):
                self.set_header("Content-Length", "42")
                try:
                    self.finish("ok")
                except Exception as e:
                    test.server_error = e
                    raise

        class TooLow(RequestHandler):
            def get(self):
                self.set_header("Content-Length", "2")
                try:
                    self.finish("hello")
                except Exception as e:
                    test.server_error = e
                    raise

        return [("/high", TooHigh), ("/low", TooLow)]

    def test_content_length_too_high(self):
        # When the content-length is too high, the connection is simply
        # closed without completing the response.  An error is logged on
        # the server.
        with ExpectLog(app_log, "(Uncaught exception|Exception in callback)"):
            with ExpectLog(
                gen_log,
                "(Cannot send error response after headers written"
                "|Failed to flush partial response)",
            ):
                with self.assertRaises(HTTPClientError):
                    self.fetch("/high", raise_error=True)
        self.assertEqual(
            str(self.server_error), "Tried to write 40 bytes less than Content-Length"
        )

    def test_content_length_too_low(self):
        # When the content-length is too low, the connection is closed
        # without writing the last chunk, so the client never sees the request
        # complete (which would be a framing error).
        with ExpectLog(app_log, "(Uncaught exception|Exception in callback)"):
            with ExpectLog(
                gen_log,
                "(Cannot send error response after headers written"
                "|Failed to flush partial response)",
            ):
                with self.assertRaises(HTTPClientError):
                    self.fetch("/low", raise_error=True)
        self.assertEqual(
            str(self.server_error), "Tried to write more data than Content-Length"
        )


class ClientCloseTest(SimpleHandlerTestCase):
    class Handler(RequestHandler):
        def get(self):
            if self.request.version.startswith("HTTP/1"):
                # Simulate a connection closed by the client during
                # request processing.  The client will see an error, but the
                # server should respond gracefully (without logging errors
                # because we were unable to write out as many bytes as
                # Content-Length said we would)
                self.request.connection.stream.close()
                self.write("hello")
            else:
                # TODO: add a HTTP2-compatible version of this test.
                self.write("requires HTTP/1.x")

    def test_client_close(self):
        with self.assertRaises((HTTPClientError, unittest.SkipTest)):
            response = self.fetch("/", raise_error=True)
            if response.body == b"requires HTTP/1.x":
                self.skipTest("requires HTTP/1.x")
            self.assertEqual(response.code, 599)


class SignedValueTest(unittest.TestCase):
    SECRET = "It's a secret to everybody"
    SECRET_DICT = {0: "asdfbasdf", 1: "12312312", 2: "2342342"}

    def past(self):
        return self.present() - 86400 * 32

    def present(self):
        return 1300000000

    def test_known_values(self):
        signed_v1 = create_signed_value(
            SignedValueTest.SECRET, "key", "value", version=1, clock=self.present
        )
        self.assertEqual(
            signed_v1, b"dmFsdWU=|1300000000|31c934969f53e48164c50768b40cbd7e2daaaa4f"
        )

        signed_v2 = create_signed_value(
            SignedValueTest.SECRET, "key", "value", version=2, clock=self.present
        )
        self.assertEqual(
            signed_v2,
            b"2|1:0|10:1300000000|3:key|8:dmFsdWU=|"
            b"3d4e60b996ff9c5d5788e333a0cba6f238a22c6c0f94788870e1a9ecd482e152",
        )

        signed_default = create_signed_value(
            SignedValueTest.SECRET, "key", "value", clock=self.present
        )
        self.assertEqual(signed_default, signed_v2)

        decoded_v1 = decode_signed_value(
            SignedValueTest.SECRET, "key", signed_v1, min_version=1, clock=self.present
        )
        self.assertEqual(decoded_v1, b"value")

        decoded_v2 = decode_signed_value(
            SignedValueTest.SECRET, "key", signed_v2, min_version=2, clock=self.present
        )
        self.assertEqual(decoded_v2, b"value")

    def test_name_swap(self):
        signed1 = create_signed_value(
            SignedValueTest.SECRET, "key1", "value", clock=self.present
        )
        signed2 = create_signed_value(
            SignedValueTest.SECRET, "key2", "value", clock=self.present
        )
        # Try decoding each string with the other's "name"
        decoded1 = decode_signed_value(
            SignedValueTest.SECRET, "key2", signed1, clock=self.present
        )
        self.assertIs(decoded1, None)
        decoded2 = decode_signed_value(
            SignedValueTest.SECRET, "key1", signed2, clock=self.present
        )
        self.assertIs(decoded2, None)

    def test_expired(self):
        signed = create_signed_value(
            SignedValueTest.SECRET, "key1", "value", clock=self.past
        )
        decoded_past = decode_signed_value(
            SignedValueTest.SECRET, "key1", signed, clock=self.past
        )
        self.assertEqual(decoded_past, b"value")
        decoded_present = decode_signed_value(
            SignedValueTest.SECRET, "key1", signed, clock=self.present
        )
        self.assertIs(decoded_present, None)

    def test_payload_tampering(self):
        # These cookies are variants of the one in test_known_values.
        sig = "3d4e60b996ff9c5d5788e333a0cba6f238a22c6c0f94788870e1a9ecd482e152"

        def validate(prefix):
            return b"value" == decode_signed_value(
                SignedValueTest.SECRET, "key", prefix + sig, clock=self.present
            )

        self.assertTrue(validate("2|1:0|10:1300000000|3:key|8:dmFsdWU=|"))
        # Change key version
        self.assertFalse(validate("2|1:1|10:1300000000|3:key|8:dmFsdWU=|"))
        # length mismatch (field too short)
        self.assertFalse(validate("2|1:0|10:130000000|3:key|8:dmFsdWU=|"))
        # length mismatch (field too long)
        self.assertFalse(validate("2|1:0|10:1300000000|3:keey|8:dmFsdWU=|"))

    def test_signature_tampering(self):
        prefix = "2|1:0|10:1300000000|3:key|8:dmFsdWU=|"

        def validate(sig):
            return b"value" == decode_signed_value(
                SignedValueTest.SECRET, "key", prefix + sig, clock=self.present
            )

        self.assertTrue(
            validate("3d4e60b996ff9c5d5788e333a0cba6f238a22c6c0f94788870e1a9ecd482e152")
        )
        # All zeros
        self.assertFalse(validate("0" * 32))
        # Change one character
        self.assertFalse(
            validate("4d4e60b996ff9c5d5788e333a0cba6f238a22c6c0f94788870e1a9ecd482e152")
        )
        # Change another character
        self.assertFalse(
            validate("3d4e60b996ff9c5d5788e333a0cba6f238a22c6c0f94788870e1a9ecd482e153")
        )
        # Truncate
        self.assertFalse(
            validate("3d4e60b996ff9c5d5788e333a0cba6f238a22c6c0f94788870e1a9ecd482e15")
        )
        # Lengthen
        self.assertFalse(
            validate(
                "3d4e60b996ff9c5d5788e333a0cba6f238a22c6c0f94788870e1a9ecd482e1538"
            )
        )

    def test_non_ascii(self):
        value = b"\xe9"
        signed = create_signed_value(
            SignedValueTest.SECRET, "key", value, clock=self.present
        )
        decoded = decode_signed_value(
            SignedValueTest.SECRET, "key", signed, clock=self.present
        )
        self.assertEqual(value, decoded)

    def test_key_versioning_read_write_default_key(self):
        value = b"\xe9"
        signed = create_signed_value(
            SignedValueTest.SECRET_DICT, "key", value, clock=self.present, key_version=0
        )
        decoded = decode_signed_value(
            SignedValueTest.SECRET_DICT, "key", signed, clock=self.present
        )
        self.assertEqual(value, decoded)

    def test_key_versioning_read_write_non_default_key(self):
        value = b"\xe9"
        signed = create_signed_value(
            SignedValueTest.SECRET_DICT, "key", value, clock=self.present, key_version=1
        )
        decoded = decode_signed_value(
            SignedValueTest.SECRET_DICT, "key", signed, clock=self.present
        )
        self.assertEqual(value, decoded)

    def test_key_versioning_invalid_key(self):
        value = b"\xe9"
        signed = create_signed_value(
            SignedValueTest.SECRET_DICT, "key", value, clock=self.present, key_version=0
        )
        newkeys = SignedValueTest.SECRET_DICT.copy()
        newkeys.pop(0)
        decoded = decode_signed_value(newkeys, "key", signed, clock=self.present)
        self.assertEqual(None, decoded)

    def test_key_version_retrieval(self):
        value = b"\xe9"
        signed = create_signed_value(
            SignedValueTest.SECRET_DICT, "key", value, clock=self.present, key_version=1
        )
        key_version = get_signature_key_version(signed)
        self.assertEqual(1, key_version)


class XSRFTest(SimpleHandlerTestCase):
    class Handler(RequestHandler):
        def get(self):
            version = int(self.get_argument("version", "2"))
            # This would be a bad idea in a real app, but in this test
            # it's fine.
            self.settings["xsrf_cookie_version"] = version
            self.write(self.xsrf_token)

        def post(self):
            self.write("ok")

    def get_app_kwargs(self):
        return dict(xsrf_cookies=True)

    def setUp(self):
        super(XSRFTest, self).setUp()
        self.xsrf_token = self.get_token()

    def get_token(self, old_token=None, version=None):
        if old_token is not None:
            headers = self.cookie_headers(old_token)
        else:
            headers = None
        response = self.fetch(
            "/" if version is None else ("/?version=%d" % version), headers=headers
        )
        response.rethrow()
        return native_str(response.body)

    def cookie_headers(self, token=None):
        if token is None:
            token = self.xsrf_token
        return {"Cookie": "_xsrf=" + token}

    def test_xsrf_fail_no_token(self):
        with ExpectLog(gen_log, ".*'_xsrf' argument missing"):
            response = self.fetch("/", method="POST", body=b"")
        self.assertEqual(response.code, 403)

    def test_xsrf_fail_body_no_cookie(self):
        with ExpectLog(gen_log, ".*XSRF cookie does not match POST"):
            response = self.fetch(
                "/",
                method="POST",
                body=urllib.parse.urlencode(dict(_xsrf=self.xsrf_token)),
            )
        self.assertEqual(response.code, 403)

    def test_xsrf_fail_argument_invalid_format(self):
        with ExpectLog(gen_log, ".*'_xsrf' argument has invalid format"):
            response = self.fetch(
                "/",
                method="POST",
                headers=self.cookie_headers(),
                body=urllib.parse.urlencode(dict(_xsrf="3|")),
            )
        self.assertEqual(response.code, 403)

    def test_xsrf_fail_cookie_invalid_format(self):
        with ExpectLog(gen_log, ".*XSRF cookie does not match POST"):
            response = self.fetch(
                "/",
                method="POST",
                headers=self.cookie_headers(token="3|"),
                body=urllib.parse.urlencode(dict(_xsrf=self.xsrf_token)),
            )
        self.assertEqual(response.code, 403)

    def test_xsrf_fail_cookie_no_body(self):
        with ExpectLog(gen_log, ".*'_xsrf' argument missing"):
            response = self.fetch(
                "/", method="POST", body=b"", headers=self.cookie_headers()
            )
        self.assertEqual(response.code, 403)

    def test_xsrf_success_short_token(self):
        response = self.fetch(
            "/",
            method="POST",
            body=urllib.parse.urlencode(dict(_xsrf="deadbeef")),
            headers=self.cookie_headers(token="deadbeef"),
        )
        self.assertEqual(response.code, 200)

    def test_xsrf_success_non_hex_token(self):
        response = self.fetch(
            "/",
            method="POST",
            body=urllib.parse.urlencode(dict(_xsrf="xoxo")),
            headers=self.cookie_headers(token="xoxo"),
        )
        self.assertEqual(response.code, 200)

    def test_xsrf_success_post_body(self):
        response = self.fetch(
            "/",
            method="POST",
            body=urllib.parse.urlencode(dict(_xsrf=self.xsrf_token)),
            headers=self.cookie_headers(),
        )
        self.assertEqual(response.code, 200)

    def test_xsrf_success_query_string(self):
        response = self.fetch(
            "/?" + urllib.parse.urlencode(dict(_xsrf=self.xsrf_token)),
            method="POST",
            body=b"",
            headers=self.cookie_headers(),
        )
        self.assertEqual(response.code, 200)

    def test_xsrf_success_header(self):
        response = self.fetch(
            "/",
            method="POST",
            body=b"",
            headers=dict(
                {"X-Xsrftoken": self.xsrf_token},  # type: ignore
                **self.cookie_headers()
            ),
        )
        self.assertEqual(response.code, 200)

    def test_distinct_tokens(self):
        # Every request gets a distinct token.
        NUM_TOKENS = 10
        tokens = set()
        for i in range(NUM_TOKENS):
            tokens.add(self.get_token())
        self.assertEqual(len(tokens), NUM_TOKENS)

    def test_cross_user(self):
        token2 = self.get_token()
        # Each token can be used to authenticate its own request.
        for token in (self.xsrf_token, token2):
            response = self.fetch(
                "/",
                method="POST",
                body=urllib.parse.urlencode(dict(_xsrf=token)),
                headers=self.cookie_headers(token),
            )
            self.assertEqual(response.code, 200)
        # Sending one in the cookie and the other in the body is not allowed.
        for cookie_token, body_token in (
            (self.xsrf_token, token2),
            (token2, self.xsrf_token),
        ):
            with ExpectLog(gen_log, ".*XSRF cookie does not match POST"):
                response = self.fetch(
                    "/",
                    method="POST",
                    body=urllib.parse.urlencode(dict(_xsrf=body_token)),
                    headers=self.cookie_headers(cookie_token),
                )
            self.assertEqual(response.code, 403)

    def test_refresh_token(self):
        token = self.xsrf_token
        tokens_seen = set([token])
        # A user's token is stable over time.  Refreshing the page in one tab
        # might update the cookie while an older tab still has the old cookie
        # in its DOM.  Simulate this scenario by passing a constant token
        # in the body and re-querying for the token.
        for i in range(5):
            token = self.get_token(token)
            # Tokens are encoded uniquely each time
            tokens_seen.add(token)
            response = self.fetch(
                "/",
                method="POST",
                body=urllib.parse.urlencode(dict(_xsrf=self.xsrf_token)),
                headers=self.cookie_headers(token),
            )
            self.assertEqual(response.code, 200)
        self.assertEqual(len(tokens_seen), 6)

    def test_versioning(self):
        # Version 1 still produces distinct tokens per request.
        self.assertNotEqual(self.get_token(version=1), self.get_token(version=1))

        # Refreshed v1 tokens are all identical.
        v1_token = self.get_token(version=1)
        for i in range(5):
            self.assertEqual(self.get_token(v1_token, version=1), v1_token)

        # Upgrade to a v2 version of the same token
        v2_token = self.get_token(v1_token)
        self.assertNotEqual(v1_token, v2_token)
        # Each v1 token can map to many v2 tokens.
        self.assertNotEqual(v2_token, self.get_token(v1_token))

        # The tokens are cross-compatible.
        for cookie_token, body_token in ((v1_token, v2_token), (v2_token, v1_token)):
            response = self.fetch(
                "/",
                method="POST",
                body=urllib.parse.urlencode(dict(_xsrf=body_token)),
                headers=self.cookie_headers(cookie_token),
            )
            self.assertEqual(response.code, 200)


class XSRFCookieKwargsTest(SimpleHandlerTestCase):
    class Handler(RequestHandler):
        def get(self):
            self.write(self.xsrf_token)

    def get_app_kwargs(self):
        return dict(
            xsrf_cookies=True, xsrf_cookie_kwargs=dict(httponly=True, expires_days=2)
        )

    def test_xsrf_httponly(self):
        response = self.fetch("/")
        self.assertIn("httponly;", response.headers["Set-Cookie"].lower())
        self.assertIn("expires=", response.headers["Set-Cookie"].lower())
        header = response.headers.get("Set-Cookie")
        match = re.match(".*; expires=(?P<expires>.+);.*", header)
        assert match is not None

        expires = datetime.datetime.utcnow() + datetime.timedelta(days=2)
        parsed = email.utils.parsedate(match.groupdict()["expires"])
        assert parsed is not None
        header_expires = datetime.datetime(*parsed[:6])
        self.assertTrue(abs((expires - header_expires).total_seconds()) < 10)


class FinishExceptionTest(SimpleHandlerTestCase):
    class Handler(RequestHandler):
        def get(self):
            self.set_status(401)
            self.set_header("WWW-Authenticate", 'Basic realm="something"')
            if self.get_argument("finish_value", ""):
                raise Finish("authentication required")
            else:
                self.write("authentication required")
                raise Finish()

    def test_finish_exception(self):
        for u in ["/", "/?finish_value=1"]:
            response = self.fetch(u)
            self.assertEqual(response.code, 401)
            self.assertEqual(
                'Basic realm="something"', response.headers.get("WWW-Authenticate")
            )
            self.assertEqual(b"authentication required", response.body)


class DecoratorTest(WebTestCase):
    def get_handlers(self):
        class RemoveSlashHandler(RequestHandler):
            @removeslash
            def get(self):
                pass

        class AddSlashHandler(RequestHandler):
            @addslash
            def get(self):
                pass

        return [("/removeslash/", RemoveSlashHandler), ("/addslash", AddSlashHandler)]

    def test_removeslash(self):
        response = self.fetch("/removeslash/", follow_redirects=False)
        self.assertEqual(response.code, 301)
        self.assertEqual(response.headers["Location"], "/removeslash")

        response = self.fetch("/removeslash/?foo=bar", follow_redirects=False)
        self.assertEqual(response.code, 301)
        self.assertEqual(response.headers["Location"], "/removeslash?foo=bar")

    def test_addslash(self):
        response = self.fetch("/addslash", follow_redirects=False)
        self.assertEqual(response.code, 301)
        self.assertEqual(response.headers["Location"], "/addslash/")

        response = self.fetch("/addslash?foo=bar", follow_redirects=False)
        self.assertEqual(response.code, 301)
        self.assertEqual(response.headers["Location"], "/addslash/?foo=bar")


class CacheTest(WebTestCase):
    def get_handlers(self):
        class EtagHandler(RequestHandler):
            def get(self, computed_etag):
                self.write(computed_etag)

            def compute_etag(self):
                return self._write_buffer[0]

        return [("/etag/(.*)", EtagHandler)]

    def test_wildcard_etag(self):
        computed_etag = '"xyzzy"'
        etags = "*"
        self._test_etag(computed_etag, etags, 304)

    def test_strong_etag_match(self):
        computed_etag = '"xyzzy"'
        etags = '"xyzzy"'
        self._test_etag(computed_etag, etags, 304)

    def test_multiple_strong_etag_match(self):
        computed_etag = '"xyzzy1"'
        etags = '"xyzzy1", "xyzzy2"'
        self._test_etag(computed_etag, etags, 304)

    def test_strong_etag_not_match(self):
        computed_etag = '"xyzzy"'
        etags = '"xyzzy1"'
        self._test_etag(computed_etag, etags, 200)

    def test_multiple_strong_etag_not_match(self):
        computed_etag = '"xyzzy"'
        etags = '"xyzzy1", "xyzzy2"'
        self._test_etag(computed_etag, etags, 200)

    def test_weak_etag_match(self):
        computed_etag = '"xyzzy1"'
        etags = 'W/"xyzzy1"'
        self._test_etag(computed_etag, etags, 304)

    def test_multiple_weak_etag_match(self):
        computed_etag = '"xyzzy2"'
        etags = 'W/"xyzzy1", W/"xyzzy2"'
        self._test_etag(computed_etag, etags, 304)

    def test_weak_etag_not_match(self):
        computed_etag = '"xyzzy2"'
        etags = 'W/"xyzzy1"'
        self._test_etag(computed_etag, etags, 200)

    def test_multiple_weak_etag_not_match(self):
        computed_etag = '"xyzzy3"'
        etags = 'W/"xyzzy1", W/"xyzzy2"'
        self._test_etag(computed_etag, etags, 200)

    def _test_etag(self, computed_etag, etags, status_code):
        response = self.fetch(
            "/etag/" + computed_etag, headers={"If-None-Match": etags}
        )
        self.assertEqual(response.code, status_code)


class RequestSummaryTest(SimpleHandlerTestCase):
    class Handler(RequestHandler):
        def get(self):
            # remote_ip is optional, although it's set by
            # both HTTPServer and WSGIAdapter.
            # Clobber it to make sure it doesn't break logging.
            self.request.remote_ip = None
            self.finish(self._request_summary())

    def test_missing_remote_ip(self):
        resp = self.fetch("/")
        self.assertEqual(resp.body, b"GET / (None)")


class HTTPErrorTest(unittest.TestCase):
    def test_copy(self):
        e = HTTPError(403, reason="Go away")
        e2 = copy.copy(e)
        self.assertIsNot(e, e2)
        self.assertEqual(e.status_code, e2.status_code)
        self.assertEqual(e.reason, e2.reason)


class ApplicationTest(AsyncTestCase):
    def test_listen(self):
        app = Application([])
        server = app.listen(0, address="127.0.0.1")
        server.stop()


class URLSpecReverseTest(unittest.TestCase):
    def test_reverse(self):
        self.assertEqual("/favicon.ico", url(r"/favicon\.ico", None).reverse())
        self.assertEqual("/favicon.ico", url(r"^/favicon\.ico$", None).reverse())

    def test_non_reversible(self):
        # URLSpecs are non-reversible if they include non-constant
        # regex features outside capturing groups. Currently, this is
        # only strictly enforced for backslash-escaped character
        # classes.
        paths = [r"^/api/v\d+/foo/(\w+)$"]
        for path in paths:
            # A URLSpec can still be created even if it cannot be reversed.
            url_spec = url(path, None)
            try:
                result = url_spec.reverse()
                self.fail(
                    "did not get expected exception when reversing %s. "
                    "result: %s" % (path, result)
                )
            except ValueError:
                pass

    def test_reverse_arguments(self):
        self.assertEqual(
            "/api/v1/foo/bar", url(r"^/api/v1/foo/(\w+)$", None).reverse("bar")
        )


class RedirectHandlerTest(WebTestCase):
    def get_handlers(self):
        return [
            ("/src", WebRedirectHandler, {"url": "/dst"}),
            ("/src2", WebRedirectHandler, {"url": "/dst2?foo=bar"}),
            (r"/(.*?)/(.*?)/(.*)", WebRedirectHandler, {"url": "/{1}/{0}/{2}"}),
        ]

    def test_basic_redirect(self):
        response = self.fetch("/src", follow_redirects=False)
        self.assertEqual(response.code, 301)
        self.assertEqual(response.headers["Location"], "/dst")

    def test_redirect_with_argument(self):
        response = self.fetch("/src?foo=bar", follow_redirects=False)
        self.assertEqual(response.code, 301)
        self.assertEqual(response.headers["Location"], "/dst?foo=bar")

    def test_redirect_with_appending_argument(self):
        response = self.fetch("/src2?foo2=bar2", follow_redirects=False)
        self.assertEqual(response.code, 301)
        self.assertEqual(response.headers["Location"], "/dst2?foo=bar&foo2=bar2")

    def test_redirect_pattern(self):
        response = self.fetch("/a/b/c", follow_redirects=False)
        self.assertEqual(response.code, 301)
        self.assertEqual(response.headers["Location"], "/b/a/c")

Youez - 2016 - github.com/yon3zu
LinuXploit