Import the fixtures into the test database for queries

lib/db_revision.py

@@ -18,6 +18,7 @@ class DBRevision:
             self.request_number,
             self.request_id,
         ) = row
+        self.rev = int(self.rev)
 
     def __str__(self):
         return f"Rev {self.project}/{self.rev} Md5 {self.unexpanded_srcmd5} {self.commit_time} {self.userid} {self.request_number}"
@@ -25,6 +26,16 @@ class DBRevision:
     def __repr__(self):
         return f"[{self.__str__()}]"
 
+    def __eq__(self, other):
+        return self.dbid == other.dbid
+
+    def __lt__(self, other):
+        if self.project != other.project:
+            return self.project < other.project
+        if self.package != other.package:
+            return self.package < other.package
+        return self.rev < other.rev
+
     def as_dict(self, db):
         """Return a dict we can put into YAML for test cases"""
         ret = {
@@ -164,3 +175,55 @@ class DBRevision:
                 (project, package),
             )
             return [row[0] for row in cur.fetchall()]
+
+    @staticmethod
+    def import_fixture_dict(db, rev_dict):
+        """Used in test cases to read a revision from fixtures into the test database"""
+        with db.cursor() as cur:
+            cur.execute(
+                """INSERT INTO revisions (project, package, rev, unexpanded_srcmd5, expanded_srcmd5, commit_time, userid, comment, broken)  
+                        VALUES(%s, %s, %s, %s, %s, %s, %s, %s, %s) RETURNING id""",
+                (
+                    rev_dict["project"],
+                    rev_dict["package"],
+                    rev_dict["rev"],
+                    rev_dict["unexpanded_srcmd5"],
+                    rev_dict["expanded_srcmd5"],
+                    rev_dict["commit_time"],
+                    rev_dict["userid"],
+                    rev_dict["comment"],
+                    rev_dict["broken"],
+                ),
+            )
+            rev_id = cur.fetchone()[0]
+            for file_dict in rev_dict["files"]:
+                cur.execute(
+                    "INSERT INTO files (md5, mtime, name, size, revision_id) VALUES(%s, %s, %s, %s, %s)",
+                    (
+                        file_dict["md5"],
+                        file_dict["mtime"],
+                        file_dict["name"],
+                        file_dict["size"],
+                        rev_id,
+                    ),
+                )
+            request = rev_dict.get("request")
+            if request:
+                cur.execute(
+                    """INSERT INTO requests (creator, number, source_project, source_package,
+                            source_rev, state, type) VALUES (%s, %s, %s, %s, %s, %s, %s) RETURNING id""",
+                    (
+                        request["creator"],
+                        request["number"],
+                        request.get("source_project"),
+                        request.get("source_package"),
+                        request.get("source_rev"),
+                        request["state"],
+                        request["type"],
+                    ),
+                )
+                request_id = cur.fetchone()[0]
+                cur.execute(
+                    "UPDATE revisions SET request_id=%s, request_number=%s WHERE id=%s",
+                    (request_id, request["number"], rev_id),
+                )

tests/tree_test.py

@@ -0,0 +1,26 @@
+import os
+import unittest
+from pprint import pprint
+
+import yaml
+
+from lib.db import DB
+from lib.db_revision import DBRevision
+
+
+class TestTreeMethods(unittest.TestCase):
+    def setUp(self):
+        self.db = DB(section="test")
+        path = os.path.join(os.path.dirname(__file__), "fixtures/zsh-data.yaml")
+        with open(path, "r") as f:
+            zsh_data = yaml.safe_load(f)
+        for rev in zsh_data["revisions"]:
+            DBRevision.import_fixture_dict(self.db, rev)
+
+    def test_create_tree(self):
+        factory_revisions = DBRevision.all_revisions(self.db, "openSUSE:Factory", "zsh")
+        pprint(sorted(factory_revisions))
+
+
+if __name__ == "__main__":
+    unittest.main()