Files
matchmaking-v2/tests/test_pool.py
2026-07-17 19:27:33 +05:30

265 lines
10 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""Warm-pool gate — the reorder/floor decision that keeps the shelf stocked without over-fetching."""
import asyncio
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock
import pytest
from app.engine import search
from app.engine.search import pool_decision
from app.db.repo import search_signature
FLOOR, REORDER = 90, 100 # the configured thresholds (POOL_SAFETY_FLOOR / POOL_REORDER_AT)
def test_pool_healthy_serves_no_fetch():
assert pool_decision(150, False, FLOOR, REORDER) == "pool"
assert pool_decision(100, False, FLOOR, REORDER) == "pool" # exactly at reorder → still no fetch
def test_pool_dipping_serves_plus_background_refill():
assert pool_decision(99, False, FLOOR, REORDER) == "pool+refill" # below reorder → top up in bg
assert pool_decision(90, False, FLOOR, REORDER) == "pool+refill" # at the floor → still serve, refill
def test_pool_below_floor_blocks_and_fetches():
assert pool_decision(89, False, FLOOR, REORDER) == "sweep" # under the hard floor → blocking fetch
assert pool_decision(0, False, FLOOR, REORDER) == "sweep" # empty pool (first search) → fetch
def test_forced_fresh_always_fetches():
assert pool_decision(999, True, FLOOR, REORDER) == "sweep" # dev _fresh toggle bypasses the pool
assert pool_decision(5, True, FLOOR, REORDER, exhausted=True) == "sweep" # force overrides exhaustion
def test_exhausted_query_serves_instead_of_block_fetching():
# tapped-out niche query below the floor: serve what's pooled, DON'T re-buy duplicates
assert pool_decision(30, False, FLOOR, REORDER, exhausted=True) == "pool" # below floor but exhausted
assert pool_decision(1, False, FLOOR, REORDER, exhausted=True) == "pool" # even 1 job → serve it
# but an EMPTY exhausted pool must still try a fetch (nothing to serve)
assert pool_decision(0, False, FLOOR, REORDER, exhausted=True) == "sweep"
# exhaustion does NOT suppress the healthy/refill paths
assert pool_decision(200, False, FLOOR, REORDER, exhausted=True) == "pool"
def test_floor_never_above_reorder_invariant():
# the gate only makes sense when floor ≤ reorder (reorder early, floor as the hard backstop)
from app.config import get_settings
s = get_settings()
assert s.POOL_SAFETY_FLOOR <= s.POOL_REORDER_AT
def test_target_title_changes_search_signature():
base = {
"title": "Product Manager",
"role": ["Product"],
"targetTitles": ["Backend Engineer"],
"location": ["Bangalore · India"],
"workMode": ["Hybrid"],
}
changed = {**base, "targetTitles": ["Data Analyst"]}
assert search_signature(base) != search_signature(changed)
def test_work_mode_changes_search_signature_using_dashboard_contract():
base = {"title": "Data Analyst", "location": ["Bangalore · India"], "workMode": ["Hybrid"]}
changed = {**base, "workMode": ["Remote"]}
assert search_signature(base) != search_signature(changed)
def test_legacy_work_mode_does_not_change_adapter_cache_identity():
common = {"title": "Data Analyst", "location": ["Bangalore · India"]}
assert search_signature(common) == search_signature({**common, "work_mode": ["Hybrid"]})
def test_retrieval_inputs_change_search_signature():
base = {
"title": "Data Analyst",
"location": ["Bangalore · India"],
"workMode": ["Hybrid"],
"industry": ["Fintech"],
"experience": ["Mid-level"],
"years": 5,
"user_context": {"skills": ["Python", "SQL"]},
}
changes = [
{"title": "Product Manager"},
{"location": ["Mumbai · India"]},
{"workMode": ["Remote"]},
{"industry": ["Healthcare"]},
{"experience": ["Senior"]},
{"years": 8},
{"user_context": {"skills": ["Rust", "SQL"]}},
]
for change in changes:
assert search_signature(base) != search_signature({**base, **change})
def test_rank_only_preferences_do_not_cold_start_raw_pool():
base = {"title": "Data Analyst", "location": ["Bangalore · India"]}
changed = {**base, "targetComp": "₹2230L", "dealBreakers": ["Travel"], "stretch": "reach"}
assert search_signature(base) == search_signature(changed)
def test_search_signature_ignores_unrelated_user_context():
base = {"title": "Data Analyst", "location": ["Bangalore · India"], "user_context": {"skills": ["Python"]}}
changed = {**base, "user_context": {"skills": ["Python"], "education": ["MBA"]}}
assert search_signature(base) == search_signature(changed)
def test_sweep_marks_total_board_exception_failure(monkeypatch):
async def fail_board(*_args, **_kwargs):
raise RuntimeError("provider blocked")
monkeypatch.setattr(search, "_fetch_board", fail_board)
monkeypatch.setattr(search, "get_settings", lambda: SimpleNamespace(
BOARDS_ENABLED="naukri,foundit",
BOARD_TIMEOUT_S=1,
))
result = asyncio.run(search.run_sweep({"title": "Data Analyst", "location": ["Bangalore · India"]}))
assert result["all_boards_failed"] is True
assert result["failed_boards"] == 2
assert result["sources"] == {}
def test_sweep_marks_total_board_timeout_failure(monkeypatch):
async def timeout_board(*_args, **_kwargs):
await asyncio.sleep(0.05)
return "naukri", []
monkeypatch.setattr(search, "_fetch_board", timeout_board)
monkeypatch.setattr(search, "get_settings", lambda: SimpleNamespace(
BOARDS_ENABLED="naukri,foundit",
BOARD_TIMEOUT_S=0.001,
))
result = asyncio.run(search.run_sweep({"title": "Data Analyst", "location": ["Bangalore · India"]}))
assert result["all_boards_failed"] is True
assert result["failed_boards"] == 2
def test_sweep_marks_empty_or_unknown_board_config_as_failure(monkeypatch):
monkeypatch.setattr(search, "get_settings", lambda: SimpleNamespace(
BOARDS_ENABLED="unknown-board",
BOARD_TIMEOUT_S=1,
))
result = asyncio.run(search.run_sweep({"title": "Data Analyst", "location": ["Bangalore · India"]}))
assert result["all_boards_failed"] is True
assert result["enabled_boards"] == 0
assert result["configured_boards"] == ["unknown-board"]
def test_total_outage_emits_frontend_readable_agent_error(monkeypatch):
from app.agent.session import MatchmakingAgentSession
ws = MagicMock()
ws.send_json = AsyncMock()
session = MatchmakingAgentSession(ws, "user-1")
async def outage_heartbeat(awaitable, _messages):
awaitable.close()
return {
"opportunities": [],
"sources": {},
"all_boards_failed": True,
"failed_boards": 2,
}
session._with_heartbeat = outage_heartbeat
monkeypatch.setattr("app.agent.session.get_settings", lambda: SimpleNamespace(
POOL_ENABLED=False,
POOL_SAFETY_FLOOR=90,
POOL_REORDER_AT=100,
), raising=False)
monkeypatch.setattr("app.config.get_settings", lambda: SimpleNamespace(
POOL_ENABLED=False,
POOL_SAFETY_FLOOR=90,
POOL_REORDER_AT=100,
))
monkeypatch.setattr("app.db.repo.get_seen_ids", AsyncMock(return_value=set()))
asyncio.run(session.handle_run_search({"title": "Data Analyst", "location": ["Bangalore · India"]}))
message = ws.send_json.await_args.args[0]
assert message["type"] == "agent_error"
assert message["message"] == "All enabled job boards are temporarily unavailable."
assert message["data"]["code"] == "job_boards_unavailable"
def test_background_refill_does_not_mark_outage_as_exhausted(monkeypatch):
from app.agent.session import MatchmakingAgentSession
session = MatchmakingAgentSession(MagicMock(), "user-1")
monkeypatch.setattr("app.engine.search.run_sweep", AsyncMock(return_value={
"opportunities": [],
"sources": {},
"all_boards_failed": True,
}))
save = AsyncMock()
exhausted = AsyncMock()
monkeypatch.setattr("app.db.repo.pool_save", save)
monkeypatch.setattr("app.db.repo.mark_pool_exhausted", exhausted)
asyncio.run(session._background_refill({"title": "Data Analyst"}, "signature"))
save.assert_not_awaited()
exhausted.assert_not_awaited()
def test_background_refill_preserves_signature_retrieval_skill(monkeypatch):
from app.agent.session import MatchmakingAgentSession
ws = MagicMock()
ws.send_json = AsyncMock()
session = MatchmakingAgentSession(ws, "user-1")
prefs = {
"title": "Data Analyst",
"location": ["Bangalore · India"],
"user_context": {"skills": ["Python", "SQL"], "education": ["MBA"]},
}
sig = search_signature(prefs)
captured = {}
monkeypatch.setattr("app.config.get_settings", lambda: SimpleNamespace(
POOL_ENABLED=True,
POOL_SAFETY_FLOOR=90,
POOL_REORDER_AT=100,
))
monkeypatch.setattr("app.db.repo.get_seen_ids", AsyncMock(return_value=set()))
monkeypatch.setattr("app.db.repo.pool_load_fresh_unseen", AsyncMock(return_value=[{"id": f"job-{i}"} for i in range(95)]))
monkeypatch.setattr("app.db.repo.is_pool_exhausted", AsyncMock(return_value=False))
monkeypatch.setattr("app.db.repo.save_feed", AsyncMock())
monkeypatch.setattr("app.db.repo.update_salary_band", AsyncMock())
monkeypatch.setattr("app.db.repo.mark_seen", AsyncMock())
monkeypatch.setattr("app.db.repo.get_opportunity_states", AsyncMock(return_value={}))
monkeypatch.setattr("app.agent.session.emit_grow_event", AsyncMock())
monkeypatch.setattr("app.engine.sift.sift", lambda *_args: ([{"id": "job-1"}], {"mode": "test", "scored": 1}))
monkeypatch.setattr("app.engine.curate.curate", lambda *_args: [{"id": "job-1"}])
async def capture_refill(refill_prefs, refill_sig):
captured["prefs"] = refill_prefs
captured["sig"] = refill_sig
async def direct_heartbeat(awaitable, _messages):
return await awaitable
session._background_refill = capture_refill
session._with_heartbeat = direct_heartbeat
session._grow_activity_fields = AsyncMock(return_value={})
asyncio.run(session.handle_run_search(prefs))
asyncio.run(asyncio.sleep(0))
assert captured["sig"] == sig
assert captured["prefs"]["_skills"] == ["Python", "SQL"]
assert "user_context" not in captured["prefs"]
assert search_signature(captured["prefs"]) == sig