Merge pull request #676 from opensafely-core/check-codelists-ok

Check for out-of-date codelists and fail the job request if any actions require the database

jobrunner/cli/local_run.py

@@ -448,6 +448,7 @@ def create_job_request_and_jobs(project_dir, actions, force_run_dependencies):
         requested_actions=actions,
         cancelled_actions=[],
         workspace=project_dir.name,
+        codelists_ok=True,
         database_name="dummy",
         force_run_dependencies=force_run_dependencies,
         branch="",

jobrunner/create_or_update_jobs.py

@@ -90,6 +90,11 @@ def create_jobs(job_request):
     new_jobs = get_new_jobs_to_run(job_request, pipeline_config, latest_jobs)
     assert_new_jobs_created(job_request, new_jobs, latest_jobs)
     resolve_reusable_action_references(new_jobs)
+
+    # check for database actions in the new jobs, and raise an exception if
+    # codelists are out of date
+    assert_codelists_ok(job_request, new_jobs)
+
     # There is a delay between getting the current jobs (which we fetch from
     # the database and the disk) and inserting our new jobs below. This means
     # the state of the world may have changed in the meantime. Why is this OK?
@@ -299,6 +304,18 @@ def assert_new_jobs_created(job_request, new_jobs, current_jobs):
     raise Exception(f"Unexpected job states after scheduling: {current_job_states}")
 
 
+def assert_codelists_ok(job_request, new_jobs):
+    if job_request.codelists_ok:
+        return True
+    for job in new_jobs:
+        # Codelists are out of date; fail the entire job request if any job
+        # requires database access
+        if job.requires_db:
+            raise JobRequestError(
+                f"Codelists are out of date (required by action {job.action})"
+            )
+
+
 def create_failed_job(job_request, exception):
     """
     Sometimes we want to say to the job-server (and the user): your JobRequest

jobrunner/models.py

@@ -101,6 +101,7 @@ class JobRequest:
     requested_actions: list
     cancelled_actions: list
     workspace: str
+    codelists_ok: bool
     database_name: str
     force_run_dependencies: bool = False
     branch: str = None

jobrunner/sync.py

@@ -126,6 +126,7 @@ def job_request_from_remote_format(job_request):
         requested_actions=job_request["requested_actions"],
         cancelled_actions=job_request["cancelled_actions"],
         workspace=job_request["workspace"]["name"],
+        codelists_ok=job_request["codelists_ok"],
         # Transitional code while we move the location of the `database_name` attribute
         database_name=(
             job_request["database_name"]

tests/factories.py

@@ -19,6 +19,7 @@
     "requested_actions": ["action"],
     "cancelled_actions": [],
     "workspace": "workspace",
+    "codelists_ok": True,
     "database_name": "default",
     "original": {
         "created_by": "testuser",

tests/lib/test_log_utils.py

@@ -24,6 +24,7 @@
     commit="commit",
     requested_actions=["action"],
     cancelled_actions=[],
+    codelists_ok=True,
     database_name="dummy",
 )
 

tests/test_create_or_update_jobs.py

@@ -1,3 +1,4 @@
+import re
 import uuid
 from pathlib import Path
 from unittest import mock
@@ -36,6 +37,7 @@ def test_create_or_update_jobs(tmp_work_dir, db):
         requested_actions=["generate_cohort"],
         cancelled_actions=[],
         workspace="1",
+        codelists_ok=True,
         database_name="dummy",
         original=dict(
             created_by="user",
@@ -77,6 +79,7 @@ def test_create_or_update_jobs_with_git_error(tmp_work_dir):
         requested_actions=["generate_cohort"],
         cancelled_actions=[],
         workspace="1",
+        codelists_ok=True,
         database_name="dummy",
         original=dict(
             created_by="user",
@@ -245,6 +248,7 @@ def test_validate_job_request(params, exc_msg, monkeypatch):
         requested_actions=["generate_cohort"],
         cancelled_actions=[],
         workspace="1",
+        codelists_ok=True,
         database_name="default",  # note db from from job-server is 'default'
         original=dict(
             created_by="user",
@@ -271,6 +275,7 @@ def make_job_request(action=None, actions=None, **kwargs):
         repo_url="https://example.com/repo.git",
         commit="abcdef0123456789",
         workspace="1",
+        codelists_ok=True,
         database_name="default",
         requested_actions=actions,
         cancelled_actions=[],
@@ -341,3 +346,34 @@ def test_create_failed_job_nothing_to_do(db):
     assert spans[0].status.status_code == trace.StatusCode.UNSET
     assert spans[1].name == "JOB"
     assert spans[1].status.status_code == trace.StatusCode.UNSET
+
+
+@pytest.mark.parametrize(
+    "requested_action,expect_error",
+    [("generate_cohort", True), ("analyse_data", True), ("standalone_action", False)],
+)
+def test_create_or_update_jobs_with_out_of_date_codelists(
+    tmp_work_dir, requested_action, expect_error
+):
+    project = TEST_PROJECT + (
+        """
+  standalone_action:
+    run: python:latest analysis/do_something.py
+    outputs:
+      moderately_sensitive:
+        something: done.txt
+"""
+    )
+    job_request = make_job_request(action=requested_action, codelists_ok=False)
+    if expect_error:
+        # The error reports the action that needed the up-to-date codelists, even if that
+        # wasn't the action explicitly requested
+        with pytest.raises(
+            JobRequestError,
+            match=re.escape(
+                "Codelists are out of date (required by action generate_cohort)"
+            ),
+        ):
+            create_jobs_with_project_file(job_request, project)
+    else:
+        assert create_jobs_with_project_file(job_request, project) == 1

tests/test_integration.py

@@ -70,6 +70,7 @@ def test_integration_with_cohortextractor(
             "repo": str(test_repo.path),
             "branch": "HEAD",
         },
+        "codelists_ok": True,
         "database_name": "dummy",
         "sha": test_repo.commit,
         "created_by": "user",
@@ -121,6 +122,7 @@ def test_integration_with_cohortextractor(
             "repo": str(test_repo.path),
             "branch": "HEAD",
         },
+        "codelists_ok": True,
         "database_name": "dummy",
         "sha": test_repo.commit,
         "created_by": "user",
@@ -244,6 +246,7 @@ def test_integration_with_databuilder(
             "repo": str(test_repo.path),
             "branch": "HEAD",
         },
+        "codelists_ok": True,
         "database_name": "dummy",
         "sha": test_repo.commit,
         "created_by": "user",
@@ -292,6 +295,7 @@ def test_integration_with_databuilder(
             "repo": str(test_repo.path),
             "branch": "HEAD",
         },
+        "codelists_ok": True,
         "database_name": "dummy",
         "sha": test_repo.commit,
         "created_by": "user",

tests/test_sync.py

@@ -19,6 +19,7 @@ def test_job_request_from_remote_format():
         "cancelled_actions": ["analyse"],
         "force_run_dependencies": True,
         "sha": "abcdef",
+        "codelists_ok": True,
         "created_by": "user",
         "project": "project",
         "orgs": ["org"],
@@ -29,6 +30,7 @@ def test_job_request_from_remote_format():
         commit="abcdef",
         branch="master",
         workspace="testing",
+        codelists_ok=True,
         database_name="default",
         requested_actions=["generate_cohort"],
         cancelled_actions=["analyse"],
@@ -52,6 +54,7 @@ def test_job_request_from_remote_format_database_name_fallback():
         "cancelled_actions": ["analyse"],
         "force_run_dependencies": True,
         "sha": "abcdef",
+        "codelists_ok": True,
         "created_by": "user",
         "project": "project",
         "orgs": ["org"],
@@ -62,6 +65,7 @@ def test_job_request_from_remote_format_database_name_fallback():
         commit="abcdef",
         branch="master",
         workspace="testing",
+        codelists_ok=True,
         database_name="default",
         requested_actions=["generate_cohort"],
         cancelled_actions=["analyse"],