Experimenting with testing incremental ingestion

plugins/incremental-ingestion-backend/README.md

@@ -39,15 +39,15 @@ This approach has the following benefits,
   ```ts
   const builder = CatalogBuilder.create(env);
   // incremental builder receives builder because it'll register
-  // incremental entity providers with the builder 
+  // incremental entity providers with the builder
   const incrementalBuilder = IncrementalCatalogBuilder.create(env, builder);
   ```
-3. Last step, add `await incrementBuilder.build()` after `await builder.build()` to ensure that all `CatalogBuider` migration run before running `incrementBuilder.build()` migrations.
+3. Last step, add `await incrementalBuilder.build()` after `await builder.build()` to ensure that all `CatalogBuider` migration run before running `incrementalBuilder.build()` migrations.
   ```ts
   const { processingEngine, router } = await builder.build();
 
-  // this has to run after `await builder.build()` so ensure that catalog migrations are completed 
-  // before incremental builder migrations are executed 
+  // this has to run after `await builder.build()` so ensure that catalog migrations are completed
+  // before incremental builder migrations are executed
   await incrementalBuilder.build();
   ```
 
@@ -67,18 +67,18 @@ import { PluginEnvironment } from '../types';
 export default async function createPlugin(
   env: PluginEnvironment,
 ): Promise<Router> {
-  
+
   const builder = CatalogBuilder.create(env);
   // incremental builder receives builder because it'll register
-  // incremental entity providers with the builder 
+  // incremental entity providers with the builder
   const incrementalBuilder = IncrementalCatalogBuilder.create(env, builder);
 
   builder.addProcessor(new ScaffolderEntitiesProcessor());
 
   const { processingEngine, router } = await builder.build();
 
-  // this has to run after `await builder.build()` so ensure that catalog migrations are completed 
-  // before incremental builder migrations are executed 
+  // this has to run after `await builder.build()` so ensure that catalog migrations are completed
+  // before incremental builder migrations are executed
   await incrementalBuilder.build();
 
   await processingEngine.start();
@@ -186,9 +186,9 @@ If you need to pass a token to your API, then you can create a constructor that
 
 ```ts
 export class MyIncrementalEntityProvider implements IncrementalEntityProvider<Cursor, Context> {
-  
+
   token: string;
-  
+
   construtor(token: string) {
     this.token = token;
   }
@@ -211,9 +211,9 @@ The last step is to implement the actual `next` method that will accept the curs
 
 ```ts
 export class MyIncrementalEntityProvider implements IncrementalEntityProvider<Cursor, Context> {
-  
+
   token: string;
-  
+
   construtor(token: string) {
     this.token = token;
   }
@@ -276,7 +276,7 @@ export class MyIncrementalEntityProvider implements IncrementalEntityProvider<Cu
 }
 ```
 
-Now that you have your new Incremental Entity Provider, we can connect it to the catalog. 
+Now that you have your new Incremental Entity Provider, we can connect it to the catalog.
 
 ## Adding an Incremental Entity Provider to the catalog
 

plugins/incremental-ingestion-backend/package.json

@@ -37,12 +37,17 @@
     "yn": "^4.0.0"
   },
   "devDependencies": {
+    "@backstage/backend-tasks": "^0.3.0",
     "@backstage/cli": "^0.17.0",
-    "@types/supertest": "^2.0.8",
+    "@backstage/plugin-permission-node": "^0.6.0",
+    "@effection/vitest": "^2.0.0",
     "@types/luxon": "^2.0.4",
+    "@types/supertest": "^2.0.8",
     "@types/uuid": "^8.3.4",
+    "effection": "^2.0.4",
+    "msw": "^0.35.0",
     "supertest": "^4.0.2",
-    "msw": "^0.35.0"
+    "vitest": "^0.23.2"
   },
   "files": [
     "dist",

plugins/incremental-ingestion-backend/src/tests/config.ts

@@ -0,0 +1,79 @@
+export const backstageConfig = {
+  backend: {
+    listen: { port: 8800 },
+    database: {
+      prefix: 'graphql_tests_',
+      client: 'pg',
+      connection: {
+        host: 'localhost',
+        port: '5432',
+        user: 'postgres',
+        password: 'postgres',
+      },
+    },
+    baseUrl: 'http://localhost:8800',
+  },
+  catalog: {
+    rules: [
+      {
+        allow: [
+          'Component',
+          'System',
+          'API',
+          'Group',
+          'User',
+          'Resource',
+          'Location'
+        ],
+      },
+    ],
+    locations: [
+      {
+        type: 'url',
+        target:
+          'https://github.com/thefrontside/backstage/blob/main/catalog-info.yaml',
+      },
+      {
+        type: 'url',
+        target:
+          'https://github.com/backstage/backstage/blob/master/packages/catalog-model/examples/all-components.yaml',
+      },
+      {
+        type: 'url',
+        target:
+          'https://github.com/backstage/backstage/blob/master/packages/catalog-model/examples/all-systems.yaml',
+      },
+      {
+        type: 'url',
+        target:
+          'https://github.com/backstage/backstage/blob/master/packages/catalog-model/examples/all-apis.yaml',
+      },
+      {
+        type: 'url',
+        target:
+          'https://github.com/backstage/backstage/blob/master/packages/catalog-model/examples/all-resources.yaml',
+      },
+      {
+        type: 'url',
+        target:
+          'https://github.com/backstage/backstage/blob/master/packages/catalog-model/examples/acme/org.yaml',
+      },
+      {
+        type: 'url',
+        target:
+          'https://github.com/backstage/software-templates/blob/main/scaffolder-templates/react-ssr-template/template.yaml',
+        rules: [{ allow: ['Template'] }]
+      },{
+        type: 'url',
+        target:
+          'https://github.com/backstage/software-templates/blob/main/scaffolder-templates/springboot-grpc-template/template.yaml',
+        rules: [{ allow: ['Template'] }]
+      },{
+        type: 'url',
+        target:
+          'https://github.com/backstage/software-templates/blob/main/scaffolder-templates/docs-template/template.yaml',
+        rules: [{ allow: ['Template'] }]
+      },
+    ],
+  },
+};

plugins/incremental-ingestion-backend/src/tests/setupTests.ts

@@ -0,0 +1,108 @@
+/* eslint-disable func-names */
+import { describe, beforeAll, it } from '@effection/vitest';
+import { TaskScheduler } from '@backstage/backend-tasks';
+import { CatalogBuilder } from '@backstage/plugin-catalog-backend';
+import { ServerPermissionClient } from '@backstage/plugin-permission-node';
+import { ensure, once, Operation } from 'effection';
+import { Duration } from 'luxon';
+import { createLogger, Logger, transports } from 'winston';
+import { EntityIteratorResult, IncrementalCatalogBuilder } from '..';
+import { IncrementalEntityProvider, IncrementalEntityProviderOptions, PluginEnvironment } from '../types';
+import { ConfigReader } from '@backstage/config';
+import { backstageConfig } from './config';
+import { DatabaseManager, ServerTokenManager, SingleHostDiscovery, UrlReaders } from '@backstage/backend-common';
+
+interface Page {
+  data: string[];
+  retries?: number;
+  timeout?: number;
+}
+
+class EntityProvider implements IncrementalEntityProvider<number, void> {
+  private pages: Page[] = [];
+
+  getProviderName() { return 'EntityProvider' }
+
+  async around(burst: () => Promise<void>): Promise<void> {
+    burst();
+  }
+
+  async next(_context: void, page: number): Promise<EntityIteratorResult<number>> {
+    // TODO Handle pages
+    return this.pages
+  }
+
+  setData(data: Page[]) { this.pages = data }
+}
+
+function useCatalogPlugin(env: PluginEnvironment): Operation<EntityProvider> {
+  return {
+    name: "CatalogPlugin",
+    *init() {
+      const builder = CatalogBuilder.create(env);
+      const incrementalBuilder = IncrementalCatalogBuilder.create(env, builder);
+      const { processingEngine } = yield builder.build();
+      yield incrementalBuilder.build()
+
+      const provider = new EntityProvider();
+      const schedule: IncrementalEntityProviderOptions = {
+        burstInterval: Duration.fromObject({ milliseconds: 100 }),
+        burstLength: Duration.fromObject({ milliseconds: 100 }),
+        restLength: Duration.fromObject({ seconds: 1 }),
+      }
+
+      incrementalBuilder.addIncrementalEntityProvider(provider, schedule);
+
+      yield processingEngine.start();
+      yield ensure(() => processingEngine.stop());
+
+      return provider;
+     }
+  }
+}
+
+function useLogger(): Operation<Logger> {
+  return {
+    name: "Logger",
+    *init() {
+      const transport = new transports.Console();
+      const logger = createLogger({
+        level: 'error',
+        transports: [transport],
+      });
+      yield ensure(function* () {
+        logger.end();
+        logger.on('error', () => { /* noop */ });
+        yield once(logger, 'finish')
+      });
+      return logger
+    }
+  }
+}
+
+describe('incrementally ingest entities', () => {
+  let provider: EntityProvider
+
+  beforeAll(function* () {
+    const logger = yield useLogger()
+    const config = new ConfigReader(backstageConfig);
+    const reader = UrlReaders.default({ logger, config });
+    const databaseManager = DatabaseManager.fromConfig(config);
+    const discovery = SingleHostDiscovery.fromConfig(config);
+    const tokenManager = ServerTokenManager.noop();
+    const permissions = ServerPermissionClient.fromConfig(config, { discovery, tokenManager });
+    const scheduler = TaskScheduler.fromConfig(config).forPlugin('catalog');
+    provider = yield useCatalogPlugin({
+      logger,
+      database: databaseManager.forPlugin('catalog'),
+      config,
+      reader,
+      permissions,
+      scheduler
+    })
+  });
+
+  it.eventually('successfuly ingest data', function* () {
+    provider.setData([]);
+  })
+})