xunit: Create the `xfstestsdb xunit gc` command

This command is used to garbage collect the xunit table by removing
xunits that have no testcases. This could be expanded on later to add
more removal conditions, such as xunits older than some age.

Implements: #15 (`xfstestsdb xunit gc`)
Signed-off-by: Anna Schumaker <anna@nowheycreamery.com>

tests/xunit/test_gc.py

@@ -0,0 +1,70 @@
+# Copyright 2023 (c) Anna Schumaker.
+"""Tests the `xfstestsdb xunit gc` command."""
+import io
+import unittest
+import unittest.mock
+import xfstestsdb.xunit.gc
+import tests.xunit
+
+
+class TestXunitGC(unittest.TestCase):
+    """Tests the `xfstestsdb xunit gc` command."""
+
+    def setUp(self):
+        """Set up common variables."""
+        self.xfstestsdb = xfstestsdb.Command()
+        self.xunit = self.xfstestsdb.commands["xunit"]
+        self.gc = self.xunit.commands["gc"]
+
+    def setup_runs(self, mock_stdout: io.StringIO):
+        """Set up runs in the database and clear stdout."""
+        self.xfstestsdb.run(["new", "/dev/vda1"])
+        self.xfstestsdb.run(["new", "/dev/vda1"])
+
+        self.xfstestsdb.run(["xunit", "read", "1", str(tests.xunit.XUNIT_1)])
+        self.xfstestsdb.run(["xunit", "read", "1", str(tests.xunit.XUNIT_1),
+                             "--name", "test-2"])
+        self.xfstestsdb.run(["xunit", "read", "2", str(tests.xunit.XUNIT_1)])
+
+        self.xfstestsdb.sql("DELETE FROM testcases WHERE xunitid=?", 2)
+        self.xfstestsdb.sql("DELETE FROM testcases WHERE xunitid=?", 3)
+
+        mock_stdout.seek(0)
+        mock_stdout.truncate(0)
+
+    def test_init(self):
+        """Check that the xunit gc command was set up properly."""
+        self.assertIsInstance(self.gc, xfstestsdb.commands.Command)
+        self.assertIsInstance(self.gc, xfstestsdb.xunit.gc.Command)
+        self.assertEqual(self.xunit.subparser.choices["gc"],
+                         self.gc.parser)
+
+    @unittest.mock.patch("sys.stdout", new_callable=io.StringIO)
+    def test_gc_empty(self, mock_stdout: io.StringIO):
+        """Test garbage collecting an empty database."""
+        self.xfstestsdb.run(["xunit", "gc"])
+        self.assertEqual(mock_stdout.getvalue(), "")
+
+    @unittest.mock.patch("sys.stdout", new_callable=io.StringIO)
+    def test_gc_xunits(self, mock_stdout: io.StringIO):
+        """Test garbage collecting xunits with default options."""
+        self.setup_runs(mock_stdout)
+        self.xfstestsdb.run(["xunit", "gc"])
+        self.assertRegex(mock_stdout.getvalue(),
+                         "run #1 xunit 'test-2' has been deleted\n"
+                         "run #2 xunit 'test-1' has been deleted")
+
+        cur = self.xfstestsdb.sql("SELECT COUNT(*) FROM xunits")
+        self.assertEqual(cur.fetchone()["COUNT(*)"], 1)
+
+    @unittest.mock.patch("sys.stdout", new_callable=io.StringIO)
+    def test_gc_xunits_dry_run(self, mock_stdout: io.StringIO):
+        """Test garbage collecting xunits with the --dry-run option."""
+        self.setup_runs(mock_stdout)
+        self.xfstestsdb.run(["xunit", "gc", "--dry-run"])
+        self.assertRegex(mock_stdout.getvalue(),
+                         "run #1 xunit 'test-2' would be deleted\n"
+                         "run #2 xunit 'test-1' would be deleted")
+
+        cur = self.xfstestsdb.sql("SELECT COUNT(*) FROM xunits")
+        self.assertEqual(cur.fetchone()["COUNT(*)"], 3)

xfstestsdb/xunit/__init__.py

@@ -4,6 +4,7 @@ import argparse
 from .. import commands
 from .. import sqlite
 from . import delete
+from . import gc
 from . import list
 from . import properties
 from . import read
@@ -20,6 +21,7 @@ class Command(commands.Command):
                          help="xfstestsdb xunit commands")
         self.subparser = self.parser.add_subparsers(title="xunit commands")
         self.commands = {"delete": delete.Command(self.subparser, sql),
+                         "gc": gc.Command(self.subparser, sql),
                          "list": list.Command(self.subparser, sql),
                          "properties": properties.Command(self.subparser, sql),
                          "read": read.Command(self.subparser, sql),

xfstestsdb/xunit/gc.py

@@ -0,0 +1,29 @@
+# Copyright 2023 (c) Anna Schumaker.
+"""The `xfstestsdb xunit gc` command."""
+import argparse
+from .. import commands
+from .. import sqlite
+
+
+class Command(commands.Command):
+    """The `xfstestsdb xunit gc` command."""
+
+    def __init__(self, subparser: argparse.Action,
+                 sql: sqlite.Connection) -> None:
+        """Set up the xunit gc command."""
+        super().__init__(subparser, sql, "gc",
+                         help="garbage collect xunit entries")
+        self.parser.add_argument("--dry-run", action="store_true",
+                                 help="do not remove the xunit entries")
+
+    def do_command(self, args: argparse.Namespace) -> None:
+        """Clean up the xunit table."""
+        how = "would be" if args.dry_run else "has been"
+        cur = self.sql("""SELECT xunitid, runid, name FROM xunits
+                       WHERE NOT EXISTS (SELECT 1 FROM testcases
+                                         WHERE xunitid = xunits.xunitid)""")
+
+        for row in cur.fetchall():
+            if not args.dry_run:
+                self.sql("DELETE FROM xunits WHERE xunitid=?", row['xunitid'])
+            print(f"run #{row['runid']} xunit '{row['name']}' {how} deleted")