Add tests for message endpoint

tests/conftest.py

@@ -8,11 +8,12 @@
 from sqlalchemy_helpers import get_or_create
 from webhook_to_fedora_messaging.main import create_app
 from os import environ
+from webhook_to_fedora_messaging.models.user import User
+from webhook_to_fedora_messaging.models.service import Service
 from os.path import abspath, basename
 from webhook_to_fedora_messaging.models.user import User
 from webhook_to_fedora_messaging.models.service import Service
 
-
 @pytest.fixture(scope="session")
 def client():
     root = abspath(__name__)
@@ -43,15 +44,16 @@ def create_service(client):
     with client.application.app_context():
         # Setup code to create the object in the database
         user, created = get_or_create(db.session, User, username="mehmet") # Adjust fields as necessary
-        get_or_create(db.session, Service, name="Github Demo", type="Github", desc="description", user_id=user.id)
-
+        service, created = get_or_create(db.session, Service, name="GitHub Demo", type="GitHub", desc="description", user_id=user.id)
+        service.token = "43b901cba25c458a9175ef119a9603bb"
         db.session.commit()
+        db.session.refresh(service)
 
-    yield
+    yield service
 
     with client.application.app_context():
         # Teardown code to remove the object from the database
         db.session.query(User).filter_by(username="mehmet").delete()
-        db.session.query(Service).filter_by(name="Github Demo").delete()
+        db.session.query(Service).filter_by(name="GitHub Demo").delete()
         db.session.commit()
 

tests/test_message/test_message_create.py

@@ -0,0 +1,59 @@
+import pytest
+import hmac
+import json
+import hashlib
+import pathlib
+from contextlib import nullcontext
+from .event_sample import request, headers
+from fedora_messaging.testing import mock_sends
+from webhook_to_fedora_messaging_messages.github.github import GithubMessageV1
+
+
+def signature_header(service, request):
+    hashed_value = hmac.new(service.token.encode('utf-8'), msg=json.dumps(request, sort_keys=True).encode("utf-8"), digestmod=hashlib.sha256)
+    return hashed_value.hexdigest()
+
+
+def get_headers(service, request):
+    return {
+    "Content-Type": "application/json",
+    "X-Hub-Signature-256": f"sha256={signature_header(service, request)}",
+    "X-Github-Event": "push",
+    "X-Github-Delivery": "f1064eb2-4995-11ef-82e4-18ae0022c13c",
+    "X-Github-Hook-Id": "491622597",
+    "X-Github-Hook-Installation-Target-Id": "807808293",
+    "X-Github-Hook-Installation-Target-Type": "repository",
+    "X-Hub-Signature": "sha1=0e44dae9a9c979dc05d1f5846b06fe578e581533",
+    }
+
+
+def test_message_create(client, create_service):
+    headers = get_headers(create_service, request)
+    with mock_sends(GithubMessageV1):
+        response = client.post(f"/message/{create_service.uuid}", json=request, headers=headers)
+    assert response.status_code == 200
+
+
+@pytest.fixture
+def request_data():
+    fixtures_dir = pathlib.Path(__file__).parent.joinpath("fixtures")
+    with open(fixtures_dir.joinpath("payload.json")) as fh:
+        return fh.read().strip()
+
+
+def test_message_create_400(client, create_service, request_data):
+    headers = get_headers(create_service, request)
+    headers["X-Hub-Signature-256"] = ""
+    response = client.post(f"/message/{create_service.uuid}", data=request_data, headers=headers)
+    assert response.status_code == 400
+
+
+def test_message_create_404(client):
+    response = client.post(f"/message/non-exsitent", json={})
+    assert response.status_code == 404
+
+
+def test_message_create_415(client, create_service):
+    response = client.post(f"/message/{create_service.uuid}", data="not json")
+    assert response.status_code == 415
+

webhook_to_fedora_messaging/endpoints/message.py

@@ -6,12 +6,14 @@
 from webhook_to_fedora_messaging.exceptions import SignatureMatchError
 from sqlalchemy import select
 from sqlalchemy.exc import NoResultFound
+from .util import validate_request
 
 
 message_endpoint = Blueprint("message_endpoint", __name__)
 
 
 @message_endpoint.route("/<service_uuid>", methods=["POST"])
+@validate_request([])
 def create_msg(service_uuid):
     """
     Used for creating a new message by sending a post request to /message path

webhook_to_fedora_messaging/endpoints/util.py

@@ -3,7 +3,7 @@
 
 
 def validate_request(fields=None):
-    fields = fields or ['username']
+    fields = fields if fields is not None else ['username']
 
     def decorator(func):
         @wraps(func)