more tests, more coverage

Author: Paul Philion

redmine/model.py

@@ -254,6 +254,10 @@ def set_field(self, fieldname:str, value):
     def asdict(self):
         return dataclasses.asdict(self)
 
+    @property
+    def json(self):
+        return json.dumps(self.asdict(), indent=4, default=vars)
+
 
 @dataclass
 class UserResult:

redmine/synctime.py

@@ -30,10 +30,15 @@ def parse_millis(timestamp:int) -> dt.datetime:
     return dt.datetime.fromtimestamp(timestamp, dt.timezone.utc)
 
 
+def ago(**kwargs) -> dt.datetime:
+    # so generate one for "three yeas ago"
+    return now() - dt.timedelta(**kwargs)
+
+
 def epoch_datetime() -> dt.datetime:
     # discord API fails when using 0 as a timestamp,
-    # so generate one for "three yeas ago"
-    return now() - dt.timedelta(days=3*365)
+    # so generate one for "ten years ago"
+    return ago(days=10*365)
 
 
 def parse_str(timestamp:str) -> dt.datetime:

redmine/tickets.py

@@ -329,9 +329,16 @@ def recycle(self, ticket:Ticket, team_id: int):
 
     def dusty(self) -> set[Ticket]:
         # tickets that are older than 7
-        tickets = set()
-        tickets.update(self.older_than(TICKET_DUSTY_AGE))
-        return tickets
+
+        # http://localhost/projects/scn/issues.json?v[status_id][]=2&op[updated_on]=%3Ct-&v[updated_on][]=7
+        #"status_id": "2"
+        #query = f"/issues.json?status_id=2&op[updated_on]=%3Ct-&v[updated_on][]=7&op[priority_id]=!&v[priority_id][]=14&include=children"
+        query = "/issues.json?set_filter=1&sort=id%3Adesc&f[]=priority_id&op[priority_id]=!&v[priority_id][]=14&f[]=status_id&op[status_id]=%3D&v[status_id][]=2&f[]=updated_on&op[updated_on]=%3Ct-&v[updated_on][]=7&f[]=&c[]=tracker&c[]=status&c[]=priority&c[]=subject&c[]=assigned_to&c[]=updated_on&group_by=&t[]="
+        response = self.session.get(query)
+        if response:
+            return TicketsResult(**response).issues
+        else:
+            return []
 
 
     def recyclable(self) -> set[Ticket]:

tests/mock_session.py

@@ -7,7 +7,7 @@
 from urllib.parse import urlparse
 
 from redmine.session import RedmineSession
-from redmine.model import User
+from redmine.model import User, Ticket, TicketsResult
 
 TEST_DATA = "data" # dir with test data
 
@@ -50,10 +50,32 @@ def _get(self, resource:str) -> dict:
             #super().get(query, impersonate_id)
 
 
+    def _cache(self, resource:str, item):
+        self.test_cache[resource] = json.dumps(item, indent=4, default=str)
+
+
     def cache_user(self, user:User):
-        data = {}
-        data['user'] = user.asdict()
-        self.test_cache[f"/users/{user.id}.json"] = json.dumps(data)
+        data = {
+            'user': user.asdict()
+        }
+        self._cache(f"/users/{user.id}.json", data)
+
+
+    def cache_ticket(self, ticket:Ticket):
+        data = {
+            'issue': ticket.asdict()
+        }
+        log.info(data)
+        self._cache(f"/issues/{ticket.id}.json", data)
+
+
+    def cache_results(self, tickets:list[Ticket]):
+        result = TicketsResult(
+            total_count=len(tickets),
+            limit=25,
+            offset=0,
+            issues=tickets)
+        self._cache("/issues.json", result.asdict())
 
 
     def get(self, query:str, impersonate_id:str|None=None):
@@ -80,24 +102,24 @@ def put(self, resource:str, data:str, impersonate_id:str|None=None) -> None:
         log.debug(f"PUT {path} -> {item}")
 
 
-    def post(self, resource: str, data:str, user_login: str|None = None, files: list|None = None) -> dict|None:
-        log.info(f"POST {resource}, data={data} user_login={user_login}")
-        item_id = self._next_id()
+    # def post(self, resource: str, data:str, user_login: str|None = None, files: list|None = None) -> dict|None:
+    #     log.info(f"POST {resource}, data={data} user_login={user_login}")
+    #     item_id = self._next_id()
 
-        path = urlparse(resource).path
+    #     path = urlparse(resource).path
 
-        name_a = path.rsplit('.', 1)
-        path = f"{name_a[0]}/{item_id}.{name_a[1]}"
+    #     name_a = path.rsplit('.', 1)
+    #     path = f"{name_a[0]}/{item_id}.{name_a[1]}"
 
-        data_dict = json.loads(data)
-        for _, v in data_dict.items():
-            v['id'] = item_id
-        value = json.dumps(data_dict)
-        log.debug(f"POST {path} -> {value}")
-        self.test_cache[path] = value
-
-        return data_dict
-        #raise RedmineException(f"POST failed, status=[{r.status_code}] {r.reason}", r.headers['X-Request-Id'])
+    #     data_dict = json.loads(data)
+    #     for _, v in data_dict.items():
+    #         v['id'] = item_id
+    #     value = json.dumps(data_dict)
+    #     log.debug(f"POST {path} -> {value}")
+    #     self.test_cache[path] = value
+
+    #     return data_dict
+    #     #raise RedmineException(f"POST failed, status=[{r.status_code}] {r.reason}", r.headers['X-Request-Id'])
 
 
     def delete(self, resource: str) -> None:

tests/test_tickets.py

@@ -1,32 +1,60 @@
 #!/usr/bin/env python3
 """Redmine tickets manager test cases"""
 
+import datetime
 import unittest
 import logging
 import json
 from unittest.mock import MagicMock, patch
 
 from dotenv import load_dotenv
 
-from redmine.model import ParentTicket, synctime
+from redmine.model import ParentTicket, synctime, TicketStatus
 from redmine.tickets import TICKET_DUSTY_AGE, TICKET_MAX_AGE
 
 from tests import test_utils
 
-
 log = logging.getLogger(__name__)
 
 class TestTicketManager(test_utils.MockRedmineTestCase):
     """Mocked testing of ticket manager"""
 
     def test_dusty_tickets(self):
-        for ticket in self.tickets_mgr.dusty():
+        # to find dusty tickets, a dusty ticket needs to be created.
+        ticket1 = self.create_ticket(
+            status=TicketStatus(id=2,name="In Progress",is_closed=False),
+            updated_on=synctime.ago(days=TICKET_DUSTY_AGE+1),
+        )
+        # cache a search result
+        self.session.cache_results([ticket1])
+
+        self.assertGreaterEqual(synctime.age(ticket1.updated_on), datetime.timedelta(days=TICKET_DUSTY_AGE))
+
+        dusty_tickets = self.tickets_mgr.dusty()
+        self.assertGreaterEqual(len(dusty_tickets), 1, "No dusty tickets found")
+        for ticket in dusty_tickets:
             age = synctime.age(ticket.updated_on)
             self.assertGreaterEqual(age.days, TICKET_DUSTY_AGE)
+            self.assertEqual(ticket.status.name, "In Progress")
+
+
+    def test_recyclable_tickets(self):
+        # to find dusty tickets, a dusty ticket needs to be created.
+        ticket1 = self.create_ticket(
+            status=TicketStatus(id=2,name="In Progress",is_closed=False),
+            updated_on=synctime.ago(days=TICKET_MAX_AGE+1),
+        )
+        # cache a search result
+        self.session.cache_results([ticket1])
+
+        self.assertGreaterEqual(synctime.age(ticket1.updated_on), datetime.timedelta(days=TICKET_DUSTY_AGE))
 
-        for ticket in self.tickets_mgr.recyclable():
+        dusty_tickets = self.tickets_mgr.dusty()
+        self.assertGreaterEqual(len(dusty_tickets), 1, "No recyclable tickets found")
+        for ticket in dusty_tickets:
             age = synctime.age(ticket.updated_on)
             self.assertGreaterEqual(age.days, TICKET_MAX_AGE)
+            self.assertEqual(ticket.status.name, "In Progress")
 
 
     @unittest.skip # FIXME currently breaking mock testing

tests/test_utils.py

@@ -141,14 +141,13 @@ class MockRedmineTestCase(unittest.TestCase):
     @classmethod
     def setUpClass(cls):
         cls.tag:str = tagstr()
-        session = MockSession(cls.tag)
-        cls.redmine = Client.from_session(session, default_project=1)
+        cls.session = MockSession(cls.tag)
+        cls.redmine = Client.from_session(cls.session, default_project=1)
         cls.user_mgr = cls.redmine.user_mgr
         cls.tickets_mgr = cls.redmine.ticket_mgr
 
         cls.user:User = mock_user(cls.tag)
-        #session.test_cache[f"/users/{cls.user.id}.json"] = json.dumps(cls.user)
-        session.cache_user(cls.user)
+        cls.session.cache_user(cls.user)
         cls.user_mgr.cache.cache_user(cls.user)
         log.info(f"SETUP created mock user: {cls.user}")
 
@@ -167,8 +166,11 @@ def message_from(self, ticket: Ticket) -> Message:
         return message
 
 
-    def create_ticket(self) -> Ticket:
-        return self.redmine.create_ticket(self.user, self.create_message())
+    def create_ticket(self, **kwargs) -> Ticket:
+        # create a ticket from json template
+        ticket = mock_ticket(**kwargs)
+        self.session.cache_ticket(ticket)
+        return ticket
 
 
 class MockBotTestCase(MockRedmineTestCase, unittest.IsolatedAsyncioTestCase):