openjobradar/tests/test_watchlist.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
"""Per-user org watchlist (roadmap M4, ADR-0026)."""
from __future__ import annotations

from openjobradar.tenancy import InMemoryStore, TenantContext, TenantRepository
from openjobradar.tenancy.watchlist import WatchlistEntry, WatchlistService

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


def _service() -> WatchlistService:
    return WatchlistService(TenantRepository(InMemoryStore()))


def _entry(slug: str = "gitlab", **overrides) -> WatchlistEntry:
    defaults = {
        "slug": slug,
        "name": "GitLab",
        "ats_type": "greenhouse",
        "ats_board_id": "gitlab",
        "description": "DevSecOps platform.",
    }
    return WatchlistEntry(**{**defaults, **overrides})


def test_add_and_get_roundtrip() -> None:
    watchlist = _service()
    watchlist.add(ALICE, _entry())
    fetched = watchlist.get(ALICE, "gitlab")
    assert fetched == _entry()


def test_get_missing_entry_returns_none() -> None:
    assert _service().get(ALICE, "nope") is None


def test_remove_reports_whether_an_entry_existed() -> None:
    watchlist = _service()
    watchlist.add(ALICE, _entry())
    assert watchlist.remove(ALICE, "gitlab") is True
    assert watchlist.remove(ALICE, "gitlab") is False
    assert watchlist.get(ALICE, "gitlab") is None


def test_list_all_returns_every_entry() -> None:
    watchlist = _service()
    watchlist.add(ALICE, _entry("gitlab"))
    watchlist.add(ALICE, _entry("acme", name="Acme", ats_board_id="acme"))
    entries = watchlist.list_all(ALICE)
    assert {entry.slug for entry in entries} == {"gitlab", "acme"}


def test_count_matches_list_length() -> None:
    watchlist = _service()
    assert watchlist.count(ALICE) == 0
    watchlist.add(ALICE, _entry("gitlab"))
    watchlist.add(ALICE, _entry("acme", name="Acme", ats_board_id="acme"))
    assert watchlist.count(ALICE) == 2


def test_default_tier_is_warm() -> None:
    entry = _entry()
    assert entry.tier == "warm"


def test_watchlists_are_isolated_per_tenant() -> None:
    watchlist = _service()
    watchlist.add(ALICE, _entry())
    assert watchlist.list_all(BOB) == ()
    assert watchlist.get(BOB, "gitlab") is None
    assert watchlist.count(BOB) == 0


def test_add_overwrites_an_existing_slug() -> None:
    watchlist = _service()
    watchlist.add(ALICE, _entry(tier="hot"))
    watchlist.add(ALICE, _entry(tier="cold"))
    assert watchlist.count(ALICE) == 1
    assert watchlist.get(ALICE, "gitlab").tier == "cold"