[DPE-2415] sync libraries and update src as necessary

lib/charms/mongodb/v0/helpers.py

@@ -1,16 +1,22 @@
 """Simple functions, which can be used in both K8s and VM charms."""
 # Copyright 2023 Canonical Ltd.
 # See LICENSE file for licensing details.
-
+import json
 import logging
 import os
 import secrets
 import string
 import subprocess
-from typing import List
+from typing import List, Optional, Union
 
 from charms.mongodb.v0.mongodb import MongoDBConfiguration, MongoDBConnection
-from ops.model import ActiveStatus, BlockedStatus, StatusBase, WaitingStatus
+from ops.model import (
+    ActiveStatus,
+    BlockedStatus,
+    MaintenanceStatus,
+    StatusBase,
+    WaitingStatus,
+)
 from pymongo.errors import AutoReconnect, ServerSelectionTimeoutError
 
 # The unique Charmhub library identifier, never change it
@@ -21,7 +27,7 @@
 
 # Increment this PATCH version before using `charmcraft publish-lib` or reset
 # to 0 if you are raising the major API version
-LIBPATCH = 7
+LIBPATCH = 8
 
 
 # path to store mongodb ketFile
@@ -194,3 +200,49 @@ def copy_licenses_to_unit():
     subprocess.check_output(
         "cp -r /snap/charmed-mongodb/current/licenses/* src/licenses", shell=True
     )
+
+
+_StrOrBytes = Union[str, bytes]
+
+
+def process_pbm_error(error_string: Optional[_StrOrBytes]) -> str:
+    """Parses pbm error string and returns a user friendly message."""
+    message = "couldn't configure s3 backup option"
+    if not error_string:
+        return message
+    if type(error_string) == bytes:
+        error_string = error_string.decode("utf-8")
+    if "status code: 403" in error_string:  # type: ignore
+        message = "s3 credentials are incorrect."
+    elif "status code: 404" in error_string:  # type: ignore
+        message = "s3 configurations are incompatible."
+    elif "status code: 301" in error_string:  # type: ignore
+        message = "s3 configurations are incompatible."
+    return message
+
+
+def current_pbm_op(pbm_status: str) -> str:
+    """Parses pbm status for the operation that pbm is running."""
+    pbm_status = json.loads(pbm_status)
+    return pbm_status["running"] if "running" in pbm_status else ""
+
+
+def process_pbm_status(pbm_status: str) -> StatusBase:
+    """Parses current pbm operation and returns unit status."""
+    current_op = current_pbm_op(pbm_status)
+    # no operations are currently running with pbm
+    if current_op == {}:
+        return ActiveStatus("")
+
+    if current_op["type"] == "backup":
+        backup_id = current_op["name"]
+        return MaintenanceStatus(f"backup started/running, backup id:'{backup_id}'")
+
+    if current_op["type"] == "restore":
+        backup_id = current_op["name"]
+        return MaintenanceStatus(f"restore started/running, backup id:'{backup_id}'")
+
+    if current_op["type"] == "resync":
+        return WaitingStatus("waiting to sync s3 configurations.")
+
+    return ActiveStatus()