Formatting

Author: val500

server/src/testflinger/api/v1.py

@@ -102,26 +102,29 @@ def job_post(json_data: dict):
 @v1.input(schemas.Job, location="json")
 @v1.output(schemas.JobId)
 def agent_job_post(json_data: dict):
-    """Add a child job to the queue with inherited credentials from parent job."""
+    """
+    Add a child job to the queue with inherited credentials
+    from parent job.
+    """
     try:
         job_queue = json_data.get("job_queue")
         parent_job_id = json_data.get("parent_job_id")
     except (AttributeError, BadRequest):
         job_queue = ""
         parent_job_id = None
-    
+
     if not job_queue:
         abort(422, message="Invalid data or no job_queue specified")
-    
+
     if not parent_job_id:
         abort(422, message="parent_job_id required for agent job submission")
-    
+
     if not check_valid_uuid(parent_job_id):
         abort(400, message="Invalid parent_job_id specified")
-    
+
     # Retrieve parent job permissions for credential inheritance
     inherited_permissions = database.retrieve_parent_permissions(parent_job_id)
-    
+
     try:
         job = job_builder(json_data, inherited_permissions)
     except ValueError:
@@ -147,9 +150,10 @@ def has_attachments(data: dict) -> bool:
 
 def job_builder(data: dict, inherited_permissions: dict = None):
     """Build a job from a dictionary of data.
-    
+
     :param data: Job data dictionary
-    :param inherited_permissions: Optional permissions inherited from parent job
+    :param inherited_permissions: Optional permissions inherited from
+    parent job
     """
     job = {
         "created_at": datetime.now(timezone.utc),
@@ -173,9 +177,13 @@ def job_builder(data: dict, inherited_permissions: dict = None):
         data["attachments_status"] = "waiting"
 
     priority_level = data.get("job_priority", 0)
-    
-    # Use inherited permissions if provided, otherwise use current user's permissions
-    permissions_to_check = inherited_permissions if inherited_permissions else g.permissions
+
+    # Use inherited permissions if provided, otherwise use current
+    # user's permissions
+    if inherited_permissions:
+        permissions_to_check = inherited_permissions
+    else:
+        permissions_to_check = g.permissions
     auth.check_permissions(
         permissions_to_check,
         data,

server/src/testflinger/database.py

@@ -441,7 +441,7 @@ def check_client_exists(client_id: str) -> bool:
 
 def add_client_permissions(data: dict) -> None:
     """Create a new client_id along with its permissions.
-    :param data: client_id along with its permissions
+    :param data: client_id along with its permissions.
     """
     mongo.db.client_permissions.insert_one(data)
 
@@ -540,18 +540,19 @@ def delete_refresh_token(token: str) -> None:
     """
     mongo.db.refresh_tokens.delete_one({"refresh_token": token})
 
+
 def retrieve_parent_permissions(parent_job_id: str) -> dict:
     """Retrieve auth permissions from parent job for credential inheritance.
-    
+
     :param parent_job_id: UUID of the parent job to inherit permissions from.
-    :return: Dictionary with parent job's auth permissions, or empty dict if none.
+    :return: Dictionary with parent job's auth permissions,
+    or empty dict if none.
     """
     parent_job = mongo.db.jobs.find_one(
-        {"job_id": parent_job_id}, 
-        {"auth_permissions": True, "_id": False}
+        {"job_id": parent_job_id}, {"auth_permissions": True, "_id": False}
     )
-    
+
     if not parent_job:
         return {}
-    
+
     return parent_job.get("auth_permissions", {})

server/tests/test_v1_authorization.py

@@ -946,6 +946,7 @@ def test_role_hierachy_create_permissions(mongo_app_with_permissions):
     # Cleanup
     mongo.client_permissions.delete_one({"client_id": "test_manager"})
 
+
 def test_refresh_access_token(mongo_app_with_permissions):
     """Test refreshing an access token with a valid refresh token."""
     app, _, client_id, client_key, max_priority = mongo_app_with_permissions
@@ -1232,26 +1233,27 @@ def test_refresh_token_last_accessed_update(mongo_app_with_permissions):
 
     assert last_accessed_after > last_accessed_before
 
+
 def test_retrieve_parent_permissions_valid(mongo_app):
     """Test retrieving auth permissions from valid parent job."""
     from testflinger import database
-    
+
     app, mongo = mongo_app
     parent_permissions = {
         "client_id": "test_client",
         "max_priority": {"queue1": 100},
         "allowed_queues": ["restricted_queue"],
-        "max_reservation_time": {"queue1": 7200}
+        "max_reservation_time": {"queue1": 7200},
     }
-    
+
     # Insert parent job with permissions
     parent_job = {
         "job_id": "parent-123",
         "auth_permissions": parent_permissions,
-        "job_data": {"job_queue": "queue1"}
+        "job_data": {"job_queue": "queue1"},
     }
     mongo.jobs.insert_one(parent_job)
-    
+
     # Test retrieval
     retrieved_permissions = database.retrieve_parent_permissions("parent-123")
     assert retrieved_permissions == parent_permissions
@@ -1260,16 +1262,13 @@ def test_retrieve_parent_permissions_valid(mongo_app):
 def test_retrieve_parent_permissions_no_permissions(mongo_app):
     """Test retrieving permissions from parent job with no auth_permissions."""
     from testflinger import database
-    
+
     app, mongo = mongo_app
-    
+
     # Insert parent job without permissions
-    parent_job = {
-        "job_id": "parent-456", 
-        "job_data": {"job_queue": "queue1"}
-    }
+    parent_job = {"job_id": "parent-456", "job_data": {"job_queue": "queue1"}}
     mongo.jobs.insert_one(parent_job)
-    
+
     # Should return empty dict, not error
     retrieved_permissions = database.retrieve_parent_permissions("parent-456")
     assert retrieved_permissions == {}
@@ -1278,42 +1277,40 @@ def test_retrieve_parent_permissions_no_permissions(mongo_app):
 def test_retrieve_parent_permissions_nonexistent(mongo_app):
     """Test retrieving permissions from non-existent parent job."""
     from testflinger import database
-    
+
     app, mongo = mongo_app
-    
+
     # Should return empty dict, not error
-    retrieved_permissions = database.retrieve_parent_permissions("nonexistent-job")
+    retrieved_permissions = database.retrieve_parent_permissions(
+        "nonexistent-job"
+    )
     assert retrieved_permissions == {}
 
 
 def test_agent_jobs_endpoint_with_credentials(mongo_app_with_permissions):
     """Test agent endpoint submits child job with inherited credentials."""
     app, mongo, client_id, client_key, _ = mongo_app_with_permissions
-    
+
     # Create parent job with permissions
-    authenticate_output = app.post(
-        "/v1/oauth2/token",
-        headers=create_auth_header(client_id, client_key),
-	)
-    token = authenticate_output.data.decode("utf-8")
+    token = get_access_token(app, client_id, client_key)
     parent_job = {"job_queue": "myqueue2", "job_priority": 200}
     parent_response = app.post(
         "/v1/job", json=parent_job, headers={"Authorization": token}
     )
     parent_job_id = parent_response.json.get("job_id")
-    
+
     # Submit child job via agent endpoint
     child_job = {
-        "job_queue": "myqueue2", 
+        "job_queue": "myqueue2",
         "job_priority": 200,
-        "parent_job_id": parent_job_id
+        "parent_job_id": parent_job_id,
     }
     child_response = app.post("/v1/agent/jobs", json=child_job)
     assert child_response.status_code == 200
-    
+
     child_job_id = child_response.json.get("job_id")
     assert child_job_id is not None
-    
+
     # Verify child job inherited parent permissions
     child_job_data = mongo.jobs.find_one({"job_id": child_job_id})
     assert child_job_data["parent_job_id"] == parent_job_id
@@ -1323,7 +1320,7 @@ def test_agent_jobs_endpoint_with_credentials(mongo_app_with_permissions):
 def test_agent_jobs_endpoint_missing_parent_job_id(mongo_app):
     """Test agent endpoint rejects jobs without parent_job_id."""
     app, _ = mongo_app
-    
+
     child_job = {"job_queue": "myqueue"}
     response = app.post("/v1/agent/jobs", json=child_job)
     assert response.status_code == 422
@@ -1333,11 +1330,8 @@ def test_agent_jobs_endpoint_missing_parent_job_id(mongo_app):
 def test_agent_jobs_endpoint_invalid_parent_job_id(mongo_app):
     """Test agent endpoint rejects jobs with invalid parent_job_id."""
     app, _ = mongo_app
-    
-    child_job = {
-        "job_queue": "myqueue",
-        "parent_job_id": "invalid-uuid"
-    }
+
+    child_job = {"job_queue": "myqueue", "parent_job_id": "invalid-uuid"}
     response = app.post("/v1/agent/jobs", json=child_job)
     assert response.status_code == 400
     assert "Invalid parent_job_id" in response.get_json()["message"]
@@ -1346,20 +1340,17 @@ def test_agent_jobs_endpoint_invalid_parent_job_id(mongo_app):
 def test_agent_jobs_endpoint_no_parent_permissions(mongo_app):
     """Test agent endpoint works when parent has no permissions."""
     app, mongo = mongo_app
-    
+
     # Create parent job without authentication (no permissions)
     parent_job = {"job_queue": "myqueue"}
     parent_response = app.post("/v1/job", json=parent_job)
     parent_job_id = parent_response.json.get("job_id")
-    
+
     # Submit child job via agent endpoint
-    child_job = {
-        "job_queue": "myqueue",
-        "parent_job_id": parent_job_id
-    }
+    child_job = {"job_queue": "myqueue", "parent_job_id": parent_job_id}
     child_response = app.post("/v1/agent/jobs", json=child_job)
     assert child_response.status_code == 200
-    
+
     # Verify child job was created successfully
     child_job_id = child_response.json.get("job_id")
     child_job_data = mongo.jobs.find_one({"job_id": child_job_id})