test coverage: ensure that multi insert preserves job args

.github/workflows/ci.yaml

@@ -40,7 +40,7 @@ jobs:
         uses: actions/checkout@v4
 
       - name: Install Rye
-        uses: eifinger/setup-rye@v3
+        uses: eifinger/setup-rye@v4
 
       # Needed for River's CLI. There is a version of Go on Actions' base image,
       # but it's old and can't read modern `go.mod` annotations correctly.
@@ -89,7 +89,7 @@ jobs:
         uses: actions/checkout@v4
 
       - name: Install Rye
-        uses: eifinger/setup-rye@v3
+        uses: eifinger/setup-rye@v4
 
       # Needed for River's CLI. There is a version of Go on Actions' base image,
       # but it's old and can't read modern `go.mod` annotations correctly.
@@ -122,7 +122,7 @@ jobs:
         uses: actions/checkout@v4
 
       - name: Install Rye
-        uses: eifinger/setup-rye@v3
+        uses: eifinger/setup-rye@v4
 
       - name: Rye sync
         run: rye sync

src/riverqueue/client.py

@@ -703,9 +703,9 @@ def unique_bitmask_to_states(mask: str) -> list[JobState]:
 
 def _validate_tags(tags: list[str]) -> list[str]:
     for tag in tags:
-        assert (
-            len(tag) <= 255 and tag_re.match(tag)
-        ), f"tags should be less than 255 characters in length and match regex {tag_re.pattern}"
+        assert len(tag) <= 255 and tag_re.match(tag), (
+            f"tags should be less than 255 characters in length and match regex {tag_re.pattern}"
+        )
     return tags
 
 

tests/client_test.py

@@ -507,6 +507,6 @@ def test_unique_bitmask_from_states(description, input_states, postgres_bitstrin
         input_states = []
 
     result = unique_bitmask_from_states(input_states)
-    assert (
-        result == postgres_bitstring
-    ), f"{description} For states {input_states}, expected {postgres_bitstring}, got {result}"
+    assert result == postgres_bitstring, (
+        f"{description} For states {input_states}, expected {postgres_bitstring}, got {result}"
+    )

tests/conftest.py

@@ -98,6 +98,6 @@ def check_leftover_jobs(engine) -> Iterator[None]:
 
     with engine.begin() as conn_tx:
         jobs = river_job.Querier(conn_tx).job_get_all()
-        assert (
-            list(jobs) == []
-        ), "test case should not have persisted any jobs after run"
+        assert list(jobs) == [], (
+            "test case should not have persisted any jobs after run"
+        )

tests/driver/riversqlalchemy/sqlalchemy_driver_test.py

@@ -305,6 +305,39 @@ async def test_insert_many_tx(self, client, simple_args, test_tx):
         assert results[0].unique_skipped_as_duplicated is False
         assert results[0].job.id > 0
 
+    @pytest.mark.asyncio
+    async def test_insert_many_preserves_distinct_args(self, client):
+        # Insert mixed types and ensure each row retains its own args and kind
+        from dataclasses import dataclass
+
+        @dataclass
+        class TypeA:
+            n: int
+            kind: str = "simple_a"
+
+            def to_json(self) -> str:
+                return json.dumps({"a": self.n})
+
+        @dataclass
+        class TypeB:
+            s: str
+            kind: str = "simple_b"
+
+            def to_json(self) -> str:
+                return json.dumps({"b": self.s})
+
+        batch = [TypeA(1), TypeB("x"), TypeA(2), TypeB("y")]
+        results = await client.insert_many(batch)
+
+        assert len(results) == 4
+        for res, arg in zip(results, batch):
+            if isinstance(arg, TypeA):
+                assert res.job.kind == "simple_a"
+                assert res.job.args == {"a": arg.n}
+            else:
+                assert res.job.kind == "simple_b"
+                assert res.job.args == {"b": arg.s}
+
 
 class TestSyncClient:
     #
@@ -502,3 +535,35 @@ def test_insert_many_tx(self, client, simple_args, test_tx):
         assert len(results) == 1
         assert results[0].unique_skipped_as_duplicated is False
         assert results[0].job.id > 0
+
+    def test_insert_many_preserves_distinct_args(self, client):
+        # Insert mixed types and ensure each row retains its own args and kind
+        from dataclasses import dataclass
+
+        @dataclass
+        class TypeA:
+            n: int
+            kind: str = "simple_a"
+
+            def to_json(self) -> str:
+                return json.dumps({"a": self.n})
+
+        @dataclass
+        class TypeB:
+            s: str
+            kind: str = "simple_b"
+
+            def to_json(self) -> str:
+                return json.dumps({"b": self.s})
+
+        batch = [TypeA(1), TypeB("x"), TypeA(2), TypeB("y")]
+        results = client.insert_many(batch)
+
+        assert len(results) == 4
+        for res, arg in zip(results, batch):
+            if isinstance(arg, TypeA):
+                assert res.job.kind == "simple_a"
+                assert res.job.args == {"a": arg.n}
+            else:
+                assert res.job.kind == "simple_b"
+                assert res.job.args == {"b": arg.s}