<NME> test_session.py
<BEF> import os
import platform
import threading
import time
from http.client import HTTPConnection

from distutils.spawn import find_executable
import pytest
from path import Path
from more_itertools import consume
import portend

import cherrypy
from cherrypy._cpcompat import HTTPSConnection
from cherrypy.lib import sessions
from cherrypy.lib import reprconf
from cherrypy.lib.httputil import response_codes
from cherrypy.test import helper
from cherrypy import _json as json

localDir = Path(__file__).dirname()


def http_methods_allowed(methods=['GET', 'HEAD']):
    method = cherrypy.request.method.upper()
    if method not in methods:
        cherrypy.response.headers['Allow'] = ', '.join(methods)
        raise cherrypy.HTTPError(405)


cherrypy.tools.allow = cherrypy.Tool('on_start_resource', http_methods_allowed)


def setup_server():

    @cherrypy.config(**{
        'tools.sessions.on': True,
        'tools.sessions.storage_class': sessions.RamSession,
        'tools.sessions.storage_path': localDir,
        'tools.sessions.timeout': (1.0 / 60),
        'tools.sessions.clean_freq': (1.0 / 60),
    })
    class Root:

        @cherrypy.expose
        def clear(self):
            cherrypy.session.cache.clear()

        @cherrypy.expose
        @cherrypy.tools.json_out()
        def data(self):
            cherrypy.session['aha'] = 'foo'
            return cherrypy.session._data

        @cherrypy.expose
        def testGen(self):
            counter = cherrypy.session.get('counter', 0) + 1
            cherrypy.session['counter'] = counter
            yield str(counter)

        @cherrypy.expose
        def testStr(self):
            counter = cherrypy.session.get('counter', 0) + 1
            cherrypy.session['counter'] = counter
            return str(counter)

        @cherrypy.expose
        @cherrypy.config(**{'tools.sessions.on': False})
        def set_session_cls(self, new_cls_name):
            new_cls = reprconf.attributes(new_cls_name)
            cfg = {'tools.sessions.storage_class': new_cls}
            self.__class__._cp_config.update(cfg)
            if hasattr(cherrypy, 'session'):
                del cherrypy.session
            if new_cls.clean_thread:
                new_cls.clean_thread.stop()
                new_cls.clean_thread.unsubscribe()
                del new_cls.clean_thread

        @cherrypy.expose
        def index(self):
            sess = cherrypy.session
            c = sess.get('counter', 0) + 1
            time.sleep(0.01)
            sess['counter'] = c
            return str(c)

        @cherrypy.expose
        def keyin(self, key):
            return str(key in cherrypy.session)

        @cherrypy.expose
        def delete(self):
            cherrypy.session.delete()
            sessions.expire()
            return 'done'

        @cherrypy.expose
        def delkey(self, key):
            del cherrypy.session[key]
            return 'OK'

        @cherrypy.expose
        def redir_target(self):
            return self._cp_config['tools.sessions.storage_class'].__name__

        @cherrypy.expose
        def iredir(self):
            raise cherrypy.InternalRedirect('/redir_target')

        @cherrypy.expose
        @cherrypy.config(**{
            'tools.allow.on': True,
            'tools.allow.methods': ['GET'],
        })
        def restricted(self):
            return cherrypy.request.method

        @cherrypy.expose
        def regen(self):
            cherrypy.tools.sessions.regenerate()
            return 'logged in'

        @cherrypy.expose
        def length(self):
            return str(len(cherrypy.session))

        @cherrypy.expose
        @cherrypy.config(**{
            'tools.sessions.path': '/session_cookie',
            'tools.sessions.name': 'temp',
            'tools.sessions.persistent': False,
        })
        def session_cookie(self):
            # Must load() to start the clean thread.
            cherrypy.session.load()
            return cherrypy.session.id

    cherrypy.tree.mount(Root())


class SessionTest(helper.CPWebCase):
    setup_server = staticmethod(setup_server)

    @classmethod
    def teardown_class(cls):
        """Clean up sessions."""
        super(cls, cls).teardown_class()
        consume(
            file.remove_p()
            for file in localDir.listdir()
            if file.basename().startswith(
                sessions.FileSession.SESSION_PREFIX
            )
        )

    def test_0_Session(self):
        self.getPage('/set_session_cls/cherrypy.lib.sessions.RamSession')
        self.getPage('/clear')

        # Test that a normal request gets the same id in the cookies.
        # Note: this wouldn't work if /data didn't load the session.
        self.getPage('/data')
        assert self.body == b'{"aha": "foo"}'
        c = self.cookies[0]
        self.getPage('/data', self.cookies)
        self.cookies[0] == c

        self.getPage('/testStr')
        assert self.body == b'1'
        cookie_parts = dict([p.strip().split('=')
                             for p in self.cookies[0][1].split(';')])
        # Assert there is an 'expires' param
        expected_cookie_keys = {'session_id', 'expires', 'Path', 'Max-Age'}
        assert set(cookie_parts.keys()) == expected_cookie_keys
        self.getPage('/testGen', self.cookies)
        assert self.body == b'2'
        self.getPage('/testStr', self.cookies)
        assert self.body == b'3'
        self.getPage('/data', self.cookies)
        expected_data = {'counter': 3, 'aha': 'foo'}
        assert json.decode(self.body.decode('utf-8')) == expected_data
        self.getPage('/length', self.cookies)
        assert self.body == b'2'
        self.getPage('/delkey?key=counter', self.cookies)
        assert self.status_code == 200

        self.getPage('/set_session_cls/cherrypy.lib.sessions.FileSession')
        self.getPage('/testStr')
        assert self.body == b'1'
        self.getPage('/testGen', self.cookies)
        assert self.body == b'2'
        self.getPage('/testStr', self.cookies)
        assert self.body == b'3'
        self.getPage('/delkey?key=counter', self.cookies)
        assert self.status_code == 200

        # Wait for the session.timeout (1 second)
        time.sleep(2)
        self.getPage('/')
        assert self.body == b'1'
        self.getPage('/length', self.cookies)
        assert self.body == b'1'

        # Test session __contains__
        self.getPage('/keyin?key=counter', self.cookies)
        assert self.body == b'True'
        cookieset1 = self.cookies

        # Make a new session and test __len__ again
        self.getPage('/')
        self.getPage('/length', self.cookies)
        assert self.body == b'2'

        # Test session delete
        self.getPage('/delete', self.cookies)
        assert self.body == b'done'
        self.getPage('/delete', cookieset1)
        assert self.body == b'done'

        def f():
            return [
                x
                for x in os.listdir(localDir)
                if x.startswith('session-') and not x.endswith('.lock')
            ]
        assert f() == []

        # Wait for the cleanup thread to delete remaining session files
        self.getPage('/')
        assert f() != []
        time.sleep(2)
        assert f() == []

    def test_1_Ram_Concurrency(self):
        self.getPage('/set_session_cls/cherrypy.lib.sessions.RamSession')
        self._test_Concurrency()

    def test_2_File_Concurrency(self):
        self.getPage('/set_session_cls/cherrypy.lib.sessions.FileSession')
        self._test_Concurrency()

    def _test_Concurrency(self):
        client_thread_count = 5
        request_count = 30

        # Get initial cookie
        self.getPage('/')
        assert self.body == b'1'
        cookies = self.cookies

        data_dict = {}
        errors = []

        def request(index):
            if self.scheme == 'https':
                c = HTTPSConnection('%s:%s' % (self.interface(), self.PORT))
            else:
                c = HTTPConnection('%s:%s' % (self.interface(), self.PORT))
            for i in range(request_count):
                c.putrequest('GET', '/')
                for k, v in cookies:
                    c.putheader(k, v)
                c.endheaders()
                response = c.getresponse()
                body = response.read()
                if response.status != 200 or not body.isdigit():
                    errors.append((response.status, body))
                else:
                    data_dict[index] = max(data_dict[index], int(body))
                # Uncomment the following line to prove threads overlap.
                # sys.stdout.write("%d " % index)

        # Start <request_count> requests from each of
        # <client_thread_count> concurrent clients
        ts = []
        for c in range(client_thread_count):
            data_dict[c] = 0
            t = threading.Thread(target=request, args=(c,))
            ts.append(t)
            t.start()

        for t in ts:
            t.join()

        hitcount = max(data_dict.values())
        expected = 1 + (client_thread_count * request_count)

        for e in errors:
            print(e)
        assert len(errors) == 0
        assert hitcount == expected

    def test_3_Redirect(self):
        # Start a new session
        self.getPage('/testStr')
        self.getPage('/iredir', self.cookies)
        assert self.body == b'FileSession'

    def test_4_File_deletion(self):
        # Start a new session
        self.getPage('/testStr')
        # Delete the session file manually and retry.
        id = self.cookies[0][1].split(';', 1)[0].split('=', 1)[1]
        path = os.path.join(localDir, 'session-' + id)
        os.unlink(path)
        self.getPage('/testStr', self.cookies)

    def test_5_Error_paths(self):
        self.getPage('/unknown/page')
        self.assertErrorPage(404, "The path '/unknown/page' was not found.")

        # Note: this path is *not* the same as above. The above
        # takes a normal route through the session code; this one
        # skips the session code's before_handler and only calls
        # before_finalize (save) and on_end (close). So the session
        # code has to survive calling save/close without init.
        self.getPage('/restricted', self.cookies, method='POST')
        self.assertErrorPage(405, response_codes[405][1])

    def test_6_regenerate(self):
        self.getPage('/testStr')
        # grab the cookie ID
        id1 = self.cookies[0][1].split(';', 1)[0].split('=', 1)[1]
        self.getPage('/regen')
        assert self.body == b'logged in'
        id2 = self.cookies[0][1].split(';', 1)[0].split('=', 1)[1]
        assert id1 != id2

        self.getPage('/testStr')
        # grab the cookie ID
        id1 = self.cookies[0][1].split(';', 1)[0].split('=', 1)[1]
        self.getPage('/testStr',
                     headers=[
                         ('Cookie',
                          'session_id=maliciousid; '
                          'expires=Sat, 27 Oct 2017 04:18:28 GMT; Path=/;')])
        id2 = self.cookies[0][1].split(';', 1)[0].split('=', 1)[1]
        assert id1 != id2
        assert id2 != 'maliciousid'

    def test_7_session_cookies(self):
        self.getPage('/set_session_cls/cherrypy.lib.sessions.RamSession')
        self.getPage('/clear')
        self.getPage('/session_cookie')
        # grab the cookie ID
        cookie_parts = dict([p.strip().split('=')
                            for p in self.cookies[0][1].split(';')])
        # Assert there is no 'expires' param
        assert set(cookie_parts.keys()) == {'temp', 'Path'}
        id1 = cookie_parts['temp']
        assert list(sessions.RamSession.cache) == [id1]

        # Send another request in the same "browser session".
        self.getPage('/session_cookie', self.cookies)
        cookie_parts = dict([p.strip().split('=')
                            for p in self.cookies[0][1].split(';')])
        # Assert there is no 'expires' param
        assert set(cookie_parts.keys()) == {'temp', 'Path'}
        assert self.body.decode('utf-8') == id1
        assert list(sessions.RamSession.cache) == [id1]

        # Simulate a browser close by just not sending the cookies
        self.getPage('/session_cookie')
        # grab the cookie ID
        cookie_parts = dict([p.strip().split('=')
                            for p in self.cookies[0][1].split(';')])
        # Assert there is no 'expires' param
        assert set(cookie_parts.keys()) == {'temp', 'Path'}
        # Assert a new id has been generated...
        id2 = cookie_parts['temp']
        assert id1 != id2
        assert set(sessions.RamSession.cache.keys()) == {id1, id2}

        # Wait for the session.timeout on both sessions
        time.sleep(2.5)
        cache = list(sessions.RamSession.cache)
        if cache:
            if cache == [id2]:
                self.fail('The second session did not time out.')
            else:
                self.fail('Unknown session id in cache: %r', cache)

    def test_8_Ram_Cleanup(self):
        def lock():
            s1 = sessions.RamSession()
            s1.acquire_lock()
            time.sleep(1)
            s1.release_lock()

        t = threading.Thread(target=lock)
        t.start()
        start = time.time()
        while not sessions.RamSession.locks and time.time() - start < 5:
            time.sleep(0.01)
        assert len(sessions.RamSession.locks) == 1, 'Lock not acquired'
        s2 = sessions.RamSession()
        s2.clean_up()
        msg = 'Clean up should not remove active lock'
        assert len(sessions.RamSession.locks) == 1, msg
        t.join()


def is_memcached_present():
    executable = find_executable('memcached')
    return bool(executable)


@pytest.fixture(scope='session')
def memcached_server_present():
    is_memcached_present() or pytest.skip('memcached not available')


@pytest.fixture()
def memcached_client_present():
    pytest.importorskip('memcache')


@pytest.fixture(scope='session')
def memcached_instance(request, watcher_getter, memcached_server_present):
    """
    Start up an instance of memcached.
    monkeypatch.setattr(
        sessions.MemcachedSession,
        'servers',
        [str(memcached_connection)],
    )


        except Exception:
            return True
        return False

    proc = watcher_getter(
        name='memcached',
        arguments=['-p', str(port)],
        checker=is_occupied,
        request=request,
    )
    return locals()


@pytest.fixture
def memcached_configured(
    memcached_instance, monkeypatch,
    memcached_client_present,
):
    server = 'localhost:{port}'.format_map(memcached_instance)
    monkeypatch.setattr(
        sessions.MemcachedSession,
        'servers',
        [server],
    )


@pytest.mark.skipif(
    platform.system() == 'Windows',
    reason='pytest-services helper does not work under Windows',
)
@pytest.mark.usefixtures('memcached_configured')
class MemcachedSessionTest(helper.CPWebCase):
    setup_server = staticmethod(setup_server)

    def test_0_Session(self):
        self.getPage(
            '/set_session_cls/cherrypy.lib.sessions.MemcachedSession'
        )

        self.getPage('/testStr')
        assert self.body == b'1'
        self.getPage('/testGen', self.cookies)
        assert self.body == b'2'
        self.getPage('/testStr', self.cookies)
        assert self.body == b'3'
        self.getPage('/length', self.cookies)
        self.assertErrorPage(500)
        assert b'NotImplementedError' in self.body
        self.getPage('/delkey?key=counter', self.cookies)
        assert self.status_code == 200

        # Wait for the session.timeout (1 second)
        time.sleep(1.25)
        self.getPage('/')
        assert self.body == b'1'

        # Test session __contains__
        self.getPage('/keyin?key=counter', self.cookies)
        assert self.body == b'True'

        # Test session delete
        self.getPage('/delete', self.cookies)
        assert self.body == b'done'

    def test_1_Concurrency(self):
        client_thread_count = 5
        request_count = 30

        # Get initial cookie
        self.getPage('/')
        assert self.body == b'1'
        cookies = self.cookies

        data_dict = {}

        def request(index):
            for i in range(request_count):
                self.getPage('/', cookies)
                # Uncomment the following line to prove threads overlap.
                # sys.stdout.write("%d " % index)
            if not self.body.isdigit():
                self.fail(self.body)
            data_dict[index] = int(self.body)

        # Start <request_count> concurrent requests from
        # each of <client_thread_count> clients
        ts = []
        for c in range(client_thread_count):
            data_dict[c] = 0
            t = threading.Thread(target=request, args=(c,))
            ts.append(t)
            t.start()

        for t in ts:
            t.join()

        hitcount = max(data_dict.values())
        expected = 1 + (client_thread_count * request_count)
        assert hitcount == expected

    def test_3_Redirect(self):
        # Start a new session
        self.getPage('/testStr')
        self.getPage('/iredir', self.cookies)
        assert self.body == b'MemcachedSession'

    def test_5_Error_paths(self):
        self.getPage('/unknown/page')
        self.assertErrorPage(
            404, "The path '/unknown/page' was not found.")

        # Note: this path is *not* the same as above. The above
        # takes a normal route through the session code; this one
        # skips the session code's before_handler and only calls
        # before_finalize (save) and on_end (close). So the session
        # code has to survive calling save/close without init.
        self.getPage('/restricted', self.cookies, method='POST')
        self.assertErrorPage(405, response_codes[405][1])

<MSG> Revert "Forcefully stringify memcached socket path"

This reverts commit 43c83f36f36ce0f04d590b83017526429cf217b2.

<DFF> @@ -423,7 +423,7 @@ def memcached_configured(memcached_connection, monkeypatch):
     monkeypatch.setattr(
         sessions.MemcachedSession,
         'servers',
-        [str(memcached_connection)],
+        [memcached_connection],
     )
 
 

{'commit': "diff --git a/cherrypy/test/test_session.py b/cherrypy/test/test_session.py\nindex 135b1df2..f3d47ba0 100755\n--- a/cherrypy/test/test_session.py\n+++ b/cherrypy/test/test_session.py\n@@ -423,7 +423,7 @@ def memcached_configured(memcached_connection, monkeypatch):\n     monkeypatch.setattr(\n         sessions.MemcachedSession,\n         'servers',\n-        [str(memcached_connection)],\n+        [memcached_connection],\n     )\n \n \n", 'commit_subject': 'Revert "Forcefully stringify memcached socket path"', 'repo_name': 'cherrypy/cherrypy', 'committer': {'name': 'Sviatoslav Sydorenko', 'email': '13bd1fc5137f503b0a8355ad400e7a87ebb45b25@sydorenko.org.ua', 'time_sec': '1536536629', 'tz_offset': '120', 'date': {'seconds': '1536536629'}}, 'commit_tree': 'b0dd6acfd60255d85f13b9b8f1db0072e71abf6c', 'commit_message': 'Revert "Forcefully stringify memcached socket path"\n\nThis reverts commit 43c83f36f36ce0f04d590b83017526429cf217b2.\n', 'language_name': 'Python', 'license': 'bsd-3-clause', 'hunks': "@@ -423,7 +423,7 @@ def memcached_configured(memcached_connection, monkeypatch):\n     monkeypatch.setattr(\n         sessions.MemcachedSession,\n         'servers',\n-        [str(memcached_connection)],\n+        [memcached_connection],\n     )\n \n \n", 'addition_count': 1, 'deletion_count': 1, 'src_file': 'a/cherrypy/test/test_session.py', 'tgt_file': 'b/cherrypy/test/test_session.py', 'file_extension': '.py', 'before_file': 'import os\nimport platform\nimport threading\nimport time\nfrom http.client import HTTPConnection\n\nfrom distutils.spawn import find_executable\nimport pytest\nfrom path import Path\nfrom more_itertools import consume\nimport portend\n\nimport cherrypy\nfrom cherrypy._cpcompat import HTTPSConnection\nfrom cherrypy.lib import sessions\nfrom cherrypy.lib import reprconf\nfrom cherrypy.lib.httputil import response_codes\nfrom cherrypy.test import helper\nfrom cherrypy import _json as json\n\nlocalDir = Path(__file__).dirname()\n\n\ndef http_methods_allowed(methods=[\'GET\', \'HEAD\']):\n    method = cherrypy.request.method.upper()\n    if method not in methods:\n        cherrypy.response.headers[\'Allow\'] = \', \'.join(methods)\n        raise cherrypy.HTTPError(405)\n\n\ncherrypy.tools.allow = cherrypy.Tool(\'on_start_resource\', http_methods_allowed)\n\n\ndef setup_server():\n\n    @cherrypy.config(**{\n        \'tools.sessions.on\': True,\n        \'tools.sessions.storage_class\': sessions.RamSession,\n        \'tools.sessions.storage_path\': localDir,\n        \'tools.sessions.timeout\': (1.0 / 60),\n        \'tools.sessions.clean_freq\': (1.0 / 60),\n    })\n    class Root:\n\n        @cherrypy.expose\n        def clear(self):\n            cherrypy.session.cache.clear()\n\n        @cherrypy.expose\n        @cherrypy.tools.json_out()\n        def data(self):\n            cherrypy.session[\'aha\'] = \'foo\'\n            return cherrypy.session._data\n\n        @cherrypy.expose\n        def testGen(self):\n            counter = cherrypy.session.get(\'counter\', 0) + 1\n            cherrypy.session[\'counter\'] = counter\n            yield str(counter)\n\n        @cherrypy.expose\n        def testStr(self):\n            counter = cherrypy.session.get(\'counter\', 0) + 1\n            cherrypy.session[\'counter\'] = counter\n            return str(counter)\n\n        @cherrypy.expose\n        @cherrypy.config(**{\'tools.sessions.on\': False})\n        def set_session_cls(self, new_cls_name):\n            new_cls = reprconf.attributes(new_cls_name)\n            cfg = {\'tools.sessions.storage_class\': new_cls}\n            self.__class__._cp_config.update(cfg)\n            if hasattr(cherrypy, \'session\'):\n                del cherrypy.session\n            if new_cls.clean_thread:\n                new_cls.clean_thread.stop()\n                new_cls.clean_thread.unsubscribe()\n                del new_cls.clean_thread\n\n        @cherrypy.expose\n        def index(self):\n            sess = cherrypy.session\n            c = sess.get(\'counter\', 0) + 1\n            time.sleep(0.01)\n            sess[\'counter\'] = c\n            return str(c)\n\n        @cherrypy.expose\n        def keyin(self, key):\n            return str(key in cherrypy.session)\n\n        @cherrypy.expose\n        def delete(self):\n            cherrypy.session.delete()\n            sessions.expire()\n            return \'done\'\n\n        @cherrypy.expose\n        def delkey(self, key):\n            del cherrypy.session[key]\n            return \'OK\'\n\n        @cherrypy.expose\n        def redir_target(self):\n            return self._cp_config[\'tools.sessions.storage_class\'].__name__\n\n        @cherrypy.expose\n        def iredir(self):\n            raise cherrypy.InternalRedirect(\'/redir_target\')\n\n        @cherrypy.expose\n        @cherrypy.config(**{\n            \'tools.allow.on\': True,\n            \'tools.allow.methods\': [\'GET\'],\n        })\n        def restricted(self):\n            return cherrypy.request.method\n\n        @cherrypy.expose\n        def regen(self):\n            cherrypy.tools.sessions.regenerate()\n            return \'logged in\'\n\n        @cherrypy.expose\n        def length(self):\n            return str(len(cherrypy.session))\n\n        @cherrypy.expose\n        @cherrypy.config(**{\n            \'tools.sessions.path\': \'/session_cookie\',\n            \'tools.sessions.name\': \'temp\',\n            \'tools.sessions.persistent\': False,\n        })\n        def session_cookie(self):\n            # Must load() to start the clean thread.\n            cherrypy.session.load()\n            return cherrypy.session.id\n\n    cherrypy.tree.mount(Root())\n\n\nclass SessionTest(helper.CPWebCase):\n    setup_server = staticmethod(setup_server)\n\n    @classmethod\n    def teardown_class(cls):\n        """Clean up sessions."""\n        super(cls, cls).teardown_class()\n        consume(\n            file.remove_p()\n            for file in localDir.listdir()\n            if file.basename().startswith(\n                sessions.FileSession.SESSION_PREFIX\n            )\n        )\n\n    def test_0_Session(self):\n        self.getPage(\'/set_session_cls/cherrypy.lib.sessions.RamSession\')\n        self.getPage(\'/clear\')\n\n        # Test that a normal request gets the same id in the cookies.\n        # Note: this wouldn\'t work if /data didn\'t load the session.\n        self.getPage(\'/data\')\n        assert self.body == b\'{"aha": "foo"}\'\n        c = self.cookies[0]\n        self.getPage(\'/data\', self.cookies)\n        self.cookies[0] == c\n\n        self.getPage(\'/testStr\')\n        assert self.body == b\'1\'\n        cookie_parts = dict([p.strip().split(\'=\')\n                             for p in self.cookies[0][1].split(\';\')])\n        # Assert there is an \'expires\' param\n        expected_cookie_keys = {\'session_id\', \'expires\', \'Path\', \'Max-Age\'}\n        assert set(cookie_parts.keys()) == expected_cookie_keys\n        self.getPage(\'/testGen\', self.cookies)\n        assert self.body == b\'2\'\n        self.getPage(\'/testStr\', self.cookies)\n        assert self.body == b\'3\'\n        self.getPage(\'/data\', self.cookies)\n        expected_data = {\'counter\': 3, \'aha\': \'foo\'}\n        assert json.decode(self.body.decode(\'utf-8\')) == expected_data\n        self.getPage(\'/length\', self.cookies)\n        assert self.body == b\'2\'\n        self.getPage(\'/delkey?key=counter\', self.cookies)\n        assert self.status_code == 200\n\n        self.getPage(\'/set_session_cls/cherrypy.lib.sessions.FileSession\')\n        self.getPage(\'/testStr\')\n        assert self.body == b\'1\'\n        self.getPage(\'/testGen\', self.cookies)\n        assert self.body == b\'2\'\n        self.getPage(\'/testStr\', self.cookies)\n        assert self.body == b\'3\'\n        self.getPage(\'/delkey?key=counter\', self.cookies)\n        assert self.status_code == 200\n\n        # Wait for the session.timeout (1 second)\n        time.sleep(2)\n        self.getPage(\'/\')\n        assert self.body == b\'1\'\n        self.getPage(\'/length\', self.cookies)\n        assert self.body == b\'1\'\n\n        # Test session __contains__\n        self.getPage(\'/keyin?key=counter\', self.cookies)\n        assert self.body == b\'True\'\n        cookieset1 = self.cookies\n\n        # Make a new session and test __len__ again\n        self.getPage(\'/\')\n        self.getPage(\'/length\', self.cookies)\n        assert self.body == b\'2\'\n\n        # Test session delete\n        self.getPage(\'/delete\', self.cookies)\n        assert self.body == b\'done\'\n        self.getPage(\'/delete\', cookieset1)\n        assert self.body == b\'done\'\n\n        def f():\n            return [\n                x\n                for x in os.listdir(localDir)\n                if x.startswith(\'session-\') and not x.endswith(\'.lock\')\n            ]\n        assert f() == []\n\n        # Wait for the cleanup thread to delete remaining session files\n        self.getPage(\'/\')\n        assert f() != []\n        time.sleep(2)\n        assert f() == []\n\n    def test_1_Ram_Concurrency(self):\n        self.getPage(\'/set_session_cls/cherrypy.lib.sessions.RamSession\')\n        self._test_Concurrency()\n\n    def test_2_File_Concurrency(self):\n        self.getPage(\'/set_session_cls/cherrypy.lib.sessions.FileSession\')\n        self._test_Concurrency()\n\n    def _test_Concurrency(self):\n        client_thread_count = 5\n        request_count = 30\n\n        # Get initial cookie\n        self.getPage(\'/\')\n        assert self.body == b\'1\'\n        cookies = self.cookies\n\n        data_dict = {}\n        errors = []\n\n        def request(index):\n            if self.scheme == \'https\':\n                c = HTTPSConnection(\'%s:%s\' % (self.interface(), self.PORT))\n            else:\n                c = HTTPConnection(\'%s:%s\' % (self.interface(), self.PORT))\n            for i in range(request_count):\n                c.putrequest(\'GET\', \'/\')\n                for k, v in cookies:\n                    c.putheader(k, v)\n                c.endheaders()\n                response = c.getresponse()\n                body = response.read()\n                if response.status != 200 or not body.isdigit():\n                    errors.append((response.status, body))\n                else:\n                    data_dict[index] = max(data_dict[index], int(body))\n                # Uncomment the following line to prove threads overlap.\n                # sys.stdout.write("%d " % index)\n\n        # Start <request_count> requests from each of\n        # <client_thread_count> concurrent clients\n        ts = []\n        for c in range(client_thread_count):\n            data_dict[c] = 0\n            t = threading.Thread(target=request, args=(c,))\n            ts.append(t)\n            t.start()\n\n        for t in ts:\n            t.join()\n\n        hitcount = max(data_dict.values())\n        expected = 1 + (client_thread_count * request_count)\n\n        for e in errors:\n            print(e)\n        assert len(errors) == 0\n        assert hitcount == expected\n\n    def test_3_Redirect(self):\n        # Start a new session\n        self.getPage(\'/testStr\')\n        self.getPage(\'/iredir\', self.cookies)\n        assert self.body == b\'FileSession\'\n\n    def test_4_File_deletion(self):\n        # Start a new session\n        self.getPage(\'/testStr\')\n        # Delete the session file manually and retry.\n        id = self.cookies[0][1].split(\';\', 1)[0].split(\'=\', 1)[1]\n        path = os.path.join(localDir, \'session-\' + id)\n        os.unlink(path)\n        self.getPage(\'/testStr\', self.cookies)\n\n    def test_5_Error_paths(self):\n        self.getPage(\'/unknown/page\')\n        self.assertErrorPage(404, "The path \'/unknown/page\' was not found.")\n\n        # Note: this path is *not* the same as above. The above\n        # takes a normal route through the session code; this one\n        # skips the session code\'s before_handler and only calls\n        # before_finalize (save) and on_end (close). So the session\n        # code has to survive calling save/close without init.\n        self.getPage(\'/restricted\', self.cookies, method=\'POST\')\n        self.assertErrorPage(405, response_codes[405][1])\n\n    def test_6_regenerate(self):\n        self.getPage(\'/testStr\')\n        # grab the cookie ID\n        id1 = self.cookies[0][1].split(\';\', 1)[0].split(\'=\', 1)[1]\n        self.getPage(\'/regen\')\n        assert self.body == b\'logged in\'\n        id2 = self.cookies[0][1].split(\';\', 1)[0].split(\'=\', 1)[1]\n        assert id1 != id2\n\n        self.getPage(\'/testStr\')\n        # grab the cookie ID\n        id1 = self.cookies[0][1].split(\';\', 1)[0].split(\'=\', 1)[1]\n        self.getPage(\'/testStr\',\n                     headers=[\n                         (\'Cookie\',\n                          \'session_id=maliciousid; \'\n                          \'expires=Sat, 27 Oct 2017 04:18:28 GMT; Path=/;\')])\n        id2 = self.cookies[0][1].split(\';\', 1)[0].split(\'=\', 1)[1]\n        assert id1 != id2\n        assert id2 != \'maliciousid\'\n\n    def test_7_session_cookies(self):\n        self.getPage(\'/set_session_cls/cherrypy.lib.sessions.RamSession\')\n        self.getPage(\'/clear\')\n        self.getPage(\'/session_cookie\')\n        # grab the cookie ID\n        cookie_parts = dict([p.strip().split(\'=\')\n                            for p in self.cookies[0][1].split(\';\')])\n        # Assert there is no \'expires\' param\n        assert set(cookie_parts.keys()) == {\'temp\', \'Path\'}\n        id1 = cookie_parts[\'temp\']\n        assert list(sessions.RamSession.cache) == [id1]\n\n        # Send another request in the same "browser session".\n        self.getPage(\'/session_cookie\', self.cookies)\n        cookie_parts = dict([p.strip().split(\'=\')\n                            for p in self.cookies[0][1].split(\';\')])\n        # Assert there is no \'expires\' param\n        assert set(cookie_parts.keys()) == {\'temp\', \'Path\'}\n        assert self.body.decode(\'utf-8\') == id1\n        assert list(sessions.RamSession.cache) == [id1]\n\n        # Simulate a browser close by just not sending the cookies\n        self.getPage(\'/session_cookie\')\n        # grab the cookie ID\n        cookie_parts = dict([p.strip().split(\'=\')\n                            for p in self.cookies[0][1].split(\';\')])\n        # Assert there is no \'expires\' param\n        assert set(cookie_parts.keys()) == {\'temp\', \'Path\'}\n        # Assert a new id has been generated...\n        id2 = cookie_parts[\'temp\']\n        assert id1 != id2\n        assert set(sessions.RamSession.cache.keys()) == {id1, id2}\n\n        # Wait for the session.timeout on both sessions\n        time.sleep(2.5)\n        cache = list(sessions.RamSession.cache)\n        if cache:\n            if cache == [id2]:\n                self.fail(\'The second session did not time out.\')\n            else:\n                self.fail(\'Unknown session id in cache: %r\', cache)\n\n    def test_8_Ram_Cleanup(self):\n        def lock():\n            s1 = sessions.RamSession()\n            s1.acquire_lock()\n            time.sleep(1)\n            s1.release_lock()\n\n        t = threading.Thread(target=lock)\n        t.start()\n        start = time.time()\n        while not sessions.RamSession.locks and time.time() - start < 5:\n            time.sleep(0.01)\n        assert len(sessions.RamSession.locks) == 1, \'Lock not acquired\'\n        s2 = sessions.RamSession()\n        s2.clean_up()\n        msg = \'Clean up should not remove active lock\'\n        assert len(sessions.RamSession.locks) == 1, msg\n        t.join()\n\n\ndef is_memcached_present():\n    executable = find_executable(\'memcached\')\n    return bool(executable)\n\n\n@pytest.fixture(scope=\'session\')\ndef memcached_server_present():\n    is_memcached_present() or pytest.skip(\'memcached not available\')\n\n\n@pytest.fixture()\ndef memcached_client_present():\n    pytest.importorskip(\'memcache\')\n\n\n@pytest.fixture(scope=\'session\')\ndef memcached_instance(request, watcher_getter, memcached_server_present):\n    """\n    Start up an instance of memcached.\n    monkeypatch.setattr(\n        sessions.MemcachedSession,\n        \'servers\',\n        [str(memcached_connection)],\n    )\n\n\n        except Exception:\n            return True\n        return False\n\n    proc = watcher_getter(\n        name=\'memcached\',\n        arguments=[\'-p\', str(port)],\n        checker=is_occupied,\n        request=request,\n    )\n    return locals()\n\n\n@pytest.fixture\ndef memcached_configured(\n    memcached_instance, monkeypatch,\n    memcached_client_present,\n):\n    server = \'localhost:{port}\'.format_map(memcached_instance)\n    monkeypatch.setattr(\n        sessions.MemcachedSession,\n        \'servers\',\n        [server],\n    )\n\n\n@pytest.mark.skipif(\n    platform.system() == \'Windows\',\n    reason=\'pytest-services helper does not work under Windows\',\n)\n@pytest.mark.usefixtures(\'memcached_configured\')\nclass MemcachedSessionTest(helper.CPWebCase):\n    setup_server = staticmethod(setup_server)\n\n    def test_0_Session(self):\n        self.getPage(\n            \'/set_session_cls/cherrypy.lib.sessions.MemcachedSession\'\n        )\n\n        self.getPage(\'/testStr\')\n        assert self.body == b\'1\'\n        self.getPage(\'/testGen\', self.cookies)\n        assert self.body == b\'2\'\n        self.getPage(\'/testStr\', self.cookies)\n        assert self.body == b\'3\'\n        self.getPage(\'/length\', self.cookies)\n        self.assertErrorPage(500)\n        assert b\'NotImplementedError\' in self.body\n        self.getPage(\'/delkey?key=counter\', self.cookies)\n        assert self.status_code == 200\n\n        # Wait for the session.timeout (1 second)\n        time.sleep(1.25)\n        self.getPage(\'/\')\n        assert self.body == b\'1\'\n\n        # Test session __contains__\n        self.getPage(\'/keyin?key=counter\', self.cookies)\n        assert self.body == b\'True\'\n\n        # Test session delete\n        self.getPage(\'/delete\', self.cookies)\n        assert self.body == b\'done\'\n\n    def test_1_Concurrency(self):\n        client_thread_count = 5\n        request_count = 30\n\n        # Get initial cookie\n        self.getPage(\'/\')\n        assert self.body == b\'1\'\n        cookies = self.cookies\n\n        data_dict = {}\n\n        def request(index):\n            for i in range(request_count):\n                self.getPage(\'/\', cookies)\n                # Uncomment the following line to prove threads overlap.\n                # sys.stdout.write("%d " % index)\n            if not self.body.isdigit():\n                self.fail(self.body)\n            data_dict[index] = int(self.body)\n\n        # Start <request_count> concurrent requests from\n        # each of <client_thread_count> clients\n        ts = []\n        for c in range(client_thread_count):\n            data_dict[c] = 0\n            t = threading.Thread(target=request, args=(c,))\n            ts.append(t)\n            t.start()\n\n        for t in ts:\n            t.join()\n\n        hitcount = max(data_dict.values())\n        expected = 1 + (client_thread_count * request_count)\n        assert hitcount == expected\n\n    def test_3_Redirect(self):\n        # Start a new session\n        self.getPage(\'/testStr\')\n        self.getPage(\'/iredir\', self.cookies)\n        assert self.body == b\'MemcachedSession\'\n\n    def test_5_Error_paths(self):\n        self.getPage(\'/unknown/page\')\n        self.assertErrorPage(\n            404, "The path \'/unknown/page\' was not found.")\n\n        # Note: this path is *not* the same as above. The above\n        # takes a normal route through the session code; this one\n        # skips the session code\'s before_handler and only calls\n        # before_finalize (save) and on_end (close). So the session\n        # code has to survive calling save/close without init.\n        self.getPage(\'/restricted\', self.cookies, method=\'POST\')\n        self.assertErrorPage(405, response_codes[405][1])\n', 'message': 'Revert "Forcefully stringify memcached socket path"\n\nThis reverts commit 43c83f36f36ce0f04d590b83017526429cf217b2.\n', 'author_name': 'Sviatoslav Sydorenko', 'lang': 'py'}