openjobradar/tests/test_source_yield.py at dependabot/pip/pytest-9.1.1 · ghzhost/openjobradar · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
"""Per-user source-yield tracking (roadmap M4, ADR-0020)."""
from __future__ import annotations

import pytest
from hypothesis import given
from hypothesis import strategies as st

from openjobradar.tenancy import InMemoryStore, TenantContext, TenantRepository
from openjobradar.tenancy.source_yield import (
    OUTCOME_KINDS,
    SourceYieldLedger,
    UnknownOutcomeKindError,
)

ALICE = TenantContext.from_sub("alice-sub-0000001")
BOB = TenantContext.from_sub("bob-sub-00000002")


def _ledger() -> SourceYieldLedger:
    return SourceYieldLedger(TenantRepository(InMemoryStore()))


def test_summary_of_unknown_org_is_the_neutral_default() -> None:
    summary = _ledger().summary(ALICE, "acme")
    assert summary.total_outcomes == 0
    assert summary.has_signal is False
    assert summary.recommended_tier == "warm"


def test_record_outcome_rejects_unknown_kind() -> None:
    with pytest.raises(UnknownOutcomeKindError):
        _ledger().record_outcome(ALICE, "acme", "ghosted", at="2026-08-23T00:00:00Z")


def test_record_outcome_rejects_empty_timestamp() -> None:
    with pytest.raises(ValueError):
        _ledger().record_outcome(ALICE, "acme", "applied", at="")


def test_all_catalogued_kinds_are_accepted() -> None:
    ledger = _ledger()
    for kind in OUTCOME_KINDS:
        ledger.record_outcome(ALICE, "acme", kind, at="2026-08-23T00:00:00Z")
    summary = ledger.summary(ALICE, "acme")
    assert summary.total_outcomes == len(OUTCOME_KINDS)


def test_few_outcomes_stay_at_neutral_default_regardless_of_rate() -> None:
    ledger = _ledger()
    for _ in range(3):
        ledger.record_outcome(ALICE, "acme", "interview", at="2026-08-23T00:00:00Z")
    summary = ledger.summary(ALICE, "acme")
    assert summary.has_signal is False
    assert summary.recommended_tier == "warm"  # 3 outcomes is below the minimum sample


def test_high_yield_org_is_recommended_hot() -> None:
    ledger = _ledger()
    for i in range(5):
        kind = "interview" if i < 2 else "posting_seen"  # 2/5 = 40% >= 0.20
        ledger.record_outcome(ALICE, "acme", kind, at=f"2026-08-{i + 1:02d}T00:00:00Z")
    summary = ledger.summary(ALICE, "acme")
    assert summary.has_signal is True
    assert summary.yield_rate == pytest.approx(0.4)
    assert summary.recommended_tier == "hot"


def test_zero_yield_org_with_enough_history_is_recommended_cold() -> None:
    ledger = _ledger()
    for i in range(10):
        ledger.record_outcome(ALICE, "acme", "posting_seen", at=f"2026-08-{i + 1:02d}T00:00:00Z")
    summary = ledger.summary(ALICE, "acme")
    assert summary.has_signal is True
    assert summary.yield_rate == 0.0
    assert summary.recommended_tier == "cold"


def test_history_is_capped_at_72_outcomes() -> None:
    ledger = _ledger()
    # first 28 evicted by the cap, then exactly 72 "interview" records fill the window
    for i in range(100):
        kind = "posting_seen" if i < 28 else "interview"
        ledger.record_outcome(ALICE, "acme", kind, at=f"2026-08-23T00:{i % 60:02d}:00Z")
    summary = ledger.summary(ALICE, "acme")
    assert summary.total_outcomes == 72
    assert summary.positive_outcomes == 72  # the evicted posting_seen records don't count


def test_yield_ledgers_are_isolated_per_user() -> None:
    ledger = _ledger()
    for _ in range(10):
        ledger.record_outcome(ALICE, "acme", "offer", at="2026-08-23T00:00:00Z")
    bob_summary = ledger.summary(BOB, "acme")
    assert bob_summary.total_outcomes == 0
    assert bob_summary.recommended_tier == "warm"


def test_orgs_are_tracked_independently_per_user() -> None:
    ledger = _ledger()
    for _ in range(10):
        ledger.record_outcome(ALICE, "acme", "offer", at="2026-08-23T00:00:00Z")
    globex_summary = ledger.summary(ALICE, "globex")
    assert globex_summary.total_outcomes == 0


@given(st.lists(st.sampled_from(OUTCOME_KINDS), min_size=0, max_size=120))
def test_recommendation_is_always_one_of_the_three_tiers(kinds) -> None:
    ledger = _ledger()
    for i, kind in enumerate(kinds):
        ledger.record_outcome(ALICE, "acme", kind, at=f"2026-08-23T00:{i % 60:02d}:{i % 60:02d}Z")
    summary = ledger.summary(ALICE, "acme")
    assert summary.recommended_tier in ("hot", "warm", "cold")
    assert summary.total_outcomes == min(len(kinds), 72)
    assert 0.0 <= summary.yield_rate <= 1.0