<s> from datetime import timedelta <EOL> from mock import patch <EOL> import re <EOL> from django . core import management <EOL> from django . test import TestCase <EOL> from django . test . utils import override_settings <EOL> from django . utils . timezone import now <EOL> import requests <EOL> from . factories import ConsumedProxyGrantingTicketFactory <EOL> from . factories import ConsumedProxyTicketFactory <EOL> from . factories import ConsumedServiceTicketFactory <EOL> from . factories import ExpiredProxyGrantingTicketFactory <EOL> from . factories import ExpiredServiceTicketFactory <EOL> from . factories import ProxyGrantingTicketFactory <EOL> from . factories import ProxyTicketFactory <EOL> from . factories import ServiceTicketFactory <EOL> from . factories import UserFactory <EOL> from mama_cas . models import ProxyGrantingTicket <EOL> from mama_cas . models import ProxyTicket <EOL> from mama_cas . models import ServiceTicket <EOL> from mama_cas . exceptions import InvalidProxyCallback <EOL> from mama_cas . exceptions import InvalidRequest <EOL> from mama_cas . exceptions import InvalidService <EOL> from mama_cas . exceptions import InvalidTicket <EOL> from mama_cas . exceptions import UnauthorizedServiceProxy <EOL> class TicketManagerTests ( TestCase ) : <EOL> """<STR_LIT>""" <EOL> url = '<STR_LIT>' <EOL> def setUp ( self ) : <EOL> self . user = UserFactory ( ) <EOL> def test_create_ticket ( self ) : <EOL> """<STR_LIT>""" <EOL> st = ServiceTicket . objects . create_ticket ( user = self . user ) <EOL> self . assertTrue ( re . search ( st . TICKET_RE , st . ticket ) ) <EOL> def test_create_ticket_ticket ( self ) : <EOL> """<STR_LIT>""" <EOL> ticket = '<STR_LIT>' <EOL> st = ServiceTicket . objects . create_ticket ( ticket = ticket , user = self . user ) <EOL> self . assertEqual ( st . ticket , ticket ) <EOL> def test_create_ticket_service ( self ) : <EOL> """<STR_LIT>""" <EOL> service = '<STR_LIT>' <EOL> st = ServiceTicket . objects . create_ticket ( service = service , user = self . user ) <EOL> self . assertEqual ( st . service , '<STR_LIT>' ) <EOL> def test_create_ticket_no_expires ( self ) : <EOL> """<STR_LIT>""" <EOL> st = ServiceTicket . objects . create_ticket ( user = self . user ) <EOL> self . assertTrue ( st . expires > now ( ) ) <EOL> def test_create_ticket_expires ( self ) : <EOL> """<STR_LIT>""" <EOL> expires = now ( ) + timedelta ( seconds = <NUM_LIT:30> ) <EOL> st = ServiceTicket . objects . create_ticket ( expires = expires , user = self . user ) <EOL> self . assertEqual ( st . expires , expires ) <EOL> def test_create_ticket_str ( self ) : <EOL> """<STR_LIT>""" <EOL> str = ServiceTicket . objects . create_ticket_str ( ) <EOL> self . assertTrue ( re . search ( '<STR_LIT>' , str ) ) <EOL> def test_create_ticket_str_prefix ( self ) : <EOL> """<STR_LIT>""" <EOL> str = ProxyGrantingTicket . objects . create_ticket_str ( prefix = '<STR_LIT>' ) <EOL> self . assertTrue ( re . search ( '<STR_LIT>' , str ) ) <EOL> def test_validate_ticket ( self ) : <EOL> """<STR_LIT>""" <EOL> st = ServiceTicketFactory ( ) <EOL> ticket = ServiceTicket . objects . validate_ticket ( st . ticket , self . url ) <EOL> self . assertEqual ( ticket , st ) <EOL> self . assertTrue ( ticket . is_consumed ( ) ) <EOL> def test_validate_ticket_no_ticket ( self ) : <EOL> """<STR_LIT>""" <EOL> with self . assertRaises ( InvalidRequest ) : <EOL> ServiceTicket . objects . validate_ticket ( None , self . url ) <EOL> def test_validate_ticket_invalid_ticket ( self ) : <EOL> """<STR_LIT>""" <EOL> with self . assertRaises ( InvalidTicket ) : <EOL> ServiceTicket . objects . validate_ticket ( '<STR_LIT>' , self . url ) <EOL> def test_validate_ticket_does_not_exist ( self ) : <EOL> """<STR_LIT>""" <EOL> ticket = '<STR_LIT>' <EOL> with self . assertRaises ( InvalidTicket ) : <EOL> ServiceTicket . objects . validate_ticket ( ticket , self . url ) <EOL> def test_validate_ticket_consumed_ticket ( self ) : <EOL> """<STR_LIT>""" <EOL> st = ConsumedServiceTicketFactory ( ) <EOL> with self . assertRaises ( InvalidTicket ) : <EOL> ServiceTicket . objects . validate_ticket ( st . ticket , self . url ) <EOL> def test_validate_ticket_expired_ticket ( self ) : <EOL> """<STR_LIT>""" <EOL> st = ExpiredServiceTicketFactory ( ) <EOL> with self . assertRaises ( InvalidTicket ) : <EOL> ServiceTicket . objects . validate_ticket ( st . ticket , self . url ) <EOL> def test_validate_ticket_no_service ( self ) : <EOL> """<STR_LIT>""" <EOL> st = ServiceTicketFactory ( ) <EOL> with self . assertRaises ( InvalidRequest ) : <EOL> ServiceTicket . objects . validate_ticket ( st . ticket , None ) <EOL> st = ServiceTicket . objects . get ( ticket = st . ticket ) <EOL> self . assertTrue ( st . is_consumed ( ) ) <EOL> def test_validate_ticket_invalid_service ( self ) : <EOL> """<STR_LIT>""" <EOL> service = '<STR_LIT>' <EOL> st = ServiceTicketFactory ( ) <EOL> with self . assertRaises ( InvalidService ) : <EOL> ServiceTicket . objects . validate_ticket ( st . ticket , service ) <EOL> def test_validate_ticket_service_mismatch ( self ) : <EOL> """<STR_LIT>""" <EOL> service = '<STR_LIT>' <EOL> st = ServiceTicketFactory ( ) <EOL> with self . assertRaises ( InvalidService ) : <EOL> ServiceTicket . objects . validate_ticket ( st . ticket , service ) <EOL> def test_validate_ticket_renew ( self ) : <EOL> """<STR_LIT>""" <EOL> st = ServiceTicketFactory ( primary = True ) <EOL> ticket = ServiceTicket . objects . validate_ticket ( st . ticket , self . url , <EOL> renew = True ) <EOL> self . assertEqual ( ticket , st ) <EOL> def test_validate_ticket_renew_secondary ( self ) : <EOL> """<STR_LIT>""" <EOL> st = ServiceTicketFactory ( ) <EOL> with self . assertRaises ( InvalidTicket ) : <EOL> ServiceTicket . objects . validate_ticket ( st . ticket , self . url , <EOL> renew = True ) <EOL> def test_delete_invalid_tickets ( self ) : <EOL> """<STR_LIT>""" <EOL> ServiceTicketFactory ( ) <EOL> expired = ExpiredServiceTicketFactory ( ) <EOL> consumed = ConsumedServiceTicketFactory ( ) <EOL> referenced = ConsumedServiceTicketFactory ( ) <EOL> ProxyGrantingTicketFactory ( granted_by_st = referenced ) <EOL> ServiceTicket . objects . delete_invalid_tickets ( ) <EOL> self . assertEqual ( ServiceTicket . objects . count ( ) , <NUM_LIT:2> ) <EOL> self . assertRaises ( ServiceTicket . DoesNotExist , <EOL> ServiceTicket . objects . get , <EOL> ticket = expired . ticket ) <EOL> self . assertRaises ( ServiceTicket . DoesNotExist , <EOL> ServiceTicket . objects . get , <EOL> ticket = consumed . ticket ) <EOL> def test_consume_tickets ( self ) : <EOL> """<STR_LIT>""" <EOL> st1 = ServiceTicketFactory ( ) <EOL> st2 = ServiceTicketFactory ( ) <EOL> ServiceTicket . objects . consume_tickets ( self . user ) <EOL> self . assertTrue ( ServiceTicket . objects . get ( ticket = st1 ) . is_consumed ( ) ) <EOL> self . assertTrue ( ServiceTicket . objects . get ( ticket = st2 ) . is_consumed ( ) ) <EOL> class TicketTests ( TestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_ticket_consumed ( self ) : <EOL> """<STR_LIT>""" <EOL> st = ServiceTicketFactory ( ) <EOL> st . consume ( ) <EOL> st = ServiceTicket . objects . get ( ticket = st . ticket ) <EOL> self . assertTrue ( st . is_consumed ( ) ) <EOL> def test_ticket_not_consumed ( self ) : <EOL> """<STR_LIT>""" <EOL> st = ServiceTicketFactory ( ) <EOL> self . assertFalse ( st . is_consumed ( ) ) <EOL> def test_ticket_expired ( self ) : <EOL> """<STR_LIT>""" <EOL> st = ExpiredServiceTicketFactory ( ) <EOL> self . assertTrue ( st . is_expired ( ) ) <EOL> def test_ticket_not_expired ( self ) : <EOL> """<STR_LIT>""" <EOL> st = ServiceTicketFactory ( ) <EOL> self . assertFalse ( st . is_expired ( ) ) <EOL> @ override_settings ( MAMA_CAS_ENABLE_SINGLE_SIGN_OUT = True ) <EOL> class ServiceTicketManagerTests ( TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> self . user = UserFactory ( ) <EOL> def test_request_sign_out ( self ) : <EOL> """<STR_LIT>""" <EOL> ConsumedServiceTicketFactory ( ) <EOL> ConsumedServiceTicketFactory ( ) <EOL> with patch ( '<STR_LIT>' ) as mock : <EOL> mock . return_value . status_code = <NUM_LIT:200> <EOL> ServiceTicket . objects . request_sign_out ( self . user ) <EOL> self . assertEqual ( mock . call_count , <NUM_LIT:2> ) <EOL> @ override_settings ( MAMA_CAS_ASYNC_CONCURRENCY = <NUM_LIT:0> ) <EOL> def test_request_sign_out_no_pool ( self ) : <EOL> """<STR_LIT>""" <EOL> ConsumedServiceTicketFactory ( ) <EOL> ConsumedServiceTicketFactory ( ) <EOL> with patch ( '<STR_LIT>' ) as mock : <EOL> mock . return_value . status_code = <NUM_LIT:200> <EOL> ServiceTicket . objects . request_sign_out ( self . user ) <EOL> self . assertEqual ( mock . call_count , <NUM_LIT:2> ) <EOL> class ServiceTicketTests ( TestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_create_service_ticket ( self ) : <EOL> """<STR_LIT>""" <EOL> st = ServiceTicketFactory ( ) <EOL> self . assertTrue ( st . ticket . startswith ( st . TICKET_PREFIX ) ) <EOL> def test_primary ( self ) : <EOL> """<STR_LIT>""" <EOL> st = ServiceTicketFactory ( primary = True ) <EOL> self . assertTrue ( st . is_primary ( ) ) <EOL> def test_secondary ( self ) : <EOL> """<STR_LIT>""" <EOL> st = ServiceTicketFactory ( ) <EOL> self . assertFalse ( st . is_primary ( ) ) <EOL> def test_request_sign_out ( self ) : <EOL> """<STR_LIT>""" <EOL> st = ServiceTicketFactory ( ) <EOL> with patch ( '<STR_LIT>' ) as mock : <EOL> mock . return_value . status_code = <NUM_LIT:200> <EOL> st . request_sign_out ( ) <EOL> def test_request_sign_out_exception ( self ) : <EOL> """<STR_LIT>""" <EOL> st = ServiceTicketFactory ( ) <EOL> with patch ( '<STR_LIT>' ) as mock : <EOL> mock . side_effect = requests . exceptions . RequestException <EOL> st . request_sign_out ( ) <EOL> def test_request_sign_out_invalid_status ( self ) : <EOL> """<STR_LIT>""" <EOL> st = ServiceTicketFactory ( ) <EOL> with patch ( '<STR_LIT>' ) as mock : <EOL> mock . return_value . status_code = <NUM_LIT> <EOL> st . request_sign_out ( ) <EOL> class ProxyTicketTests ( TestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_create_proxy_ticket ( self ) : <EOL> """<STR_LIT>""" <EOL> pt = ProxyTicketFactory ( ) <EOL> self . assertTrue ( pt . ticket . startswith ( pt . TICKET_PREFIX ) ) <EOL> class ProxyGrantingTicketManager ( TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> self . user = UserFactory ( ) <EOL> self . pt = ProxyTicketFactory ( ) <EOL> self . pgtid = ProxyGrantingTicket . objects . create_ticket_str ( ) <EOL> self . pgtiou = ProxyGrantingTicket . objects . create_ticket_str ( prefix = ProxyGrantingTicket . IOU_PREFIX ) <EOL> def test_create_ticket ( self ) : <EOL> """<STR_LIT>""" <EOL> with patch ( '<STR_LIT>' ) as mock : <EOL> mock . return_value . status_code = <NUM_LIT:200> <EOL> pgt = ProxyGrantingTicket . objects . create_ticket ( '<STR_LIT>' , '<STR_LIT>' , <EOL> user = self . user , granted_by_pt = self . pt ) <EOL> self . assertTrue ( re . search ( pgt . TICKET_RE , pgt . ticket ) ) <EOL> self . assertTrue ( pgt . iou . startswith ( pgt . IOU_PREFIX ) ) <EOL> def test_create_ticket_invalid_pgturl ( self ) : <EOL> """<STR_LIT>""" <EOL> with patch ( '<STR_LIT>' ) as mock : <EOL> mock . side_effect = requests . exceptions . ConnectionError <EOL> pgt = ProxyGrantingTicket . objects . create_ticket ( '<STR_LIT>' , '<STR_LIT>' , <EOL> user = self . user , granted_by_pt = self . pt ) <EOL> self . assertIsNone ( pgt ) <EOL> def test_validate_callback ( self ) : <EOL> """<STR_LIT>""" <EOL> with patch ( '<STR_LIT>' ) as mock : <EOL> mock . return_value . status_code = <NUM_LIT:200> <EOL> try : <EOL> ProxyGrantingTicket . objects . validate_callback ( '<STR_LIT>' , '<STR_LIT>' , <EOL> self . pgtid , self . pgtiou ) <EOL> except InvalidProxyCallback : <EOL> self . fail ( "<STR_LIT>" ) <EOL> def test_validate_callback_unauthorized_service ( self ) : <EOL> """<STR_LIT>""" <EOL> with self . assertRaises ( UnauthorizedServiceProxy ) : <EOL> ProxyGrantingTicket . objects . validate_callback ( '<STR_LIT>' , '<STR_LIT>' , <EOL> self . pgtid , self . pgtiou ) <EOL> def test_validate_callback_http_pgturl ( self ) : <EOL> """<STR_LIT>""" <EOL> with self . assertRaises ( InvalidProxyCallback ) : <EOL> ProxyGrantingTicket . objects . validate_callback ( '<STR_LIT>' , '<STR_LIT>' , <EOL> self . pgtid , self . pgtiou ) <EOL> def test_validate_callback_invalid_pgturl ( self ) : <EOL> """<STR_LIT>""" <EOL> with self . assertRaises ( InvalidProxyCallback ) : <EOL> ProxyGrantingTicket . objects . validate_callback ( '<STR_LIT>' , '<STR_LIT>' , <EOL> self . pgtid , self . pgtiou ) <EOL> def test_validate_callback_ssl_error ( self ) : <EOL> """<STR_LIT>""" <EOL> with patch ( '<STR_LIT>' ) as mock : <EOL> mock . side_effect = requests . exceptions . SSLError <EOL> with self . assertRaises ( InvalidProxyCallback ) : <EOL> ProxyGrantingTicket . objects . validate_callback ( '<STR_LIT>' , '<STR_LIT>' , <EOL> self . pgtid , self . pgtiou ) <EOL> def test_validate_callback_connection_error ( self ) : <EOL> """<STR_LIT>""" <EOL> with patch ( '<STR_LIT>' ) as mock : <EOL> mock . side_effect = requests . exceptions . ConnectionError <EOL> with self . assertRaises ( InvalidProxyCallback ) : <EOL> ProxyGrantingTicket . objects . validate_callback ( '<STR_LIT>' , '<STR_LIT>' , <EOL> self . pgtid , self . pgtiou ) <EOL> def test_validate_callback_timeout ( self ) : <EOL> """<STR_LIT>""" <EOL> with patch ( '<STR_LIT>' ) as mock : <EOL> mock . side_effect = requests . exceptions . Timeout <EOL> with self . assertRaises ( InvalidProxyCallback ) : <EOL> ProxyGrantingTicket . objects . validate_callback ( '<STR_LIT>' , '<STR_LIT>' , <EOL> self . pgtid , self . pgtiou ) <EOL> def test_validate_callback_invalid_status ( self ) : <EOL> """<STR_LIT>""" <EOL> with patch ( '<STR_LIT>' ) as mock : <EOL> mock . return_value . raise_for_status . side_effect = requests . exceptions . HTTPError <EOL> with self . assertRaises ( InvalidProxyCallback ) : <EOL> ProxyGrantingTicket . objects . validate_callback ( '<STR_LIT>' , '<STR_LIT>' , <EOL> self . pgtid , self . pgtiou ) <EOL> def test_validate_ticket ( self ) : <EOL> """<STR_LIT>""" <EOL> pgt = ProxyGrantingTicketFactory ( ) <EOL> ticket = ProxyGrantingTicket . objects . validate_ticket ( pgt . ticket , '<STR_LIT>' ) <EOL> self . assertEqual ( ticket , pgt ) <EOL> self . assertFalse ( ticket . is_consumed ( ) ) <EOL> def test_validate_ticket_no_ticket ( self ) : <EOL> """<STR_LIT>""" <EOL> with self . assertRaises ( InvalidRequest ) : <EOL> ProxyGrantingTicket . objects . validate_ticket ( None , '<STR_LIT>' ) <EOL> def test_validate_ticket_no_service ( self ) : <EOL> """<STR_LIT>""" <EOL> pgt = ProxyGrantingTicketFactory ( ) <EOL> with self . assertRaises ( InvalidRequest ) : <EOL> ProxyGrantingTicket . objects . validate_ticket ( pgt . ticket , None ) <EOL> def test_validate_ticket_invalid_ticket ( self ) : <EOL> """<STR_LIT>""" <EOL> with self . assertRaises ( InvalidTicket ) : <EOL> ProxyGrantingTicket . objects . validate_ticket ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> def test_validate_ticket_does_not_exist ( self ) : <EOL> """<STR_LIT>""" <EOL> ticket = '<STR_LIT>' <EOL> with self . assertRaises ( InvalidTicket ) : <EOL> ProxyGrantingTicket . objects . validate_ticket ( ticket , '<STR_LIT>' ) <EOL> def test_validate_ticket_consumed_ticket ( self ) : <EOL> """<STR_LIT>""" <EOL> pgt = ConsumedProxyGrantingTicketFactory ( ) <EOL> with self . assertRaises ( InvalidTicket ) : <EOL> ProxyGrantingTicket . objects . validate_ticket ( pgt . ticket , '<STR_LIT>' ) <EOL> def test_validate_ticket_expired_ticket ( self ) : <EOL> """<STR_LIT>""" <EOL> pgt = ExpiredProxyGrantingTicketFactory ( ) <EOL> with self . assertRaises ( InvalidTicket ) : <EOL> ProxyGrantingTicket . objects . validate_ticket ( pgt . ticket , '<STR_LIT>' ) <EOL> def test_validate_ticket_invalid_service ( self ) : <EOL> """<STR_LIT>""" <EOL> pgt = ProxyGrantingTicketFactory ( ) <EOL> with self . assertRaises ( InvalidService ) : <EOL> ProxyGrantingTicket . objects . validate_ticket ( pgt . ticket , '<STR_LIT>' ) <EOL> class ProxyGrantingTicketTests ( TestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_create_proxy_granting_ticket ( self ) : <EOL> """<STR_LIT>""" <EOL> pgt = ProxyGrantingTicketFactory ( ) <EOL> self . assertTrue ( pgt . ticket . startswith ( pgt . TICKET_PREFIX ) ) <EOL> class ManagementCommandTests ( TestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_cleanupcas_management_command ( self ) : <EOL> """<STR_LIT>""" <EOL> st = ConsumedServiceTicketFactory ( ) <EOL> pgt = ExpiredProxyGrantingTicketFactory ( granted_by_st = st ) <EOL> ConsumedProxyTicketFactory ( granted_by_pgt = pgt ) <EOL> management . call_command ( '<STR_LIT>' ) <EOL> self . assertEqual ( ServiceTicket . objects . count ( ) , <NUM_LIT:0> ) <EOL> self . assertEqual ( ProxyGrantingTicket . objects . count ( ) , <NUM_LIT:0> ) <EOL> self . assertEqual ( ProxyTicket . objects . count ( ) , <NUM_LIT:0> ) <EOL> def test_cleanupcas_management_command_chain ( self ) : <EOL> """<STR_LIT>""" <EOL> st = ConsumedServiceTicketFactory ( ) <EOL> pgt = ExpiredProxyGrantingTicketFactory ( expires = now ( ) - timedelta ( seconds = <NUM_LIT:5> ) , <EOL> granted_by_st = st ) <EOL> pt = ConsumedProxyTicketFactory ( granted_by_pgt = pgt ) <EOL> pgt2 = ExpiredProxyGrantingTicketFactory ( granted_by_st = None , <EOL> granted_by_pt = pt ) <EOL> ConsumedProxyTicketFactory ( granted_by_pgt = pgt2 ) <EOL> management . call_command ( '<STR_LIT>' ) <EOL> self . assertEqual ( ServiceTicket . objects . count ( ) , <NUM_LIT:0> ) <EOL> self . assertEqual ( ProxyGrantingTicket . objects . count ( ) , <NUM_LIT:0> ) <EOL> self . assertEqual ( ProxyTicket . objects . count ( ) , <NUM_LIT:0> ) </s>
