graphile_worker 0.14.0

High performance Rust/PostgreSQL job queue (also suitable for getting jobs generated by PostgreSQL triggers/functions out into a different work queue)
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216

"""Regression tests for preserving API coverage and failure signals in release checks."""
import importlib.util
import json
import os
from pathlib import Path
import subprocess
import tempfile
import unittest
from unittest.mock import patch

SPEC = importlib.util.spec_from_file_location("semver_checks", Path(__file__).with_name("semver_checks.py"))
adapter = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(adapter)


class ApiCheckTests(unittest.TestCase):
    def test_profiles_cover_each_supported_version_and_other_stable_features(self):
        """Every supported telemetry API and every other stable feature must be checked."""
        current = {"features": {name: [] for name in (
            "default", "driver-sqlx", "driver-tokio-postgres", "opentelemetry_0_30",
            "opentelemetry_0_31", "opentelemetry_0_32", "_private", "nightly",
        )}}
        baseline = {"features": {name: [] for name in (
            "default", "driver-sqlx", "opentelemetry_0_30", "opentelemetry_0_31",
        )}}
        profiles = adapter.profiles(current, baseline)
        self.assertEqual(len(profiles), 4)
        for _, current_set, baseline_set in profiles:
            self.assertLessEqual(len(current_set & adapter.TELEMETRY), 1)
            self.assertLessEqual(len(baseline_set & adapter.TELEMETRY), 1)
            self.assertTrue({"default", "driver-sqlx", "driver-tokio-postgres"} <= current_set)
            self.assertTrue({"default", "driver-sqlx"} <= baseline_set)
        self.assertEqual(set.union(*(item[1] for item in profiles)), adapter.stable_features(current["features"]))
        self.assertEqual(set.union(*(item[2] for item in profiles)), adapter.stable_features(baseline["features"]))
        self.assertEqual(profiles[-1][2] & adapter.TELEMETRY, set())

    def test_packages_without_telemetry_keep_native_feature_selection(self):
        """Ordinary crates keep the native checker feature policy."""
        self.assertEqual(adapter.profiles({"features": {"default": []}}, {"features": {}}), [])

    def test_option_replacement_preserves_paths_and_other_checker_flags(self):
        """Paths with spaces and release policy flags survive argument adaptation."""
        arguments = ["--release-type", "minor", "--baseline-root=/a path/Cargo.toml", "--package", "worker"]
        result = adapter.replace_option(arguments, "--baseline-root", "/another path/Cargo.toml")
        self.assertEqual(adapter.option(result, "--baseline-root"), "/another path/Cargo.toml")
        self.assertEqual(result[:2], arguments[:2])
        self.assertEqual(result[-2:], arguments[-2:])
        with self.assertRaises(ValueError):
            adapter.option(["--package", "one", "--package", "two"], "--package")

    def test_incompatibility_from_any_profile_is_not_hidden_by_later_success(self):
        """A later successful profile must not erase an earlier API incompatibility."""
        calls = []
        def checker(command, check):
            calls.append(command)
            return subprocess.CompletedProcess(command, [0, 100, 0][len(calls) - 1])
        self.assertEqual(adapter.run_checks([(str(i), [str(i)]) for i in range(3)], checker), 100)
        self.assertEqual(len(calls), 3)

    def test_checker_build_failure_is_never_treated_as_an_api_result(self):
        """A build error must remain a fatal error, not an API compatibility result."""
        calls = []
        def checker(command, check):
            calls.append(command)
            return subprocess.CompletedProcess(command, 101)
        self.assertEqual(adapter.run_checks([("first", ["one"]), ("second", ["two"])], checker), 101)
        self.assertEqual(calls, [["one"]])

    def test_all_native_checks_must_succeed(self):
        """Only success from every executed profile produces overall success."""
        self.assertEqual(adapter.run_checks([("one", ["one"]), ("two", ["two"])],
                         lambda command, check: subprocess.CompletedProcess(command, 0)), 0)

    def test_baseline_fix_changes_only_known_missing_build_features(self):
        """Only known build features change; source, license and original baseline stay intact."""
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            source = root / "published"
            (source / "src").mkdir(parents=True)
            manifest = source / "Cargo.toml"
            original = ('[package]\nname = "graphile_worker_database"\nversion = "0.1.6"\n'
                        '[dependencies.tokio]\nversion = "1.53.0"\nfeatures = ["rt", "sync"]\noptional = true\n')
            manifest.write_text(original)
            rust = b"pub fn existing_api() {}\n"
            (source / "src/lib.rs").write_bytes(rust)
            (source / "LICENSE").write_text("retained license fixture")
            with adapter.buildable_baseline(manifest, root / "scratch") as copied:
                self.assertNotEqual(copied, manifest)
                repaired = adapter.tomllib.loads(copied.read_text())
                self.assertEqual(repaired["dependencies"]["tokio"]["features"], ["rt", "sync", "macros", "time"])
                self.assertEqual(repaired["workspace"], {})
                self.assertEqual((copied.parent / "src/lib.rs").read_bytes(), rust)
                self.assertEqual((copied.parent / "LICENSE").read_text(), "retained license fixture")
                self.assertEqual(manifest.read_text(), original)
            self.assertFalse(copied.exists())
            self.assertEqual(manifest.read_text(), original)
            manifest.write_text(original.replace('version = "0.1.6"', 'version = "0.1.7"'))
            with adapter.buildable_baseline(manifest, root / "scratch") as unchanged:
                self.assertEqual(unchanged, manifest)

    def test_changed_baseline_layout_fails_instead_of_silently_skipping_checks(self):
        """Unknown published manifest layouts must fail visibly."""
        with tempfile.TemporaryDirectory() as directory:
            manifest = Path(directory) / "Cargo.toml"
            manifest.write_text('[package]\nname="graphile_worker_database"\nversion="0.1.6"\n'
                                '[dependencies.tokio]\nversion="1.53"\n')
            with self.assertRaises(ValueError), adapter.buildable_baseline(manifest, Path(directory) / "scratch"):
                self.fail("an unknown manifest shape must fail closed")

    def test_main_invokes_real_checker_for_each_profile_with_original_policy(self):
        """Every generated command retains the real checker and the original release policy."""
        with tempfile.TemporaryDirectory() as directory:
            manifest = Path(directory) / "Cargo.toml"
            manifest.write_text('[package]\nname="worker"\nversion="1.0.0"\n')
            args = ["semver-checks", "check-release", "--release-type", "minor", "--color", "never",
                    "--manifest-path", str(manifest), "--package", "worker", "--baseline-root", str(manifest)]
            calls = []
            def checks(commands, **kwargs):
                calls.extend(commands)
                return 100
            current = {"features": {name: [] for name in (
                "default", "driver-sqlx", "opentelemetry_0_30", "opentelemetry_0_32",
            )}}
            baseline = {"features": {name: [] for name in (
                "default", "driver-tokio-postgres", "opentelemetry_0_31",
            )}}
            with patch.dict(os.environ, {"GRAPHILE_SEMVER_CHECKS_BIN": "/real/checker"}), \
                    patch.object(adapter, "metadata", side_effect=[current, baseline]), \
                    patch.object(adapter, "run_checks", side_effect=checks):
                self.assertEqual(adapter.main(args), 100)
            self.assertEqual(
                [(label, adapter.option(command, "--current-features"),
                  adapter.option(command, "--baseline-features")) for label, command in calls],
                [
                    ("without telemetry", "default,driver-sqlx", "default,driver-tokio-postgres"),
                    ("opentelemetry_0_30", "default,driver-sqlx,opentelemetry_0_30",
                     "default,driver-tokio-postgres"),
                    ("opentelemetry_0_31", "default,driver-sqlx",
                     "default,driver-tokio-postgres,opentelemetry_0_31"),
                    ("opentelemetry_0_32", "default,driver-sqlx,opentelemetry_0_32",
                     "default,driver-tokio-postgres"),
                ],
            )
            for _, command in calls:
                self.assertEqual(command[0], "/real/checker")
                self.assertEqual(adapter.option(command, "--release-type"), "minor")
                self.assertIn("--only-explicit-features", command)
                self.assertIn("--current-features", command)
                self.assertIn("--baseline-features", command)

    def test_unexpected_feature_overrides_fail_instead_of_reducing_coverage(self):
        """Changed caller feature policy requires an explicit adapter update."""
        with patch.dict(os.environ, {"GRAPHILE_SEMVER_CHECKS_BIN": "/real/checker"}):
            with self.assertRaises(ValueError):
                adapter.main(["semver-checks", "check-release", "--default-features"])

    def test_transitive_repair_uses_only_published_source_and_temporary_config(self):
        """A transitive repair must preserve the checked manifest and all Rust bytes."""
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            database = root / "published-database"
            (database / "src").mkdir(parents=True)
            manifest = database / "Cargo.toml"
            original = ('[package]\nname="graphile_worker_database"\nversion="0.1.6"\n'
                        '[dependencies.tokio]\nversion="1.53"\nfeatures=["rt","sync"]\n')
            manifest.write_text(original)
            (database / "src/lib.rs").write_text("pub fn published_api() {}\n")
            checked = root / "Cargo.toml"
            checked_source = '[package]\nname="graphile_worker_queries"\nversion="0.1.2"\n'
            checked.write_text(checked_source)
            with patch.object(adapter, "published_database_dependency", return_value=manifest):
                with adapter.baseline_environment(checked, root / "scratch") as cwd:
                    config = adapter.tomllib.loads((cwd / ".cargo/config.toml").read_text())
                    copied = Path(config["patch"]["crates-io"]["graphile_worker_database"]["path"])
                    self.assertEqual((copied / "src/lib.rs").read_bytes(), (database / "src/lib.rs").read_bytes())
                    self.assertEqual(checked.read_text(), checked_source)
                    self.assertEqual(manifest.read_text(), original)
                self.assertFalse(cwd.exists())
                self.assertFalse(copied.exists())

    def test_transitive_lookup_is_scoped_to_the_known_crates_io_version(self):
        """Other releases and other registries must never receive the baseline patch."""
        with tempfile.TemporaryDirectory() as directory:
            manifest = Path(directory) / "Cargo.toml"
            manifest.write_text('[package]\nname="graphile_worker_queries"\nversion="0.1.2"\n')
            known = {"name": "graphile_worker_database", "version": "0.1.6",
                     "source": "registry+https://github.com/rust-lang/crates.io-index",
                     "manifest_path": "/published/database/Cargo.toml"}
            others = [{**known, "version": "0.1.7"}, {**known, "source": "registry+https://elsewhere.invalid"}]
            result = subprocess.CompletedProcess([], 0, stdout=json.dumps({"packages": others}))
            with patch.object(adapter.subprocess, "run", return_value=result):
                self.assertIsNone(adapter.published_database_dependency(manifest))
            result.stdout = json.dumps({"packages": [known, *others]})
            with patch.object(adapter.subprocess, "run", return_value=result):
                self.assertEqual(adapter.published_database_dependency(manifest), Path(known["manifest_path"]))

    def test_temporary_config_directory_reaches_every_checker_profile(self):
        """The generated Cargo configuration must cover all real checker subprocesses."""
        calls = []
        def checker(command, **kwargs):
            calls.append(kwargs)
            return subprocess.CompletedProcess(command, 0)
        self.assertEqual(adapter.run_checks([("one", ["one"]), ("two", ["two"])],
                                           checker, cwd=Path("/temporary/config")), 0)
        self.assertEqual(calls, [{"check": False, "cwd": Path("/temporary/config")}] * 2)

    def test_version_probe_delegates_to_installed_checker(self):
        """Tool installation probes must reach the installed checker unchanged."""
        with patch.dict(os.environ, {"GRAPHILE_SEMVER_CHECKS_BIN": "/real/checker"}), \
                patch.object(adapter.subprocess, "run", return_value=subprocess.CompletedProcess([], 0)) as run:
            self.assertEqual(adapter.main(["--version"]), 0)
            run.assert_called_once_with(["/real/checker", "--version"], check=False)


if __name__ == "__main__":
    unittest.main()