Rework models

api/db.py

@@ -10,7 +10,7 @@
 from beanie import init_beanie
 from fastapi_pagination.ext.motor import paginate
 from motor import motor_asyncio
-from kernelci.api.models import Hierarchy, Node, Regression
+from kernelci.api.models import Hierarchy, Node, parse_node_obj
 from .models import User, UserGroup
 
 
@@ -25,7 +25,6 @@
     COLLECTIONS = {
         User: 'user',
         Node: 'node',
-        Regression: 'regression',
         UserGroup: 'usergroup',
     }
 
@@ -164,7 +163,7 @@
 
     async def _create_recursively(self, hierarchy: Hierarchy, parent: Node,
                                   cls, col):
-        obj, nodes = hierarchy.node, hierarchy.child_nodes
+        obj = parse_node_obj(hierarchy.node)
         if parent:
             obj.parent = parent.id
         if obj.id:
@@ -180,7 +179,7 @@
             obj.id = res.inserted_id
         obj = cls(**await col.find_one(ObjectId(obj.id)))
         obj_list = [obj]
-        for node in nodes:
+        for node in hierarchy.child_nodes:
             child_nodes = await self._create_recursively(node, obj, cls, col)
             obj_list.extend(child_nodes)
         return obj_list

api/main.py

@@ -10,7 +10,7 @@
 
 import os
 import re
-from typing import List, Union
+from typing import List
 from fastapi import (
     Depends,
     FastAPI,
@@ -27,12 +27,11 @@
 from pymongo.errors import DuplicateKeyError
 from fastapi_users import FastAPIUsers
 from beanie import PydanticObjectId
 from kernelci.api.models import (
     Node,
     Hierarchy,
-    Regression,
     PublishEvent,
-    get_model_from_kind
+    parse_node_obj,
 )
 from .auth import Authentication
 from .db import Database
@@ -428,13 +427,14 @@
     return {
         'op': operation,
         'id': str(node.id),
+        'kind': node.kind,
         'name': node.name,
         'path': node.path,
         'group': node.group,
         'state': node.state,
         'result': node.result,
-        'revision': node.revision.dict(),
         'owner': node.owner,
+        'data': node.data,
     }
 
 
@@ -447,13 +447,12 @@
     return translated
 
 
-@app.get('/node/{node_id}', response_model=Union[Regression, Node],
+@app.get('/node/{node_id}', response_model=Node,
          response_model_by_alias=False)
-async def get_node(node_id: str, kind: str = "node"):
+async def get_node(node_id: str):
     """Get node information from the provided node id"""
     try:
-        model = get_model_from_kind(kind)
-        return await db.find_by_id(model, node_id)
+        return await db.find_by_id(Node, node_id)
     except KeyError as error:
         raise HTTPException(
             status_code=status.HTTP_404_NOT_FOUND,
@@ -477,7 +476,7 @@
 
 
 @app.get('/nodes', response_model=PageModel)
-async def get_nodes(request: Request, kind: str = "node"):
+async def get_nodes(request: Request):
     """Get all the nodes if no request parameters have passed.
        Get all the matching nodes otherwise, within the pagination limit."""
     query_params = dict(request.query_params)
@@ -489,7 +488,9 @@
     query_params = await translate_null_query_params(query_params)
 
     try:
-        model = get_model_from_kind(kind)
+        # Query using the base Node model, regardless of the specific
+        # node type
+        model = Node
         translated_params = model.translate_fields(query_params)
         paginated_resp = await db.find_by_attributes(model, translated_params)
         paginated_resp.items = serialize_paginated_data(
@@ -505,15 +506,17 @@
 
 
 @app.get('/count', response_model=int)
-async def get_nodes_count(request: Request, kind: str = "node"):
+async def get_nodes_count(request: Request):
     """Get the count of all the nodes if no request parameters have passed.
        Get the count of all the matching nodes otherwise."""
     query_params = dict(request.query_params)
 
     query_params = await translate_null_query_params(query_params)
 
     try:
-        model = get_model_from_kind(kind)
+        # Query using the base Node model, regardless of the specific
+        # node type
+        model = Node
         translated_params = model.translate_fields(query_params)
         return await db.count(model, translated_params)
     except KeyError as error:
@@ -536,6 +539,10 @@
 async def post_node(node: Node,
                     current_user: User = Depends(get_current_user)):
     """Create a new node"""
+    # Explicit pydantic model validation
+    parse_node_obj(node)
+
+    # [TODO] Implement sanity checks depending on the node kind
     if node.parent:
         parent = await db.find_by_id(Node, node.parent)
         if not parent:
@@ -546,6 +553,9 @@
 
     await _verify_user_group_existence(node.user_groups)
     node.owner = current_user.username
+    # The node is handled as a generic Node by the DB, regardless of its
+    # specific kind. The concrete Node submodel (Kbuild, Checkout, etc.)
+    # is only used for data format validation
     obj = await db.create(node)
     data = _get_node_event_data('created', obj)
     attributes = {}
@@ -566,19 +576,27 @@
             status_code=status.HTTP_404_NOT_FOUND,
             detail=f"Node not found with id: {node.id}"
         )
-    is_valid, message = node_from_id.validate_node_state_transition(
+    # Sanity checks
+    # Note: do not update node ownership fields, don't update 'state'
+    # until we've checked the state transition is valid.
+    update_data = node.dict(exclude={'user', 'user_groups', 'state'})
+    new_node_def = node_from_id.copy(update=update_data)
+    # 1- Parse and validate node to specific subtype
+    specialized_node = parse_node_obj(new_node_def)
+
+    # 2 - State transition checks
+    is_valid, message = specialized_node.validate_node_state_transition(
         node.state)
     if not is_valid:
         raise HTTPException(
             status_code=status.HTTP_400_BAD_REQUEST,
             detail=message
         )
+    # Now we can update the state
+    new_node_def.state = node.state
 
-    # Do not update node ownership fields
-    update_data = node.dict(exclude={'user', 'user_groups'})
-    node = node_from_id.copy(update=update_data)
-
-    obj = await db.update(node)
+    # Update node in the DB
+    obj = await db.update(new_node_def)
     data = _get_node_event_data('updated', obj)
     attributes = {}
     if data.get('owner', None):
@@ -696,34 +714,6 @@
     return await pubsub.subscription_stats()
 
 
-# -----------------------------------------------------------------------------
-# Regression
-
-@app.post('/regression', response_model=Regression,
-          response_model_by_alias=False)
-async def post_regression(regression: Regression,
-                          user: str = Depends(get_current_user)):
-    """Create a new regression"""
-    obj = await db.create(regression)
-    operation = 'created'
-    await pubsub.publish_cloudevent('regression', {'op': operation,
-                                                   'id': str(obj.id)})
-    return obj
-
-
-@app.put('/regression/{regression_id}', response_model=Regression,
-         response_model_by_alias=False)
-async def put_regression(regression_id: str, regression: Regression,
-                         user: str = Depends(get_current_user)):
-    """Update an already added regression"""
-    regression.id = ObjectId(regression_id)
-    obj = await db.update(regression)
-    operation = 'updated'
-    await pubsub.publish_cloudevent('regression', {'op': operation,
-                                                   'id': str(obj.id)})
-    return obj
-
-
 versioned_app = VersionedFastAPI(
         app,
         version_format='{major}',

migrations/20231215122000_node_models.py

@@ -0,0 +1,137 @@
+# SPDX-License-Identifier: LGPL-2.1-or-later
+#
+# Copyright (C) 2023 Collabora Limited
+# Author: Ricardo Cañuelo <[email protected]>
+
+"""Migration for Node objects to comply with the models after commits:
+
+    api.models: basic definitions of Node submodels
+    api.main: use node endpoints for all type of Node subtypes
+    api.db: remove regression collection
+
+"""
+
+from bson.objectid import ObjectId
+
+name = '20231215122000_node_models'
+dependencies = ['20231102101356_user']
+
+
+def node_upgrade_needed(node):
+    """Checks if a DB Node passed as a parameter needs to be migrated
+    with this script.
+
+    Parameters:
+      user: a mongodb document (dict) defining a KernelCI Node
+
+    Returns:
+      True if the node needs to be migrated, False otherwise
+
+    """
+    # The existence of a 'revision' key seems to be enough to detect a
+    # pre-migration Node
+    if 'revision' in node:
+        return True
+    else:
+        return False
+
+
+def upgrade(db: "pymongo.database.Database"):
+    # Update nodes
+    nodes = db.node.find()
+    for node in nodes:
+        # Skip any node that's not in the old format
+        if not node_upgrade_needed(node):
+            continue
+        if not node.get('data'):
+            # Initialize 'data' field if it's empty: a generic Node
+            # with no specific type may have an emtpy 'data' field
+            db.node.update_one(
+                {'_id': node['_id']},
+                {'$set': {'data': {}}}
+            )
+        # move 'revision' to 'data.kernel_revision'
+        db.node.update_one(
+            {'_id': node['_id']},
+            {
+                '$set': {
+                    'data.kernel_revision': node['revision']
+                },
+                '$unset': {'revision': ''}
+            }
+        )
+
+    # Re-format regressions: move them from "regression" to "node"
+    regressions = db.regression.find()
+    for regression in regressions:
+        db.node.insert_one(
+            {
+                'name': regression.get('name'),
+                'group': regression.get('group'),
+                'path': regression.get('path'),
+                'kind': 'regression',
+                'data': {
+                    'pass_node': ObjectId(regression['regression_data'][0]),
+                    'fail_node': ObjectId(regression['regression_data'][1])
+                },
+                'artifacts': regression.get('artifacts'),
+                'created': regression.get('created'),
+                'updated': regression.get('updated'),
+                'timeout': regression.get('timeout'),
+                'owner': regression.get('owner'),
+            }
+        )
+        db.regression.delete_one({'_id': regression['_id']})
+
+
+def downgrade(db: 'pymongo.database.Database'):
+    # Move regressions back to "regression"
+    regressions = db.node.find({'kind': 'regression'})
+    for regression in regressions:
+        fail_node = db.node.find_one(
+            {'_id': ObjectId(regression['data']['fail_node'])}
+        )
+        db.regression.insert_one(
+            {
+                'name': regression.get('name'),
+                'group': regression.get('group'),
+                'path': regression.get('path'),
+                'kind': 'regression',
+                'parent': regression['data']['fail_node'],
+                'regression_data': [
+                    regression['data']['pass_node'],
+                    regression['data']['fail_node']
+                ],
+                'revision': fail_node['data']['kernel_revision'],
+                'artifacts': regression.get('artifacts'),
+                'created': regression.get('created'),
+                'updated': regression.get('updated'),
+                'timeout': regression.get('timeout'),
+                'owner': regression.get('owner'),
+            }
+        )
+        db.node.delete_one({'_id': regression['_id']})
+
+    # Downgrade node format
+    nodes = db.node.find()
+    for node in nodes:
+        # Skip any node that's already in the old format
+        if node_upgrade_needed(node):
+            continue
+        # move 'data.kernel_revision' to 'revision'
+        db.node.update_one(
+            {'_id': node['_id']},
+            {
+                '$set': {
+                    'revision': node['data']['kernel_revision']
+                },
+                '$unset': {'data.kernel_revision': ''}
+            }
+        )
+        # unset 'data' if it's empty
+        node['data'].pop('kernel_revision', None)
+        if len(node['data']) == 0:
+            db.node.update_one(
+                {'_id': node['_id']},
+                {'$unset': {'data': ''}}
+            )

tests/e2e_tests/test_pipeline.py

@@ -59,8 +59,8 @@ async def test_node_pipeline(test_async_client):
     await task_listen
     event_data = from_json(task_listen.result().json().get('data')).data
     assert event_data != 'BEEP'
-    keys = {'op', 'id', 'name', 'path', 'group', 'state', 'result', 'revision',
-            'owner'}
+    keys = {'op', 'id', 'kind', 'name', 'path',
+            'group', 'state', 'result', 'owner', 'data'}
     assert keys == event_data.keys()
     assert event_data.get('op') == 'created'
     assert event_data.get('id') == response.json()['id']
@@ -82,7 +82,7 @@ async def test_node_pipeline(test_async_client):
     await task_listen
     event_data = from_json(task_listen.result().json().get('data')).data
     assert event_data != 'BEEP'
-    keys = {'op', 'id', 'name', 'path', 'group', 'state', 'result', 'revision',
-            'owner'}
+    keys = {'op', 'id', 'kind', 'name', 'path',
+            'group', 'state', 'result', 'owner', 'data'}
     assert keys == event_data.keys()
     assert event_data.get('op') == 'updated'