265 lines
10 KiB
Python
265 lines
10 KiB
Python
"""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": "₹22–30L", "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
|