feat:support headers

mockingbird/handler.py

@@ -1,6 +1,6 @@
 import http.server
 import json
-from typing import Any, Tuple, Optional, Dict, Callable
+from typing import Any, Tuple, Optional, Dict, Callable, List
 
 from mockingbird.stub_group import StubGroup
 
@@ -23,24 +23,43 @@ def do_DELETE(self):
         self._handle_request("DELETE")
 
     def _handle_request(self, method: str):
-        matched_stub, path_params = self.stub_matcher.match(method, self.path)
+        request_headers = {key.lower(): value for key, value in self.headers.items()}
+        matched_stub, path_params = self.stub_matcher.match(
+            method, self.path, request_headers=request_headers
+        )
 
         if matched_stub:
             self._send_response(matched_stub, path_params)
         else:
             self._send_404_response(method)
 
-    def _send_response(self,
-                       matched_stub: Tuple[int, Any, Optional[Callable]],
-                       path_params: Dict[str, str]):
-        status_code, response, response_func = matched_stub
+    def _send_response(
+            self,
+            matched_stub: Tuple[
+                int, Any, Optional[Callable], Optional[List[Tuple[str, str]]]
+            ],
+            path_params: Dict[str, str]
+    ):
+        status_code, response, response_func, headers = matched_stub
         if response_func:
             status_code, response = response_func()
 
         self.send_response(status_code)
-        self.send_header('Content-Type', 'application/json')
+        self._set_headers(headers)
+
         self.end_headers()
+        self._write_response(response, path_params)
+
+    def _set_headers(self, headers: Optional[List[Tuple[str, str]]]):
+        if headers:
+            for header_name, header_value in headers:
+                self.send_header(header_name, header_value)
+            if not any(header[0].lower() == 'content-type' for header in headers):
+                self.send_header('Content-Type', 'application/json')
+        else:
+            self.send_header('Content-Type', 'application/json')
 
+    def _write_response(self, response: Any, path_params: Dict[str, str]):
         if isinstance(response, dict):
             response = self._format_response(response, path_params)
             self.wfile.write(json.dumps(response).encode('utf-8'))

mockingbird/route.py

@@ -1,6 +1,6 @@
 import re
 
-from typing import Dict, Tuple, Any, Callable, Optional, Pattern, Union
+from typing import Dict, Tuple, Any, Callable, Optional, Pattern, Union, List
 
 
 class Route:
@@ -9,6 +9,7 @@ def __init__(self, method: str, path: str):
         self.path = path
         self._body = {}
         self._status = 200
+        self._headers: List[Tuple[str, str]] = []
         self._response_func: Optional[Callable[[], Tuple[int, Any]]] = None
 
         escaped_path = re.escape(path)
@@ -24,6 +25,10 @@ def status(self, status_code: int):
         self._status = status_code
         return self
 
+    def headers(self, headers: List[Tuple[str, str]]):
+        self._headers = headers
+        return self
+
     def response_func(self, func: Callable[[], Tuple[int, Any]]):
         self._response_func = func
         return self
@@ -35,5 +40,6 @@ def build(self):
             "compiled_path": self._compiled_path,
             "body": self._body,
             "status": self._status,
+            "headers": self._headers,
             "response_func": self._response_func
         }

mockingbird/server.py

@@ -26,6 +26,7 @@ def routes(self, *routes: Route):
                 pattern=route_config["compiled_path"],
                 status_code=route_config["status"],
                 response=route_config["body"],
+                headers=route_config["headers"],
                 response_func=route_config["response_func"]
             )
         return self

mockingbird/stub_group.py

@@ -1,5 +1,5 @@
 import re
-from typing import Pattern, Dict, Tuple, Any, Optional, Callable, Union
+from typing import Pattern, Dict, Tuple, Any, Optional, Callable, Union, List
 
 
 class StubGroup:
@@ -8,31 +8,42 @@ def __init__(self):
             str,
             Dict[
                 Pattern,
-                Tuple[int, Any, Optional[
-                    Callable[[], Tuple[int, Any]]]
-                ]]] = {}
-
-    def add(self, method: str,
-            pattern: Union[str, Pattern], status_code: int, response: Any,
-            response_func: Optional[Callable[[], Tuple[int, Any]]] = None):
+                Tuple[int, Any, Optional[Callable], Optional[List[Tuple[str, str]]]]
+            ]
+        ] = {}
+
+    def add(self, method: str, pattern: Union[str, Pattern],
+            status_code: int, response: Any,
+            response_func: Optional[Callable[[], Tuple[int, Any]]] = None,
+            headers: Optional[List[Tuple[str, str]]] = None):
         if method not in self.stubs:
             self.stubs[method] = {}
 
         if isinstance(pattern, str):
             pattern = re.compile(f"^{re.sub(r'\\{(\\w+)\\}',
                                             r'(?P<\\1>[^/]+)', pattern)}$")
 
-        self.stubs[method][pattern] = (status_code, response, response_func)
+        self.stubs[method][pattern] = (status_code, response, response_func, headers)
 
-    def match(self, method: str, path: str):
+    def match(self, method: str, path: str,
+              request_headers: Optional[Dict[str, str]] = None):
         matched_stub = None
         path_params = {}
 
-        for compiled_path, (status_code, response, response_func)\
+        for compiled_path, (status_code, response, response_func, headers)\
                 in self.stubs.get(method, {}).items():
             match = compiled_path.match(path)
             if match:
-                matched_stub = (status_code, response, response_func)
+                if headers and request_headers:
+                    headers_included = all(
+                        header_name in request_headers
+                        and request_headers[header_name] == header_value
+                        for header_name, header_value in headers
+                    )
+                    if not headers_included:
+                        continue
+
+                matched_stub = (status_code, response, response_func, headers)
                 path_params = match.groupdict()
                 break
 

tests/support/client.py

@@ -6,5 +6,5 @@ class Client:
     def __init__(self, root='http://localhost:8080'):
         self.root = root
 
-    def get(self, path):
-        return requests.get(f"{self.root + path}")
+    def get(self, path, headers=None):
+        return requests.get(f"{self.root + path}", headers=headers)

tests/test_mockingbird_get.py

@@ -69,3 +69,13 @@ def test_get_empty_response(mockingbird_server):
     resp = Client().get('/empty')
     assert_that(resp.status_code, is_(204))
     assert_that(resp.text, is_(""))
+
+
+def test_get_headers(mockingbird_server):
+    mockingbird_server.routes(
+        get("/hello").
+        body("hi there").
+        headers([("Content-Type", "text/plain")])
+    )
+    resp = Client().get('/hello', {"Content-Type": "application/json"})
+    assert_that(resp.status_code, is_(404))

tests/test_stub_group.py

@@ -19,7 +19,7 @@ def test_match():
     stub_group = StubGroup()
     stub_group.add("GET", "/hi", 200, {"message": "hello"})
     matched, _ = stub_group.match("GET", "/hi")
-    assert_that(matched, is_((200, {"message": "hello"}, None)))
+    assert_that(matched, is_((200, {"message": "hello"}, None, None)))
 
 
 def test_none_on_partial_match():
@@ -38,7 +38,7 @@ def test_match_w_path_param():
 
     matched, path_param = stub_group.match("GET", "/hello/mockingbird")
 
-    assert_that(matched, is_((200, {"message": "Hello, {name}!"}, None)))
+    assert_that(matched, is_((200, {"message": "Hello, {name}!"}, None, None)))
     assert_that(path_param, is_({"name": "mockingbird"}))
 
 
@@ -51,4 +51,30 @@ def dynamic_response():
     stub_group.add("GET", r"^/dynamic$",
                    200, {}, response_func=dynamic_response)
     matched, _ = stub_group.match("GET", "/dynamic")
-    assert_that(matched, is_((200, {}, dynamic_response)))
+    assert_that(matched, is_((200, {}, dynamic_response, None)))
+
+
+def test_no_match_given_unexpected_header():
+    stub_group = StubGroup()
+    stub_group.add(
+        "GET", "/hi", 200,
+        {"message": "hello"}, headers=[("Content-Type", "text/plain")])
+    matched, _ = stub_group.match(
+        "GET", "/hi",
+        request_headers={"Content-Type": "application/json"})
+    assert_that(matched, none())
+
+
+def test_match_given_expected_headers():
+    stub_group = StubGroup()
+    stub_group.add("GET", "/hi", 200,
+                   {"message": "hello"},
+                   headers=[
+                       ("Content-Type", "application/json"),
+                       ("Authorization", "Bearer YOUR_TOKEN"),
+                       ("Custom-Header", "CustomValue")
+                   ])
+    matched, _ = stub_group.match(
+        "GET", "/hi",
+        request_headers={"Content-Type": "application/json"})
+    assert_that(matched, none())