add message schema

Signed-off-by: Ryan Lerch <[email protected]>

pyproject.toml

@@ -38,7 +38,10 @@ ruff = ">=0.0.253"
 reuse = ">=1.1.0"
 
 [tool.poetry.plugins."fedora.messages"]
-"tahrir.new" = "tahrir_messages.thing:NewThingV1"
+"person.login.first" = "tahrir_messages:PersonLoginFirstV1"
+"badge.award" = "tahrir_messages:BadgeAwardV1"
+"person.rank.advance" = "tahrir_messages:PersonRankAdvanceV1"
+
 
 [tool.black]
 line-length = 100

tahrir_messages/__init__.py

@@ -2,4 +2,141 @@
 #
 # SPDX-License-Identifier: LGPL-3.0-or-later
 
-from .thing import NewThingV1
+import typing
+
+from fedora_messaging import message
+
+USER_SCHEMA = {
+    "type": "object",
+    "required": ["username", "badges_user_id"],
+    "properties": {
+        "username": {"type": "string"},
+        "badges_user_id": {"type": "integer"},
+    },
+}
+
+
+class TahrirMessage(message.Message):
+    @property
+    def app_name(self):
+        return "tahrir"
+
+    @property
+    def app_icon(self):
+        return "https://apps.fedoraproject.org/img/icons/badges.png"
+
+    @property
+    def usernames(self):
+        return [self.agent_name]
+
+    @property
+    def groups(self):
+        return []
+
+    @property
+    def url(self):
+        return None
+
+    def __str__(self):
+        return self.summary
+
+
+class PersonLoginFirstV1(TahrirMessage):
+    """The message sent when a user logs into tahrir for the first time"""
+
+    @property
+    def agent_name(self):
+        return self.body["user"]["username"]
+
+    @property
+    def summary(self):
+        return f"{self.agent_name} logged into badges for the first time"
+
+    topic = "person.login.first"
+    body_schema: typing.ClassVar = {
+        "id": "http://fedoraproject.org/message-schema/tahrir",
+        "$schema": "http://json-schema.org/draft-04/schema#",
+        "description": "The message sent when a user logs into tahrir for the first time",
+        "type": "object",
+        "required": ["user"],
+        "properties": {
+            "user": USER_SCHEMA,
+        },
+    }
+
+
+class BadgeAwardV1(TahrirMessage):
+    """The message sent when a badge is awarded"""
+
+    @property
+    def agent_name(self):
+        return self.body["user"]["username"]
+
+    @property
+    def summary(self):
+        return f"{self.agent_name} was awarded the badge `{self.body['badge']['name']}`"
+
+    badge_schema: typing.ClassVar = {
+        "type": "object",
+        "required": ["badge_id", "description", "image_url", "name"],
+        "properties": {
+            "badge_id": {"type": "string"},
+            "description": {"type": "string"},
+            "image_url": {"type": "string"},
+            "name": {"type": "string"},
+        },
+    }
+
+    topic = "person.login.first"
+    body_schema: typing.ClassVar = {
+        "id": "http://fedoraproject.org/message-schema/tahrir",
+        "$schema": "http://json-schema.org/draft-04/schema#",
+        "description": "The message sent when a user logs into tahrir for the first time",
+        "type": "object",
+        "required": ["user", "badge"],
+        "properties": {
+            "user": USER_SCHEMA,
+            "badge": badge_schema,
+        },
+    }
+
+
+class PersonRankAdvanceV1(TahrirMessage):
+    """The message sent when a user's rank changes"""
+
+    @property
+    def agent_name(self):
+        return self.body["person"]["nickname"]
+
+    @property
+    def summary(self):
+        return (
+            f"{self.agent_name}'s Badges rank changed from {self.body['old_rank']} "
+            f"to {self.body['person']['rank']}"
+        )
+
+    person_schema: typing.ClassVar = {
+        "type": "object",
+        "required": ["bio", "email", "id", "nickname", "rank", "website"],
+        "properties": {
+            "bio": {"type": ["string", "null"]},
+            "email": {"type": "string"},
+            "id": {"type": "integer"},
+            "nickname": {"type": "string"},
+            "rank": {"type": "integer"},
+            "website": {"type": ["string", "null"]},
+        },
+    }
+
+    topic = "person.login.first"
+    body_schema: typing.ClassVar = {
+        "id": "http://fedoraproject.org/message-schema/tahrir",
+        "$schema": "http://json-schema.org/draft-04/schema#",
+        "description": "The message sent when a user's rank changes",
+        "type": "object",
+        "required": ["old_rank", "person"],
+        "properties": {
+            "old_rank": {"type": "integer"},
+            "person": person_schema,
+        },
+    }

tests/test_messages.py

@@ -0,0 +1,88 @@
+# SPDX-FileCopyrightText: 2024 Contributors to the Fedora Project
+#
+# SPDX-License-Identifier: LGPL-3.0-or-later
+
+"""Unit tests for the message schema."""
+
+import pytest
+from jsonschema import ValidationError
+
+from tahrir_messages import BadgeAwardV1, PersonLoginFirstV1, PersonRankAdvanceV1
+
+from .utils import DUMMY_BADGE, DUMMY_PERSON, DUMMY_USER
+
+
+def test_person_login_first():
+    body = {
+        "user": DUMMY_USER,
+    }
+    message = PersonLoginFirstV1(body=body)
+    message.validate()
+
+    expected_summary = "dudemcpants logged into badges for the first time"
+    assert message.summary == expected_summary
+    assert str(message) == expected_summary
+    assert message.agent_name == "dudemcpants"
+    assert message.usernames == ["dudemcpants"]
+    assert message.app_name == "tahrir"
+    assert message.url is None
+    assert message.app_icon == "https://apps.fedoraproject.org/img/icons/badges.png"
+
+
+def test_person_login_first_missing_fields():
+    message = PersonLoginFirstV1(body={})
+    with pytest.raises(ValidationError):
+        message.validate()
+
+
+def test_badge_award():
+    body = {"user": DUMMY_USER, "badge": DUMMY_BADGE}
+    message = BadgeAwardV1(body=body)
+    message.validate()
+
+    expected_summary = "dudemcpants was awarded the badge `White Hat`"
+    assert message.summary == expected_summary
+    assert str(message) == expected_summary
+    assert message.agent_name == "dudemcpants"
+    assert message.usernames == ["dudemcpants"]
+    assert message.app_name == "tahrir"
+    assert message.url is None
+    assert message.app_icon == "https://apps.fedoraproject.org/img/icons/badges.png"
+
+
+def test_badge_award_missing_fields():
+    message = BadgeAwardV1(body={})
+    with pytest.raises(ValidationError):
+        message.validate()
+
+    body = {"badge": DUMMY_BADGE}
+    message = BadgeAwardV1(body=body)
+    with pytest.raises(ValidationError):
+        message.validate()
+
+    body = {"user": DUMMY_BADGE}
+    message = BadgeAwardV1(body=body)
+    with pytest.raises(ValidationError):
+        message.validate()
+
+
+def test_person_rank_advance():
+    body = {"old_rank": 2, "person": DUMMY_PERSON}
+    message = PersonRankAdvanceV1(body=body)
+    message.validate()
+
+    expected_summary = "aaronhale's Badges rank changed from 2 to 1"
+    assert message.summary == expected_summary
+    assert str(message) == expected_summary
+    assert message.agent_name == "aaronhale"
+    assert message.usernames == ["aaronhale"]
+    assert message.app_name == "tahrir"
+    assert message.url is None
+    assert message.app_icon == "https://apps.fedoraproject.org/img/icons/badges.png"
+
+
+def test_person_rank_advance_missing_fields():
+    """Assert an exception is actually raised on validation failure."""
+    message = BadgeAwardV1(body={})
+    with pytest.raises(ValidationError):
+        message.validate()