Update tests and fix gitpython version

The gitpython version that was in use has a problem (see issue here:
gitpython-developers/GitPython#983). As
mentioned in the thread, this can be resolved by pinning the version to
3.1.x

Author: danielvdende

Diff for: setup.py

@@ -16,7 +16,7 @@
     "databricks-cli==0.9.0",
     "docker==4.0.2",
     "flake8==3.7.2",
-    "gitpython==3.0.2",
+    "gitpython==3.1.1",
     "jinja2==2.10.1",
     "kubernetes==10.0.1",
     "py4j==0.10.7",

Diff for: takeoff/azure/deploy_to_databricks.py

@@ -29,7 +29,7 @@
                     vol.Optional("config_file", default="databricks.json.j2"): str,
                     vol.Optional("name", default=""): str,
                     vol.Optional("lang", default="python"): vol.All(str, vol.In(["python", "scala"])),
-                    vol.Optional("run_immediately", default=True): bool,
+                    vol.Optional("run_stream_job_immediately", default=True): bool,
                     vol.Optional("arguments", default=[{}]): [{}],
                     vol.Optional("schedule"): {
                         vol.Required("quartz_cron_expression"): str,
@@ -88,14 +88,14 @@ def deploy_to_databricks(self):
             job_name = f"{app_name}-{self.env.artifact_tag}"
             job_config = self.create_config(job_name, job)
             is_streaming = self._job_is_streaming(job_config)
-            run_immediately = job["run_immediately"]
+            run_stream_job_immediately = job["run_stream_job_immediately"]
 
             logger.info("Removing old job")
             self.remove_job(self.env.artifact_tag, job_config=job, is_streaming=is_streaming)
 
             logger.info("Submitting new job with configuration:")
             logger.info(pprint.pformat(job_config))
-            self._submit_job(job_config, is_streaming, run_immediately)
+            self._submit_job(job_config, is_streaming, run_stream_job_immediately)
 
     def create_config(self, job_name: str, job_config: dict):
         common_arguments = dict(
@@ -192,11 +192,11 @@ def _kill_it_with_fire(self, job_id):
             logger.info(f"Canceling active runs {active_run_ids}")
             [self.runs_api.cancel_run(_) for _ in active_run_ids]
 
-    def _submit_job(self, job_config: dict, is_streaming: bool, run_immediately: bool):
+    def _submit_job(self, job_config: dict, is_streaming: bool, run_stream_job_immediately: bool):
         job_resp = self.jobs_api.create_job(job_config)
         logger.info(f"Created Job with ID {job_resp['job_id']}")
 
-        if is_streaming or run_immediately:
+        if is_streaming and run_stream_job_immediately:
             resp = self.jobs_api.run_now(
                 job_id=job_resp["job_id"],
                 jar_params=None,

Diff for: tests/azure/test_deploy_to_databricks.py

@@ -599,12 +599,14 @@ def test_kill_it_with_fire(self, victim):
         victim.runs_api.cancel_run.assert_has_calls(calls)
 
     def test_submit_job_batch(self, victim):
-        victim._submit_job({}, False)
+        victim._submit_job({}, False, True)
 
         victim.jobs_api.create_job.assert_called_with({})
 
+        victim.jobs_api.run_now.assert_not_called()
+
     def test_submit_job_streaming(self, victim):
-        victim._submit_job({}, True)
+        victim._submit_job({}, True, True)
 
         victim.jobs_api.create_job.assert_called_with({})
 
@@ -615,3 +617,10 @@ def test_submit_job_streaming(self, victim):
             python_params=None,
             spark_submit_params=None,
         )
+
+    def test_submit_job_streaming_without_immediate_run(self, victim):
+        victim._submit_job({}, True, False)
+
+        victim.jobs_api.create_job.assert_called_with({})
+
+        victim.jobs_api.run_now.assert_not_called()