Install black

Author: gerardsegarra

src/app.py

@@ -24,7 +24,7 @@
 if hasattr(logging, loglevel_flask):
     loglevel_flask = getattr(logging, loglevel_flask)
     log.setLevel(loglevel_flask)
-logging.getLogger('apscheduler.executors.default').setLevel(logging.WARNING)
+logging.getLogger("apscheduler.executors.default").setLevel(logging.WARNING)
 
 jobs = dict()
 job_handler = JobEventsHandler()
@@ -82,9 +82,7 @@ def process_workflow_job():
                 app.logger.error(f"Job {job.id} was in progress before being queued")
                 del jobs[job.id]
             else:
-                time_to_start = (
-                    job.time_start - job_requested.time_start
-                ).seconds
+                time_to_start = (job.time_start - job_requested.time_start).seconds
 
         context_details = {
             **context_details,
@@ -105,9 +103,7 @@ def process_workflow_job():
             app.logger.warning(f"Job {job.id} is {job.action} but not stored!")
             time_to_finish = 0
         else:
-            time_to_finish = (
-                job.time_completed - job.time_start
-            ).seconds
+            time_to_finish = (job.time_completed - job.time_start).seconds
             # delete from memory
             del jobs[job.id]
 
@@ -129,7 +125,7 @@ def process_workflow_job():
     return True
 
 
-@scheduler.task('interval', id='monitor_queued', seconds=30)
+@scheduler.task("interval", id="monitor_queued", seconds=30)
 def monitor_queued_jobs():
     """Return the job that has been queued and not starting for long time."""
     app.logger.debug("Starting monitor_queued_jobs")

src/github.py

@@ -1,7 +1,7 @@
 from utils import parse_datetime
 
 
-class GithubJob():
+class GithubJob:
     def __init__(self, json_body: str):
         self.data = json_body
 

tests/test_jobs.py

@@ -12,12 +12,12 @@ def new_job_event():
 
 @pytest.fixture
 def in_progress_job_event():
-    return {"workflow_job": {"id": "workflow_id"},  "action": "in_progress"}
+    return {"workflow_job": {"id": "workflow_id"}, "action": "in_progress"}
 
 
 @pytest.fixture
 def completed_job_event():
-    return {"workflow_job": {"id": "workflow_id"},  "action": "completed"}
+    return {"workflow_job": {"id": "workflow_id"}, "action": "completed"}
 
 
 def test_new_job(new_job_event):

tests/tests.py

@@ -41,7 +41,7 @@ def test_method_not_allowed(client):
 
 
 def test_headers_not_correct(client, caplog):
-    response = client.post(HOOK_ENDPOINT, headers={'User-Agent': 'foo'})
+    response = client.post(HOOK_ENDPOINT, headers={"User-Agent": "foo"})
     assert response.status_code == 401
     assert caplog.messages == [
         "User-Agent is foo",
@@ -197,5 +197,5 @@ def test_line_break_in_job_name(client, caplog):
     assert caplog.messages == [
         'action=queued repository=foo/foo branch=new-feature-branch job_id=6 run_id=10 job_name="Build and push '
         'images (actions-runner-dind, NPROC=2 , runner-images/devops/actions-runner-dind, l..."'
-        ' workflow=CI requestor=testerbot'
+        " workflow=CI requestor=testerbot"
     ]