# Path: sockjs/cyclone/session.py
# class SessionMixin(object):
#     """ Represents one session object stored in the session container.
#         Derive from this object to store additional data.
#     """
# 
#     def __init__(self, session_id=None, expiry=None, time_module=time):
#         """ Constructor.
# 
#         @param session_id: Optional session id. If not provided, will generate
#                            new session id.
# 
#         @param expiry: Expiration time in seconds. If not provided, will never
#                        expire.
# 
#         @param time_module: only used for unit testing. A provider for a time()
#                             function
#         """
#         self.time_module = time_module
# 
#         self.session_id = session_id or self._random_key()
#         self.promoted = None
#         self.expiry = expiry
# 
#         if self.expiry is not None:
#             self.expiry_date = self.time_module.time() + self.expiry
# 
#     def _random_key(self):
#         """ Return random session key """
#         hashstr = '%s%s' % (random.random(), self.time_module.time())
#         return hashlib.md5(hashstr).hexdigest()
# 
#     def is_alive(self):
#         """ Check if session is still alive """
#         return self.expiry_date > self.time_module.time()
# 
#     def promote(self):
#         """ Mark object as alive, so it won't be collected during next
#         run of the garbage collector.
#         """
#         if self.expiry is not None:
#             self.promoted = self.time_module.time() + self.expiry
# 
#     def on_delete(self, forced):
#         """ Triggered when object was expired or deleted. """
#         pass
# 
#     def __cmp__(self, other):
#         return cmp(self.expiry_date, other.expiry_date)
# 
#     def __repr__(self):
#         return '%f %s %d' % (getattr(self, 'expiry_date', -1),
#                              self.session_id,
#                              self.promoted or 0)
# 
# Path: sockjs/cyclone/conn.py
# class ConnectionInfo(object):
#     """ Connection information object.
# 
#     Will be passed to the C{connectionMade} handler of your connection class.
# 
#     Has few properties:
# 
#     @cvar ip: Caller IP address
# 
#     @cvar cookies: Collection of cookies
# 
#     @cvar arguments: Collection of the query string arguments
# 
#     @cvar headers: a selection of the request's headers
# 
#     @cvar path: uri's path of the request
#     """
# 
#     _exposed_headers = set( ('origin', 'referer', 'x-client-ip',
#                              'x-forwarded-for', 'x-cluster-client-ip',
#                              'user-agent')
#                           )
# 
#     def __init__(self, ip, cookies, arguments, headers, path):
#         self.ip = ip
#         self.cookies = cookies
#         self.arguments = arguments
#         self.path = path
#         self._expose_headers(headers)
# 
#     def _expose_headers(self, headers):
#         self.headers = {}
#         for header_name, header_value in headers.iteritems():
#             if header_name.lower() in self._exposed_headers:
#                 self.headers[header_name] = header_value
# 
#     def get_header(self, name):
#         """ Return a single header by name
#         """
#         return self.headers.get(name)
# 
#     def get_argument(self, name):
#         """ Return single argument by name """
#         val = self.arguments.get(name)
#         if val:
#             return val[0]
#         return None
# 
#     def get_cookie(self, name):
#         """ Return single cookie by its name """
#         return self.cookies.get(name)


from twisted.trial import unittest
from sockjs.cyclone.session import SessionMixin
from sockjs.cyclone.conn import ConnectionInfo
    def test_get_argument(self):
        c = ConnectionInfo(None, None, dict(arg=('myarg', '')), {}, None)
        self.assertEquals(c.get_argument('arg'), 'myarg')

    def test_get_argument_returns_None_on_not_found(self):
        c = ConnectionInfo(None, None, dict(), {}, None)
        self.assertTrue(c.get_argument('arg') is None)

    def test_dont_expose_unknown_headers(self):
        c = ConnectionInfo(None, None, {}, {'StrangeHeader': '42'}, None)
        self.assertTrue(c.get_header('StrangeHeader') is None)

    def test_expose_whitelisted_headers(self):
        headers = dict((h, '42') for h in ConnectionInfo._exposed_headers)
        c = ConnectionInfo(None, None, {}, headers, None)
        for h in headers.keys():
            self.assertTrue(c.get_header(h) is not None)


class TimeMock(object):
    def __init__(self, returned_time):
        self.returned_time = returned_time
    def set(self, time):
        self.returned_time = time
    def time(self):
        return self.returned_time


class SessionMixinTest(unittest.TestCase):
    def test_repr(self):

# Path: sockjs/cyclone/session.py
# class SessionMixin(object):
#     """ Represents one session object stored in the session container.
#         Derive from this object to store additional data.
#     """
# 
#     def __init__(self, session_id=None, expiry=None, time_module=time):
#         """ Constructor.
# 
#         @param session_id: Optional session id. If not provided, will generate
#                            new session id.
# 
#         @param expiry: Expiration time in seconds. If not provided, will never
#                        expire.
# 
#         @param time_module: only used for unit testing. A provider for a time()
#                             function
#         """
#         self.time_module = time_module
# 
#         self.session_id = session_id or self._random_key()
#         self.promoted = None
#         self.expiry = expiry
# 
#         if self.expiry is not None:
#             self.expiry_date = self.time_module.time() + self.expiry
# 
#     def _random_key(self):
#         """ Return random session key """
#         hashstr = '%s%s' % (random.random(), self.time_module.time())
#         return hashlib.md5(hashstr).hexdigest()
# 
#     def is_alive(self):
#         """ Check if session is still alive """
#         return self.expiry_date > self.time_module.time()
# 
#     def promote(self):
#         """ Mark object as alive, so it won't be collected during next
#         run of the garbage collector.
#         """
#         if self.expiry is not None:
#             self.promoted = self.time_module.time() + self.expiry
# 
#     def on_delete(self, forced):
#         """ Triggered when object was expired or deleted. """
#         pass
# 
#     def __cmp__(self, other):
#         return cmp(self.expiry_date, other.expiry_date)
# 
#     def __repr__(self):
#         return '%f %s %d' % (getattr(self, 'expiry_date', -1),
#                              self.session_id,
#                              self.promoted or 0)
# 
# Path: sockjs/cyclone/conn.py
# class ConnectionInfo(object):
#     """ Connection information object.
# 
#     Will be passed to the C{connectionMade} handler of your connection class.
# 
#     Has few properties:
# 
#     @cvar ip: Caller IP address
# 
#     @cvar cookies: Collection of cookies
# 
#     @cvar arguments: Collection of the query string arguments
# 
#     @cvar headers: a selection of the request's headers
# 
#     @cvar path: uri's path of the request
#     """
# 
#     _exposed_headers = set( ('origin', 'referer', 'x-client-ip',
#                              'x-forwarded-for', 'x-cluster-client-ip',
#                              'user-agent')
#                           )
# 
#     def __init__(self, ip, cookies, arguments, headers, path):
#         self.ip = ip
#         self.cookies = cookies
#         self.arguments = arguments
#         self.path = path
#         self._expose_headers(headers)
# 
#     def _expose_headers(self, headers):
#         self.headers = {}
#         for header_name, header_value in headers.iteritems():
#             if header_name.lower() in self._exposed_headers:
#                 self.headers[header_name] = header_value
# 
#     def get_header(self, name):
#         """ Return a single header by name
#         """
#         return self.headers.get(name)
# 
#     def get_argument(self, name):
#         """ Return single argument by name """
#         val = self.arguments.get(name)
#         if val:
#             return val[0]
#         return None
# 
#     def get_cookie(self, name):
#         """ Return single cookie by its name """
#         return self.cookies.get(name)


# Path: tests/test_session.py
from twisted.trial import unittest
from sockjs.cyclone.session import SessionMixin
from sockjs.cyclone.conn import ConnectionInfo

    def test_get_argument(self):
        c = ConnectionInfo(None, None, dict(arg=('myarg', '')), {}, None)
        self.assertEquals(c.get_argument('arg'), 'myarg')

    def test_get_argument_returns_None_on_not_found(self):
        c = ConnectionInfo(None, None, dict(), {}, None)
        self.assertTrue(c.get_argument('arg') is None)

    def test_dont_expose_unknown_headers(self):
        c = ConnectionInfo(None, None, {}, {'StrangeHeader': '42'}, None)
        self.assertTrue(c.get_header('StrangeHeader') is None)

    def test_expose_whitelisted_headers(self):
        headers = dict((h, '42') for h in ConnectionInfo._exposed_headers)
        c = ConnectionInfo(None, None, {}, headers, None)
        for h in headers.keys():
            self.assertTrue(c.get_header(h) is not None)


class TimeMock(object):
    def __init__(self, returned_time):
        self.returned_time = returned_time
    def set(self, time):
        self.returned_time = time
    def time(self):
        return self.returned_time


class SessionMixinTest(unittest.TestCase):
    def test_repr(self):

        s = SessionMixin()