
    G/jV]                     0   d dl Z d dlZd dlZd dlZd dlZd dlmZmZmZ d dl	m
Z
mZmZ d dlZd dlZd dlZd dlmZ d dlmZ d dlmZ d dlmZmZmZmZmZmZ  G d d	ej:                        Z G d
 deej>                        Z  G d dej:                        Z!y)    N)castUnionDict)patchMock	AsyncMock)ec)serialization)default_backend)	WebPusherNoDataWebPushExceptionCaseInsensitiveDictwebpushwebpush_asyncc                   "   e Zd ZdZddZd Zd Zd Zd Z e	d      d	        Z
 e	d      d
        Z e	j                  ed       e	j                  ej                  d      d               Z e	j                  ed       e	j                  ej                  d      d               Z e	d      d        Z e	d      d        Z e	d      d        Zd Zd Z e	d      d        Z e	d      d        Zd Z e	d      d        Z e	d      d        Zy)WebpushTestUtilszMHcCAQEEIPeN1iAipHbt8+/KZ2NIF8NeN24jqAmnMLFZEMocY8RboAoGCCqGSM49AwEHoUQDQgAEEJwJZq/GN8jJbo1GGpyU70hmP2hbWAUpQFKDByKB81yldJ9GTklBM5xqEwuPM7VuQcyiLDhvovthPIXx+gsQRQ==Nc                     |s0t        j                  t        j                         t                     }|t	        j
                  t        j                  d            j                  d      | j                  |      ddS )N      =)authp256dhendpointkeys)
r	   generate_private_key	SECP256R1r   base64urlsafe_b64encodeosurandomstrip_get_pubkey_str)selfrecv_keyr   s      Z/var/www/html/maxservice/venv/lib/python3.12/site-packages/pywebpush/tests/test_webpush.py_gen_subscription_infoz'WebpushTestUtils._gen_subscription_info#   sa    ..r||~?PQH 00B@FFtL..x8
 	
    c                     t        j                  |j                         j                  t        j
                  j                  t        j                  j                              j                  d      S )N)encodingformatr   )
r   r   
public_keypublic_bytesr
   EncodingX962PublicFormatUncompressedPointr"   )r$   priv_keys     r&   r#   z WebpushTestUtils._get_pubkey_str.   sZ    ''!..&//44$11CC / 

 %+	r(   c                    ddddd}d}| j                  t        t        ddddi       | j                  t        t        d	d
did       | j                  t        t        d	ddid       | j                  t        t        d	dddd       t        |      }|j                  |k7  sJ |j                  d   |d   k7  sJ |j                  d   |d   k(  sJ |j                  |k(  sJ |j
                  dk(  sJ y )Nhttps://example.com/zWBOrnIslXrUow2VAzKCUAE4sIbK00daEZCswOcf8m3TF8V82B-OpOg5JbmYLg44kRcvQC1E2gMJshsUYA-_zMPR8k8JV6sjdbhAi1n3_LDBLvA)r   r   r   sA   "WJ0P3(% l4u
q&1|ẂN[rM0!F =r   zAAA=https://example.comr   r   r   s   Un"},0K)assertRaisesr   r   subscription_inforeceiver_keyauth_key)r$   r8   	rk_decodepushs       r&   	test_initzWebpushTestUtils.test_init6   s=    /D 1	
! 	 	i&VV2T)U	
 	.68JK	

 	.8HI	

 	1#)6:	
 *+%%)::::%%f-1B61JJJJ%%j15Fz5RRRR  I---}} NNNNr(   c           	      h   dD ],  }t        j                  t        j                         t                     }| j	                  |      }d}t        |      }|j                  |j                         |      }	 d }d|v r't        j                  |j                  |d               }d }|dk7  r't        j                  |j                  |d               }t        j                  |j                  |d   d               }	t        j                  |d	   ||||	|
      }
|
j                  d      |k(  r-J  y )N)aesgcm	aes128gcm1Mary had a little lamb, with some nice mint jellycontent_encodingsaltr@   
crypto_keyr   r   body)rD   dhprivate_keyauth_secretversionutf8)r	   r   r   r   r'   r   encoder   urlsafe_b64decode_repadhttp_ecedecryptdecode)r$   rC   r%   r8   datar<   encodedraw_saltraw_dhraw_authdecodeds              r&   test_encodezWebpushTestUtils.test_encodef   s+    7 	2..r||~?PQH $ ; ;H EFD./Dkk$++-BRkSG H !33DKK4PQF;.11$++gl>S2TU//-f5f=>H &&$$(G >>&)T111;	2r(   c                     | j                         }d}t        |      }| j                  t        |j                  |d       y )NrA   	aesgcm128rB   r'   r   r7   r   rL   )r$   r8   rR   r<   s       r&   test_bad_content_encodingz*WebpushTestUtils.test_bad_content_encoding   sA     779B*+dkk4+ 	 	
r(   zrequests.postc                    | j                         }ddd}d}t        |      j                  ||       |j                  d      |j                  d   d   k(  sJ |j                  d   j                  d      }|j                  d	      d
k(  sJ |j                  d      |j                  d      k(  sJ |j                  d      }d|v sJ |j                  d      dk(  sJ y Npre-existingbearer vapidz
Crypto-KeyAuthenticationMary had a little lambr   r      headersttl0AUTHENTICATIONrb   
crypto-keycontent-encodingr@   r'   r   sendget	call_argsr$   	mock_postr8   re   rR   pheadersckeys          r&   	test_sendzWebpushTestUtils.test_send   s     779!/>R'#$))$8 $$Z0I4G4G4J14MMMM&&q)--i8||E"c)))||,-=M1NNNN||L)%%%||./;>>>r(   c           	         d|j                   _        | j                         }d}t        ||| j                  ddidddi       |j                  d	      |j                  d
   d
   k(  sJ |j                  d   j                  d      }|j                  d      dk(  sJ d }t        j                  t        j                   ||d   j                  d      d               j                  d            }|j                  d	d      j                  |d         sJ d|j                  d      v sJ |j                  d      }d|v sJ |j                  d      dk(  sJ |j                  d      dk(  sJ y )N   rc   submailto:ops@example.comr?   Test-Header
test-valuer8   rR   vapid_private_keyvapid_claimsrC   re   r   r   rd   re   rf   rg   c                 *    | dd t        |       dz   z   S Nz====   lenstrs    r&   repadz/WebpushTestUtils.test_send_vapid.<locals>.repad       #c(Q,///r(   authorization.rK    audvapidri   dh=rj   test-header)return_valuestatus_coder'   r   	vapid_keyrm   rn   jsonloadsr   rM   splitrQ   
startswithr$   rp   r8   rR   rq   r   r   rr   s           r&   test_send_vapidz WebpushTestUtils.test_send_vapid   sz   -0	* 779'/"nn!9:%"L1	
 !$$Z0I4G4G4J14MMMM&&q)--i8||E"c)))	0 zz$$h/55c:1=>fVn

 !$$Z4??ULLL(,,7777||L)}}||./8;;;||M*l:::r(   rl   signc                    d|j                   _        | j                         }d}t        j                  j                  | j                        }t        dd      }t        ||||       |j                  |       |j                          y Nru   rc   rw   r6   )rv   r   r8   rR   r{   r|   )r   r   r'   py_vapidVapidfrom_stringr   dictr   assert_called_once_withassert_called_oncer$   
vapid_signpusher_sendr8   rR   r   claimss          r&   test_webpush_vapid_instancez,WebpushTestUtils.test_webpush_vapid_instance   s     03  , 779'NN..t~~>	-1(.C.
 	/'		
 	**62&&(r(   c                    d|j                   _        | j                         }d}t        j                  j                  | j                        }t        ddt        t        j                         dz
              }t        ||||       |j                  |       |j                          t        |d         t        t        j                               kD  sJ y 	Nru   rc   rw   r6   iؽ  )rv   r   expr   r   )r   r   r'   r   r   r   r   r   inttimer   r   r   r   s          r&   test_webpush_vapid_expz'WebpushTestUtils.test_webpush_vapid_exp   s     03  , 779'NN..t~~>	(%DIIK%'(

 	/'		
 	**62&&(6%=!C		$4444r(   c           	          d|j                   _        | j                         }d}| j                  t        t
        ||ddd       y )Nru   rc   r6   rw   r   rv   r8   rR   r|   )r   r   r'   r7   r   r   r$   rp   r8   rR   s       r&   test_send_bad_vapid_no_keyz+WebpushTestUtils.test_send_bad_vapid_no_key   sN    -0	* 779'/,/ 	 		
r(   c           	          d|j                   _        | j                         }d}| j                  t        t
        ||ddd| j                         y )N  rc   r6   rw   r   r8   rR   r|   r{   )r   r   r'   r7   r   r   r   r   s       r&   test_send_bad_vapid_bad_returnz/WebpushTestUtils.test_send_bad_vapid_bad_return   sW    -0	* 779'/,/ #nn 	 
	
r(   c                    | j                         }ddd}t        |      j                  d|       |j                  d      |j                  d   d   k(  sJ |j                  d   j                  d      }|j                  d	      d
k(  sJ d|vsJ |j                  d      |j                  d      k(  sJ |j                  d      }d|v sJ y Nr_   r`   ra   r   r   r   rd   re   rf   rg   
encryptionrh   rb   ri   rk   r$   rp   r8   re   rq   rr   s         r&   test_send_emptyz WebpushTestUtils.test_send_empty  s     779!/>R#$))"g6 $$Z0I4G4G4J14MMMM&&q)--i8||E"c)))8+++||,-=M1NNNN||L)%%%r(   c                     | j                         }ddd}t        |      }| j                  t        |j                  d|       y )Nr_   r`   ra   r   )r'   r   r7   r   rL   )r$   r8   re   pushers       r&   test_encode_emptyz"WebpushTestUtils.test_encode_empty  s>     779!/>R,-&&--W=r(   c                     | j                         }|d= ddd}d}t        |      }| j                  t        |j                  ||       y )Nr   r_   r`   ra   	Somethingr[   )r$   r8   re   rR   r   s        r&   test_encode_no_cryptoz&WebpushTestUtils.test_encode_no_crypto  sL     779f%!/>R,-*FMM4Ir(   c                 <   | j                         }d}t        |      j                  |       |j                  d      |j                  d   d   k(  sJ |j                  d   j                  d      }|j                  d      dk(  sJ |j                  d      d	k(  sJ y 
Nrc   r   r   rd   re   rf   rg   rj   r@   rk   r$   rp   r8   rR   rq   s        r&   test_send_no_headersz%WebpushTestUtils.test_send_no_headers#  s     779'#$))$/ $$Z0I4G4G4J14MMMM&&q)--i8||E"c)))||./;>>>r(   zpywebpush.openc                     | j                         }t        |dddd| j                  d      }t        t        |      }dD ]  }||v rJ d|         y 	Nrc   r6   rw   r   T)rR   r|   r{   curl)z!curl -vX POST https://example.comz -H "content-encoding: aes128gcm"z-H "authorization: vapid z-H "ttl: 0"z-H "content-length:z	missing: )r'   r   r   r   r   )r$   openerr8   resultss        r&   test_as_curlzWebpushTestUtils.test_as_curl-  sp     779),/ #nn	
 c6"
 	0A ;/)A3/;	0r(   c                     t        ddd      }d|d   k(  sJ d|j                  d      k(  sJ d|j                  d      k(  sJ |d= |j                  d      J y )Napplebanana)FoobarfooFOOr   )r   rm   )r$   cis     r&   test_ci_dictzWebpushTestUtils.test_ci_dictD  sj     !BC"U)###"&&-'''"&&-'''uIvve}$$$r(   c                 "   d|j                   _        | j                         }t        |      j	                  d       |j
                  d   j                  d      dk(  sJ t        |d       |j
                  d   j                  d      dk(  sJ y )Nru   @timeoutrd   r   獗n $@)r   r   r'   r   rl   rn   rm   r   r$   rp   r8   s      r&   test_timeoutzWebpushTestUtils.test_timeoutL  s    -0	* 779#$))#)6""1%)))4;;;!62""1%)))4>>>r(   zrequests.Sessionc                    | j                         }ddd}d}t        ||      j                  ||       |j                  d      |j                  j
                  d   d   k(  sJ |j                  j
                  d   j                  d	      }|j                  d
      dk(  sJ |j                  d      |j                  d      k(  sJ |j                  d      }d|v sJ |j                  d      dk(  sJ y )Nr_   r`   ra   rc   )requests_sessionr   r   rd   re   rf   rg   rh   rb   ri   rj   r@   )r'   r   rl   rm   postrn   r$   mock_sessionr8   re   rR   rq   rr   s          r&    test_send_using_requests_sessionz1WebpushTestUtils.test_send_using_requests_sessionU  s     779!/>R'#lCHHwW $$Z0L4E4E4O4OPQ4RST4UUUU$$..q155i@||E"c)))||,-=M1NNNN||L)%%%||./;>>>r(   )Nr4   )__name__
__module____qualname__r   r'   r#   r=   rX   r\   r   rs   r   objectr   r   r   r   r   r   r   r   r   r   r   r   r   r   r    r(   r&   r   r      s   	/ 	
.O`2@
 ?? ? ?; ;> U\\)V$U\\(..&)) * %)" U\\)V$U\\(..&)5 * %5( ?
 
  ?
 
" ?
& 
&>J ?? ? 0 0,% ?? ? ? ?r(   r   c                      e Zd Z ede      d        Z ede      d        Z ede      d        Z ede      d        Z ede      d        Z	 ede      d	        Z
 ej                  ed
       ej                  ej                  d      d               Z ej                  ed
       ej                  ej                  d      d               Zd Z ede      d        Z ede      d        Zd Zy)WebPusherAsyncTestCasezaiohttp.ClientSession.post)new_callablec                   K   | j                         }ddd}d}t        |      j                  ||       d {    |j                  d      |j                  d   d   k(  sJ |j                  d   j                  d      }|j                  d	      d
k(  sJ |j                  d      |j                  d      k(  sJ |j                  d      }d|v sJ |j                  d      dk(  sJ y 7 wr^   r'   r   
send_asyncrm   rn   ro   s          r&   rs   z WebPusherAsyncTestCase.test_sende  s      779!/>R')*55dGDDD $$Z0I4G4G4J14MMMM&&q)--i8||E"c)))||,-=M1NNNN||L)%%%||./;>>> 	Es   6C,C*B2C,c                   K   | j                         }ddd}t        |      j                  d|       d {    |j                  d      |j                  d   d   k(  sJ |j                  d   j                  d      }|j                  d	      d
k(  sJ d|vsJ |j                  d      |j                  d      k(  sJ |j                  d      }d|v sJ y 7 wr   r   r   s         r&   r   z&WebPusherAsyncTestCase.test_send_emptys  s      779!/>R)*55b'BBB $$Z0I4G4G4J14MMMM&&q)--i8||E"c)))8+++||,-=M1NNNN||L)%%% 	Cs   4CCB"Cc                 X  K   | j                         }d}t        |      j                  |       d {    |j                  d      |j                  d   d   k(  sJ |j                  d   j                  d      }|j                  d      dk(  sJ |j                  d      d	k(  sJ y 7 uwr   r   r   s        r&   r   z+WebPusherAsyncTestCase.test_send_no_headers  s      779')*55d;;; $$Z0I4G4G4J14MMMM&&q)--i8||E"c)))||./;>>>	 	<s   0B*B(A6B*c                    K   d|j                   _        | j                         }t        |      j	                  d       d {    |j
                  d   j                  d      dk(  sJ y 7 (w)Nru   r   r   rd   r   )r   r   r'   r   r   rn   rm   r   s      r&   r   z#WebPusherAsyncTestCase.test_timeout  sj     -0	* 779)*55c5BBB""1%)))4;;; 	Cs   A A-A+)A-zaiohttp.ClientSessionc                   K   | j                         }ddd}d}t        ||      j                  ||       d {    |j                  d      |j                  j
                  d   d   k(  sJ |j                  j
                  d   j                  d	      }|j                  d
      dk(  sJ |j                  d      |j                  d      k(  sJ |j                  d      }d|v sJ |j                  d      dk(  sJ y 7 ŭw)Nr_   r`   ra   rc   )aiohttp_sessionr   r   rd   re   rf   rg   rh   rb   ri   rj   r@   )r'   r   r   rm   r   rn   r   s          r&   r   z7WebPusherAsyncTestCase.test_send_using_requests_session  s     779!/>R')<HSS'
 	
 	
 !$$Z0L4E4E4O4OPQ4RST4UUUU$$..q155i@||E"c)))||,-=M1NNNN||L)%%%||./;>>>	
s   8DD CDc           	      6  K   d|j                   _        t        d      |j                   _        | j	                         }d}t        ||| j                  ddiddd	i
       d {    |j                  d      |j                  d   d   k(  sJ |j                  d   j                  d      }|j                  d      dk(  sJ d }t        j                  t        j                   ||d   j                  d      d               j                  d            }|j                  dd      j                  |d         sJ d|j                  d      v sJ |j                  d      }d|v sJ |j                  d      dk(  sJ |j                  d      d	k(  sJ y 7 3w)Nru   r   r   rc   rv   rw   r?   rx   ry   rz   r   r   rd   re   rf   rg   c                 *    | dd t        |       dz   z   S r~   r   r   s    r&   r   z>WebPusherAsyncTestCase.test_webpush_async_vapid.<locals>.repad  r   r(   r   r   rK   r   r   ri   r   rj   r   )r   statusr   textr'   r   r   rm   rn   r   r   r   rM   r   rQ   r   r   s           r&   test_webpush_async_vapidz/WebPusherAsyncTestCase.test_webpush_async_vapid  s    (+	%&/R&@	# 779'/"nn!9:%"L1
 	
 	
 !$$Z0I4G4G4J14MMMM&&q)--i8||E"c)))	0 zz$$h/55c:1=>fVn

 !$$Z4??ULLL(,,7777||L)}}||./8;;;||M*l:::3	
s   A!F#F$D3Fr   r   c                 @  K   t               }d|_        ||_        | j                         }d}t        j
                  j                  | j                        }t        dd      }t        ||||       d {    |j                  |       |j                          y 7 &wr   )r   r   r   r'   r   r   r   r   r   r   r   r   r$   r   r   mock_responser8   rR   r   r   s           r&   !test_webpush_async_vapid_instancez8WebPusherAsyncTestCase.test_webpush_async_vapid_instance  s      "#0  779'NN..t~~>	-1(.C.
 /'	
 	
 	
 	**62&&(	
s   A3B5B6'Bc                   K   t               }d|_        ||_        | j                         }d}t        j
                  j                  | j                        }t        ddt        t        j                         dz
              }t        ||||       d {    |j                  |       |j                          t        |d         t        t        j                               kD  sJ y 7 Twr   )r   r   r   r'   r   r   r   r   r   r   r   r   r   r   r   s           r&   test_webpush_async_vapid_expz3WebPusherAsyncTestCase.test_webpush_async_vapid_exp  s      "#0  779'NN..t~~>	(%DIIK%'(

 /'	
 	
 	
 	**62&&(6%=!C		$4444	
s   BC+C)AC+c                    K   | j                         }d}| j                  t              5  t        ||ddd       d {    d d d        y 7 # 1 sw Y   y xY ww)Nrc   r6   rw   r   r   )r'   r7   r   r   )r$   r8   rR   s      r&   #test_webpush_async_bad_vapid_no_keyz:WebPusherAsyncTestCase.test_webpush_async_bad_vapid_no_key  sk      779'/0 	"303  	 		 	s-   (AAA A	AAAAc                 J  K   d|j                   _        d|j                   _        t        d      |j                   _        | j                         }d}| j                  t              5  t        ||ddd| j                  	       d {    d d d        y 7 # 1 sw Y   y xY ww)
Nr   GonezSubscription expiredr   rc   r6   rw   r   r   )
r   r   reasonr   r   r'   r7   r   r   r   r   s       r&   'test_webpush_async_bad_vapid_bad_returnz>WebPusherAsyncTestCase.test_webpush_async_bad_vapid_bad_return  s     (+	%(.	%&/=S&T	# 779'/0 		"303 #'..  		 				 		s0   A%B#' BBB	B#BB B#c                    K   t               }d|_        t        d      |_        ||_        | j                         }t        |d       d {    |j                  d   j                  d      dk(  sJ y 7 (w)Nru   r   r   r   r   rd   r   )	r   r   r   r   r   r'   r   rn   rm   )r$   rp   r   r8   s       r&   test_webpush_async_timeoutz1WebPusherAsyncTestCase.test_webpush_async_timeout  ss     "&B7!.	 779-v>>>""1%)))4>>> 	?s   A
A7A5)A7c                    K   | j                         }t        |dddd| j                  d       d {   }t        t        |      }dD ]  }||v rJ d|         y 7 *wr   )r'   r   r   r   r   )r$   r8   r   r   s       r&   test_webpush_async_as_curlz1WebPusherAsyncTestCase.test_webpush_async_as_curl  s~      779$),/ #nn	
 	
 c6"
 	0A ;/)A3/;	0	
s   1A AA A N)r   r   r   r   r   rs   r   r   r   r   r   r   r   r   r   r   r   r   r  r  r  r   r(   r&   r   r   d  sf   
'i@? A? 'i@
& A
& 'i@? A? 'i@< A< ";? <? 'i@; A;@ U\\)\*U\\(..&)) * +)& U\\)\*U\\(..&)5 * +5, 'i@ A$ 'i@? A?0r(   r   c                       e Zd Zd Zy)WebpushExceptionTestCasec                    ddl m} t        d      }| dk(  sJ t        |      }d|_        t        j                  |j                        |j
                  _        d|_        d|_	        t        d|      }| d	j                  |j                        k(  sJ |j                   sJ d
       t        t         j                  |j                        j                         j                  d      dk(  sJ t        dg d      }| dk(  sJ y )Nr   )Responser   zWebPushException: foo)specz{"code": 401, "errno": 109, "error": "Unauthorized", "more_info": "http://autopush.readthedocs.io/en/latest/http.html#error-codes", "message": "Request did not validate missing authorization header"}i  Unauthorizedz"WebPushException: foo, Response {}z<Response [401]>errnom   )rd         z)WebPushException: foo, Response [1, 2, 3])requestsr	  r   r   r   r   r   r   r   r   r+   responser   rm   )r$   r	  r   r  s       r&   test_exceptionz'WebpushExceptionTestCase.test_exception1  s   %u%2222 X& 	 &*ZZ%>""(uh/?FFMM
 
 	
 
 ,, 4"44 H%%s||499;??HCOOOui0FFFFr(   N)r   r   r   r  r   r(   r&   r  r  0  s    Gr(   r  )"r   r   r    unittestr   typingr   r   r   unittest.mockr   r   r   rO   r   r  )cryptography.hazmat.primitives.asymmetricr	   cryptography.hazmat.primitivesr
   cryptography.hazmat.backendsr   	pywebpushr   r   r   r   r   r   TestCaser   IsolatedAsyncioTestCaser   r  r   r(   r&   <module>r     s}      	   $ $ 0 0    8 8 8 G?x(( G?T
I0-x/O/O I0XGx00 Gr(   