Block exec until sync is complete (#134)

This PR adds code to parse logs from bricks sync to compute the state of
sync completeness. If sync is in progress (ie upload/delete requests are
inflight to the workspace), we block execution in the databricks runner

Video for exec:

https://user-images.githubusercontent.com/88374338/197779331-4b560bcf-63b2-4239-a3fb-4ee7a0814b62.mov

Video for exec as workflow


https://user-images.githubusercontent.com/88374338/198030331-737e3502-3276-4d68-acb5-4b1229431fca.mov

Co-authored-by: Fabian Jakobs <fabian.jakobs@databricks.com>

Author: shreyas-goenka

packages/databricks-vscode/src/cli/BricksTasks.test.ts

@@ -1,10 +1,10 @@
 import {ApiClient, Repo} from "@databricks/databricks-sdk";
 import * as assert from "assert";
 import {anything, instance, mock, when, verify} from "ts-mockito";
-import {ProcessExecution, Uri} from "vscode";
+import {ProcessExecution, Uri, EventEmitter} from "vscode";
 import {ConnectionManager} from "../configuration/ConnectionManager";
-import {SyncDestination} from "../configuration/SyncDestination";
-import {BricksTaskProvider, SyncTask} from "./BricksTasks";
+import {SyncState} from "../sync/CodeSynchronizer";
+import {BricksTaskProvider, SyncTask, BricksSyncParser} from "./BricksTasks";
 import {CliWrapper} from "./CliWrapper";
 
 describe(__filename, () => {
@@ -26,49 +26,120 @@ describe(__filename, () => {
     });
 
     it("should create a sync task", () => {
-        let task = new SyncTask(connection, cli, "incremental");
+        let task = new SyncTask(
+            connection,
+            cli,
+            "incremental",
+            (state: SyncState) => {}
+        );
 
         assert.equal(task.definition.type, "databricks");
         assert.equal(task.definition.task, "sync");
         assert.equal(task.isBackground, true);
         assert.deepEqual(task.problemMatchers, ["$bricks-sync"]);
     });
+});
 
-    it("should lazily create a process execution", async () => {
-        let connectionMock = mock(ConnectionManager);
-        const testSyncDestination = new SyncDestination(
-            instance(mock(Repo)),
-            Uri.from({
-                scheme: "dbws",
-                path: "/Workspace/notebook-best-practices",
-            }),
-            Uri.file("/Desktop/notebook-best-practices")
+describe("tests for BricksSycnParser", () => {
+    let syncState: SyncState = "STOPPED";
+    let bricksSycnParser: BricksSyncParser;
+
+    const syncStateCallback = (state: SyncState) => {
+        syncState = state;
+    };
+
+    beforeEach(() => {
+        syncState = "STOPPED";
+        bricksSycnParser = new BricksSyncParser(
+            syncStateCallback,
+            mock(EventEmitter<string>)
         );
-        when(connectionMock.profile).thenReturn("DEFAULT");
-        when(connectionMock.me).thenReturn("fabian.jakobs@databricks.com");
-        when(connectionMock.syncDestination).thenReturn(testSyncDestination);
+    });
 
-        let cliMock = mock(CliWrapper);
-        when(cliMock.getSyncCommand(anything())).thenReturn({
-            command: "bricks",
-            args: ["sync"],
-        });
+    it("processing empty logs transitions sync status from INACTIVE -> WATCHING_FOR_CHANGES", () => {
+        assert.equal(syncState, "STOPPED");
+        bricksSycnParser.process("");
+        assert.equal(syncState, "WATCHING_FOR_CHANGES");
+    });
 
-        let task = new SyncTask(
-            instance(connectionMock),
-            instance(cliMock),
-            "incremental"
+    it("processing action log transitions sync status from INACTIVE -> INPROGRESS", () => {
+        assert.equal(syncState, "STOPPED");
+        bricksSycnParser.process("Action: PUT: hello.txt");
+        assert.equal(syncState, "IN_PROGRESS");
+    });
+
+    it("test bricksSycnParser.process correctly keeps track of state of inflight requests", () => {
+        // recieving some random logs from bricks sync
+        assert.equal(syncState, "STOPPED");
+        bricksSycnParser.process("some random logs");
+        assert.equal(syncState, "WATCHING_FOR_CHANGES");
+
+        // upload  hello.txt
+        bricksSycnParser.process("Action: PUT: hello.txt");
+        assert.equal(syncState, "IN_PROGRESS");
+        bricksSycnParser.process("Uploaded hello.txt");
+        assert.equal(syncState, "WATCHING_FOR_CHANGES");
+
+        // delete  bye.txt
+        bricksSycnParser.process("Action: DELETE: bye.txt");
+        assert.equal(syncState, "IN_PROGRESS");
+        bricksSycnParser.process("Deleted bye.txt");
+        assert.equal(syncState, "WATCHING_FOR_CHANGES");
+
+        // both upload and delete some random prefix string that should be ignored
+        bricksSycnParser.process(
+            "[INFO] foo bar Action: PUT: a.txt DELETE: b.txt"
+        );
+        bricksSycnParser.process("Uploaded a.txt");
+        assert.equal(syncState, "IN_PROGRESS");
+        bricksSycnParser.process("Deleted b.txt");
+        assert.equal(syncState, "WATCHING_FOR_CHANGES");
+
+        // upload and delete multiple files
+        bricksSycnParser.process(
+            "Action: PUT: a.txt, c.txt DELETE: b.txt, d.txt"
         );
-        assert.ok(task.execution);
+        bricksSycnParser.process("Uploaded a.txt");
+        assert.equal(syncState, "IN_PROGRESS");
+        bricksSycnParser.process("Deleted b.txt");
+        assert.equal(syncState, "IN_PROGRESS");
+        bricksSycnParser.process("Deleted d.txt");
+        assert.equal(syncState, "IN_PROGRESS");
+        bricksSycnParser.process("Uploaded c.txt");
+        assert.equal(syncState, "WATCHING_FOR_CHANGES");
 
-        let execution = task.execution as ProcessExecution;
+        // multi line logs
+        bricksSycnParser.process(
+            "Action: PUT: a.txt, c.txt DELETE: b.txt, d.txt\n" +
+                "Uploaded a.txt\n" +
+                "some random text\n" +
+                "Uploaded c.txt"
+        );
+        bricksSycnParser.process("Deleted b.txt");
+        assert.equal(syncState, "IN_PROGRESS");
+        bricksSycnParser.process("Deleted d.txt");
+        assert.equal(syncState, "WATCHING_FOR_CHANGES");
+    });
 
-        const syncCommandMock = cliMock.getSyncCommand(anything());
+    it("uploaded logs for untracked files throw errors", () => {
+        assert.throws(
+            () => {
+                bricksSycnParser.process("Uploaded a.txt");
+            },
+            {
+                message: /untracked file uploaded/,
+            }
+        );
+    });
 
-        verify(syncCommandMock).never();
-        assert.equal(execution.process, "bricks");
-        verify(syncCommandMock).once();
-        assert.deepEqual(execution.args, ["sync"]);
-        verify(syncCommandMock).once();
+    it("delete logs for untracked files throw errors", () => {
+        assert.throws(
+            () => {
+                bricksSycnParser.process("Deleted a.txt");
+            },
+            {
+                message: /untracked file deleted/,
+            }
+        );
     });
 });