t: add tests for --scrub

Problem: There exists no tests for the scrub_old_jobs() function.

Add some tests.

t/Makefile.am

@@ -36,7 +36,8 @@ TESTSCRIPTS = \
 	t1033-mf-priority-update-job.t \
 	t1034-mf-priority-config.t \
 	t5000-valgrind.t \
-	python/t1000-example.py
+	python/t1000-example.py \
+	python/t1001-scrub-old-jobs.py
 
 dist_check_SCRIPTS = \
 	$(TESTSCRIPTS) \

t/python/t1001-scrub-old-jobs.py

@@ -0,0 +1,127 @@
+#!/usr/bin/env python3
+
+###############################################################
+# Copyright 2024 Lawrence Livermore National Security, LLC
+# (c.f. AUTHORS, NOTICE.LLNS, COPYING)
+#
+# This file is part of the Flux resource manager framework.
+# For details, see https://github.com/flux-framework.
+#
+# SPDX-License-Identifier: LGPL-3.0
+###############################################################
+import unittest
+import os
+import sqlite3
+import sys
+
+from unittest import mock
+
+from fluxacct.accounting import job_archive_interface as jobs
+from fluxacct.accounting import create_db as c
+
+
+class TestAccountingCLI(unittest.TestCase):
+    # create accounting database and populate the jobs table with job records
+    @classmethod
+    def setUpClass(self):
+        c.create_db("FluxAccountingTest.db")
+        global conn
+        global cur
+        try:
+            conn = sqlite3.connect("file:FluxAccountingTest.db?mode=rw", uri=True)
+            cur = conn.cursor()
+        except sqlite3.OperationalError:
+            print(f"Unable to open test database file", file=sys.stderr)
+            sys.exit(-1)
+
+        # We choose 604801 for t_inactive here because scrub_old_jobs() will
+        # look for jobs that have finished at a minimum of 604800. We want to
+        # test calling the function when no old-enough jobs are found to make
+        # sure nothing gets deleted
+        userid = 9999
+        t_submit = t_run = 0
+        t_inactive = 604801
+        ranks = r = jobspec = ""
+        insert_stmt = "INSERT INTO jobs VALUES (?, ?, ?, ?, ?, ?, ?, ?)"
+
+        cur.execute(
+            insert_stmt,
+            (
+                "1",
+                userid,
+                t_submit,
+                t_run,
+                t_inactive,
+                ranks,
+                r,
+                jobspec,
+            ),
+        )
+        cur.execute(
+            insert_stmt,
+            (
+                "2",
+                userid,
+                t_submit,
+                t_run,
+                t_inactive,
+                ranks,
+                r,
+                jobspec,
+            ),
+        )
+        cur.execute(
+            insert_stmt,
+            (
+                "3",
+                userid,
+                t_submit,
+                t_run,
+                t_inactive,
+                ranks,
+                r,
+                jobspec,
+            ),
+        )
+
+        conn.commit()
+
+    # ensure jobs are added to jobs table
+    def test_01_check_jobs_table(self):
+        cur.execute("SELECT * FROM jobs")
+        rows = cur.fetchall()
+
+        self.assertEqual(len(rows), 3)
+
+    # if no old job records are found when scrubbing the job-archive,
+    # none are removed
+    def test_02_scrub_old_jobs_none_found(self):
+        jobs.scrub_old_jobs(conn, cur, num_weeks=1)
+        cur.execute("SELECT * FROM jobs")
+        rows = cur.fetchall()
+
+        self.assertEqual(len(rows), 3)
+
+    # if we find job records older than our cutoff date, remove them
+    # from the DB
+    def test_03_scrub_old_jobs_success(self):
+        jobs.scrub_old_jobs(conn, cur, num_weeks=2)
+        cur.execute("SELECT * FROM jobs")
+        rows = cur.fetchall()
+
+        self.assertEqual(len(rows), 0)
+
+    # remove database and log file
+    @classmethod
+    def tearDownClass(self):
+        conn.close()
+        os.remove("FluxAccountingTest.db")
+
+
+if __name__ == "__main__":
+    from pycotap import TAPTestRunner
+
+    unittest.main(testRunner=TAPTestRunner())
+
+
+# vi: ts=4 sw=4 expandtab