LITE-28071 Retrieve deployment request data

connect_ext_ppr/utils.py

@@ -114,6 +114,11 @@ def get_product_items(client, prd_id):
     return client.products[prd_id].items.all()
 
 
+@connect_error
+def get_hub(client, hub_id):
+    return client.hubs[hub_id].get()
+
+
 def namespaced_media_client(client, account_id, deployment_id, file_collection):
     return (
         client

connect_ext_ppr/webapp.py

@@ -59,6 +59,7 @@
     get_deployment_by_id,
     get_deployment_request_schema,
     get_deployment_schema,
+    get_hub,
     get_hubs,
     get_instance_by_id,
     get_marketplace_schema,
@@ -120,6 +121,36 @@ def list_deployment_requests(
 
         return response_list
 
+    @router.get(
+        '/deployments/requests/{depl_req_id}',
+        summary='Deployment Request details',
+        response_model=DeploymentRequestSchema,
+    )
+    def get_deployment_request(
+        self,
+        depl_req_id: str,
+        db: VerboseBaseSession = Depends(get_db),
+        client: ConnectClient = Depends(get_installation_client),
+        installation: dict = Depends(get_installation),
+    ):
+        dr = (
+            db.query(DeploymentRequest).options(
+                joinedload(DeploymentRequest.ppr),
+                joinedload(DeploymentRequest.deployment),
+            )
+            .filter(
+                DeploymentRequest.deployment.has(account_id=installation['owner']['id']),
+                DeploymentRequest.id == depl_req_id)
+            .first()
+        )
+        if dr:
+            hub = get_hub(client, dr.deployment.hub_id)
+            return get_deployment_request_schema(dr, hub)
+        raise ExtensionHttpError.EXT_001(
+            format_kwargs={'obj_id': depl_req_id},
+            status_code=status.HTTP_404_NOT_FOUND,
+        )
+
     @router.get(
         '/deployments/{deployment_id}',
         summary='Deployment details',

tests/test_deployment_requests.py

@@ -1,4 +1,4 @@
-def test_get_deployments_requests(
+def test_list_deployments_requests(
     dbsession,
     mocker,
     deployment_factory,
@@ -52,3 +52,91 @@ def test_get_deployments_requests(
         }
         assert list(events.keys()) == ['created']
         assert list(events['created'].keys()) == ['at', 'by']
+
+
+def test_get_deployment_request(
+    dbsession,
+    mocker,
+    deployment_factory,
+    deployment_request_factory,
+    installation,
+    api_client,
+    connect_client,
+):
+    hub_data = {
+        'id': 'HB-0000-0001',
+        'name': 'Another Hub for the best',
+    }
+
+    mocker.patch(
+        'connect_ext_ppr.webapp.get_hub',
+        return_value=hub_data,
+    )
+
+    dep1 = deployment_factory(account_id=installation['owner']['id'], hub_id=hub_data['id'])
+    dep2 = deployment_factory(account_id='PA-123-456')
+
+    dr1 = deployment_request_factory(deployment=dep1)
+    deployment_request_factory(deployment=dep1)
+    deployment_request_factory(deployment=dep2)
+
+    response = api_client.get(
+        f'/api/deployments/requests/{dr1.id}',
+        installation=installation,
+    )
+    response_item = response.json()
+    events = response_item.pop('events')
+    assert response.status_code == 200
+    assert response_item == {
+        'id': dr1.id,
+        'deployment': {
+            'id': dep1.id,
+            'product': {
+                'id': dep1.product.id,
+                'name': dep1.product.name,
+                'icon': dep1.product.logo,
+            },
+            'hub': hub_data,
+        },
+        'ppr': {
+            'id': dr1.ppr_id,
+            'version': dr1.ppr.version,
+        },
+        'status': dr1.status.value,
+        'manually': dr1.manually,
+        'delegate_l2': dr1.delegate_l2,
+
+    }
+    assert list(events.keys()) == ['created']
+    assert list(events['created'].keys()) == ['at', 'by']
+
+
+def test_get_deployment_request_not_found(
+    dbsession,
+    deployment_factory,
+    deployment_request_factory,
+    installation,
+    api_client,
+):
+    hub_data = {
+        'id': 'HB-0000-0001',
+        'name': 'Another Hub for the best',
+    }
+
+    dep1 = deployment_factory(account_id=installation['owner']['id'], hub_id=hub_data['id'])
+    dep2 = deployment_factory(account_id='PA-123-456')
+
+    deployment_request_factory(deployment=dep1)
+    deployment_request_factory(deployment=dep1)
+    bad_dr = deployment_request_factory(deployment=dep2)
+
+    response = api_client.get(
+        f'/api/deployments/requests/{bad_dr.id}',
+        installation=installation,
+    )
+    error = response.json()
+
+    assert response.status_code == 404
+    assert error == {
+        'error_code': 'EXT_001', 'errors': [f'Object `{bad_dr.id}` not found.'],
+    }