Implement handling cookies (#39)

Implement handling cookies

kobin/environs.py

@@ -1,9 +1,10 @@
 import threading
 import cgi
 import json
-from typing import Dict, List, Tuple
+from typing import Dict, List, Tuple, Any
 import http.client as http_client
 from urllib.parse import SplitResult
+from http.cookies import SimpleCookie  # type: ignore
 
 
 def _local_property():
@@ -89,6 +90,15 @@ def url(self) -> str:
         url_split_result = SplitResult(protocol, host, self.path, query_params, '')  # type: ignore
         return url_split_result.geturl()
 
+    @property
+    def cookies(self) -> Dict[str, str]:
+        cookies = SimpleCookie(self.environ.get('HTTP_COOKIE', '')).values()
+        return {c.key: c.value for c in cookies}
+
+    def get_cookie(self, key: str, default: str=None, secret=None) -> str:
+        value = self.cookies.get(key)
+        return value or default
+
     def __getitem__(self, key):
         return self.environ[key]
 
@@ -145,6 +155,7 @@ def __init__(self, body: str='', status: int=None, headers: Dict=None,
         self._headers = {}  # type: Dict[str, List[str]]
         self.body = body
         self._status_code = status or self.default_status
+        self._cookies = SimpleCookie()  # type: SimpleCookie
 
         if headers:
             for name, value in headers.items():
@@ -182,14 +193,44 @@ def headerlist(self) -> List[Tuple[str, str]]:
         out += [(name, val)
                 for (name, vals) in headers
                 for val in vals]
+        if self._cookies:
+            for c in self._cookies.values():
+                out.append(('Set-Cookie', c.OutputString()))
         return [(k, v.encode('utf8').decode('latin1')) for (k, v) in out]
 
+    def set_cookie(self, key: str, value: Any, expires: str=None, path: str=None, **options: Dict[str, Any]) -> None:
+        from datetime import timedelta, datetime, date
+        import time
+        self._cookies[key] = value
+        if expires:
+            self._cookies[key]['expires'] = expires
+        if path:
+            self._cookies[key]['path'] = path
+
+        for k, v in options.items():
+            if k == 'max_age':
+                if isinstance(v, timedelta):
+                    v = v.seconds + v.days * 24 * 3600
+            if k == 'expires':
+                if isinstance(v, (date, datetime)):
+                    v = v.timetuple()
+                elif isinstance(v, (int, float)):
+                    v = v.gmtime(value)
+                v = time.strftime("%a, %d %b %Y %H:%M:%S GMT", v)  # type: ignore
+            self._cookies[key][k.replace('_', '-')] = v
+
+    def delete_cookie(self, key, **kwargs) -> None:
+        kwargs['max_age'] = -1
+        kwargs['expires'] = 0
+        self.set_cookie(key, '', **kwargs)
+
     def add_header(self, key: str, value: str) -> None:
         self._headers.setdefault(key, []).append(value)
 
     def apply(self, other):
         self.status = other._status_code
         self._headers = other._headers
+        self._cookies = other._cookies
         self.body = other.body
 
 
@@ -200,6 +241,7 @@ class LocalResponse(Response):
     bind = Response.__init__
     _status_code = _local_property()
     _headers = _local_property()
+    _cookies = _local_property()
     body = _local_property()
 
 

kobin/server_adapters.py

@@ -1,6 +1,6 @@
 from typing import Dict, Any
 
-from . import Kobin
+from .app import Kobin
 
 
 class ServerAdapter:
@@ -49,7 +49,7 @@ def load(self) -> Kobin:
 
         GunicornApplication().run()
 
-servers = {
+servers = {  # type: Dict[str, Any]
     'wsgiref': WSGIRefServer,
     'gunicorn': GunicornServer,
-}  # type: Dict[str, Any]
+}

kobin/static_files.py

@@ -3,15 +3,15 @@
 import os
 import time
 
-from kobin.templates import load_file
+from .templates import load_file
 from .environs import request, response
 
 
 def static_file(filename: str,
                 mimetype: str='auto',
                 download: str='',
                 charset: str='UTF-8') -> bytes:
-    from . import current_config
+    from .app import current_config
     static_dirs = current_config()['STATICFILES_DIRS']  # type: List[str]
     filename = load_file(filename, static_dirs)  # Get abs path
     headers = dict()  # type: Dict[str, str]

tests/test_environs.py

@@ -22,6 +22,20 @@ def test_get_default_value(self):
         request = Request({})
         self.assertEqual(request.get('hoge', 'HOGE'), 'HOGE')
 
+    def test_cookies_property_has_nothing(self):
+        request = Request({})
+        self.assertEqual(len(request.cookies), 0)
+
+    def test_cookies_property_has_an_item(self):
+        request = Request({'HTTP_COOKIE': 'foo="bar"'})
+        self.assertEqual(len(request.cookies), 1)
+
+    def test_get_cookie(self):
+        request = Request({'HTTP_COOKIE': 'foo="bar"'})
+        actual = request.get_cookie("foo")
+        expected = 'bar'
+        self.assertEqual(actual, expected)
+
     def test_path_property(self):
         request = Request({'PATH_INFO': '/hoge'})
         self.assertEqual(request.path, '/hoge')
@@ -136,6 +150,18 @@ def test_constructor_headerlist(self):
         expected_content_type = ('Content-Type', 'text/html; charset=UTF-8')
         self.assertIn(expected_content_type, response.headerlist)
 
+    def test_set_cookie(self):
+        response = Response()
+        response.set_cookie('foo', 'bar')
+        expected_set_cookie = ('Set-Cookie', 'foo=bar')
+        self.assertIn(expected_set_cookie, response.headerlist)
+
+    def test_delete_cookie(self):
+        response = Response()
+        response.delete_cookie('foo')
+        expected_set_cookie = ('Set-Cookie', 'foo=""; Max-Age=-1')
+        self.assertIn(expected_set_cookie, response.headerlist)
+
     def test_constructor_headerlist_has_already_content_type(self):
         response = Response()
         response.add_header('Content-Type', 'application/json')

tests/test_static_files.py

@@ -1,5 +1,5 @@
 import os
-from unittest import TestCase, skip
+from unittest import TestCase
 from unittest.mock import patch
 
 from kobin.static_files import static_file
@@ -33,33 +33,33 @@ class StaticFilesTests(TestCase):
         'STATICFILES_DIRS': [os.path.join(os.path.dirname(os.path.abspath(__file__)), 'static')]
     }
 
-    @patch('kobin.current_config')
+    @patch('kobin.app.current_config')
     def test_return_static_file(self, mock_current_config):
         mock_current_config.return_value = self.dummy_current_config
         actual = static_file('static.css')
         expected = 'body { color: black; }\n'.encode('utf-8')
         self.assertEqual(actual, expected)
 
-    @patch('kobin.current_config')
+    @patch('kobin.app.current_config')
     def test_static_file_404_not_found(self, mock_current_config):
         mock_current_config.return_value = self.dummy_current_config
         self.assertRaises(HTTPError, static_file, 'not_found.png')
 
-    @patch('kobin.current_config')
+    @patch('kobin.app.current_config')
     def test_exist_last_modified_in_headers(self, mock_current_config):
         mock_current_config.return_value = self.dummy_current_config
         static_file('static.css')
         self.assertIn('Last-Modified', response._headers)
 
-    @patch('kobin.current_config')
+    @patch('kobin.app.current_config')
     def test_content_length(self, mock_current_config):
         mock_current_config.return_value = self.dummy_current_config
         expected_content_length = str(23)
         static_file('static.css')
         actual_content_length = response._headers['Content-Length']
         self.assertIn(expected_content_length, actual_content_length)
 
-    @patch('kobin.current_config')
+    @patch('kobin.app.current_config')
     def test_content_type(self, mock_current_config):
         mock_current_config.return_value = self.dummy_current_config
         expected_content_type = 'text/css; charset=UTF-8'