Feature/glue start job (#4959)

moto/glue/exceptions.py

@@ -73,3 +73,8 @@ class CrawlerRunningException(GlueClientError):
 class CrawlerNotRunningException(GlueClientError):
     def __init__(self, msg):
         super().__init__("CrawlerNotRunningException", msg)
+
+
+class ConcurrentRunsExceededException(GlueClientError):
+    def __init__(self, msg):
+        super().__init__("ConcurrentRunsExceededException", msg)

moto/glue/models.py

@@ -16,6 +16,7 @@ from .exceptions import (
     PartitionNotFoundException,
     VersionNotFoundException,
     JobNotFoundException,
+    ConcurrentRunsExceededException,
 )
 from ..utilities.paginator import paginate
 
@@ -34,6 +35,7 @@ class GlueBackend(BaseBackend):
         self.databases = OrderedDict()
         self.crawlers = OrderedDict()
         self.jobs = OrderedDict()
+        self.job_runs = OrderedDict()
 
     @staticmethod
     def default_vpc_endpoint_service(service_region, zones):
@@ -205,6 +207,14 @@ class GlueBackend(BaseBackend):
         except KeyError:
             raise JobNotFoundException(name)
 
+    def start_job_run(self, name):
+        job = self.get_job(name)
+        return job.start_job_run()
+
+    def get_job_run(self, name, run_id):
+        job = self.get_job(name)
+        return job.get_job_run(run_id)
+
     @paginate(pagination_model=PAGINATION_MODEL)
     def list_jobs(self):
         return [job for _, job in self.jobs.items()]
@@ -464,6 +474,7 @@ class FakeJob:
         self.max_retries = max_retries
         self.allocated_capacity = allocated_capacity
         self.timeout = timeout
+        self.state = "READY"
         self.max_capacity = max_capacity
         self.security_configuration = security_configuration
         self.tags = tags
@@ -501,5 +512,70 @@ class FakeJob:
             "GlueVersion": self.glue_version,
         }
 
+    def start_job_run(self):
+        if self.state == "RUNNING":
+            raise ConcurrentRunsExceededException(
+                f"Job with name {self.name} already running"
+            )
+        fake_job_run = FakeJobRun(job_name=self.name)
+        self.state = "RUNNING"
+        return fake_job_run.job_run_id
+
+    def get_job_run(self, run_id):
+        fake_job_run = FakeJobRun(job_name=self.name, job_run_id=run_id)
+        return fake_job_run
+
+
+class FakeJobRun:
+    def __init__(
+        self,
+        job_name: int,
+        job_run_id: str = "01",
+        arguments: dict = None,
+        allocated_capacity: int = None,
+        timeout: int = None,
+        worker_type: str = "Standard",
+    ):
+        self.job_name = job_name
+        self.job_run_id = job_run_id
+        self.arguments = arguments
+        self.allocated_capacity = allocated_capacity
+        self.timeout = timeout
+        self.worker_type = worker_type
+        self.started_on = datetime.utcnow()
+        self.modified_on = datetime.utcnow()
+        self.completed_on = datetime.utcnow()
+
+    def get_name(self):
+        return self.job_name
+
+    def as_dict(self):
+        return {
+            "Id": self.job_run_id,
+            "Attempt": 1,
+            "PreviousRunId": "01",
+            "TriggerName": "test_trigger",
+            "JobName": self.job_name,
+            "StartedOn": self.started_on.isoformat(),
+            "LastModifiedOn": self.modified_on.isoformat(),
+            "CompletedOn": self.completed_on.isoformat(),
+            "JobRunState": "SUCCEEDED",
+            "Arguments": self.arguments or {"runSpark": "spark -f test_file.py"},
+            "ErrorMessage": "",
+            "PredecessorRuns": [
+                {"JobName": "string", "RunId": "string"},
+            ],
+            "AllocatedCapacity": self.allocated_capacity or 123,
+            "ExecutionTime": 123,
+            "Timeout": self.timeout or 123,
+            "MaxCapacity": 123.0,
+            "WorkerType": self.worker_type,
+            "NumberOfWorkers": 123,
+            "SecurityConfiguration": "string",
+            "LogGroupName": "test/log",
+            "NotificationProperty": {"NotifyDelayAfter": 123},
+            "GlueVersion": "0.9",
+        }
+
 
 glue_backend = GlueBackend()

moto/glue/responses.py

@@ -371,6 +371,17 @@ class GlueResponse(BaseResponse):
         job = self.glue_backend.get_job(name)
         return json.dumps({"Job": job.as_dict()})
 
+    def start_job_run(self):
+        name = self.parameters.get("JobName")
+        job_run_id = self.glue_backend.start_job_run(name)
+        return json.dumps(dict(JobRunId=job_run_id))
+
+    def get_job_run(self):
+        name = self.parameters.get("JobName")
+        run_id = self.parameters.get("RunId")
+        job_run = self.glue_backend.get_job_run(name, run_id)
+        return json.dumps({"JobRun": job_run.as_dict()})
+
     def list_jobs(self):
         next_token = self._get_param("NextToken")
         max_results = self._get_int_param("MaxResults")

tests/test_glue/test_glue.py

@@ -108,6 +108,55 @@ def test_get_job_exists():
     assert response["Job"]["GlueVersion"]
 
 
+@mock_glue
+def test_start_job_run():
+    client = create_glue_client()
+    job_name = create_test_job(client)
+    response = client.start_job_run(JobName=job_name)
+    assert response["JobRunId"]
+
+
+@mock_glue
+def test_start_job_run_already_running():
+    client = create_glue_client()
+    job_name = create_test_job(client)
+    client.start_job_run(JobName=job_name)
+    with pytest.raises(ClientError) as exc:
+        client.start_job_run(JobName=job_name)
+    exc.value.response["Error"]["Code"].should.equal("ConcurrentRunsExceededException")
+    exc.value.response["Error"]["Message"].should.match(
+        f"Job with name {job_name} already running"
+    )
+
+
+@mock_glue
+def test_get_job_run():
+    client = create_glue_client()
+    job_name = create_test_job(client)
+    response = client.get_job_run(JobName=job_name, RunId="01")
+    assert response["JobRun"]["Id"]
+    assert response["JobRun"]["Attempt"]
+    assert response["JobRun"]["PreviousRunId"]
+    assert response["JobRun"]["TriggerName"]
+    assert response["JobRun"]["StartedOn"]
+    assert response["JobRun"]["LastModifiedOn"]
+    assert response["JobRun"]["CompletedOn"]
+    assert response["JobRun"]["JobRunState"]
+    assert response["JobRun"]["Arguments"]
+    assert response["JobRun"]["ErrorMessage"] == ""
+    assert response["JobRun"]["PredecessorRuns"]
+    assert response["JobRun"]["AllocatedCapacity"]
+    assert response["JobRun"]["ExecutionTime"]
+    assert response["JobRun"]["Timeout"]
+    assert response["JobRun"]["MaxCapacity"]
+    assert response["JobRun"]["WorkerType"]
+    assert response["JobRun"]["NumberOfWorkers"]
+    assert response["JobRun"]["SecurityConfiguration"]
+    assert response["JobRun"]["LogGroupName"]
+    assert response["JobRun"]["NotificationProperty"]
+    assert response["JobRun"]["GlueVersion"]
+
+
 @mock_glue
 def test_list_jobs_with_max_results():
     client = create_glue_client()