From c36ac7f948344b644b2f46659dfbda4b489e8425 Mon Sep 17 00:00:00 2001 From: Manish Kumar Date: Tue, 25 Aug 2026 22:12:05 -0500 Subject: [PATCH] test: add SDK edge case coverage --- tests/test_edge_cases.py | 196 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 196 insertions(+) create mode 100644 tests/test_edge_cases.py diff --git a/tests/test_edge_cases.py b/tests/test_edge_cases.py new file mode 100644 index 0000000..77c6f50 --- /dev/null +++ b/tests/test_edge_cases.py @@ -0,0 +1,196 @@ +"""Additional contract tests for SDK boundaries and edge-case request shapes.""" + +from __future__ import annotations + +from unittest.mock import MagicMock + +import pytest +import responses + +from omnibioai import OmniBioAI +from omnibioai._base import BaseServiceClient, _extract_message, _safe_json +from omnibioai.auth.session import AuthenticatedSession +from omnibioai.auth.tokens import TokenPair +from omnibioai.exceptions import ValidationError +from omnibioai.tes.client import TESClient + +GATEWAY = "https://gateway.example.com" +AUTH = "https://auth.example.com" + + +def _client() -> OmniBioAI: + return OmniBioAI(access_token="access-1", refresh_token="refresh-1", base_url=GATEWAY) + + +def _response(status_code: int, body=None, reason: str = "Reason") -> MagicMock: + response = MagicMock() + response.status_code = status_code + response.reason = reason + if body is None: + response.json.side_effect = ValueError("not json") + else: + response.json.return_value = body + return response + + +class TestBaseBoundaryCases: + def test_request_normalizes_url_and_forwards_kwargs(self): + session = MagicMock() + session.last_trace_id = "trace-1" + session.request.return_value = _response(200, {"ok": True}) + service = BaseServiceClient("https://api.example.com/", session) + + result = service._request("POST", "/v1/items", json={"name": "x"}, timeout=3) + + assert result == {"ok": True} + session.request.assert_called_once_with( + "POST", "https://api.example.com/v1/items", json={"name": "x"}, timeout=3 + ) + + @pytest.mark.parametrize("status", [204, 302]) + def test_non_error_statuses_return_empty_body(self, status): + session = MagicMock() + session.last_trace_id = None + assert BaseServiceClient("https://api.example.com", session)._parse(_response(status)) is None + + def test_malformed_json_falls_back_to_reason_for_client_error(self): + session = MagicMock() + session.last_trace_id = "trace-2" + + with pytest.raises(ValidationError) as exc: + BaseServiceClient("https://api.example.com", session)._parse( + _response(400, reason="Bad Request") + ) + + assert str(exc.value) == "Bad Request" + assert exc.value.trace_id == "trace-2" + + def test_safe_json_returns_none_when_response_is_not_json(self): + assert _safe_json(_response(200)) is None + + @pytest.mark.parametrize( + ("body", "expected"), + [ + ({"error": "bad"}, "bad"), + ({"error": "bad", "reason": "because"}, "bad: because"), + ({"detail": {"reason": "nested reason"}}, "nested reason"), + ({"detail": ""}, ""), + ({"error": ""}, None), + ], + ) + def test_extract_message_normalizes_known_shapes(self, body, expected): + assert _extract_message(body) == expected + + +class TestAuthRequestBoundaryCases: + @responses.activate + def test_custom_headers_survive_and_authorization_is_sdk_owned(self): + session = AuthenticatedSession(TokenPair("access-1", "refresh-1"), AUTH, timeout=7) + responses.add(responses.GET, f"{GATEWAY}/health", json={"ok": True}, status=200) + + session.request( + "GET", f"{GATEWAY}/health", + headers={"X-Client": "demo", "Authorization": "Bearer caller-token"}, + ) + + sent = responses.calls[0].request + assert sent.headers["X-Client"] == "demo" + assert sent.headers["Authorization"] == "Bearer access-1" + assert sent.headers["X-Trace-Id"] + + @responses.activate + def test_retry_flag_prevents_a_second_refresh(self): + session = AuthenticatedSession(TokenPair("access-1", "refresh-1"), AUTH) + responses.add(responses.GET, f"{GATEWAY}/health", json={"detail": "expired"}, status=401) + + result = session.request("GET", f"{GATEWAY}/health", _is_retry=True) + + assert result.status_code == 401 + assert len(responses.calls) == 1 + + @responses.activate + def test_refresh_preserves_trace_id_across_original_and_retry(self): + session = AuthenticatedSession(TokenPair("access-1", "refresh-1"), AUTH) + responses.add(responses.GET, f"{GATEWAY}/health", json={}, status=401) + responses.add(responses.POST, f"{AUTH}/auth/refresh", json={"access_token": "access-2"}, status=200) + responses.add(responses.GET, f"{GATEWAY}/health", json={"ok": True}, status=200) + + session.request("GET", f"{GATEWAY}/health", trace_id="fixed-trace") + + assert responses.calls[0].request.headers["X-Trace-Id"] == "fixed-trace" + assert responses.calls[2].request.headers["X-Trace-Id"] == "fixed-trace" + assert session.last_trace_id == "fixed-trace" + + +class TestServiceParameterEdges: + @responses.activate + def test_rag_query_keeps_zero_top_k_and_encodes_query(self): + c = _client() + responses.add(responses.POST, f"{GATEWAY}/rag/v1/query", json={"results": []}, status=200) + + c.rag.query("BRCA1 & TP53", top_k=0, mode="structured", hybrid_search=True) + + body = responses.calls[0].request.body.decode() + assert '"top_k": 0' in body + assert '"hybrid_search": true' in body + assert "BRCA1 & TP53" in body + + @responses.activate + def test_rag_entity_includes_explicit_empty_type(self): + c = _client() + responses.add(responses.GET, f"{GATEWAY}/rag/v1/kg/entity", json={"name": "X"}, status=200) + + c.rag.kg_entity("X", type="") + + assert "type=" in responses.calls[0].request.url + + @responses.activate + def test_models_list_sends_only_the_metric_filter_when_requested(self): + c = _client() + responses.add(responses.GET, f"{GATEWAY}/model-registry/v1/models", json={"models": []}, status=200) + + c.models.list(metric_gte="accuracy:0.95") + + url = responses.calls[0].request.url + assert "metric_gte=accuracy%3A0.95" in url + assert "task=" not in url + assert "model_name=" not in url + + @responses.activate + def test_models_resolve_can_disable_verification(self): + c = _client() + responses.add(responses.GET, f"{GATEWAY}/model-registry/v1/resolve", json={"path": "/x"}, status=200) + + c.models.resolve("task", "latest", verify=False) + + assert "verify=False" in responses.calls[0].request.url + + def test_tes_run_request_normalizes_none_inputs(self): + assert TESClient._run_request("tool", None, None, None) == { + "tool_id": "tool", "inputs": {}, "resources": {}, "constraints": {} + } + + @responses.activate + def test_tes_omits_empty_server_id_query_parameter(self): + c = _client() + responses.add(responses.POST, f"{GATEWAY}/tes/api/runs/validate", json={"ok": True}, status=200) + + c.tes.validate("tool", server_id="") + + assert responses.calls[0].request.url == f"{GATEWAY}/tes/api/runs/validate" + + @responses.activate + def test_workflow_run_preserves_explicit_empty_inputs_and_engine(self): + c = _client() + responses.add( + responses.GET, f"{GATEWAY}/workflow-bundles/v1/workflows/wf", + json=[{"id": 4, "version": "1.0"}], status=200, + ) + responses.add( + responses.POST, f"{GATEWAY}/workflow-bundles/v1/workflows/4/run", + json={"run_id": "r1"}, status=201, + ) + + c.workflows.run("wf", inputs={}, engine="") + + assert responses.calls[1].request.body.decode() == '{"inputs": {}, "engine": ""}'