CI: Better validate logging

test/validate_ci_marklog.py

@@ -1,28 +1,37 @@
-# Check the mark.log file that is generated by the CI to make sure it contains the expected values
 import argparse
 import os
+import sys
+from loguru import logger
+from collections import Counter
+
+
+class MarkLogError(Exception):
+    """Custom exception for mark.log validation failures."""
+
+    pass
 
 
 def parse_args():
     parser = argparse.ArgumentParser(
         description="Check the mark.log file that is generated by the CI to make sure it contains the expected values"
     )
-    parser.add_argument(
+    group = parser.add_mutually_exclusive_group(required=True)
+    group.add_argument(
         "--guids", action="store_true", help="Check the mark.log file for guids"
     )
-    parser.add_argument(
+    group.add_argument(
         "--locations", action="store_true", help="Check the mark.log file for locations"
     )
-    parser.add_argument(
+    group.add_argument(
         "--write", action="store_true", help="Check the mark.log file for write-run"
     )
-    parser.add_argument(
+    group.add_argument(
         "--plex", action="store_true", help="Check the mark.log file for Plex"
     )
-    parser.add_argument(
+    group.add_argument(
         "--jellyfin", action="store_true", help="Check the mark.log file for Jellyfin"
     )
-    parser.add_argument(
+    group.add_argument(
         "--emby", action="store_true", help="Check the mark.log file for Emby"
     )
 
@@ -31,51 +40,47 @@ def parse_args():
 
 def read_marklog():
     marklog = os.path.join(os.getcwd(), "mark.log")
-    with open(marklog, "r") as f:
-        lines = f.readlines()
-    return lines
+    try:
+        with open(marklog, "r") as f:
+            lines = [line.strip() for line in f if line.strip()]
+        return lines
+    except Exception as e:
+        raise MarkLogError(f"Error reading {marklog}: {e}")
 
 
 def check_marklog(lines, expected_values):
-    try:
-        # Check to make sure the marklog contains all the expected values and nothing else
-        found_values = []
-        for line in lines:
-            # Remove the newline character
-            line = line.strip()
-            if line not in expected_values:
-                raise Exception("Line not found in marklog: " + line)
-
-            found_values.append(line)
-
-        # Check to make sure the marklog contains the same number of values as the expected values
-        if len(found_values) != len(expected_values):
-            raise Exception(
-                "Marklog did not contain the same number of values as the expected values, found "
-                + str(len(found_values))
-                + " values, expected "
-                + str(len(expected_values))
-                + " values\n"
-                + "\n".join(found_values)
-            )
-
-        # Check that the two lists contain the same values
-        if sorted(found_values) != sorted(expected_values):
-            raise Exception(
-                "Marklog did not contain the same values as the expected values, found:\n"
-                + "\n".join(sorted(found_values))
-                + "\n\nExpected:\n"
-                + "\n".join(sorted(expected_values))
-            )
-
-        return True
-    except Exception as e:
-        print(e)
-        return False
+    found_counter = Counter(lines)
+    expected_counter = Counter(expected_values)
+
+    # Determine missing and extra items by comparing counts
+    missing = expected_counter - found_counter
+    extra = found_counter - expected_counter
+
+    if missing or extra:
+        if missing:
+            logger.error("Missing expected entries (with counts):")
+            for entry, count in missing.items():
+                logger.error(f"  {entry}: missing {count} time(s)")
+        if extra:
+            logger.error("Unexpected extra entries found (with counts):")
+            for entry, count in extra.items():
+                logger.error(f"  {entry}: found {count} extra time(s)")
+
+        logger.error(
+            f"Entry count mismatch: found {len(lines)} entries, expected {len(expected_values)} entries."
+        )
+        logger.error("Full mark.log content:")
+        for line in sorted(lines):
+            logger.error(f"  {line}")
+        raise MarkLogError("mark.log validation failed.")
+
+    return True
 
 
 def main():
     args = parse_args()
+
+    # Expected values defined for each check
     expected_jellyfin = [
         "Plex/JellyPlex-CI/jellyplex_watched/Custom Movies/Movie Two (2021)",
         "Plex/JellyPlex-CI/jellyplex_watched/Custom TV Shows/Greatest Show Ever 3000/Episode 2",
@@ -173,38 +178,39 @@ def main():
         "Jellyfin/Jellyfin-Server/JellyUser/Shows/Monarch: Legacy of Monsters/Parallels and Interiors/4",
     ]
 
-    # Expected values for the mark.log file, dry-run is slightly different than write-run
-    # due to some of the items being copied over from one server to another and now being there
-    # for the next server run.
+    # Determine which expected values to use based on the command-line flag
     if args.guids:
         expected_values = expected_guids
+        check_type = "GUIDs"
     elif args.locations:
         expected_values = expected_locations
+        check_type = "locations"
     elif args.write:
         expected_values = expected_write
+        check_type = "write-run"
     elif args.plex:
         expected_values = expected_plex
+        check_type = "Plex"
     elif args.jellyfin:
         expected_values = expected_jellyfin
+        check_type = "Jellyfin"
     elif args.emby:
         expected_values = expected_emby
+        check_type = "Emby"
     else:
-        print("No server specified")
-        exit(1)
-
-    lines = read_marklog()
-    if not check_marklog(lines, expected_values):
-        print("Failed to validate marklog")
-        for line in lines:
-            # Remove the newline character
-            line = line.strip()
+        raise MarkLogError("No server specified")
 
-            print(line)
+    logger.info(f"Validating mark.log for {check_type}...")
 
-        exit(1)
-
-    print("Successfully validated marklog")
-    exit(0)
+    try:
+        lines = read_marklog()
+        check_marklog(lines, expected_values)
+    except MarkLogError as e:
+        logger.error(e)
+        sys.exit(1)
+
+    logger.success("Successfully validated mark.log")
+    sys.exit(0)
 
 
 if __name__ == "__main__":