Merge pull request #235 from canonical/improve-server-logic

Improve server logic

server/hwapi/data_models/repository.py

@@ -16,9 +16,9 @@
 #
 # Written by:
 #        Nadzeya Hutsko <[email protected]>
+"""Functions for working with the DB using SQLAlchemy ORM"""
 
-
-from typing import Any
+from typing import Any, Sequence
 from sqlalchemy import select, and_, null
 from sqlalchemy.orm import Session, selectinload
 
@@ -77,9 +77,7 @@ def get_vendor_by_name(db: Session, name: str) -> models.Vendor | None:
     return db.execute(stmt).scalars().first()
 
 
-def get_board(
-    db: Session, vendor_name: str, product_name: str, version: str
-) -> models.Device | None:
+def get_board(db: Session, vendor_name: str, product_name: str) -> models.Device | None:
     """Return device object (category==BOARD) matching given board data"""
     stmt = (
         select(models.Device)
@@ -88,7 +86,6 @@ def get_board(
             and_(
                 models.Vendor.name.ilike(_clean_vendor_name(vendor_name)),
                 models.Device.name.ilike(product_name),
-                models.Device.version.ilike(version),
                 models.Device.category.in_(
                     [DeviceCategory.BOARD.value, DeviceCategory.OTHER.value]
                 ),
@@ -98,46 +95,44 @@ def get_board(
     return db.execute(stmt).scalars().first()
 
 
-def get_bios(
-    db: Session, vendor_name: str, version: str, firmware_revision: str | None
-) -> models.Bios | None:
-    """Return bios object matching given bios data"""
+def get_bios_list(db: Session, vendor_name: str, version: str) -> Sequence[models.Bios]:
+    """Return a list of bios objects matching the given vendor name and version"""
     stmt = (
         select(models.Bios)
         .join(models.Vendor)
         .where(
             and_(
-                models.Vendor.name.ilike(_clean_vendor_name(vendor_name)),
+                models.Vendor.name.ilike(_clean_vendor_name(f"%{vendor_name}%")),
                 models.Bios.version.ilike(version),
             )
         )
     )
 
-    if firmware_revision:
-        stmt = stmt.filter(models.Bios.firmware_revision.ilike(firmware_revision))
-
-    return db.execute(stmt).scalars().first()
+    return db.execute(stmt).scalars().all()
 
 
 def get_machine_with_same_hardware_params(
-    db: Session, arch: str, board: models.Device, bios: models.Bios | None
+    db: Session, arch: str, board: models.Device, bios_ids: list[int]
 ) -> models.Machine | None:
     """
-    Get a machines that have the given architecture, motherboard, and optionally bios
+    Get a machine that has the given architecture, motherboard, and one of the specified BIOSes.
     """
     stmt = (
         select(models.Machine)
         .select_from(models.Machine)
         .join(models.Certificate)
         .join(models.Report, models.Certificate.reports)
         .join(models.Device, models.Report.devices)
-        .filter(and_(models.Device.id == board.id, models.Report.architecture == arch))
+        .filter(
+            and_(
+                models.Device.id == board.id,
+                models.Report.architecture == arch,
+            )
+        )
     )
 
-    if bios:
-        stmt = stmt.join(models.Bios, models.Report.bios_id == models.Bios.id).filter(
-            models.Bios.id == bios.id
-        )
+    if bios_ids:
+        stmt = stmt.filter(models.Report.bios_id.in_(bios_ids))
     else:
         stmt = stmt.filter(models.Report.bios_id.is_(null()))
 
@@ -171,6 +166,26 @@ def get_machine_architecture(db: Session, machine_id: int) -> str:
     return result if result else ""
 
 
+def get_machine_bios(db: Session, machine_id: int) -> models.Bios | None:
+    """
+    Retrieve the BIOS associated with a given machine.
+
+    :param db: Database session
+    :param machine_id: ID of the machine
+    :return: BIOS object if found, None otherwise
+    """
+    stmt = (
+        select(models.Bios)
+        .join(models.Report, models.Bios.id == models.Report.bios_id)
+        .join(models.Certificate, models.Report.certificate_id == models.Certificate.id)
+        .join(models.Machine, models.Certificate.machine_id == models.Machine.id)
+        .where(models.Machine.id == machine_id)
+        .order_by(models.Certificate.created_at.desc())
+    )
+
+    return db.execute(stmt).scalars().first()
+
+
 def get_certificate_by_name(
     db: Session, machine_id: int, cert_name: str
 ) -> models.Certificate | None:

server/hwapi/endpoints/certification/certification.py

@@ -18,7 +18,7 @@
 #        Nadzeya Hutsko <[email protected]>
 """The endpoints for working with certification status"""
 
-
+import logging
 from fastapi import APIRouter, Depends
 from sqlalchemy.orm import Session
 
@@ -61,24 +61,39 @@ def check_certification(
     vendor = repository.get_vendor_by_name(db, system_info.vendor)
     if not vendor:
         return NotCertifiedResponse()
+
     # Match against board and bios
     try:
-        board, bios = logic.find_main_hardware_components(
-            db, system_info.board, system_info.bios
-        )
+        board = logic.find_board(db, system_info.board)
+        bioses = logic.find_bioses(db, system_info.bios) if system_info.bios else []
         related_machine = logic.find_certified_machine(
-            db, system_info.architecture, board, bios
+            db, system_info.architecture, board, bioses
         )
     except ValueError:
+        logging.error(
+            (
+                "Hardware cannot be found. Machine vendor: %s, model: %s"
+                ", board model: %s, board version: %s, bios version: %s"
+            ),
+            system_info.vendor,
+            system_info.model,
+            system_info.board.product_name,
+            system_info.board.version,
+            system_info.bios.version if system_info.bios else None,
+        )
         return NotCertifiedResponse()
+
+    bios = repository.get_machine_bios(db, related_machine.id)
     related_releases, kernels = repository.get_releases_and_kernels_for_machine(
         db, related_machine.id
     )
+
     # Match against CPU codename
     if not logic.check_cpu_compatibility(db, related_machine, system_info.processor):
         return response_builders.build_related_certified_response(
             db, related_machine, board, bios, related_releases, kernels
         )
+
     # Check OS release
     release_from_request = repository.get_release_object(
         db, system_info.os.version, system_info.os.codename

server/hwapi/endpoints/certification/logic.py

@@ -18,7 +18,6 @@
 #        Nadzeya Hutsko <[email protected]>
 """The algorithms for determining certification status"""
 
-
 from sqlalchemy.orm import Session
 
 from hwapi.data_models import repository, models
@@ -29,32 +28,35 @@
 )
 
 
-def find_main_hardware_components(
-    db: Session, board_data: BoardValidator, bios_data: BiosValidator | None
-) -> tuple[models.Device, models.Bios | None]:
+def find_board(db: Session, board_data: BoardValidator) -> models.Device:
     """
-    A function to get "main hardware components" like board and bios. Can be extended
-    in future
+    Find the board device based on the given board data.
+    Raises ValueError if the board is not found.
     """
-    board = repository.get_board(
-        db, board_data.manufacturer, board_data.product_name, board_data.version
-    )
+    board = repository.get_board(db, board_data.manufacturer, board_data.product_name)
     if not board:
-        raise ValueError("Hardware not certified")
-    if bios_data:
-        bios = repository.get_bios(
-            db, bios_data.vendor, bios_data.version, bios_data.firmware_revision
-        )
-        if not bios:
-            raise ValueError("Hardware not certified")
-        return board, bios
-    return board, None
+        raise ValueError("Hardware not certified: Board not found")
+    return board
+
+
+def find_bioses(db: Session, bios_data: BiosValidator) -> list[models.Bios]:
+    """
+    Find the BIOS list based on the given BIOS data.
+    Raises ValueError if no matching BIOS is found.
+    """
+    bios_list = repository.get_bios_list(db, bios_data.vendor, bios_data.version)
+    if not bios_list:
+        raise ValueError("Hardware not certified: BIOS not found")
+    return list(bios_list)
 
 
 def find_certified_machine(
-    db: Session, arch: str, board: models.Device, bios: models.Bios | None
+    db: Session, arch: str, board: models.Device, bios_list: list[models.Bios]
 ) -> models.Machine:
-    machine = repository.get_machine_with_same_hardware_params(db, arch, board, bios)
+    bios_ids = [bios.id for bios in bios_list] if bios_list else []
+    machine = repository.get_machine_with_same_hardware_params(
+        db, arch, board, bios_ids
+    )
     if not machine:
         raise ValueError("No certified machine matches the hardware specifications")
     return machine

server/hwapi/endpoints/certification/response_validators.py

@@ -18,7 +18,6 @@
 #        Nadzeya Hutsko <[email protected]>
 """Validator models for request/response bodies"""
 
-
 from typing import Literal
 from pydantic import BaseModel
 

server/hwapi/external/c3/client.py

@@ -192,7 +192,7 @@ def _load_certified_configurations_from_response(
             self.db,
             models.Release,
             codename=response.release.codename,
-            release=response.release.release,
+            release=response.release.release.replace("LTS", "").strip(),
             release_date=response.release.release_date,
             i_version=response.release.i_version,
             supported_until=response.release.supported_until,