Rawls attempts to fetch "rawls" workspaces from WSM in addition to "MC"

core/src/main/scala/org/broadinstitute/dsde/rawls/workspace/AggregatedWorkspaceService.scala

@@ -103,20 +103,6 @@ class AggregatedWorkspaceService(workspaceManagerDAO: WorkspaceManagerDAO) exten
       }
     }
 
-  /**
-   * Optimized version of [[fetchAggregatedWorkspace]]
-   *
-   * If the provided workspace is not of type "MC", returns the provided "rawls" workspace with no WSM information, as
-   * it can be assumed to be GCP and does not need to call out to WSM for this.
-   */
-  def optimizedFetchAggregatedWorkspace(workspace: Workspace, ctx: RawlsRequestContext): AggregatedWorkspace =
-    workspace.workspaceType match {
-      case WorkspaceType.RawlsWorkspace =>
-        AggregatedWorkspace(workspace, Some(workspace.googleProjectId), azureCloudContext = None, policies = List.empty)
-      case WorkspaceType.McWorkspace =>
-        fetchAggregatedWorkspace(workspace, ctx)
-    }
-
   private def aggregateMCWorkspaceWithWSMInfo(workspace: Workspace,
                                               wsmInfo: WorkspaceDescription
   ): AggregatedWorkspace =

core/src/main/scala/org/broadinstitute/dsde/rawls/workspace/WorkspaceService.scala

@@ -311,7 +311,24 @@ class WorkspaceService(protected val ctx: RawlsRequestContext,
       getV2WorkspaceContextAndPermissions(workspaceName, SamWorkspaceActions.read, Option(attrSpecs)) flatMap {
         workspaceContext =>
           dataSource.inTransaction { dataAccess =>
-            val wsmContext = wsmService.optimizedFetchAggregatedWorkspace(workspaceContext, ctx)
+            // some GCP workspaces, like those with linked snapshots, have a stub WSM workspace
+            val wsmContext =
+              try
+                wsmService.fetchAggregatedWorkspace(workspaceContext, ctx)
+              catch {
+                case e: AggregateWorkspaceNotFoundException =>
+                  // return workspace with no WSM information for gcp workspace
+                  if (workspaceContext.workspaceType == WorkspaceType.RawlsWorkspace) {
+                    AggregatedWorkspace(workspaceContext,
+                                        Some(workspaceContext.googleProjectId),
+                                        azureCloudContext = None,
+                                        policies = List.empty
+                    )
+                  } else {
+                    // bubble up an MC workspace exception
+                    throw e
+                  }
+              }
 
             // maximum access level is required to calculate canCompute and canShare. Therefore, if any of
             // accessLevel, canCompute, canShare is specified, we have to get it.

core/src/test/scala/org/broadinstitute/dsde/rawls/workspace/AggregatedWorkspaceServiceSpec.scala

@@ -217,24 +217,6 @@ class AggregatedWorkspaceServiceSpec extends AnyFlatSpec with MockitoTestUtils {
     verify(wsmDao).getWorkspace(ArgumentMatchers.eq(legacyRawlsWorkspace.workspaceIdAsUUID), any[RawlsRequestContext])
   }
 
-  behavior of "optimizedFetchAggregatedWorkspace"
-
-  it should "not reach out to WSM for legacy GCP Rawls workspaces" in {
-    val wsmDao = mock[WorkspaceManagerDAO]
-    when(wsmDao.getWorkspace(any[UUID], any[RawlsRequestContext])).thenReturn(
-      new WorkspaceDescription().stage(WorkspaceStageModel.RAWLS_WORKSPACE)
-    )
-    val svc = new AggregatedWorkspaceService(wsmDao)
-
-    val aggregatedWorkspace = svc.optimizedFetchAggregatedWorkspace(legacyRawlsWorkspace, defaultRequestContext)
-
-    aggregatedWorkspace.baseWorkspace shouldBe legacyRawlsWorkspace
-    aggregatedWorkspace.getCloudPlatform shouldBe Some(WorkspaceCloudPlatform.Gcp)
-    aggregatedWorkspace.azureCloudContext shouldBe None
-    aggregatedWorkspace.googleProjectId shouldBe Some(legacyRawlsWorkspace.googleProjectId)
-    verify(wsmDao, never()).getWorkspace(any[UUID], any[RawlsRequestContext])
-  }
-
   behavior of "fetchAggregatedWorkspaces"
 
   it should "match an MC workspace with the correct WSM information" in {

core/src/test/scala/org/broadinstitute/dsde/rawls/workspace/WorkspaceServiceSpec.scala

@@ -3042,6 +3042,28 @@ class WorkspaceServiceSpec
     )
   }
 
+  private def createGcpWorkspaceStub(services: TestApiService,
+                                     workspaceName: String,
+                                     policies: List[WsmPolicyInput] = List(),
+                                     workspaceService: WorkspaceService
+  ): Workspace = {
+    val workspaceRequest = WorkspaceRequest(
+      testData.testProject1Name.value,
+      workspaceName,
+      Map.empty
+    )
+    when(services.workspaceManagerDAO.getWorkspace(any[UUID], any[RawlsRequestContext])).thenReturn(
+      new WorkspaceDescription()
+        .stage(WorkspaceStageModel.RAWLS_WORKSPACE)
+        .policies(policies.asJava)
+    )
+
+    Await.result(
+      services.mcWorkspaceService.createMultiCloudOrRawlsWorkspace(workspaceRequest, workspaceService),
+      Duration.Inf
+    )
+  }
+
   it should "get the details of an Azure workspace" in withTestDataServices { services =>
     val managedAppCoordinates = AzureManagedAppCoordinates(UUID.randomUUID(), UUID.randomUUID(), "fake_mrg_id")
     val workspace = createAzureWorkspace(services, managedAppCoordinates)
@@ -3148,6 +3170,47 @@ class WorkspaceServiceSpec
     additionalData.tail.head.getOrElse("pair2Key", "fail") shouldEqual "pair2Val"
   }
 
+  it should "return the policies of a GCP workspace" in withTestDataServices { services =>
+    val workspaceName = s"rawls-test-workspace-${UUID.randomUUID().toString}"
+    val workspaceRequest = WorkspaceRequest(
+      testData.testProject1Name.value,
+      workspaceName,
+      Map.empty
+    )
+    val wsmPolicyInput = new WsmPolicyInput()
+      .name("test_name")
+      .namespace("test_namespace")
+      .additionalData(
+        List(
+          new WsmPolicyPair().value("pair1Val").key("pair1Key"),
+          new WsmPolicyPair().value("pair2Val").key("pair2Key")
+        ).asJava
+      )
+    val workspace = createGcpWorkspaceStub(services, workspaceName, List(wsmPolicyInput), services.workspaceService)
+    val readWorkspace = Await.result(services.workspaceService.getWorkspace(
+                                       WorkspaceName(workspace.namespace, workspace.name),
+                                       WorkspaceFieldSpecs()
+                                     ),
+                                     Duration.Inf
+    )
+
+    val response = readWorkspace.convertTo[WorkspaceResponse]
+
+    response.workspace.name shouldBe workspaceName
+    response.azureContext shouldEqual None
+    response.workspace.cloudPlatform shouldBe Some(WorkspaceCloudPlatform.Gcp)
+    response.policies should not be empty
+    val policies: List[WorkspacePolicy] = response.policies.get
+    policies should not be empty
+    val policy: WorkspacePolicy = policies.head
+    policy.name shouldBe wsmPolicyInput.getName
+    policy.namespace shouldBe wsmPolicyInput.getNamespace
+    val additionalData = policy.additionalData
+    additionalData.length shouldEqual 2
+    additionalData.head.getOrElse("pair1Key", "fail") shouldEqual "pair1Val"
+    additionalData.tail.head.getOrElse("pair2Key", "fail") shouldEqual "pair2Val"
+  }
+
   it should "return correct canCompute permission for Azure workspaces" in withTestDataServices { services =>
     val managedAppCoordinates = AzureManagedAppCoordinates(UUID.randomUUID(), UUID.randomUUID(), "fake_mrg_id")
     val workspace = createAzureWorkspace(services, managedAppCoordinates)