Overwrite row with identical job id as pending (#142)

Author: evanugarte

printer/modules/sqlite_helpers.py

@@ -35,14 +35,27 @@ def maybe_create_table(sqlite_file: str) -> bool:
 
 def insert_print_job(sqlite_file: str, job_id: str):
     try:
-        with sqlite3.connect(sqlite_file, timeout=10.0) as db:
-            cursor = db.cursor()
-            timestamp = datetime.datetime.now()
-            db = sqlite3.connect(sqlite_file)
-            sql = "INSERT INTO logs (job_id) VALUES (?)"
+        with sqlite3.connect(sqlite_file, timeout=10.0) as conn:
+            cursor = conn.cursor()
+
+            """
+            yes i know we could do 
+
+            INSERT INTO logs (job_id, status, date) 
+            VALUES (?, 'created', CURRENT_TIMESTAMP)
+            ON CONFLICT(job_id) DO UPDATE SET 
+                status = excluded.status,
+                date = excluded.date;
+
+            but i dont care i dont feel like updating the schema on the real server
+            """
+            cursor.execute("DELETE FROM logs WHERE job_id = ?", (job_id,))
+
+            sql = "INSERT INTO logs (job_id, status, date) VALUES (?, 'created', CURRENT_TIMESTAMP)"
             cursor.execute(sql, (job_id,))
-            db.commit()
-            return timestamp
+
+            conn.commit()
+            return datetime.datetime.now()
     except sqlite3.IntegrityError:
         return None
     except Exception:

printer/test/test_lpstat.py

@@ -87,9 +87,15 @@ def test_poll_lpstat(
         self.assertEqual(str(cm.exception), "stop loop")
 
         # Check correct jobs marked as completed and acknowledged
-        mock_mark_completed.assert_called_once_with(
-            "dummy.db", ["HP_LaserJet_p2015dn_Right-53", "HP_LaserJet_p2015dn_Right-54"],
-        )
+        mock_mark_completed.assert_called_once()
+
+        # Extract the arguments from the call
+        # call_args is a tuple of (args, kwargs)
+        args, _ = mock_mark_completed.call_args
+        db_name, job_list = args
+
+        self.assertEqual(db_name, "dummy.db")
+        self.assertCountEqual(job_list, ["HP_LaserJet_p2015dn_Right-53", "HP_LaserJet_p2015dn_Right-54"])
 
 
 if __name__ == "__main__":

printer/test/test_sqlite.py

@@ -57,6 +57,45 @@ def test_insert_log(self):
         self.assertEqual(job_id, self.EXAMPLE_JOB_ID)
         self.assertEqual(status, "created")
 
+    def test_insert_log_overwrites_existing(self):
+        tmp = tempfile.NamedTemporaryFile(delete=False)
+        db_path = tmp.name
+        tmp.close()
+        sqlite_helpers.maybe_create_table(db_path)
+
+        date_in_2023 = '2023-12-24 23:50:00'
+        
+        with sqlite3.connect(db_path) as db:
+            cursor = db.cursor()
+            cursor.execute(
+                "INSERT INTO logs (job_id, status, date) VALUES (?, ?, '2023-12-24 23:50:00')",
+                (self.EXAMPLE_JOB_ID, "completed")
+            )
+            db.commit()
+
+        new_timestamp = sqlite_helpers.insert_print_job(db_path, self.EXAMPLE_JOB_ID)
+        self.assertIsNotNone(new_timestamp)
+
+        with sqlite3.connect(db_path) as db:
+            cursor = db.cursor()
+            
+            cursor.execute("SELECT date, job_id, status FROM logs WHERE job_id = ?", (self.EXAMPLE_JOB_ID,))
+            rows = cursor.fetchall()
+
+            self.assertEqual(len(rows), 1, "Database should not have duplicate job_ids")
+
+            (db_date_str, _, db_status) = rows[0]
+
+            self.assertEqual(db_status, "created", "Status has been reset to 'created'")
+
+            self.assertNotEqual(db_date_str, date_in_2023, "The date should have been updated to now")
+            
+            current_year = str(datetime.datetime.now().year)
+            self.assertIn(current_year, db_date_str, f"Expected timestamp to contain {current_year}")
+            
+            # Cleanup
+            os.unlink(db_path)
+
     def test_mark_jobs_acknowledged(self):
         tmp = tempfile.NamedTemporaryFile(delete=False)
         db_path = tmp.name