Skip to content

fix(tests): align SSE and quantum integration tests with current APIs #475

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Bryan-Roe merged 1 commit into from
Jun 21, 2026
+30 −67
Merged

fix(tests): align SSE and quantum integration tests with current APIs #475

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
87 changes: 25 additions & 62 deletions tests/test_function_app_endpoints.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,26 @@ def _mock_request(
return req


def _capture_sse_http_response(monkeypatch, app_module, captured: dict) -> None:
"""Patch HttpResponse so SSE bodies are captured as bytes or generators."""
import inspect

import azure.functions as _af

_real_HttpResponse = _af.HttpResponse

def _capturing_HttpResponse(body=None, **kwargs):
if body is not None and inspect.isgenerator(body):
consumed = b"".join(body)
captured["sse_body"] = consumed
return _real_HttpResponse(consumed, **kwargs)
if isinstance(body, (bytes, bytearray)):
captured["sse_body"] = bytes(body)
return _real_HttpResponse(body, **kwargs)

monkeypatch.setattr(app_module.func, "HttpResponse", _capturing_HttpResponse)


def _install_fake_quantum_trainer_module(
monkeypatch: pytest.MonkeyPatch,
capture: dict | None = None,
Expand Down Expand Up @@ -526,22 +546,8 @@ def test_chat_stream_whitespace_only_input_text_block_message(self, app_module):

def test_chat_stream_guardrail_blocks_prompt_injection(self, app_module, monkeypatch):
"""POST /api/chat/stream should emit safe fallback SSE when prompt is blocked."""
import inspect

import azure.functions as _af

captured: dict = {"sse_body": b""}
_real_HttpResponse = _af.HttpResponse

def _capturing_HttpResponse(body=None, **kwargs):
if body is not None and inspect.isgenerator(body):
consumed = b"".join(body)
captured["sse_body"] = consumed
return _real_HttpResponse(consumed, **kwargs)
return _real_HttpResponse(body, **kwargs)

monkeypatch.setattr(app_module.func, "HttpResponse",
_capturing_HttpResponse)
_capture_sse_http_response(monkeypatch, app_module, captured)

@sourcery-ai sourcery-ai Bot Jun 21, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

suggestion (testing): Assert on the min_similarity argument passed into fetch_similar_messages to fully cover the updated signature.

Right now the mock accepts min_similarity but the test doesn’t check what value is passed. Capture the received min_similarity in captured (e.g., captured['min_similarity'] = min_similarity) and assert on it so the test explicitly verifies the intended default/configured value and guards against future signature drift.


req = _mock_request(
"POST",
Expand All @@ -562,10 +568,6 @@ def _capturing_HttpResponse(body=None, **kwargs):

def test_chat_stream_memory_injection(self, app_module, monkeypatch):
"""POST /api/chat/stream should call memory helpers and include count in meta SSE event."""
import inspect

import azure.functions as _af

captured: dict = {"embedding": None,
"session_id": None, "sse_body": b""}

Expand All @@ -577,18 +579,7 @@ def _fake_similar(query_emb, top_k=5, session_id=None, min_similarity=0.0):
captured["session_id"] = session_id
return [{"content": "Previous answer about widgets", "similarity": 0.88}]

# Patch func.HttpResponse inside function_app so streaming body (generator) is consumed
_real_HttpResponse = _af.HttpResponse

def _capturing_HttpResponse(body=None, **kwargs):
if body is not None and inspect.isgenerator(body):
consumed = b"".join(body)
captured["sse_body"] = consumed
return _real_HttpResponse(consumed, **kwargs)
return _real_HttpResponse(body, **kwargs)

monkeypatch.setattr(app_module.func, "HttpResponse",
_capturing_HttpResponse)
_capture_sse_http_response(monkeypatch, app_module, captured)
monkeypatch.setattr(app_module, "generate_embedding", _fake_embedding)
monkeypatch.setattr(
app_module, "fetch_similar_messages", _fake_similar)
Expand Down Expand Up @@ -623,10 +614,6 @@ def _capturing_HttpResponse(body=None, **kwargs):

def test_chat_stream_emits_done_sentinel(self, app_module, monkeypatch):
"""POST /api/chat/stream should terminate SSE with data: [DONE]."""
import inspect

import azure.functions as _af

captured: dict = {"sse_body": b""}

class _FakeProvider:
Expand All @@ -647,20 +634,10 @@ def complete(self, messages, stream=False):
monkeypatch.setattr(
app_module,
"fetch_similar_messages",
lambda query_emb, top_k=5, session_id=None: [],
lambda query_emb, top_k=5, session_id=None, min_similarity=0.0: [],
)

_real_HttpResponse = _af.HttpResponse

def _capturing_HttpResponse(body=None, **kwargs):
if body is not None and inspect.isgenerator(body):
consumed = b"".join(body)
captured["sse_body"] = consumed
return _real_HttpResponse(consumed, **kwargs)
return _real_HttpResponse(body, **kwargs)

monkeypatch.setattr(app_module.func, "HttpResponse",
_capturing_HttpResponse)
_capture_sse_http_response(monkeypatch, app_module, captured)

req = _mock_request(
"POST",
Expand Down Expand Up @@ -829,10 +806,6 @@ def test_agi_reason_validation_error_when_missing_input(self, app_module):
assert "validation error" in data["error"].lower()

def test_agi_stream_emits_done_sentinel(self, app_module, monkeypatch):
import inspect

import azure.functions as _af

captured: dict = {"sse_body": b""}

class _FakeAgiProvider:
Expand All @@ -853,17 +826,7 @@ def set_goal(self, _goal: str):
),
)

_real_HttpResponse = _af.HttpResponse

def _capturing_HttpResponse(body=None, **kwargs):
if body is not None and inspect.isgenerator(body):
consumed = b"".join(body)
captured["sse_body"] = consumed
return _real_HttpResponse(consumed, **kwargs)
return _real_HttpResponse(body, **kwargs)

monkeypatch.setattr(app_module.func, "HttpResponse",
_capturing_HttpResponse)
_capture_sse_http_response(monkeypatch, app_module, captured)

req = _mock_request(
"POST",
Expand Down
10 changes: 5 additions & 5 deletions tests/test_quantum_integration.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,8 +33,8 @@ def _stub_subprocess(returncode: int = 0, stdout: str = "ok", stderr: str = ""):
@pytest.mark.parametrize(
"jobs_payload,expected_preset",
[
([{"name": "another", "preset": "heart", "status": "completed"}], "heart"),
({"another": {"name": "another", "preset": "heart", "status": "completed"}}, "heart"),
([{"name": "smoke", "preset": "heart", "status": "completed"}], "heart"),
({"smoke": {"name": "smoke", "preset": "heart", "status": "completed"}}, "heart"),
],
)
def test_run_autorun_job_reads_status_for_list_and_dict_shapes(tmp_path: Path, monkeypatch, jobs_payload, expected_preset):
Expand All @@ -49,10 +49,10 @@ def test_run_autorun_job_reads_status_for_list_and_dict_shapes(tmp_path: Path, m
monkeypatch.setattr("mount.quantum_integration.subprocess.run",
lambda *args, **kwargs: _stub_subprocess())

result = asyncio.run(integration.run_autorun_job("another", dry_run=True))
result = asyncio.run(integration.run_autorun_job("smoke", dry_run=True))

assert result["success"] is True
assert result["job_name"] == "another"
assert result["job_name"] == "smoke"
assert result["dry_run"] is True
assert result["status"]["preset"] == expected_preset
assert result["status"]["name"] == "another"
assert result["status"]["name"] == "smoke"
Loading