Merge pull request #7653 from tdesveaux/typing/db/test_result_sets

typing: Add TestResultSetModel dataclass

master/buildbot/data/test_result_sets.py

@@ -13,28 +13,33 @@
 #
 # Copyright Buildbot Team Members
 
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
 
 from twisted.internet import defer
 
 from buildbot.data import base
 from buildbot.data import types
 
+if TYPE_CHECKING:
+    from buildbot.db.test_result_sets import TestResultSetModel
+
 
 class Db2DataMixin:
-    def db2data(self, dbdict):
-        data = {
-            'test_result_setid': dbdict['id'],
-            'builderid': dbdict['builderid'],
-            'buildid': dbdict['buildid'],
-            'stepid': dbdict['stepid'],
-            'description': dbdict['description'],
-            'category': dbdict['category'],
-            'value_unit': dbdict['value_unit'],
-            'tests_passed': dbdict['tests_passed'],
-            'tests_failed': dbdict['tests_failed'],
-            'complete': bool(dbdict['complete']),
+    def db2data(self, model: TestResultSetModel):
+        return {
+            'test_result_setid': model.id,
+            'builderid': model.builderid,
+            'buildid': model.buildid,
+            'stepid': model.stepid,
+            'description': model.description,
+            'category': model.category,
+            'value_unit': model.value_unit,
+            'tests_passed': model.tests_passed,
+            'tests_failed': model.tests_failed,
+            'complete': model.complete,
         }
-        return defer.succeed(data)
 
 
 class TestResultSetsEndpoint(Db2DataMixin, base.BuildNestingMixin, base.Endpoint):
@@ -77,10 +82,7 @@ def get(self, resultSpec, kwargs):
                 builderid, complete=complete, result_spec=resultSpec
             )
 
-        results = []
-        for dbdict in sets:
-            results.append((yield self.db2data(dbdict)))
-        return results
+        return [self.db2data(model) for model in sets]
 
 
 class TestResultSetEndpoint(Db2DataMixin, base.BuildNestingMixin, base.Endpoint):
@@ -91,8 +93,8 @@ class TestResultSetEndpoint(Db2DataMixin, base.BuildNestingMixin, base.Endpoint)
 
     @defer.inlineCallbacks
     def get(self, resultSpec, kwargs):
-        dbdict = yield self.master.db.test_result_sets.getTestResultSet(kwargs['test_result_setid'])
-        return (yield self.db2data(dbdict)) if dbdict else None
+        model = yield self.master.db.test_result_sets.getTestResultSet(kwargs['test_result_setid'])
+        return self.db2data(model) if model else None
 
 
 class TestResultSet(base.ResourceType):

master/buildbot/data/test_results.py

@@ -57,7 +57,7 @@ def get(self, resultSpec, kwargs):
             return []
 
         result_dbdicts = yield self.master.db.test_results.getTestResults(
-            set_dbdict['builderid'], kwargs['test_result_setid'], result_spec=resultSpec
+            set_dbdict.builderid, kwargs['test_result_setid'], result_spec=resultSpec
         )
 
         return [self.db2data(result) for result in result_dbdicts]

master/buildbot/db/test_result_sets.py

@@ -13,12 +13,51 @@
 #
 # Copyright Buildbot Team Members
 
+from __future__ import annotations
+
+from dataclasses import dataclass
+
 from twisted.internet import defer
+from twisted.python import deprecate
+from twisted.python import versions
 
 from buildbot.db import base
+from buildbot.warnings import warn_deprecated
+
+
+@dataclass
+class TestResultSetModel:
+    id: int
+    builderid: int
+    buildid: int
+    stepid: int
+    description: str | None
+    category: str
+    value_unit: str
+    tests_passed: int | None
+    tests_failed: int | None
+    complete: bool = False
+
+    # For backward compatibility
+    def __getitem__(self, key: str):
+        warn_deprecated(
+            '4.1.0',
+            (
+                'TestResultSetsConnectorComponent '
+                'getTestResultSet, and getTestResultSets '
+                'no longer return TestResultSet as dictionnaries. '
+                'Usage of [] accessor is deprecated: please access the member directly'
+            ),
+        )
+
+        if hasattr(self, key):
+            return getattr(self, key)
+
+        raise KeyError(key)
 
 
-class TestResultSetDict(dict):
+@deprecate.deprecated(versions.Version("buildbot", 4, 1, 0), TestResultSetModel)
+class TestResultSetDict(TestResultSetModel):
     pass
 
 
@@ -27,10 +66,11 @@ class TestResultSetAlreadyCompleted(Exception):
 
 
 class TestResultSetsConnectorComponent(base.DBConnectorComponent):
-    @defer.inlineCallbacks
-    def addTestResultSet(self, builderid, buildid, stepid, description, category, value_unit):
+    def addTestResultSet(
+        self, builderid, buildid, stepid, description, category, value_unit
+    ) -> defer.Deferred[int]:
         # Returns the id of the new result set
-        def thd(conn):
+        def thd(conn) -> int:
             sets_table = self.db.model.test_result_sets
 
             insert_values = {
@@ -47,28 +87,29 @@ def thd(conn):
             r = conn.execute(q)
             return r.inserted_primary_key[0]
 
-        res = yield self.db.pool.do(thd)
-        return res
+        return self.db.pool.do(thd)
 
-    @defer.inlineCallbacks
-    def getTestResultSet(self, test_result_setid):
-        def thd(conn):
+    def getTestResultSet(self, test_result_setid: int) -> defer.Deferred[TestResultSetModel | None]:
+        def thd(conn) -> TestResultSetModel | None:
             sets_table = self.db.model.test_result_sets
             q = sets_table.select().where(sets_table.c.id == test_result_setid)
             res = conn.execute(q)
             row = res.fetchone()
             if not row:
                 return None
-            return self._thd_row2dict(conn, row)
+            return self._model_from_row(row)
 
-        res = yield self.db.pool.do(thd)
-        return res
+        return self.db.pool.do(thd)
 
-    @defer.inlineCallbacks
     def getTestResultSets(
-        self, builderid, buildid=None, stepid=None, complete=None, result_spec=None
-    ):
-        def thd(conn):
+        self,
+        builderid: int,
+        buildid: int | None = None,
+        stepid: int | None = None,
+        complete: bool | None = None,
+        result_spec=None,
+    ) -> defer.Deferred[list[TestResultSetModel]]:
+        def thd(conn) -> list[TestResultSetModel]:
             sets_table = self.db.model.test_result_sets
             q = sets_table.select().where(sets_table.c.builderid == builderid)
             if buildid is not None:
@@ -78,16 +119,16 @@ def thd(conn):
             if complete is not None:
                 q = q.where(sets_table.c.complete == (1 if complete else 0))
             if result_spec is not None:
-                return result_spec.thd_execute(conn, q, lambda x: self._thd_row2dict(conn, x))
+                return result_spec.thd_execute(conn, q, self._model_from_row)
             res = conn.execute(q)
-            return [self._thd_row2dict(conn, row) for row in res.fetchall()]
+            return [self._model_from_row(row) for row in res.fetchall()]
 
-        res = yield self.db.pool.do(thd)
-        return res
+        return self.db.pool.do(thd)
 
-    @defer.inlineCallbacks
-    def completeTestResultSet(self, test_result_setid, tests_passed=None, tests_failed=None):
-        def thd(conn):
+    def completeTestResultSet(
+        self, test_result_setid, tests_passed=None, tests_failed=None
+    ) -> defer.Deferred[None]:
+        def thd(conn) -> None:
             sets_table = self.db.model.test_result_sets
 
             values = {'complete': 1}
@@ -106,10 +147,10 @@ def thd(conn):
                     f'is already completed or does not exist'
                 )
 
-        yield self.db.pool.do(thd)
+        return self.db.pool.do(thd)
 
-    def _thd_row2dict(self, conn, row):
-        return TestResultSetDict(
+    def _model_from_row(self, row):
+        return TestResultSetModel(
             id=row.id,
             builderid=row.builderid,
             buildid=row.buildid,

master/buildbot/test/fakedb/test_result_sets.py

@@ -13,9 +13,12 @@
 #
 # Copyright Buildbot Team Members
 
+from __future__ import annotations
+
 from twisted.internet import defer
 
 from buildbot.db.test_result_sets import TestResultSetAlreadyCompleted
+from buildbot.db.test_result_sets import TestResultSetModel
 from buildbot.test.fakedb.base import FakeDBComponent
 from buildbot.test.fakedb.row import Row
 
@@ -79,21 +82,33 @@ def addTestResultSet(self, builderid, buildid, stepid, description, category, va
         }
         return defer.succeed(id)
 
-    def _row2dict(self, row):
-        row = row.copy()
-        row['complete'] = bool(row['complete'])
-        return row
+    def _model_from_row(self, row):
+        return TestResultSetModel(
+            id=row['id'],
+            builderid=row['builderid'],
+            buildid=row['buildid'],
+            stepid=row['stepid'],
+            description=row['description'],
+            category=row['category'],
+            value_unit=row['value_unit'],
+            tests_passed=row['tests_passed'],
+            tests_failed=row['tests_failed'],
+            complete=bool(row['complete']),
+        )
 
-    # returns a Deferred
-    def getTestResultSet(self, test_result_setid):
+    def getTestResultSet(self, test_result_setid: int) -> defer.Deferred[TestResultSetModel | None]:
         if test_result_setid not in self.result_sets:
             return defer.succeed(None)
-        return defer.succeed(self._row2dict(self.result_sets[test_result_setid]))
+        return defer.succeed(self._model_from_row(self.result_sets[test_result_setid]))
 
-    # returns a Deferred
     def getTestResultSets(
-        self, builderid, buildid=None, stepid=None, complete=None, result_spec=None
-    ):
+        self,
+        builderid: int,
+        buildid: int | None = None,
+        stepid: int | None = None,
+        complete: bool | None = None,
+        result_spec=None,
+    ) -> defer.Deferred[list[TestResultSetModel]]:
         ret = []
         for row in self.result_sets.values():
             if row['builderid'] != builderid:
@@ -104,7 +119,7 @@ def getTestResultSets(
                 continue
             if complete is not None and row['complete'] != complete:
                 continue
-            ret.append(self._row2dict(row))
+            ret.append(self._model_from_row(row))
 
         if result_spec is not None:
             ret = self.applyResultSpec(ret, result_spec)

master/buildbot/test/unit/data/test_test_result_sets.py

@@ -18,6 +18,7 @@
 from twisted.trial import unittest
 
 from buildbot.data import test_result_sets
+from buildbot.db.test_result_sets import TestResultSetModel
 from buildbot.test import fakedb
 from buildbot.test.fake import fakemaster
 from buildbot.test.reactor import TestReactorMixin
@@ -198,18 +199,18 @@ def test_add_test_result_set(self):
         result = yield self.master.db.test_result_sets.getTestResultSet(test_result_setid)
         self.assertEqual(
             result,
-            {
-                'id': test_result_setid,
-                'builderid': 1,
-                'buildid': 2,
-                'stepid': 3,
-                'description': 'desc',
-                'category': 'cat4',
-                'value_unit': 'ms',
-                'tests_passed': None,
-                'tests_failed': None,
-                'complete': False,
-            },
+            TestResultSetModel(
+                id=test_result_setid,
+                builderid=1,
+                buildid=2,
+                stepid=3,
+                description='desc',
+                category='cat4',
+                value_unit='ms',
+                tests_passed=None,
+                tests_failed=None,
+                complete=False,
+            ),
         )
 
     @defer.inlineCallbacks
@@ -240,18 +241,18 @@ def test_complete_test_result_set_no_results(self):
         result = yield self.master.db.test_result_sets.getTestResultSet(test_result_setid)
         self.assertEqual(
             result,
-            {
-                'id': test_result_setid,
-                'builderid': 1,
-                'buildid': 2,
-                'stepid': 3,
-                'description': 'desc',
-                'category': 'cat4',
-                'value_unit': 'ms',
-                'tests_passed': None,
-                'tests_failed': None,
-                'complete': True,
-            },
+            TestResultSetModel(
+                id=test_result_setid,
+                builderid=1,
+                buildid=2,
+                stepid=3,
+                description='desc',
+                category='cat4',
+                value_unit='ms',
+                tests_passed=None,
+                tests_failed=None,
+                complete=True,
+            ),
         )
 
     @defer.inlineCallbacks
@@ -282,16 +283,16 @@ def test_complete_test_result_set_with_results(self):
         result = yield self.master.db.test_result_sets.getTestResultSet(test_result_setid)
         self.assertEqual(
             result,
-            {
-                'id': test_result_setid,
-                'builderid': 1,
-                'buildid': 2,
-                'stepid': 3,
-                'description': 'desc',
-                'category': 'cat4',
-                'value_unit': 'ms',
-                'tests_passed': 12,
-                'tests_failed': 34,
-                'complete': True,
-            },
+            TestResultSetModel(
+                id=test_result_setid,
+                builderid=1,
+                buildid=2,
+                stepid=3,
+                description='desc',
+                category='cat4',
+                value_unit='ms',
+                tests_passed=12,
+                tests_failed=34,
+                complete=True,
+            ),
         )