test(coverage): backfill WS edge cases and small auth/type/config branches

tests/test_async_client.py

@@ -371,6 +371,14 @@ def test_no_auth_constructs_unauthenticated(self) -> None:
         client = AsyncKalshiClient()
         assert client._auth is None
 
+    def test_is_authenticated_true_with_auth(self, test_auth: KalshiAuth) -> None:
+        client = AsyncKalshiClient(auth=test_auth)
+        assert client.is_authenticated is True
+
+    def test_is_authenticated_false_without_auth(self) -> None:
+        client = AsyncKalshiClient()
+        assert client.is_authenticated is False
+
     def test_demo_flag(self, test_auth: KalshiAuth) -> None:
         client = AsyncKalshiClient(auth=test_auth, demo=True)
         assert client._config.base_url == DEMO_BASE_URL

tests/test_auth.py

@@ -251,6 +251,47 @@ def test_invalid_pem_content(self) -> None:
                 KalshiAuth.from_key_path("my-key", f.name)
         os.unlink(f.name)
 
+    @pytest.mark.skipif(
+        os.geteuid() == 0, reason="root bypasses file permission checks"
+    )
+    def test_permission_denied_wraps_with_helpful_message(
+        self, pem_bytes: bytes
+    ) -> None:
+        """A key file the user can't read raises KalshiAuthError, not PermissionError."""
+        with tempfile.NamedTemporaryFile(suffix=".pem", delete=False) as f:
+            f.write(pem_bytes)
+            f.flush()
+            path = f.name
+        try:
+            os.chmod(path, 0o000)
+            with pytest.raises(KalshiAuthError, match="Permission denied"):
+                KalshiAuth.from_key_path("my-key", path)
+        finally:
+            os.chmod(path, 0o600)
+            os.unlink(path)
+
+    def test_passphrase_protected_key_raises_helpful_error(
+        self, rsa_private_key: rsa.RSAPrivateKey
+    ) -> None:
+        """Encrypted keys produce a KalshiAuthError pointing at the openssl fix."""
+        encrypted_pem = rsa_private_key.private_bytes(
+            encoding=serialization.Encoding.PEM,
+            format=serialization.PrivateFormat.PKCS8,
+            encryption_algorithm=serialization.BestAvailableEncryption(b"pw"),
+        )
+        with tempfile.NamedTemporaryFile(suffix=".pem", delete=False) as f:
+            f.write(encrypted_pem)
+            f.flush()
+            path = f.name
+        try:
+            with pytest.raises(KalshiAuthError) as exc_info:
+                KalshiAuth.from_key_path("my-key", path)
+            msg = str(exc_info.value)
+            assert "Passphrase-protected" in msg
+            assert "openssl pkey" in msg
+        finally:
+            os.unlink(path)
+
 
 class TestFromPem:
     def test_accepts_bytes(self, pem_bytes: bytes) -> None:

tests/test_base_helpers.py

@@ -48,6 +48,16 @@ def test_happy_path_still_works(self) -> None:
         assert _join_tickers(()) is None
         assert _join_tickers("") is None
 
+    def test_non_string_bool_element_raises_unhelpful_type_error(self) -> None:
+        # Pins crash path: bool fails `"," in elem` check; update if validation is added.
+        with pytest.raises(TypeError, match=r"argument of type 'bool' is not iterable"):
+            _join_tickers([True, "A"])  # type: ignore[list-item]
+
+    def test_non_string_int_element_raises_unhelpful_type_error(self) -> None:
+        """Mirror of the bool case: int element crashes the same way."""
+        with pytest.raises(TypeError, match=r"argument of type 'int' is not iterable"):
+            _join_tickers((1, "A"))  # type: ignore[arg-type]
+
 
 class TestSyncListNullItemsCoercion:
     @respx.mock

tests/test_client.py

@@ -388,6 +388,16 @@ def test_no_auth_constructs_unauthenticated(self) -> None:
         assert client._auth is None
         client.close()
 
+    def test_is_authenticated_true_with_auth(self, test_auth: KalshiAuth) -> None:
+        client = KalshiClient(auth=test_auth)
+        assert client.is_authenticated is True
+        client.close()
+
+    def test_is_authenticated_false_without_auth(self) -> None:
+        client = KalshiClient()
+        assert client.is_authenticated is False
+        client.close()
+
     def test_demo_flag(self, test_auth: KalshiAuth) -> None:
         client = KalshiClient(auth=test_auth, demo=True)
         assert client._config.base_url == DEMO_BASE_URL

tests/test_types.py

@@ -0,0 +1,35 @@
+"""Tests for DollarDecimal / FixedPointCount type-fallback branches in kalshi.types."""
+from __future__ import annotations
+
+import pytest
+from pydantic import BaseModel
+
+from kalshi.types import DollarDecimal, FixedPointCount
+
+
+class _DollarModel(BaseModel):
+    x: DollarDecimal
+
+
+class _CountModel(BaseModel):
+    x: FixedPointCount
+
+
+class TestDollarDecimalTypeFallback:
+    def test_list_input_raises_type_error_with_named_type(self) -> None:
+        with pytest.raises(TypeError, match="Cannot convert list to Decimal"):
+            _DollarModel.model_validate({"x": [1, 2]})
+
+    def test_dict_input_raises_type_error_with_named_type(self) -> None:
+        with pytest.raises(TypeError, match="Cannot convert dict to Decimal"):
+            _DollarModel.model_validate({"x": {"nested": "value"}})
+
+
+class TestFixedPointCountTypeFallback:
+    def test_list_input_raises_type_error_with_named_type(self) -> None:
+        with pytest.raises(TypeError, match="Cannot convert list to Decimal"):
+            _CountModel.model_validate({"x": [1, 2]})
+
+    def test_dict_input_raises_type_error_with_named_type(self) -> None:
+        with pytest.raises(TypeError, match="Cannot convert dict to Decimal"):
+            _CountModel.model_validate({"x": {"nested": "value"}})

tests/ws/test_client.py

@@ -197,19 +197,21 @@ async def test_on_decorator_registers(self, fake_ws, test_auth) -> None:  # type
         ws = KalshiWebSocket(auth=test_auth, config=config)
         async with ws.connect() as session:
             received: list[object] = []
+            got_one = asyncio.Event()
 
             @session.on("fill")
             async def on_fill(msg: object) -> None:
                 received.append(msg)
+                got_one.set()
 
             await session.subscribe_fill()
             await fake_ws.send_to_all({
                 "type": "fill", "sid": 1,
                 "msg": {"trade_id": "t1", "order_id": "o1"},
             })
 
-            # Give recv_loop time to dispatch
-            await asyncio.sleep(0.2)
+            # Deterministic wait: the callback signals us; we don't sleep blindly.
+            await asyncio.wait_for(got_one.wait(), timeout=2.0)
             assert len(received) == 1
 
 
@@ -241,6 +243,52 @@ async def test_two_channels_on_same_connection(self, fake_ws, test_auth) -> None
             assert ticker_msg.msg.market_ticker == "T1"
             assert fill_msg.msg.trade_id == "t1"
 
+    async def test_two_subs_same_channel_distinct_params(self, fake_ws, test_auth) -> None:  # type: ignore[no-untyped-def]
+        """Two subs to the same channel (orderbook_delta) with different tickers.
+
+        Pins that the SDK treats each ``subscribe`` call as an independent
+        subscription with its own ``server_sid`` and its own iterator queue —
+        even when channel name collides. Messages with sid=A go only to the
+        first iterator; sid=B only to the second.
+        """
+        config = KalshiConfig(ws_base_url=fake_ws.url, timeout=5.0)
+        ws = KalshiWebSocket(auth=test_auth, config=config)
+        async with ws.connect() as session:
+            stream_a = await session.subscribe_orderbook_delta(tickers=["T1"])
+            stream_b = await session.subscribe_orderbook_delta(tickers=["T2"])
+
+            # Two distinct subscribes -> two distinct server sids
+            assert session._sub_mgr is not None
+            sids = [
+                sub.server_sid
+                for sub in session._sub_mgr.active_subscriptions.values()
+            ]
+            assert len(sids) == 2
+            assert sids[0] != sids[1]
+            sid_a, sid_b = sids[0], sids[1]
+            assert sid_a is not None and sid_b is not None
+
+            # Push one message to each sid
+            await fake_ws.send_to_all({
+                "type": "orderbook_snapshot", "sid": sid_a, "seq": 1,
+                "msg": {
+                    "market_ticker": "T1", "market_id": "x",
+                    "yes": [["0.50", "100"]], "no": [],
+                },
+            })
+            await fake_ws.send_to_all({
+                "type": "orderbook_snapshot", "sid": sid_b, "seq": 1,
+                "msg": {
+                    "market_ticker": "T2", "market_id": "y",
+                    "yes": [["0.60", "200"]], "no": [],
+                },
+            })
+
+            msg_a = await asyncio.wait_for(stream_a.__anext__(), timeout=2.0)
+            msg_b = await asyncio.wait_for(stream_b.__anext__(), timeout=2.0)
+            assert msg_a.msg.market_ticker == "T1"
+            assert msg_b.msg.market_ticker == "T2"
+
 
 # ---------------------------------------------------------------------------
 # run_forever
@@ -279,9 +327,11 @@ async def test_run_forever_returns_immediately_without_subscribe(
 class TestErrorCallback:
     async def test_on_error_called(self, fake_ws, test_auth) -> None:  # type: ignore[no-untyped-def]
         errors: list[object] = []
+        got_one = asyncio.Event()
 
         async def on_err(err: object) -> None:
             errors.append(err)
+            got_one.set()
 
         config = KalshiConfig(ws_base_url=fake_ws.url, timeout=5.0)
         ws = KalshiWebSocket(auth=test_auth, config=config, on_error=on_err)
@@ -294,5 +344,6 @@ async def on_err(err: object) -> None:
                 "type": "error",
                 "msg": {"code": 400, "msg": "bad request"},
             })
-            await asyncio.sleep(0.2)
+            # Deterministic wait: the callback signals us; we don't sleep blindly.
+            await asyncio.wait_for(got_one.wait(), timeout=2.0)
             assert len(errors) == 1

tests/ws/test_integration.py

@@ -333,6 +333,66 @@ async def test_reconnect_after_server_disconnect(
             msg2 = await asyncio.wait_for(stream.__anext__(), timeout=3.0)
             assert msg2.msg.yes_bid == 75
 
+    async def test_reconnect_failure_broadcasts_sentinel_to_iterators(
+        self,
+        fake_ws: FakeKalshiWS,
+        test_auth: KalshiAuth,
+    ) -> None:
+        """Recv loop must broadcast a sentinel when reconnect permanently fails.
+
+        Covers ``ws/client.py:197-203``: when ``_connection.reconnect()`` raises,
+        iterators must receive a sentinel so ``async for`` exits cleanly via
+        ``StopAsyncIteration`` instead of hanging forever.
+
+        Flow: connect -> subscribe -> server forces disconnect after one
+        message -> fake server rejects all reconnect attempts -> recv loop
+        catches the reconnect exception -> sentinel broadcast -> iterator
+        exits cleanly.
+        """
+        # Fast retries, single attempt so the test doesn't drag.
+        config = KalshiConfig(
+            ws_base_url=fake_ws.url,
+            timeout=2.0,
+            retry_base_delay=0.01,
+            retry_max_delay=0.05,
+            ws_max_retries=1,
+        )
+        # Disconnect after the first broadcast; then make all reconnects fail.
+        fake_ws.disconnect_after = 1
+
+        ws = KalshiWebSocket(auth=test_auth, config=config)
+        async with ws.connect() as session:
+            stream = await session.subscribe_ticker(tickers=["T1"])
+
+            # Reject any subsequent handshake -> reconnect() will exhaust
+            # its single retry and raise KalshiConnectionError.
+            fake_ws.reject_auth = True
+
+            # First message arrives, then the server closes the connection.
+            await fake_ws.send_to_all({
+                "type": "ticker", "sid": 1,
+                "msg": {"market_ticker": "T1", "market_id": "x", "yes_bid": 42},
+            })
+
+            # We should get the first message via the iterator,
+            # then the sentinel — async-for exits with no further items.
+            received: list[int] = []
+            try:
+                async with asyncio.timeout(5.0):
+                    async for msg in stream:
+                        received.append(msg.msg.yes_bid)
+            except TimeoutError:
+                pytest.fail(
+                    "Iterator hung after reconnect failure; sentinel was "
+                    "not broadcast — iterators would never exit."
+                )
+
+            # The pre-disconnect message may or may not have been queued
+            # depending on dispatch timing relative to the disconnect.
+            # The contract under test is that iteration *terminates*, not
+            # what's in the buffer first.
+            assert received == [42] or received == []
+
 
 # ---------------------------------------------------------------------------
 # 5. Sequence gap detection
@@ -348,9 +408,11 @@ async def test_gap_detection_fires_callback(
     ) -> None:
         """Sequence gap: server sends seq 1, 2, 5 (skipping 3, 4)."""
         gaps: list[SequenceGap] = []
+        got_gap = asyncio.Event()
 
         async def on_gap(gap: SequenceGap) -> None:
             gaps.append(gap)
+            got_gap.set()
 
         ws = KalshiWebSocket(auth=test_auth, config=ws_config)
         async with ws.connect() as session:
@@ -404,8 +466,8 @@ async def on_gap(gap: SequenceGap) -> None:
             })
 
             # Gap message is NOT dispatched (skipped by recv loop).
-            # Give recv loop time to invoke the gap callback.
-            await asyncio.sleep(0.3)
+            # Deterministic wait: gap handler signals us; no blind sleep.
+            await asyncio.wait_for(got_gap.wait(), timeout=2.0)
 
             assert len(gaps) == 1
             assert gaps[0].sid == 1
@@ -428,12 +490,14 @@ async def test_callback_and_iterator_coexist(
         """Iterator for ticker, callback for fill -- both receive messages."""
         ws = KalshiWebSocket(auth=test_auth, config=ws_config)
         callback_received: list[object] = []
+        got_fill = asyncio.Event()
 
         async with ws.connect() as session:
             # Register callback for fill BEFORE subscribing
             @session.on("fill")
             async def on_fill(msg: object) -> None:
                 callback_received.append(msg)
+                got_fill.set()
 
             # Subscribe to ticker via iterator
             ticker_stream = await session.subscribe_ticker(tickers=["T1"])
@@ -460,8 +524,8 @@ async def on_fill(msg: object) -> None:
             assert ticker_msg.msg.market_ticker == "T1"
             assert ticker_msg.msg.yes_bid == 55
 
-            # Wait for callback to fire
-            await asyncio.sleep(0.3)
+            # Wait for callback to fire (signaled by the callback itself).
+            await asyncio.wait_for(got_fill.wait(), timeout=2.0)
             assert len(callback_received) == 1
 
 
@@ -533,9 +597,11 @@ async def test_error_message_triggers_callback(
     ) -> None:
         """Server error message fires the on_error callback."""
         errors: list[ErrorMessage] = []
+        got_error = asyncio.Event()
 
         async def on_error(err: ErrorMessage) -> None:
             errors.append(err)
+            got_error.set()
 
         ws = KalshiWebSocket(
             auth=test_auth, config=ws_config, on_error=on_error,
@@ -550,7 +616,8 @@ async def on_error(err: ErrorMessage) -> None:
                 "msg": {"code": 5, "msg": "something went wrong"},
             })
 
-            await asyncio.sleep(0.3)
+            # Deterministic wait: callback signals; no blind sleep.
+            await asyncio.wait_for(got_error.wait(), timeout=2.0)
             assert len(errors) == 1
             assert errors[0].msg.code == 5
             assert errors[0].msg.msg == "something went wrong"