FAI-441 - Implement Phabricator check connection + streams

.github/workflows/ci.yml

@@ -52,17 +52,13 @@ jobs:
         run: lerna run test-cov
 
       - name: Run Jenkins server
+        # if: github.ref == 'refs/heads/main' # Skip PRs
         run: |
           mkdir -p ${{ github.workspace }}/sources/jenkins-source/test_files/jenkins/data/secrets/
-          sudo chown -R $USER:$USER ${{ github.workspace }}/sources/jenkins-source/test_files/jenkins/data
           docker run -d -u 0 -i --name jenkins -p 8080:8080 -p 50000:50000 -v ${{ github.workspace }}/sources/jenkins-source/test_files/jenkins/data:/var/jenkins_home jenkins/jenkins:lts-jdk11
-
-      - name: Source Acceptance Test
-        if: github.ref == 'refs/heads/main' # Skip PRs
-        run: |
           echo "Waiting until Jenkins is up"
           waiting=0
-          until curl -s -I -X GET $JENKINS_URL; do
+          until curl -s -I -X GET http://127.0.0.1:8080; do
             docker ps -a
             let waiting+=3
             sleep 3
@@ -71,16 +67,21 @@ jobs:
               exit 42
             fi
           done
+          sudo chown -R $USER:$USER ${{ github.workspace }}/sources/jenkins-source/test_files/jenkins/data
 
-          sudo chown -R $USER:$USER $(find ${{ github.workspace }}/sources/jenkins-source/test_files/jenkins/data/.java/fonts -type f)
+      - name: Source Acceptance Test
+        # if: github.ref == 'refs/heads/main' # Skip PRs
+        run: |
           for i in $(ls -d sources/*/)
           do
-            ./scripts/source-acceptance-test.sh $(echo $i | cut -f2 -d'/')
+            source=$(echo $i | cut -f2 -d'/')
+            if [[ $source != "phabricator-source" ]]; then
+              ./scripts/source-acceptance-test.sh $source
+            fi
           done
-        env:
-          JENKINS_URL: http://127.0.0.1:8080
 
       - name: Stop Jenkins server
+        # if: github.ref == 'refs/heads/main' # Skip PRs
         run: |
           docker stop jenkins
           docker rm jenkins

faros-airbyte-cdk/src/sources/source-base.ts

@@ -107,19 +107,18 @@ export abstract class AirbyteSourceBase extends AirbyteSource {
     catalog: AirbyteConfiguredCatalog,
     state?: AirbyteState
   ): AsyncGenerator<AirbyteMessage> {
+    this.logger.info(`Syncing ${this.name}`);
     const connectorState = cloneDeep(state ?? {});
-    this.logger.info(`Starting syncing ${this.name}`);
     // TODO: assert all streams exist in the connector
     // get the streams once in case the connector needs to make any queries to
     // generate them
     const streamInstances = keyBy(this.streams(config), (s) => s.name);
     for (const configuredStream of catalog.streams) {
-      const streamInstance = streamInstances[configuredStream.stream.name];
+      const streamName = configuredStream.stream.name;
+      const streamInstance = streamInstances[streamName];
       if (!streamInstance) {
         throw new VError(
-          `The requested stream ${
-            configuredStream.stream.name
-          } was not found in the source. Available streams: ${Object.keys(
+          `The requested stream ${streamName} was not found in the source. Available streams: ${Object.keys(
             streamInstances
           )}`
         );

sources/README.md

@@ -58,7 +58,7 @@ A source contains one or more streams, which correspond to entity types of the
 system your source is fetching data from. For example, a GitHub source would
 have streams for users, commits, pull requests, etc. Each stream also has its
 own arbitrary state for supporting incremental mode syncs. Implement your
-streams, an example of which is in the `JenkinsBuilds` class in `src/stream.ts`,
+streams, an example of which is in the `Builds` class in `src/stream.ts`,
 and include them in your source via the `streams()` method of your source class.
 
 Each stream has a JSON-Schema object defining the schema of the records that

sources/example-source/acceptance-test-config.yml

@@ -23,4 +23,4 @@ tests:
       configured_catalog_path: "test_files/incremental_configured_catalog.json"
       future_state_path: "test_files/abnormal_state.json"
       cursor_paths:
-        jenkins_builds: [ "cutoff" ]
+        builds: [ "cutoff" ]

sources/example-source/src/index.ts

@@ -9,7 +9,7 @@ import {
 } from 'faros-airbyte-cdk';
 import VError from 'verror';
 
-import {JenkinsBuilds} from './streams';
+import {Builds} from './streams';
 
 /** The main entry point. */
 export function mainCommand(): Command {
@@ -30,6 +30,6 @@ class ExampleSource extends AirbyteSourceBase {
     return [false, new VError('User is not chris')];
   }
   streams(config: AirbyteConfig): AirbyteStreamBase[] {
-    return [new JenkinsBuilds(this.logger)];
+    return [new Builds(this.logger)];
   }
 }

sources/example-source/src/streams/builds.ts

@@ -1,7 +1,7 @@
 import {AirbyteStreamBase, StreamKey, SyncMode} from 'faros-airbyte-cdk';
 import {Dictionary} from 'ts-essentials';
 
-export class JenkinsBuilds extends AirbyteStreamBase {
+export class Builds extends AirbyteStreamBase {
   getJsonSchema(): Dictionary<any, string> {
     return require('../../resources/schemas/builds.json');
   }

sources/example-source/src/streams/index.ts

@@ -1,3 +1,3 @@
-import {JenkinsBuilds} from './builds';
+import {Builds} from './builds';
 
-export {JenkinsBuilds};
+export {Builds};

sources/example-source/test_files/abnormal_state.json

@@ -1,5 +1,5 @@
 {
-  "jenkins_builds": {
+  "builds": {
     "cutoff": 1735718400000
   }
-}
+}