diff --git a/pyproject.toml b/pyproject.toml index 0765bcf..a86f715 100644 --- a/pyproject.toml +++ b/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"] diff --git a/tests/test_smoke.py b/tests/test_smoke.py index 9773f64..5fd2a18 100644 --- a/tests/test_smoke.py +++ b/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) diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py index 02b1ffc..2856b53 100644 --- a/tests/unit/test_config.py +++ b/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"): diff --git a/tests/unit/test_fake_backend.py b/tests/unit/test_fake_backend.py index 5f8f452..6a3ce61 100644 --- a/tests/unit/test_fake_backend.py +++ b/tests/unit/test_fake_backend.py @@ -1,8 +1,14 @@ -"""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 @@ -10,6 +16,7 @@ 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" diff --git a/tests/unit/test_models.py b/tests/unit/test_models.py index fe11e1b..ebb6bc4 100644 --- a/tests/unit/test_models.py +++ b/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", + }