feat(frontend, sdk): towards namespaced pipelines. Part of #4197

frontend/src/lib/Apis.ts

@@ -324,6 +324,7 @@ export class Apis {
     pipelineName: string,
     pipelineDescription: string,
     pipelineData: File,
+    namespace?: string,
   ): Promise<ApiPipeline> {
     const fd = new FormData();
     fd.append('uploadfile', pipelineData, pipelineData.name);
@@ -332,7 +333,7 @@ export class Apis {
       v1beta1Prefix,
       `name=${encodeURIComponent(pipelineName)}&description=${encodeURIComponent(
         pipelineDescription,
-      )}`,
+      )}` + (namespace ? `&namespace=${encodeURIComponent(namespace)}` : ''),
       {
         body: fd,
         cache: 'no-cache',
@@ -346,14 +347,16 @@ export class Apis {
     pipelineId: string,
     versionData: File,
     description?: string,
+    namespace?: string,
   ): Promise<ApiPipelineVersion> {
     const fd = new FormData();
     fd.append('uploadfile', versionData, versionData.name);
     return await this._fetchAndParse<ApiPipelineVersion>(
       '/pipelines/upload_version',
       v1beta1Prefix,
       `name=${encodeURIComponent(versionName)}&pipelineid=${encodeURIComponent(pipelineId)}` +
-        (description ? `&description=${encodeURIComponent(description)}` : ''),
+        (description ? `&description=${encodeURIComponent(description)}` : '') +
+        (namespace ? `&namespace=${encodeURIComponent(namespace)}` : ''),
       {
         body: fd,
         cache: 'no-cache',

frontend/src/pages/PipelineList.test.tsx

@@ -20,9 +20,12 @@ import * as React from 'react';
 import { RoutePage, RouteParams } from '../components/Router';
 import { Apis } from '../lib/Apis';
 import { ButtonKeys } from '../lib/Buttons';
+import { render, act } from '@testing-library/react';
 import TestUtils from '../TestUtils';
 import { PageProps } from './Page';
-import PipelineList from './PipelineList';
+import EnhancedPipelineList, { PipelineList } from './PipelineList';
+import { MemoryRouter } from 'react-router-dom';
+import { NamespaceContext } from 'src/lib/KubeflowClient';
 
 describe('PipelineList', () => {
   let tree: ReactWrapper | ShallowWrapper;
@@ -60,7 +63,20 @@ describe('PipelineList', () => {
     );
   }
 
-  async function mountWithNPipelines(n: number): Promise<ReactWrapper> {
+  function mockListNPipelines(n: number = 1) {
+    return () =>
+      Promise.resolve({
+        pipelines: range(n).map(i => ({
+          id: 'test-pipeline-id' + i,
+          name: 'test pipeline name' + i,
+        })),
+      });
+  }
+
+  async function mountWithNPipelines(
+    n: number,
+    { namespace }: { namespace?: string } = {},
+  ): Promise<ReactWrapper> {
     listPipelinesSpy.mockImplementation(() => ({
       pipelines: range(n).map(i => ({
         id: 'test-pipeline-id' + i,
@@ -71,7 +87,7 @@ describe('PipelineList', () => {
         },
       })),
     }));
-    tree = TestUtils.mountWithRouter(<PipelineList {...generateProps()} />);
+    tree = TestUtils.mountWithRouter(<PipelineList {...generateProps()} namespace={namespace} />);
     await listPipelinesSpy;
     await TestUtils.flushPromises();
     tree.update(); // Make sure the tree is updated before returning it
@@ -86,7 +102,9 @@ describe('PipelineList', () => {
   afterEach(async () => {
     // unmount() should be called before resetAllMocks() in case any part of the unmount life cycle
     // depends on mocks/spies
-    await tree.unmount();
+    if (tree.exists()) {
+      await tree.unmount();
+    }
     jest.restoreAllMocks();
   });
 
@@ -146,7 +164,14 @@ describe('PipelineList', () => {
     listPipelinesSpy.mockImplementationOnce(() => ({ pipelines: [{ name: 'pipeline1' }] }));
     tree = TestUtils.mountWithRouter(<PipelineList {...generateProps()} />);
     await listPipelinesSpy;
-    expect(listPipelinesSpy).toHaveBeenLastCalledWith('', 10, 'created_at desc', '');
+    expect(listPipelinesSpy).toHaveBeenLastCalledWith(
+      '',
+      10,
+      'created_at desc',
+      '',
+      undefined,
+      undefined,
+    );
     expect(tree.state()).toHaveProperty('displayPipelines', [
       { expandState: 0, name: 'pipeline1' },
     ]);
@@ -160,7 +185,14 @@ describe('PipelineList', () => {
     expect(refreshBtn).toBeDefined();
     await refreshBtn!.action();
     expect(listPipelinesSpy.mock.calls.length).toBe(2);
-    expect(listPipelinesSpy).toHaveBeenLastCalledWith('', 10, 'created_at desc', '');
+    expect(listPipelinesSpy).toHaveBeenLastCalledWith(
+      '',
+      10,
+      'created_at desc',
+      '',
+      undefined,
+      undefined,
+    );
     expect(updateBannerSpy).toHaveBeenLastCalledWith({});
   });
 
@@ -186,7 +218,14 @@ describe('PipelineList', () => {
     TestUtils.makeErrorResponseOnce(listPipelinesSpy, 'bad stuff happened');
     await refreshBtn!.action();
     expect(listPipelinesSpy.mock.calls.length).toBe(2);
-    expect(listPipelinesSpy).toHaveBeenLastCalledWith('', 10, 'created_at desc', '');
+    expect(listPipelinesSpy).toHaveBeenLastCalledWith(
+      '',
+      10,
+      'created_at desc',
+      '',
+      undefined,
+      undefined,
+    );
     expect(updateBannerSpy).toHaveBeenLastCalledWith(
       expect.objectContaining({
         additionalInfo: 'bad stuff happened',
@@ -562,4 +601,28 @@ describe('PipelineList', () => {
       open: true,
     });
   });
+
+  it("doesn't keep error message for request from previous namespace", async () => {
+    listPipelinesSpy.mockImplementation(() => Promise.reject('namespace cannot be empty'));
+    const { rerender } = render(
+      <MemoryRouter>
+        <NamespaceContext.Provider value={undefined}>
+          <EnhancedPipelineList {...generateProps()} />
+        </NamespaceContext.Provider>
+      </MemoryRouter>,
+    );
+
+    listPipelinesSpy.mockImplementation(mockListNPipelines());
+    rerender(
+      <MemoryRouter>
+        <NamespaceContext.Provider value={'test-ns'}>
+          <EnhancedPipelineList {...generateProps()} />
+        </NamespaceContext.Provider>
+      </MemoryRouter>,
+    );
+    await act(TestUtils.flushPromises);
+    expect(updateBannerSpy).toHaveBeenLastCalledWith(
+      {}, // Empty object means banner has no error message
+    );
+  });
 });

frontend/src/pages/PipelineList.tsx

@@ -19,7 +19,12 @@ import produce from 'immer';
 import * as React from 'react';
 import { Link } from 'react-router-dom';
 import { classes } from 'typestyle';
-import { ApiListPipelinesResponse, ApiPipeline } from '../apis/pipeline';
+import {
+  ApiListPipelinesResponse,
+  ApiPipeline,
+  ApiResourceType,
+  ApiRelationship,
+} from '../apis/pipeline';
 import CustomTable, {
   Column,
   CustomRendererProps,
@@ -34,8 +39,9 @@ import { commonCss, padding } from '../Css';
 import { Apis, ListRequest, PipelineSortKeys } from '../lib/Apis';
 import Buttons, { ButtonKeys } from '../lib/Buttons';
 import { errorToMessage, formatDateString } from '../lib/Utils';
-import { Page } from './Page';
+import { Page, PageProps } from './Page';
 import PipelineVersionList from './PipelineVersionList';
+import { NamespaceContext } from 'src/lib/KubeflowClient';
 
 interface DisplayPipeline extends ApiPipeline {
   expandState?: ExpandState;
@@ -57,7 +63,7 @@ const descriptionCustomRenderer: React.FC<CustomRendererProps<string>> = (
   return <Description description={props.value || ''} forceInline={true} />;
 };
 
-class PipelineList extends Page<{}, PipelineListState> {
+export class PipelineList extends Page<{ namespace?: string }, PipelineListState> {
   private _tableRef = React.createRef<CustomTable>();
 
   constructor(props: any) {
@@ -176,6 +182,8 @@ class PipelineList extends Page<{}, PipelineListState> {
         request.pageSize,
         request.sortBy,
         request.filter,
+        this.props.namespace ? 'NAMESPACE' : undefined,
+        this.props.namespace || undefined,
       );
       displayPipelines = response.pipelines || [];
       displayPipelines.forEach(exp => (exp.expandState = ExpandState.COLLAPSED));
@@ -247,8 +255,23 @@ class PipelineList extends Page<{}, PipelineListState> {
 
     try {
       method === ImportMethod.LOCAL
-        ? await Apis.uploadPipeline(name, description || '', file!)
-        : await Apis.pipelineServiceApi.createPipeline({ name, url: { pipeline_url: url } });
+        ? await Apis.uploadPipeline(name, description || '', file!, this.props.namespace)
+        : await Apis.pipelineServiceApi.createPipeline({
+            name,
+            url: { pipeline_url: url },
+            resource_references: this.props.namespace
+              ? [
+                  {
+                    key: {
+                      id: this.props.namespace,
+                      type: ApiResourceType.NAMESPACE,
+                    },
+                    relationship: ApiRelationship.OWNER,
+                  },
+                ]
+              : [],
+          });
+
       this.setStateSafe({ uploadDialogOpen: false });
       this.refresh();
       return true;
@@ -264,4 +287,9 @@ class PipelineList extends Page<{}, PipelineListState> {
   }
 }
 
-export default PipelineList;
+const EnhancedPipelineList: React.FC<PageProps> = props => {
+  const namespace = React.useContext(NamespaceContext);
+  return <PipelineList key={namespace} {...props} namespace={namespace} />;
+};
+
+export default EnhancedPipelineList;

sdk/python/kfp/client/client.py

@@ -1338,6 +1338,7 @@ def upload_pipeline(
         pipeline_package_path: str = None,
         pipeline_name: str = None,
         description: str = None,
+        namespace: str = None,
     ) -> kfp_server_api.ApiPipeline:
         """Uploads the pipeline to the Kubeflow Pipelines cluster.
 
@@ -1352,7 +1353,7 @@ def upload_pipeline(
         """
 
         response = self._upload_api.upload_pipeline(
-            pipeline_package_path, name=pipeline_name, description=description)
+            pipeline_package_path, name=pipeline_name, description=description, namespace=namespace)
         if self._is_ipython():
             import IPython
             html = '<a href=%s/#/pipelines/details/%s>Pipeline details</a>.' % (
@@ -1367,6 +1368,7 @@ def upload_pipeline_version(
         pipeline_id: Optional[str] = None,
         pipeline_name: Optional[str] = None,
         description: Optional[str] = None,
+        namespace: Optional[str] = None,
     ) -> kfp_server_api.ApiPipelineVersion:
         """Uploads a new version of the pipeline to the KFP cluster.
 
@@ -1402,6 +1404,9 @@ def upload_pipeline_version(
         if description:
             kwargs['description'] = description
 
+        if namespace:
+            kwargs['namespace'] = namespace
+
         response = self._upload_api.upload_pipeline_version(
             pipeline_package_path, **kwargs)