test: expand test coverage

pyproject.toml

@@ -37,6 +37,7 @@ Repository = "https://github.com/DIRACGrid/intercede"
 testing = [
   "pytest",
   "pytest-asyncio",
+  "pytest-cov",
   "pytest-xdist",
   "pyyaml",
   "ty",
@@ -59,6 +60,19 @@ markers = [
   "destructive_fetch: asserts one-shot/destructive output-retrieval semantics",
 ]
 
+[tool.coverage.run]
+branch = true
+patch = ["subprocess"]  # also measure the tests that start a fresh interpreter
+source = ["intercede"]
+
+[tool.coverage.report]
+show_missing = true
+skip_covered = true
+exclude_also = [
+  "if TYPE_CHECKING:",
+  "^\\s*\\.\\.\\.$",  # protocol method bodies
+]
+
 [tool.mypy]
 strict = true
 plugins = ["pydantic.mypy"]

tests/test_smoke.py

@@ -12,3 +12,19 @@ def test_package_imports():
 def test_version_is_defined():
     assert isinstance(intercede.version, str)
     assert intercede.version
+
+
+def test_version_falls_back_when_not_installed():
+    """Running from a source checkout, with no package metadata, still imports."""
+    import subprocess
+    import sys
+
+    code = (
+        "import importlib.metadata as metadata\n"
+        "def missing(name):\n"
+        "    raise metadata.PackageNotFoundError(name)\n"
+        "metadata.version = missing\n"
+        "import intercede\n"
+        "assert intercede.__version__ == intercede.version == 'unknown', intercede.version\n"
+    )
+    subprocess.run([sys.executable, "-c", code], check=True)  # noqa: S603  (constant input)

tests/unit/test_config.py

@@ -113,6 +113,12 @@ def test_a_field_of_another_kind(
     ):
         assert f"{field}: not for {where} (expected: " in rejected(**data)
 
+    @pytest.mark.parametrize("data", [None, "arc", ["kind", "arc"]])
+    def test_something_that_is_not_a_configuration(self, data: Any):
+        # For example a config file read as text, or a missing table.
+        with pytest.raises(ConfigurationError, match="kind: must be one of"):
+            validate_config(data)
+
     def test_every_missing_field_is_named_at_once(self):
         message = rejected(kind="batch", transport="ssh", scheduler="slurm")
         for field in ("host", "user", "known_hosts"):

tests/unit/test_fake_backend.py

@@ -1,15 +1,22 @@
-"""Behaviour of the fake that needs its test hooks: completion, partial failure, refusal."""
+"""The fake's test hooks, and the rules the conformance suite cannot reach without them.
+
+``test_conformance.py`` already runs the contract against the fake; what is here
+needs a job to finish, fail or be refused on demand.
+"""
 
 from __future__ import annotations
 
+from collections.abc import Awaitable, Callable
 from pathlib import Path
+from typing import Any
 
 import pytest
 
 from intercede import (
     ContainerSpec,
     CopyVars,
     FileRef,
+    JobHandle,
     JobStatus,
     OutputMember,
     OutputSpec,
@@ -21,110 +28,165 @@
 
 SPEC = SubmissionSpec(executable="/bin/true")
 
-
-async def test_partial_submission_keeps_copy_indexes():
-    async with FakeBackend(refuse_copies={0, 2}) as backend:
-        copies = [CopyVars(environment={"STAMP": str(i)}) for i in range(4)]
-        sub = await backend.submit(SPEC, copies)
-    assert sorted(sub.handles) == [1, 3]
-    assert sorted(sub.failures) == [0, 2]
-
-
-async def test_injected_failure_is_reported_per_job():
-    async with FakeBackend(fail_jobs={"1"}) as backend:
-        sub = await backend.submit(SPEC, 2)
-        good, bad = sub.handles[0], sub.handles[1]
-        killed = await backend.kill([good, bad])
-        assert killed[good].ok
-        assert not killed[bad].ok
-
-
-async def test_whole_operation_failure_raises():
-    async with FakeBackend() as backend:
-        backend.unavailable = True
-        with pytest.raises(ResourceUnavailableError):
-            await backend.submit(SPEC)
-
-
-async def test_output_is_repeatable_and_lands_under_dest(tmp_path: Path):
-    async with FakeBackend() as backend:
-        handle = (await backend.submit(SPEC)).handles[0]
-        backend.finish(handle, stdout="hello", files={"result/out.txt": b"42"})
-        for _ in range(2):
+Operation = Callable[[FakeBackend, list[JobHandle], Path], Awaitable[Any]]
+PER_JOB: dict[str, Operation] = {
+    "get_output": lambda b, h, d: b.get_output(h, d),
+    "kill": lambda b, h, d: b.kill(h),
+    "purge": lambda b, h, d: b.purge(h),
+    "get_diagnostics": lambda b, h, d: b.get_diagnostics(h),
+}
+
+
+class TestHooks:
+    async def test_refused_copies_keep_their_indexes(self):
+        async with FakeBackend(refuse_copies={0, 2}) as backend:
+            submission = await backend.submit(SPEC, 4)
+        assert sorted(submission.handles) == [1, 3]
+        assert sorted(submission.failures) == [0, 2]
+
+    @pytest.mark.parametrize("operation", sorted(PER_JOB))
+    async def test_an_injected_failure_is_per_job(self, operation: str, tmp_path: Path):
+        async with FakeBackend(fail_jobs={"0"}) as backend:
+            handle = (await backend.submit(SPEC)).handles[0]
+            backend.finish(handle)
+            outcome = (await PER_JOB[operation](backend, [handle], tmp_path))[handle]
+            assert (await backend.get_status([handle]))[handle] == JobStatus.DONE
+        assert not outcome.ok and outcome.reason == "injected failure"
+
+    async def test_an_unavailable_resource_fails_the_whole_operation(self):
+        async with FakeBackend() as backend:
+            backend.unavailable = True
+            with pytest.raises(ResourceUnavailableError):
+                await backend.submit(SPEC)
+
+    @pytest.mark.parametrize(
+        ("options", "spec"),
+        [
+            (
+                {"resource_staging": False},
+                SubmissionSpec(
+                    executable="x",
+                    inputs=[FileRef(name="in", source="https://example.org/in")],
+                ),
+            ),
+            (
+                {"resource_staging": False},
+                SubmissionSpec(
+                    executable="x",
+                    outputs=OutputSpec(
+                        members=[
+                            OutputMember(pattern="kept"),
+                            OutputMember(pattern="sent", destination="https://e.org/o"),
+                        ]
+                    ),
+                ),
+            ),
+            (
+                {"containers": False},
+                SubmissionSpec(executable="x", container=ContainerSpec(image="a")),
+            ),
+        ],
+        ids=["url-input", "url-output", "container"],
+    )
+    async def test_what_a_resource_cannot_do_is_refused_before_anything_runs(
+        self, options: dict[str, bool], spec: SubmissionSpec
+    ):
+        async with FakeBackend(**options) as backend:
+            with pytest.raises(SpecificationRejectedError):
+                await backend.submit(spec, 3)
+            assert backend._jobs == {}
+
+
+class TestSubmit:
+    async def test_a_copy_value_wins_over_the_spec(self):
+        spec = SubmissionSpec(
+            executable="x", environment={"SITE": "a", "STAMP": "spec"}
+        )
+        copies = [
+            CopyVars(environment={"STAMP": str(i)}, secrets={"s": b"%d" % i})
+            for i in range(2)
+        ]
+        async with FakeBackend() as backend:
+            submission = await backend.submit(spec, copies)
+            jobs = [backend._jobs[submission.handles[i].id] for i in range(2)]
+        assert [job.environment for job in jobs] == [
+            {"SITE": "a", "STAMP": "0"},
+            {"SITE": "a", "STAMP": "1"},
+        ]
+        assert [job.secrets for job in jobs] == [{"s": b"0"}, {"s": b"1"}]
+
+    async def test_job_ids_are_unique_across_submissions(self):
+        async with FakeBackend() as backend:
+            first = await backend.submit(SPEC, 2)
+            second = await backend.submit(SPEC, 2)
+        ids = {h.id for s in (first, second) for h in s.handles.values()}
+        assert len(ids) == 4
+
+
+class TestKill:
+    async def test_aborts_an_active_job(self):
+        async with FakeBackend() as backend:
+            handle = (await backend.submit(SPEC)).handles[0]
+            backend.finish(handle, status=JobStatus.RUNNING)
+            assert (await backend.kill([handle]))[handle].ok
+            assert (await backend.get_status([handle]))[handle] == JobStatus.ABORTED
+
+    async def test_leaves_a_finished_job_alone(self):
+        async with FakeBackend() as backend:
+            handle = (await backend.submit(SPEC)).handles[0]
+            backend.finish(handle, status=JobStatus.FAILED)
+            outcome = (await backend.kill([handle]))[handle]
+            assert outcome.ok and outcome.reason == "already finished"
+            assert (await backend.get_status([handle]))[handle] == JobStatus.FAILED
+
+
+class TestOutput:
+    async def test_collected_again_and_again(self, tmp_path: Path):
+        async with FakeBackend() as backend:
+            handle = (await backend.submit(SPEC)).handles[0]
+            backend.finish(handle, stdout="hello", files={"deep/out.txt": b"42"})
+            for _ in range(2):
+                outcome = (await backend.get_output([handle], tmp_path))[handle]
+                assert outcome.ok and outcome.value is not None
+                assert outcome.value.stdout is not None
+                assert outcome.value.stdout.read_text() == "hello"
+                assert outcome.value.files["deep/out.txt"].read_bytes() == b"42"
+
+    async def test_refused_while_active(self, tmp_path: Path):
+        async with FakeBackend() as backend:
+            handle = (await backend.submit(SPEC)).handles[0]
             outcome = (await backend.get_output([handle], tmp_path))[handle]
-            assert outcome.ok and outcome.value is not None
-            assert outcome.value.stdout is not None
-            assert outcome.value.stdout.read_text() == "hello"
-            assert outcome.value.files["result/out.txt"].read_bytes() == b"42"
-            assert outcome.value.files["result/out.txt"].is_relative_to(
-                tmp_path.resolve()
-            )
-
-
-async def test_output_of_a_running_job_fails_per_job(tmp_path: Path):
-    async with FakeBackend() as backend:
-        handle = (await backend.submit(SPEC)).handles[0]
-        assert not (await backend.get_output([handle], tmp_path))[handle].ok
-
-
-@pytest.mark.parametrize("name", ["../../escaped", "/abs/escaped"])
-async def test_output_members_cannot_escape_dest(tmp_path: Path, name: str):
-    dest = tmp_path / "dest"
+        assert not outcome.ok and outcome.reason == "job has not finished"
+
+    @pytest.mark.parametrize("name", ["../../escaped", "/abs/escaped"])
+    async def test_an_escaping_member_writes_nothing_at_all(
+        self, tmp_path: Path, name: str
+    ):
+        dest = tmp_path / "dest"
+        async with FakeBackend() as backend:
+            handle = (await backend.submit(SPEC)).handles[0]
+            backend.finish(handle, stdout="x", files={"ok.txt": b"x", name: b"x"})
+            outcome = (await backend.get_output([handle], dest))[handle]
+        assert not outcome.ok
+        assert not dest.exists()  # not even the safe members or stdout
+
+
+async def test_a_purged_job_is_gone_for_every_operation(tmp_path: Path):
     async with FakeBackend() as backend:
         handle = (await backend.submit(SPEC)).handles[0]
-        backend.finish(handle, files={name: b"x"})
-        outcome = (await backend.get_output([handle], dest))[handle]
-    assert not outcome.ok
-    assert not any(p.name == "escaped" for p in tmp_path.rglob("*"))
+        backend.finish(handle)
+        assert (await backend.purge([handle]))[handle].ok
+        assert (await backend.get_status([handle]))[handle] == JobStatus.UNKNOWN
+        for operation in PER_JOB.values():
+            assert not (await operation(backend, [handle], tmp_path))[handle].ok
 
 
-async def test_kill_stops_active_jobs():
-    async with FakeBackend() as backend:
-        handle = (await backend.submit(SPEC)).handles[0]
-        await backend.kill([handle])
-        assert (await backend.get_status([handle]))[handle] == JobStatus.ABORTED
-
-
-async def test_diagnostics_are_repeatable():
+async def test_diagnostics_read_a_running_job_by_name():
     async with FakeBackend() as backend:
         handle = (await backend.submit(SPEC)).handles[0]
         backend.finish(handle, status=JobStatus.RUNNING, stdout="progress")
-        for _ in range(2):
-            outcome = (await backend.get_diagnostics([handle]))[handle]
-            assert outcome.value is not None
-            assert outcome.value.content == "progress"
+        default = (await backend.get_diagnostics([handle]))[handle]
         named = (await backend.get_diagnostics([handle], filename="job.log"))[handle]
-        assert outcome.value.filename == "stdout"  # the default
-        assert named.value is not None and named.value.filename == "job.log"
-
-
-@pytest.mark.parametrize(
-    "spec",
-    [
-        SubmissionSpec(
-            executable="x", inputs=[FileRef(name="in", source="https://example.org/in")]
-        ),
-        SubmissionSpec(
-            executable="x",
-            outputs=OutputSpec(
-                members=[OutputMember(pattern="out", destination="https://e.org/o")]
-            ),
-        ),
-    ],
-)
-async def test_spec_needing_resource_staging_is_rejected_at_submit(
-    spec: SubmissionSpec,
-):
-    async with FakeBackend(resource_staging=False) as backend:
-        with pytest.raises(SpecificationRejectedError):
-            await backend.submit(spec)
-
-
-async def test_spec_with_container_is_rejected_when_unsupported():
-    async with FakeBackend(containers=False) as backend:
-        with pytest.raises(SpecificationRejectedError):
-            await backend.submit(
-                SubmissionSpec(
-                    executable="x", container=ContainerSpec(image="docker://alpine")
-                )
-            )
+    assert default.value is not None and default.value.filename == "stdout"
+    assert default.value.content == "progress"
+    assert named.value is not None and named.value.filename == "job.log"

tests/unit/test_models.py

@@ -10,6 +10,7 @@
     FileRef,
     JobHandle,
     JobOutput,
+    JobStatus,
     OpOutcome,
     OutputMember,
     Submission,
@@ -37,6 +38,13 @@ def test_a_newer_format_fails_loudly(self):
     def test_the_same_id_on_two_resources_is_two_jobs(self):
         assert len({handle(resource="a"), handle(resource="b")}) == 2
 
+    def test_key_separators_inside_parts_are_escaped(self):
+        # Without escaping, ("a:b", "c") and ("a", "b:c") would share a key.
+        first = handle("c", resource="a:b")
+        second = handle("b:c", resource="a")
+        assert first.key != second.key
+        assert str(first) == first.key == "fake:a%3Ab:c"
+
 
 @pytest.mark.parametrize("name", ["", ".", "/etc/passwd", "../up", "a/../../up"])
 def test_names_must_stay_in_the_working_directory(name: str):
@@ -91,3 +99,16 @@ def test_secrets_that_clash_with_inputs_are_named():
     spec = SubmissionSpec(executable="x", inputs=[FileRef(name="in", source="a")])
     assert spec.clashes_with(CopyVars(secrets={"in": b"s", "other": b"s"})) == {"in"}
     assert spec.clashes_with(CopyVars()) == frozenset()
+
+
+def test_job_status_values_are_stable():
+    """Stored statuses are compared by value: renaming one is a breaking change."""
+    assert {status.value for status in JobStatus} == {
+        "Submitted",
+        "Waiting",
+        "Running",
+        "Done",
+        "Failed",
+        "Aborted",
+        "Unknown",
+    }