Fix crash when google creds are unavailable

pyproject.toml

@@ -7,3 +7,6 @@ line-length = 79
 # overlap with the use of a formatter, like Black, but we can override this behavior by
 # explicitly adding the rule.
 extend-select = ["E501"]
+
+[tool.ruff.lint.per-file-ignores]
+"webapp/tests/test_gdrive.py" = ["F401", "F811"]

webapp/gdrive.py

@@ -1,41 +1,62 @@
 import difflib
+from typing import Any, Optional
 
+from flask import Flask
 from google.oauth2 import service_account
 from googleapiclient.discovery import build
 from googleapiclient.errors import HttpError
 
+from webapp.models import Webpage
+
 
 class GoogleDriveClient:
     # If modifying these scopes, delete the file token.json.
     SCOPES = ["https://www.googleapis.com/auth/drive"]
 
     def __init__(
-        self, credentials, drive_folder_id=None, copydoc_template_id=None
+        self, credentials: dict, drive_folder_id: str, copydoc_template_id: str
     ):
-        self.credentials = self._get_credentials(credentials)
-        self.service = self._build_service()
+        self.service = self._build_service(credentials)
         self.GOOGLE_DRIVE_FOLDER_ID = drive_folder_id
         self.COPYD0C_TEMPLATE_ID = copydoc_template_id
 
-    def _get_credentials(self, credentials):
-        """Load credentials from an object."""
+    def _build_service(self, credentials: dict) -> Any:
+        """
+        Build a google drive service object.
+
+        Args:
+            credentials (dict): A dict containing google authentication keys.
 
-        return service_account.Credentials.from_service_account_info(
+        Returns:
+            googleapiclient.discovery.Resource: A google drive service object.
+        """
+        credentials = service_account.Credentials.from_service_account_info(
             credentials,
             scopes=self.SCOPES,
         )
-
-    def _build_service(self):
-        return build("drive", "v3", credentials=self.credentials)
+        return build("drive", "v3", credentials=credentials)
 
     def _item_exists(
         self,
-        folder_name,
+        folder_name: str,
         parent=None,
         mime_type="'application/vnd.google-apps.folder'",
-    ):
+    ) -> Optional[str]:
         """
-        Check whether an item exists in Google Drive.
+        Check whether an item exists in Google Drive. Optionally, check if the
+        item exists in a specific parent folder. If there are several matches,
+        we take the result closest to the site (project) name.
+
+        Args:
+            folder_name (str): The name of the folder to check.
+            parent (str): The parent folder to check in.
+            mime_type (str): The mime type of the item to check.
+
+        Returns:
+            str: The id of the item if it exists, otherwise None.
+
+        Raises:
+            ValueError: If an error occurs while querying the Google Drive API
         """
         query = (
             f"name = '{folder_name}' and "
@@ -59,18 +80,33 @@ def _item_exists(
             raise ValueError(f"An error occurred: Query:{query} Error:{error}")
 
         if data := results.get("files"):
-            # Get the closest match to the folder name, if there are several
             item_names = [item["name"] for item in data]
-            result = difflib.get_close_matches(folder_name, item_names)[0]
+            # Get the closest match to the folder name, or return None.
+            try:
+                result = difflib.get_close_matches(folder_name, item_names)[0]
+            except IndexError:
+                return None
+
             # Return the file id
             result_id = next(
                 item["id"] for item in data if item["name"] == result
             )
             return result_id
+        return None
 
-    def create_folder(self, name, parent):
+    def create_folder(self, name: str, parent: str) -> str:
         """
         Create a folder in the Google Drive.
+
+        Args:
+            name (str): The name of the folder to create.
+            parent (str): The parent folder to create the folder in.
+
+        Returns:
+            str: The id of the created folder.
+
+        Raises:
+            ValueError: If an error occurs while creating the folder.
         """
         try:
             folder_metadata = {
@@ -89,9 +125,17 @@ def create_folder(self, name, parent):
                 f"An error occurred when creating a new folder: {error}"
             )
 
-    def build_webpage_folder(self, webpage):
+    def build_webpage_folder(self, webpage) -> str:
         """
-        Create a folder hierarchy in Google Drive for a webpage.
+        Create a folder hierarchy in Google Drive for a webpage. The path is
+        derived from the webpage's URL, with each part of the path representing
+        a folder. The topmost folder will be the project name.
+
+        Args:
+            webpage (Webpage): The webpage object to create a folder hierarchy.
+
+        Returns:
+            str: The id of the folder, which is a leaf node in the hierarchy.
         """
         folders = [f"/{f}" for f in webpage.url.split("/")[:-1] if f != ""]
         # Check if the project folder exists, or create one
@@ -113,15 +157,25 @@ def build_webpage_folder(self, webpage):
         # Return the last parent folder
         return parent
 
-    def copy_file(self, fileID, name, parents):
+    def copy_file(self, fileID: str, name: str, parents: list[str]) -> dict:
         """
-        Create a copydoc from a template. The document is created in the folder
-        for the webpage project.
+        Copy a file in Google Drive.
+
+        Args:
+            fileID (str): The id of the file to copy.
+            name (str): The name to give the copied file.
+            parents (list[str]): Ids of folders to copy the file to.
+
+        Returns:
+            dict: The metadata of the copied file.
+
+        Raises:
+            ValueError: If an error occurs while copying the file.
         """
         try:
             copy_metadata = {
                 "name": name,
-                "parents": [parents],
+                "parents": parents,
                 "mimeType": "application/vnd.google-apps.file",
             }
             copy = (
@@ -138,10 +192,16 @@ def copy_file(self, fileID, name, parents):
                 f"An error occurred when copying copydoc template: {error}"
             )
 
-    def create_copydoc_from_template(self, webpage):
+    def create_copydoc_from_template(self, webpage: Webpage) -> dict:
         """
         Create a copydoc from a template. The document is created in the folder
         for the webpage project.
+
+        Args:
+            webpage (Webpage): The webpage object to create a copydoc for.
+
+        Returns:
+            dict: The metadata of the copied file.
         """
         # Create the folder hierarchy for the webpage
         webpage_folder = self.build_webpage_folder(webpage)
@@ -150,13 +210,16 @@ def create_copydoc_from_template(self, webpage):
         return self.copy_file(
             fileID=self.COPYD0C_TEMPLATE_ID,
             name=webpage.url,
-            parents=webpage_folder,
+            parents=[webpage_folder],
         )
 
 
-def init_gdrive(app):
-    app.config["gdrive"] = GoogleDriveClient(
-        credentials=app.config["GOOGLE_CREDENTIALS"],
-        drive_folder_id=app.config["GOOGLE_DRIVE_FOLDER_ID"],
-        copydoc_template_id=app.config["COPYD0C_TEMPLATE_ID"],
-    )
+def init_gdrive(app: Flask) -> None:
+    try:
+        app.config["gdrive"] = GoogleDriveClient(
+            credentials=app.config["GOOGLE_CREDENTIALS"],
+            drive_folder_id=app.config["GOOGLE_DRIVE_FOLDER_ID"],
+            copydoc_template_id=app.config["COPYD0C_TEMPLATE_ID"],
+        )
+    except Exception as error:
+        app.logger.info(f"Unable to initialize gdrive: {error}")

webapp/models.py

@@ -78,7 +78,7 @@ class Webpage(db.Model, DateTimeMixin):
     copy_doc_link: str = Column(String)
     parent_id: int = Column(Integer, ForeignKey("webpages.id"))
     owner_id: int = Column(Integer, ForeignKey("users.id"))
-    status: str = Column(Enum(WebpageStatus), default=WebpageStatus.NEW)
+    status: str = Column(Enum(WebpageStatus), default=WebpageStatus.AVAILABLE)
 
     project = relationship("Project", back_populates="webpages")
     owner = relationship("User", back_populates="webpages")

webapp/settings.py

@@ -1,7 +1,15 @@
 import base64
+import contextlib
 import os
+from binascii import Error
 from os import environ
 
+# Try to decode the private key from base64 before using it
+if private_key := environ.get("GOOGLE_PRIVATE_KEY"):
+    with contextlib.suppress(Error):
+        private_key = base64.b64decode(private_key).replace(b"\\n", b"\n")
+
+
 VALKEY_HOST = environ.get("VALKEY_HOST", "localhost")
 VALKEY_PORT = environ.get("VALKEY_PORT", 6379)
 REPO_ORG = environ.get("REPO_ORG", "https://github.com/canonical")
@@ -20,9 +28,7 @@
     "type": "service_account",
     "project_id": "web-engineering-436014",
     "private_key_id": environ.get("GOOGLE_PRIVATE_KEY_ID"),
-    "private_key": base64.b64decode(environ.get("GOOGLE_PRIVATE_KEY")).replace(
-        b"\\n", b"\n"
-    ),
+    "private_key": private_key,
     "client_email": "websites-copy-docs-627@web-engineering-436014.iam.gserviceaccount.com",  # noqa: E501
     "client_id": "116847960229506342511",
     "auth_uri": "https://accounts.google.com/o/oauth2/auth",

webapp/site_repository.py

@@ -384,6 +384,7 @@ def get_tree(self, no_cache: bool = False):
             if tree := self.get_tree_from_cache():
                 return tree
 
+        self.invalidate_cache()
         return self.get_new_tree()
 
     def __create_webpage_for_node__(
@@ -537,12 +538,3 @@ def add_pages_to_list(self, tree, page_list: list):
             # If child nodes exist, add their names to the list
             if child.get("children"):
                 self.add_pages_to_list(child, page_list)
-
-    def get_webpages(self):
-        """
-        Return a list of webpages from the associated parsed tree.
-        """
-        tree = self.get_tree()
-        webpages = []
-        self.add_pages_to_list(tree, webpages)
-        return webpages

webapp/tests/fixtures.py

@@ -0,0 +1,34 @@
+import pytest
+
+from webapp import create_app
+from webapp.models import Project, Webpage, db
+
+
+@pytest.fixture(scope="session")
+def db_session():
+    app = create_app()
+    with app.app_context():
+        db.create_all()
+        yield db.session
+        db.session.close()
+        db.drop_all()
+
+
+@pytest.fixture(scope="session")
+def project(db_session):
+    project = Project(name="ubuntu.com")
+    db_session.add(project)
+    db_session.commit()
+    return project
+
+
+@pytest.fixture(scope="session")
+def webpage(db_session, project):
+    webpage = Webpage(
+        name="/data/opensearch",
+        url="/data/opensearch",
+        project_id=project.id,
+    )
+    db_session.add(webpage)
+    db_session.commit()
+    return webpage