How to use session method in Cypress

...47        # NB: be careful to delete any sessions created; stale sessions fill up48        # the /tmp (with some backends) and eventually overwhelm it after lots49        # of runs (think buildbots)50        self.session.delete()51    def test_new_session(self):52        self.assertIs(self.session.modified, False)53        self.assertIs(self.session.accessed, False)54    def test_get_empty(self):55        self.assertIsNone(self.session.get('cat'))56    def test_store(self):57        self.session['cat'] = "dog"58        self.assertIs(self.session.modified, True)59        self.assertEqual(self.session.pop('cat'), 'dog')60    def test_pop(self):61        self.session['some key'] = 'exists'62        # Need to reset these to pretend we haven't accessed it:63        self.accessed = False64        self.modified = False65        self.assertEqual(self.session.pop('some key'), 'exists')66        self.assertIs(self.session.accessed, True)67        self.assertIs(self.session.modified, True)68        self.assertIsNone(self.session.get('some key'))69    def test_pop_default(self):70        self.assertEqual(self.session.pop('some key', 'does not exist'),71                         'does not exist')72        self.assertIs(self.session.accessed, True)73        self.assertIs(self.session.modified, False)74    def test_pop_default_named_argument(self):75        self.assertEqual(self.session.pop('some key', default='does not exist'), 'does not exist')76        self.assertIs(self.session.accessed, True)77        self.assertIs(self.session.modified, False)78    def test_pop_no_default_keyerror_raised(self):79        with self.assertRaises(KeyError):80            self.session.pop('some key')81    def test_setdefault(self):82        self.assertEqual(self.session.setdefault('foo', 'bar'), 'bar')83        self.assertEqual(self.session.setdefault('foo', 'baz'), 'bar')84        self.assertIs(self.session.accessed, True)85        self.assertIs(self.session.modified, True)86    def test_update(self):87        self.session.update({'update key': 1})88        self.assertIs(self.session.accessed, True)89        self.assertIs(self.session.modified, True)90        self.assertEqual(self.session.get('update key', None), 1)91    def test_has_key(self):92        self.session['some key'] = 193        self.session.modified = False94        self.session.accessed = False95        self.assertIn('some key', self.session)96        self.assertIs(self.session.accessed, True)97        self.assertIs(self.session.modified, False)98    def test_values(self):99        self.assertEqual(list(self.session.values()), [])100        self.assertIs(self.session.accessed, True)101        self.session['some key'] = 1102        self.session.modified = False103        self.session.accessed = False104        self.assertEqual(list(self.session.values()), [1])105        self.assertIs(self.session.accessed, True)106        self.assertIs(self.session.modified, False)107    def test_keys(self):108        self.session['x'] = 1109        self.session.modified = False110        self.session.accessed = False111        self.assertEqual(list(self.session.keys()), ['x'])112        self.assertIs(self.session.accessed, True)113        self.assertIs(self.session.modified, False)114    def test_items(self):115        self.session['x'] = 1116        self.session.modified = False117        self.session.accessed = False118        self.assertEqual(list(self.session.items()), [('x', 1)])119        self.assertIs(self.session.accessed, True)120        self.assertIs(self.session.modified, False)121    def test_clear(self):122        self.session['x'] = 1123        self.session.modified = False124        self.session.accessed = False125        self.assertEqual(list(self.session.items()), [('x', 1)])126        self.session.clear()127        self.assertEqual(list(self.session.items()), [])128        self.assertIs(self.session.accessed, True)129        self.assertIs(self.session.modified, True)130    def test_save(self):131        self.session.save()132        self.assertIs(self.session.exists(self.session.session_key), True)133    def test_delete(self):134        self.session.save()135        self.session.delete(self.session.session_key)136        self.assertIs(self.session.exists(self.session.session_key), False)137    def test_flush(self):138        self.session['foo'] = 'bar'139        self.session.save()140        prev_key = self.session.session_key141        self.session.flush()142        self.assertIs(self.session.exists(prev_key), False)143        self.assertNotEqual(self.session.session_key, prev_key)144        self.assertIsNone(self.session.session_key)145        self.assertIs(self.session.modified, True)146        self.assertIs(self.session.accessed, True)147    def test_cycle(self):148        self.session['a'], self.session['b'] = 'c', 'd'149        self.session.save()150        prev_key = self.session.session_key151        prev_data = list(self.session.items())152        self.session.cycle_key()153        self.assertIs(self.session.exists(prev_key), False)154        self.assertNotEqual(self.session.session_key, prev_key)155        self.assertEqual(list(self.session.items()), prev_data)156    def test_cycle_with_no_session_cache(self):157        self.session['a'], self.session['b'] = 'c', 'd'158        self.session.save()159        prev_data = self.session.items()160        self.session = self.backend(self.session.session_key)161        self.assertIs(hasattr(self.session, '_session_cache'), False)162        self.session.cycle_key()163        self.assertCountEqual(self.session.items(), prev_data)164    def test_save_doesnt_clear_data(self):165        self.session['a'] = 'b'166        self.session.save()167        self.assertEqual(self.session['a'], 'b')168    def test_invalid_key(self):169        # Submitting an invalid session key (either by guessing, or if the db has170        # removed the key) results in a new key being generated.171        try:172            session = self.backend('1')173            session.save()174            self.assertNotEqual(session.session_key, '1')175            self.assertIsNone(session.get('cat'))176            session.delete()177        finally:178            # Some backends leave a stale cache entry for the invalid179            # session key; make sure that entry is manually deleted180            session.delete('1')181    def test_session_key_empty_string_invalid(self):182        """Falsey values (Such as an empty string) are rejected."""183        self.session._session_key = ''184        self.assertIsNone(self.session.session_key)185    def test_session_key_too_short_invalid(self):186        """Strings shorter than 8 characters are rejected."""187        self.session._session_key = '1234567'188        self.assertIsNone(self.session.session_key)189    def test_session_key_valid_string_saved(self):190        """Strings of length 8 and up are accepted and stored."""191        self.session._session_key = '12345678'192        self.assertEqual(self.session.session_key, '12345678')193    def test_session_key_is_read_only(self):194        def set_session_key(session):195            session.session_key = session._get_new_session_key()196        with self.assertRaises(AttributeError):197            set_session_key(self.session)198    # Custom session expiry199    def test_default_expiry(self):200        # A normal session has a max age equal to settings201        self.assertEqual(self.session.get_expiry_age(), settings.SESSION_COOKIE_AGE)202        # So does a custom session with an idle expiration time of 0 (but it'll203        # expire at browser close)204        self.session.set_expiry(0)205        self.assertEqual(self.session.get_expiry_age(), settings.SESSION_COOKIE_AGE)206    def test_custom_expiry_seconds(self):207        modification = timezone.now()208        self.session.set_expiry(10)209        date = self.session.get_expiry_date(modification=modification)210        self.assertEqual(date, modification + timedelta(seconds=10))211        age = self.session.get_expiry_age(modification=modification)212        self.assertEqual(age, 10)213    def test_custom_expiry_timedelta(self):214        modification = timezone.now()215        # Mock timezone.now, because set_expiry calls it on this code path.216        original_now = timezone.now217        try:218            timezone.now = lambda: modification219            self.session.set_expiry(timedelta(seconds=10))220        finally:221            timezone.now = original_now222        date = self.session.get_expiry_date(modification=modification)223        self.assertEqual(date, modification + timedelta(seconds=10))224        age = self.session.get_expiry_age(modification=modification)225        self.assertEqual(age, 10)226    def test_custom_expiry_datetime(self):227        modification = timezone.now()228        self.session.set_expiry(modification + timedelta(seconds=10))229        date = self.session.get_expiry_date(modification=modification)230        self.assertEqual(date, modification + timedelta(seconds=10))231        age = self.session.get_expiry_age(modification=modification)232        self.assertEqual(age, 10)233    def test_custom_expiry_reset(self):234        self.session.set_expiry(None)235        self.session.set_expiry(10)236        self.session.set_expiry(None)237        self.assertEqual(self.session.get_expiry_age(), settings.SESSION_COOKIE_AGE)238    def test_get_expire_at_browser_close(self):239        # Tests get_expire_at_browser_close with different settings and different240        # set_expiry calls241        with override_settings(SESSION_EXPIRE_AT_BROWSER_CLOSE=False):242            self.session.set_expiry(10)243            self.assertIs(self.session.get_expire_at_browser_close(), False)244            self.session.set_expiry(0)245            self.assertIs(self.session.get_expire_at_browser_close(), True)246            self.session.set_expiry(None)247            self.assertIs(self.session.get_expire_at_browser_close(), False)248        with override_settings(SESSION_EXPIRE_AT_BROWSER_CLOSE=True):249            self.session.set_expiry(10)250            self.assertIs(self.session.get_expire_at_browser_close(), False)251            self.session.set_expiry(0)252            self.assertIs(self.session.get_expire_at_browser_close(), True)253            self.session.set_expiry(None)254            self.assertIs(self.session.get_expire_at_browser_close(), True)255    def test_decode(self):256        # Ensure we can decode what we encode257        data = {'a test key': 'a test value'}258        encoded = self.session.encode(data)259        self.assertEqual(self.session.decode(encoded), data)260    @override_settings(SECRET_KEY='django_tests_secret_key')261    def test_decode_legacy(self):262        # RemovedInDjango40Warning: pre-Django 3.1 sessions will be invalid.263        legacy_encoded = (264            'OWUzNTNmNWQxNTBjOWExZmM4MmQ3NzNhMDRmMjU4NmYwNDUyNGI2NDp7ImEgdGVzd'265            'CBrZXkiOiJhIHRlc3QgdmFsdWUifQ=='266        )267        self.assertEqual(268            self.session.decode(legacy_encoded),269            {'a test key': 'a test value'},270        )271    @ignore_warnings(category=RemovedInDjango40Warning)272    def test_default_hashing_algorith_legacy_decode(self):273        with self.settings(DEFAULT_HASHING_ALGORITHM='sha1'):274            data = {'a test key': 'a test value'}275            encoded = self.session.encode(data)276            self.assertEqual(self.session._legacy_decode(encoded), data)277    def test_decode_failure_logged_to_security(self):278        tests = [279            base64.b64encode(b'flaskdj:alkdjf').decode('ascii'),280            'bad:encoded:value',281        ]282        for encoded in tests:283            with self.subTest(encoded=encoded):284                with self.assertLogs('django.security.SuspiciousSession', 'WARNING') as cm:285                    self.assertEqual(self.session.decode(encoded), {})286                # The failed decode is logged.287                self.assertIn('Session data corrupted', cm.output[0])288    def test_actual_expiry(self):289        # this doesn't work with JSONSerializer (serializing timedelta)290        with override_settings(SESSION_SERIALIZER='django.contrib.sessions.serializers.PickleSerializer'):291            self.session = self.backend()  # reinitialize after overriding settings292            # Regression test for #19200293            old_session_key = None294            new_session_key = None295            try:296                self.session['foo'] = 'bar'297                self.session.set_expiry(-timedelta(seconds=10))298                self.session.save()299                old_session_key = self.session.session_key300                # With an expiry date in the past, the session expires instantly.301                new_session = self.backend(self.session.session_key)302                new_session_key = new_session.session_key303                self.assertNotIn('foo', new_session)304            finally:305                self.session.delete(old_session_key)306                self.session.delete(new_session_key)307    def test_session_load_does_not_create_record(self):308        """309        Loading an unknown session key does not create a session record.310        Creating session records on load is a DOS vulnerability.311        """312        session = self.backend('someunknownkey')313        session.load()314        self.assertIsNone(session.session_key)315        self.assertIs(session.exists(session.session_key), False)316        # provided unknown key was cycled, not reused317        self.assertNotEqual(session.session_key, 'someunknownkey')318    def test_session_save_does_not_resurrect_session_logged_out_in_other_context(self):319        """320        Sessions shouldn't be resurrected by a concurrent request.321        """322        # Create new session.323        s1 = self.backend()324        s1['test_data'] = 'value1'325        s1.save(must_create=True)326        # Logout in another context.327        s2 = self.backend(s1.session_key)328        s2.delete()329        # Modify session in first context.330        s1['test_data'] = 'value2'331        with self.assertRaises(UpdateError):332            # This should throw an exception as the session is deleted, not333            # resurrect the session.334            s1.save()335        self.assertEqual(s1.load(), {})336class DatabaseSessionTests(SessionTestsMixin, TestCase):337    backend = DatabaseSession338    session_engine = 'django.contrib.sessions.backends.db'339    @property340    def model(self):341        return self.backend.get_model_class()342    def test_session_str(self):343        "Session repr should be the session key."344        self.session['x'] = 1345        self.session.save()346        session_key = self.session.session_key347        s = self.model.objects.get(session_key=session_key)348        self.assertEqual(str(s), session_key)349    def test_session_get_decoded(self):350        """351        Test we can use Session.get_decoded to retrieve data stored352        in normal way353        """354        self.session['x'] = 1355        self.session.save()356        s = self.model.objects.get(session_key=self.session.session_key)357        self.assertEqual(s.get_decoded(), {'x': 1})358    def test_sessionmanager_save(self):359        """360        Test SessionManager.save method361        """362        # Create a session363        self.session['y'] = 1364        self.session.save()365        s = self.model.objects.get(session_key=self.session.session_key)366        # Change it367        self.model.objects.save(s.session_key, {'y': 2}, s.expire_date)368        # Clear cache, so that it will be retrieved from DB369        del self.session._session_cache370        self.assertEqual(self.session['y'], 2)371    def test_clearsessions_command(self):372        """373        Test clearsessions command for clearing expired sessions.374        """375        self.assertEqual(0, self.model.objects.count())376        # One object in the future377        self.session['foo'] = 'bar'378        self.session.set_expiry(3600)379        self.session.save()380        # One object in the past381        other_session = self.backend()382        other_session['foo'] = 'bar'383        other_session.set_expiry(-3600)384        other_session.save()385        # Two sessions are in the database before clearsessions...386        self.assertEqual(2, self.model.objects.count())387        with override_settings(SESSION_ENGINE=self.session_engine):388            management.call_command('clearsessions')389        # ... and one is deleted.390        self.assertEqual(1, self.model.objects.count())391@override_settings(USE_TZ=True)392class DatabaseSessionWithTimeZoneTests(DatabaseSessionTests):393    pass394class CustomDatabaseSessionTests(DatabaseSessionTests):395    backend = CustomDatabaseSession396    session_engine = 'sessions_tests.models'397    custom_session_cookie_age = 60 * 60 * 24  # One day.398    def test_extra_session_field(self):399        # Set the account ID to be picked up by a custom session storage400        # and saved to a custom session model database column.401        self.session['_auth_user_id'] = 42402        self.session.save()403        # Make sure that the customized create_model_instance() was called.404        s = self.model.objects.get(session_key=self.session.session_key)405        self.assertEqual(s.account_id, 42)406        # Make the session "anonymous".407        self.session.pop('_auth_user_id')408        self.session.save()409        # Make sure that save() on an existing session did the right job.410        s = self.model.objects.get(session_key=self.session.session_key)411        self.assertIsNone(s.account_id)412    def test_custom_expiry_reset(self):413        self.session.set_expiry(None)414        self.session.set_expiry(10)415        self.session.set_expiry(None)416        self.assertEqual(self.session.get_expiry_age(), self.custom_session_cookie_age)417    def test_default_expiry(self):418        self.assertEqual(self.session.get_expiry_age(), self.custom_session_cookie_age)419        self.session.set_expiry(0)420        self.assertEqual(self.session.get_expiry_age(), self.custom_session_cookie_age)421class CacheDBSessionTests(SessionTestsMixin, TestCase):422    backend = CacheDBSession423    def test_exists_searches_cache_first(self):424        self.session.save()425        with self.assertNumQueries(0):426            self.assertIs(self.session.exists(self.session.session_key), True)427    # Some backends might issue a warning428    @ignore_warnings(module="django.core.cache.backends.base")429    def test_load_overlong_key(self):430        self.session._session_key = (string.ascii_letters + string.digits) * 20431        self.assertEqual(self.session.load(), {})432    @override_settings(SESSION_CACHE_ALIAS='sessions')433    def test_non_default_cache(self):434        # 21000 - CacheDB backend should respect SESSION_CACHE_ALIAS.435        with self.assertRaises(InvalidCacheBackendError):436            self.backend()437@override_settings(USE_TZ=True)438class CacheDBSessionWithTimeZoneTests(CacheDBSessionTests):439    pass440class FileSessionTests(SessionTestsMixin, SimpleTestCase):441    backend = FileSession442    def setUp(self):443        # Do file session tests in an isolated directory, and kill it after we're done.444        self.original_session_file_path = settings.SESSION_FILE_PATH445        self.temp_session_store = settings.SESSION_FILE_PATH = self.mkdtemp()446        # Reset the file session backend's internal caches447        if hasattr(self.backend, '_storage_path'):448            del self.backend._storage_path449        super().setUp()450    def tearDown(self):451        super().tearDown()452        settings.SESSION_FILE_PATH = self.original_session_file_path453        shutil.rmtree(self.temp_session_store)454    def mkdtemp(self):455        return tempfile.mkdtemp()456    @override_settings(457        SESSION_FILE_PATH='/if/this/directory/exists/you/have/a/weird/computer',458    )459    def test_configuration_check(self):460        del self.backend._storage_path461        # Make sure the file backend checks for a good storage dir462        with self.assertRaises(ImproperlyConfigured):463            self.backend()464    def test_invalid_key_backslash(self):465        # Ensure we don't allow directory-traversal.466        # This is tested directly on _key_to_file, as load() will swallow467        # a SuspiciousOperation in the same way as an OSError - by creating468        # a new session, making it unclear whether the slashes were detected.469        with self.assertRaises(InvalidSessionKey):470            self.backend()._key_to_file("a\\b\\c")471    def test_invalid_key_forwardslash(self):472        # Ensure we don't allow directory-traversal473        with self.assertRaises(InvalidSessionKey):474            self.backend()._key_to_file("a/b/c")475    @override_settings(476        SESSION_ENGINE="django.contrib.sessions.backends.file",477        SESSION_COOKIE_AGE=0,478    )479    def test_clearsessions_command(self):480        """481        Test clearsessions command for clearing expired sessions.482        """483        storage_path = self.backend._get_storage_path()484        file_prefix = settings.SESSION_COOKIE_NAME485        def count_sessions():486            return len([487                session_file for session_file in os.listdir(storage_path)488                if session_file.startswith(file_prefix)489            ])490        self.assertEqual(0, count_sessions())491        # One object in the future492        self.session['foo'] = 'bar'493        self.session.set_expiry(3600)494        self.session.save()495        # One object in the past496        other_session = self.backend()497        other_session['foo'] = 'bar'498        other_session.set_expiry(-3600)499        other_session.save()500        # One object in the present without an expiry (should be deleted since501        # its modification time + SESSION_COOKIE_AGE will be in the past when502        # clearsessions runs).503        other_session2 = self.backend()504        other_session2['foo'] = 'bar'505        other_session2.save()506        # Three sessions are in the filesystem before clearsessions...507        self.assertEqual(3, count_sessions())508        management.call_command('clearsessions')509        # ... and two are deleted.510        self.assertEqual(1, count_sessions())511class FileSessionPathLibTests(FileSessionTests):512    def mkdtemp(self):513        tmp_dir = super().mkdtemp()514        return Path(tmp_dir)515class CacheSessionTests(SessionTestsMixin, SimpleTestCase):516    backend = CacheSession517    # Some backends might issue a warning518    @ignore_warnings(module="django.core.cache.backends.base")519    def test_load_overlong_key(self):520        self.session._session_key = (string.ascii_letters + string.digits) * 20521        self.assertEqual(self.session.load(), {})522    def test_default_cache(self):523        self.session.save()524        self.assertIsNotNone(caches['default'].get(self.session.cache_key))525    @override_settings(CACHES={526        'default': {527            'BACKEND': 'django.core.cache.backends.dummy.DummyCache',528        },529        'sessions': {530            'BACKEND': 'django.core.cache.backends.locmem.LocMemCache',531            'LOCATION': 'session',532        },533    }, SESSION_CACHE_ALIAS='sessions')534    def test_non_default_cache(self):535        # Re-initialize the session backend to make use of overridden settings.536        self.session = self.backend()537        self.session.save()538        self.assertIsNone(caches['default'].get(self.session.cache_key))539        self.assertIsNotNone(caches['sessions'].get(self.session.cache_key))540    def test_create_and_save(self):541        self.session = self.backend()542        self.session.create()543        self.session.save()544        self.assertIsNotNone(caches['default'].get(self.session.cache_key))545class SessionMiddlewareTests(TestCase):546    request_factory = RequestFactory()547    @staticmethod548    def get_response_touching_session(request):549        request.session['hello'] = 'world'550        return HttpResponse('Session test')551    @override_settings(SESSION_COOKIE_SECURE=True)552    def test_secure_session_cookie(self):553        request = self.request_factory.get('/')554        middleware = SessionMiddleware(self.get_response_touching_session)555        # Handle the response through the middleware556        response = middleware(request)557        self.assertIs(response.cookies[settings.SESSION_COOKIE_NAME]['secure'], True)558    @override_settings(SESSION_COOKIE_HTTPONLY=True)559    def test_httponly_session_cookie(self):560        request = self.request_factory.get('/')561        middleware = SessionMiddleware(self.get_response_touching_session)562        # Handle the response through the middleware563        response = middleware(request)564        self.assertIs(response.cookies[settings.SESSION_COOKIE_NAME]['httponly'], True)565        self.assertIn(566            cookies.Morsel._reserved['httponly'],567            str(response.cookies[settings.SESSION_COOKIE_NAME])568        )569    @override_settings(SESSION_COOKIE_SAMESITE='Strict')570    def test_samesite_session_cookie(self):571        request = self.request_factory.get('/')572        middleware = SessionMiddleware(self.get_response_touching_session)573        response = middleware(request)574        self.assertEqual(response.cookies[settings.SESSION_COOKIE_NAME]['samesite'], 'Strict')575    @override_settings(SESSION_COOKIE_HTTPONLY=False)576    def test_no_httponly_session_cookie(self):577        request = self.request_factory.get('/')578        middleware = SessionMiddleware(self.get_response_touching_session)579        response = middleware(request)580        self.assertEqual(response.cookies[settings.SESSION_COOKIE_NAME]['httponly'], '')581        self.assertNotIn(582            cookies.Morsel._reserved['httponly'],583            str(response.cookies[settings.SESSION_COOKIE_NAME])584        )585    def test_session_save_on_500(self):586        def response_500(requset):587            response = HttpResponse('Horrible error')588            response.status_code = 500589            request.session['hello'] = 'world'590            return response591        request = self.request_factory.get('/')592        SessionMiddleware(response_500)(request)593        # The value wasn't saved above.594        self.assertNotIn('hello', request.session.load())595    def test_session_update_error_redirect(self):596        def response_delete_session(request):597            request.session = DatabaseSession()598            request.session.save(must_create=True)599            request.session.delete()600            return HttpResponse()601        request = self.request_factory.get('/foo/')602        middleware = SessionMiddleware(response_delete_session)603        msg = (604            "The request's session was deleted before the request completed. "605            "The user may have logged out in a concurrent request, for example."606        )607        with self.assertRaisesMessage(SuspiciousOperation, msg):608            # Handle the response through the middleware. It will try to save609            # the deleted session which will cause an UpdateError that's caught610            # and raised as a SuspiciousOperation.611            middleware(request)612    def test_session_delete_on_end(self):613        def response_ending_session(request):614            request.session.flush()615            return HttpResponse('Session test')616        request = self.request_factory.get('/')617        middleware = SessionMiddleware(response_ending_session)618        # Before deleting, there has to be an existing cookie619        request.COOKIES[settings.SESSION_COOKIE_NAME] = 'abc'620        # Handle the response through the middleware621        response = middleware(request)622        # The cookie was deleted, not recreated.623        # A deleted cookie header looks like:624        #  Set-Cookie: sessionid=; expires=Thu, 01 Jan 1970 00:00:00 GMT; Max-Age=0; Path=/625        self.assertEqual(626            'Set-Cookie: {}=""; expires=Thu, 01 Jan 1970 00:00:00 GMT; '627            'Max-Age=0; Path=/; SameSite={}'.format(628                settings.SESSION_COOKIE_NAME,629                settings.SESSION_COOKIE_SAMESITE,630            ),631            str(response.cookies[settings.SESSION_COOKIE_NAME])632        )633        # SessionMiddleware sets 'Vary: Cookie' to prevent the 'Set-Cookie'634        # from being cached.635        self.assertEqual(response['Vary'], 'Cookie')636    @override_settings(SESSION_COOKIE_DOMAIN='.example.local', SESSION_COOKIE_PATH='/example/')637    def test_session_delete_on_end_with_custom_domain_and_path(self):638        def response_ending_session(request):639            request.session.flush()640            return HttpResponse('Session test')641        request = self.request_factory.get('/')642        middleware = SessionMiddleware(response_ending_session)643        # Before deleting, there has to be an existing cookie644        request.COOKIES[settings.SESSION_COOKIE_NAME] = 'abc'645        # Handle the response through the middleware646        response = middleware(request)647        # The cookie was deleted, not recreated.648        # A deleted cookie header with a custom domain and path looks like:649        #  Set-Cookie: sessionid=; Domain=.example.local;650        #              expires=Thu, 01 Jan 1970 00:00:00 GMT; Max-Age=0;651        #              Path=/example/652        self.assertEqual(653            'Set-Cookie: {}=""; Domain=.example.local; expires=Thu, '654            '01 Jan 1970 00:00:00 GMT; Max-Age=0; Path=/example/; SameSite={}'.format(655                settings.SESSION_COOKIE_NAME,656                settings.SESSION_COOKIE_SAMESITE,657            ),658            str(response.cookies[settings.SESSION_COOKIE_NAME])659        )660    def test_flush_empty_without_session_cookie_doesnt_set_cookie(self):661        def response_ending_session(request):662            request.session.flush()663            return HttpResponse('Session test')664        request = self.request_factory.get('/')665        middleware = SessionMiddleware(response_ending_session)666        # Handle the response through the middleware667        response = middleware(request)668        # A cookie should not be set.669        self.assertEqual(response.cookies, {})670        # The session is accessed so "Vary: Cookie" should be set.671        self.assertEqual(response['Vary'], 'Cookie')672    def test_empty_session_saved(self):673        """674        If a session is emptied of data but still has a key, it should still675        be updated.676        """677        def response_set_session(request):678            # Set a session key and some data.679            request.session['foo'] = 'bar'680            return HttpResponse('Session test')681        request = self.request_factory.get('/')682        middleware = SessionMiddleware(response_set_session)683        # Handle the response through the middleware.684        response = middleware(request)685        self.assertEqual(tuple(request.session.items()), (('foo', 'bar'),))686        # A cookie should be set, along with Vary: Cookie.687        self.assertIn(688            'Set-Cookie: sessionid=%s' % request.session.session_key,689            str(response.cookies)690        )691        self.assertEqual(response['Vary'], 'Cookie')...

...182        else:183            self.__session_key = None184    session_key = property(_get_session_key)185    _session_key = property(_get_session_key, _set_session_key)186    def _get_session(self, no_load=False):187        """188        Lazily load session from storage (unless "no_load" is True, when only189        an empty dict is stored) and store it in the current instance.190        """191        self.accessed = True192        try:193            return self._session_cache194        except AttributeError:195            if self.session_key is None or no_load:196                self._session_cache = {}197            else:198                self._session_cache = self.load()199        return self._session_cache200    _session = property(_get_session)...

...5alert_doc = inline("<script>window.alert()</script>")6# 10.7.2 Set Window Rect7def test_set_window_rect_prompt_accept(new_session):8    # Step 29    _, session = new_session({"alwaysMatch": {"unhandledPromptBehavior": "accept"}})10    session.url = inline("<title>WD doc title</title>")11    get_response = session.transport.send("GET", "session/%s/window/rect" % session.session_id)12    original = get_response.body["value"]13    create_dialog(session)("alert", text="dismiss #1", result_var="dismiss1")14    result = session.transport.send("POST",15                                    "session/%s/window/rect" % session.session_id,16                                    {"x": int(original["y"]),17                                     "y": int(original["y"])})18    assert result.status == 20019    assert_dialog_handled(session, "dismiss #1")20    create_dialog(session)("confirm", text="dismiss #2", result_var="dismiss2")21    result = session.transport.send("POST",22                                    "session/%s/window/rect" % session.session_id,23                                    {"x": int(original["y"]),...

...27#    [...]28#29# 3. Return success.30def test_title_handle_prompt_dismiss(new_session):31    _, session = new_session({"alwaysMatch": {"unhandledPromptBehavior": "dismiss"}})32    session.url = inline("<title>WD doc title</title>")33    expected_title = read_global(session, "document.title")34    create_dialog(session)("alert", text="dismiss #1", result_var="dismiss1")35    result = session.transport.send("GET",36                                    "session/%s/title" % session.session_id)37    assert_success(result, expected_title)38    assert_dialog_handled(session, "dismiss #1")39    assert read_global(session, "dismiss1") == None40    expected_title = read_global(session, "document.title")41    create_dialog(session)("confirm", text="dismiss #2", result_var="dismiss2")42    result = session.transport.send("GET",43                                    "session/%s/title" % session.session_id)44    assert_success(result, expected_title)45    assert_dialog_handled(session, "dismiss #2")46    assert read_global(session, "dismiss2") == None47    expected_title = read_global(session, "document.title")48    create_dialog(session)("prompt", text="dismiss #3", result_var="dismiss3")49    result = session.transport.send("GET",50                                    "session/%s/title" % session.session_id)51    assert_success(result, expected_title)52    assert_dialog_handled(session, "dismiss #3")53    assert read_global(session, "dismiss3") == None54# [...]55# 2. Handle any user prompts and return its value if it is an error.56# [...]57# In order to handle any user prompts a remote end must take the following58# steps:59# 2. Run the substeps of the first matching user prompt handler:60#61#    [...]62#    - accept state63#      1. Accept the current user prompt.64#    [...]65#66# 3. Return success.67def test_title_handle_prompt_accept(new_session):68    _, session = new_session({"alwaysMatch": {"unhandledPromptBehavior": "accept"}})69    session.url = inline("<title>WD doc title</title>")70    create_dialog(session)("alert", text="accept #1", result_var="accept1")71    expected_title = read_global(session, "document.title")72    result = session.transport.send("GET",73                                    "session/%s/title" % session.session_id)74    assert_success(result, expected_title)75    assert_dialog_handled(session, "accept #1")76    assert read_global(session, "accept1") == None77    expected_title = read_global(session, "document.title")78    create_dialog(session)("confirm", text="accept #2", result_var="accept2")79    result = session.transport.send("GET",80                                    "session/%s/title" % session.session_id)81    assert_success(result, expected_title)82    assert_dialog_handled(session, "accept #2")...

...35#    [...]36#37# 3. Return success.38def test_handle_prompt_dismiss(new_session):39    _, session = new_session({"alwaysMatch": {"unhandledPromptBehavior": "dismiss"}})40    session.url = inline("<body><p>Hello, World!</p></body>")41    create_dialog(session)("alert", text="dismiss #1", result_var="dismiss1")42    result = session.transport.send("GET",43                                    "session/%s/element/active" % session.session_id)44    assert_result_is_active_element(session, result)45    assert_dialog_handled(session, "dismiss #1")46    assert session.execute_script("return dismiss1;") == None47    create_dialog(session)("confirm", text="dismiss #2", result_var="dismiss2")48    result = session.transport.send("GET",49                                    "session/%s/element/active" % session.session_id)50    assert_result_is_active_element(session, result)51    assert_dialog_handled(session, "dismiss #2")52    assert read_global(session, "dismiss2") == None53    create_dialog(session)("prompt", text="dismiss #3", result_var="dismiss3")54    result = session.transport.send("GET",55                                    "session/%s/element/active" % session.session_id)56    assert_result_is_active_element(session, result)57    assert_dialog_handled(session, "dismiss #3")58    assert read_global(session, "dismiss3") == None59# [...]60# 2. Handle any user prompts and return its value if it is an error.61# [...]62# In order to handle any user prompts a remote end must take the following63# steps:64# 2. Run the substeps of the first matching user prompt handler:65#66#    [...]67#    - accept state68#      1. Accept the current user prompt.69#    [...]70#71# 3. Return success.72def test_handle_prompt_accept(new_session):73    _, session = new_session({"alwaysMatch": {"unhandledPromptBehavior": "accept"}})74    session.url = inline("<body><p>Hello, World!</p></body>")75    create_dialog(session)("alert", text="accept #1", result_var="accept1")76    result = session.transport.send("GET",77                                    "session/%s/element/active" % session.session_id)78    assert_result_is_active_element(session, result)79    assert_dialog_handled(session, "accept #1")80    assert read_global(session, "accept1") == None81    create_dialog(session)("confirm", text="accept #2", result_var="accept2")82    result = session.transport.send("GET",83                                    "session/%s/element/active" % session.session_id)84    assert_result_is_active_element(session, result)85    assert_dialog_handled(session, "accept #2")86    assert read_global(session, "accept2"), True87    create_dialog(session)("prompt", text="accept #3", result_var="accept3")...

1from core.context import Context2from core.response import Response3from core.response.error import BadRequest, TemporaryRedirect4from core.uri import Uri5from .base import BaseController6from threading import Lock7from random import randint8import time9class SessionController( BaseController ):10	sessions = {}11	sessions_lock = Lock()12	max_session_age = 15*60*1000	# 15 minutes13	CHARS = '0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ'14	def random_code( self ):15		code = ''16		for i in range( 64 ):17			index = randint( 0, len( SessionController.CHARS )-1 )18			code += SessionController.CHARS[ index ]19		return code20	def new_code( self ):21		while True:22			code = self.random_code()23			if not code in SessionController.sessions:24				break25		return code26	def handle( self, context ):27		session_id = None28		cookie = context.request.cookie()29		if cookie != None:30			print( cookie.jar, SessionController.sessions )31			if 'session' in cookie:32				session_id = cookie['session']33		SessionController.sessions_lock.acquire()34		# If not session id was provided, redirect to the session page35		if session_id == None:36			return self.create_cookie( context )37		# If no session context exists yet, create one38		if not session_id in SessionController.sessions:39			return self.create_cookie( context )40		else:41			if SessionController.sessions[ session_id ][1] + SessionController.max_session_age < time.time():42				return self.create_cookie( context )43			context.session = SessionController.sessions[ session_id ][0]44			SessionController.sessions[ session_id ] = ( context.session, time.time() )45		# Remove all old sessions46		# TODO: Make a seperate thread that does the garbage collection.47		# Right now sessions don't really get deleted when there is just one48		self.garbage_collect()49		SessionController.sessions_lock.release()50		return super( SessionController, self ).handle( context )51	def garbage_collect( self ):52		for session_id, session in SessionController.sessions.items():53			context, last_use = session54			if last_use + SessionController.max_session_age < time.time():55				del SessionController.sessions[ session_id ]56	def create_cookie( self, context ):57		code = self.new_code()58		session = Context()59		SessionController.sessions[ code ] = ( session, time.time() )60		SessionController.sessions_lock.release()61		context.session = session62		resp = CreateSessionResponse( super( SessionController, self ).handle( context ), code )63		return resp64	65class CreateSessionResponse( Response ):66	def __init__( self, response, session_id ):67		self.response = response68		self.session_id = session_id69	def body( self ):70		return self.response.body()71	def headers( self ):72		hs = self.response.headers()73		hs['Set-Cookie'] = 'session=' + self.session_id + '; HttpOnly'74		return hs75	def status_code( self ):76		return self.response.status_code()77	def status_message( self ):78		return self.response.status_message()79class CreateSessionController( BaseController ):80	# TODO: Create a login system where the user logs in by signing a (by the server generated) random nonce81	def handle_get( self, context ):82		SessionController.sessions_lock.acquire()83		code = self.new_code()84		SessionController.sessions[ code ] = ( Context(), time.time() )85		SessionController.sessions_lock.release()86		print( 'adsfdddd', SessionController.sessions )...

...10    result = session.transport.send("GET", "session/%s/window/rect" % session.session_id)11    assert_error(result, "no such window")12def test_get_window_rect_prompt_accept(new_session):13    # Step 214    _, session = new_session({"alwaysMatch": {"unhandledPromptBehavior": "accept"}})15    session.url = inline("<title>WD doc title</title>")16    create_dialog(session)("alert", text="dismiss #1", result_var="dismiss1")17    result = session.transport.send("GET",18                                    "session/%s/window/rect" % session.session_id)19    assert result.status == 20020    assert_dialog_handled(session, "dismiss #1")21    create_dialog(session)("confirm", text="dismiss #2", result_var="dismiss2")22    result = session.transport.send("GET",23                                    "session/%s/window/rect" % session.session_id)24    assert result.status == 20025    assert_dialog_handled(session, "dismiss #2")26    create_dialog(session)("prompt", text="dismiss #3", result_var="dismiss3")27    result = session.transport.send("GET",28                                    "session/%s/window/rect" % session.session_id)...

1import json2import pytest3import types4from tests.support.inline import inline5from tests.support.asserts import assert_error, assert_success6alert_doc = inline("<script>window.alert()</script>")7frame_doc = inline("<p>frame")8one_frame_doc = inline("<iframe src='%s'></iframe>" % frame_doc)9two_frames_doc = inline("<iframe src='%s'></iframe>" % one_frame_doc)10# TODO(ato): 7.1 Get11def test_get_current_url_no_browsing_context(session, create_window):12    # 7.2 step 113    session.window_handle = create_window()14    session.close()15    result = session.transport.send("GET", "session/%s/url" % session.session_id)16    assert_error(result, "no such window")17def test_get_current_url_alert_prompt(session):18    # 7.2 step 219    session.url = alert_doc20    result = session.transport.send("GET", "session/%s/url" % session.session_id)21    assert_error(result, "unexpected alert open")22def test_get_current_url_matches_location(session):23    # 7.2 step 324    url = session.execute_script("return window.location.href")25    assert session.url == url26def test_get_current_url_payload(session):27    # 7.2 step 4-528    session.start()29    result = session.transport.send("GET", "session/%s/url" % session.session_id)30    assert result.status == 20031    assert isinstance(result.body["value"], basestring)32def test_get_current_url_special_pages(session):33    session.url = "about:blank"34    result = session.transport.send("GET", "session/%s/url" % session.session_id)35    assert_success(result, "about:blank")36# TODO(ato): This test requires modification to pass on Windows37def test_get_current_url_file_protocol(session):38    # tests that the browsing context remains the same39    # when navigated privileged documents40    session.url = "file:///"41    result = session.transport.send("GET", "session/%s/url" % session.session_id)42    assert_success(result, "file:///")43# TODO(ato): Test for http:// and https:// protocols.44# We need to expose a fixture for accessing45# documents served by wptserve in order to test this.46def test_set_malformed_url(session):47    result = session.transport.send("POST",48                                    "session/%s/url" % session.session_id,49                                    {"url": "foo"})50    assert_error(result, "invalid argument")51def test_get_current_url_after_modified_location(session):52    session.execute_script("window.location.href = 'about:blank#wd_test_modification'")53    result = session.transport.send("GET", "session/%s/url" % session.session_id)54    assert_success(result, "about:blank#wd_test_modification")55def test_get_current_url_nested_browsing_context(session, create_frame):56    session.url = "about:blank#wd_from_within_frame"57    session.switch_frame(create_frame())58    result = session.transport.send("GET", "session/%s/url" % session.session_id)59    assert_success(result, "about:blank#wd_from_within_frame")60def test_get_current_url_nested_browsing_contexts(session):61    session.url = two_frames_doc62    top_level_url = session.url63    outer_frame = session.find.css("iframe", all=False)64    session.switch_frame(outer_frame)65    inner_frame = session.find.css("iframe", all=False)66    session.switch_frame(inner_frame)...