from types import SimpleNamespace OWNER = SimpleNamespace(username="someone") TOKEN = SimpleNamespace(token="test") from copy import deepcopy import gradio as gr from contrib import cache from contrib.domain import CollectionRequest from contrib.aggregation import Accumulator from contrib.ui import preferences as prefs from contrib.ui.events import collect_ui, private_event, refresh_event, cleanup_ui def snapshot(year, private=False): value = Accumulator(CollectionRequest("someone", year, private=private)).copy() value["phase"] = "complete" value["bins"]["space.commit"][0] = 7 if private: value["private_bins"]["space.commit"][0] = 3 return value def test_session_retains_multiple_years_without_fetching(monkeypatch): import contrib.ui.events as ui session = "years" monkeypatch.setattr(ui.collection, "stream", lambda *args: (_ for _ in ()).throw(AssertionError("Unexpected fetch"))) prefs.remember(session, snapshot(2024)) prefs.remember(session, snapshot(2025)) try: for year in (2024, 2025, 2024): result = next(collect_ui("someone", year, False, False, session, gr.Request(session_hash=session), OWNER, TOKEN))[0] assert result["year"] == year and result["cache_source"] == "session" assert result["cached_years"] == [2025, 2024] assert prefs.cached("another-session", "someone", 2024) is None finally: cleanup_ui(session) def test_private_cache_requires_owner_and_remains_bounded(): session = "bounded" try: prefs.remember(session, snapshot(2024, True)) assert prefs.cached(session, "someone", 2024, True, False) is None assert prefs.cached(session, "someone", 2024, False, False) is None hidden = prefs.cached(session, "someone", 2024, False, True) assert hidden["private"] is True and hidden["show_private"] is False hidden["bins"]["space.commit"][0] = 1000 assert prefs.cached(session, "someone", 2024, True, True)["bins"]["space.commit"][0] == 7 for year in range(2000, 2015): prefs.remember(session, snapshot(year)) assert len(prefs.years(session, "someone", True)) == prefs.MAX_YEARS assert prefs.cached(session, "someone", 2000) is None finally: cleanup_ui(session) assert prefs.cached(session, "someone", 2014) is None def test_private_event_only_syncs_memory_when_already_loaded(monkeypatch): import contrib.ui.events as ui session = "instant-private" prefs.remember(session, snapshot(2026, True)) monkeypatch.setattr(ui.collection, "stream", lambda *args: (_ for _ in ()).throw(AssertionError("Unexpected fetch"))) profile, token = SimpleNamespace(username="someone"), SimpleNamespace(token="test") try: for revision, show in enumerate((False, True)): result = private_event(gr.EventData(None, {"show":show,"revision":revision,"run_id":prefs.current(session)["run_id"]}), gr.Request(session_hash=session), profile, token) assert result is None assert prefs.current(session)["show_private"] is show finally: cleanup_ui(session) def test_refresh_bypasses_session_cache(monkeypatch): import contrib.ui.events as ui session = "refresh" prefs.remember(session, snapshot(2026)) calls = [] def stream(request, session, refresh): calls.append(refresh) value = snapshot(2026) value["bins"]["space.commit"][0] = 9 yield value monkeypatch.setattr(ui.collection, "stream", stream) try: result = next(refresh_event(gr.EventData(None, {"show":False}), gr.Request(session_hash=session), OWNER, TOKEN))[0] assert calls == [True] assert result["bins"]["space.commit"][0] == 9 finally: cleanup_ui(session) def test_past_year_public_cache_lasts_one_day(monkeypatch): cache._items.clear() now = [0.] monkeypatch.setattr(cache, "monotonic", lambda: now[0]) monkeypatch.setattr(cache, "current_year", lambda: 2026) query = CollectionRequest("someone", 2024) cache.put(query, snapshot(2024)) now[0] = 3600 assert cache.get(query) is not None now[0] = 86401 assert cache.get(query) is None cache._items.clear() def test_visibility_events_reject_stale_requests(): session = "visibility-order" try: assert prefs.accept_visibility_request(session, 2) assert not prefs.accept_visibility_request(session, 1) assert not prefs.accept_visibility_request(session, 2) assert not prefs.accept_visibility_request(session, None) assert prefs.accept_visibility_request(session, 3) finally: cleanup_ui(session) def test_visibility_commit_is_atomic_and_bound_to_the_observed_view(): session = "atomic-visibility" value = snapshot(2026, True) prefs.remember(session, value) try: assert prefs.set_visibility(session, 2, False, value["run_id"]) assert not prefs.set_visibility(session, 1, True, value["run_id"]) assert not prefs.set_visibility(session, 3, True, "old-view") assert prefs.current(session)["show_private"] is False assert prefs.previous(session)["show_private"] is False finally: cleanup_ui(session) def test_partial_year_is_retained_without_becoming_a_success(monkeypatch): import contrib.ui.events as ui session = "partial-year" value = snapshot(2025) value["phase"] = "partial" value["sources"]["community"]["status"] = "partial" value["warnings"] = ["Some community entries could not be classified."] prefs.remember(session, value) monkeypatch.setattr(ui.collection, "stream", lambda *args: (_ for _ in ()).throw(AssertionError("Unexpected fetch"))) try: result = next(collect_ui("someone", 2025, False, False, session, gr.Request(session_hash=session), OWNER, TOKEN))[0] assert result["phase"] == "partial" and result["cached"] assert result["warnings"] == value["warnings"] assert prefs.previous(session) is None finally: cleanup_ui(session) def test_year_change_never_starts_any_fetch(monkeypatch): from contrib.ui.events import year_ui import contrib.ui.events as ui session = "deliberate-private" profile, token = SimpleNamespace(username="someone"), SimpleNamespace(token="test") prefs.remember(session, snapshot(2026, True)) calls = [] def stream(query, *args): calls.append(query) yield snapshot(query.year) monkeypatch.setattr(ui.collection, "stream", stream) try: result = list(year_ui("someone", 2025, session, gr.Request(session_hash=session), profile, token))[0] assert calls == [] assert result[0]["year"] == 2025 and result[0]["year_preview"] assert result[4]["value"] == 2025 finally: cleanup_ui(session) def test_private_visibility_without_data_never_fetches(monkeypatch): import contrib.ui.events as ui session = "visibility-public" prefs.remember(session, snapshot(2026)) monkeypatch.setattr(ui.collection, "stream", lambda *args: (_ for _ in ()).throw(AssertionError("Unexpected fetch"))) try: private_event(gr.EventData(None, {"show":True,"revision":1}), gr.Request(session_hash=session), SimpleNamespace(username="someone"), SimpleNamespace(token="test")) assert prefs.current(session)["private"] is False finally: cleanup_ui(session) def test_cached_auth_label_uses_current_viewer_and_signout_clears_private_data(): from contrib.ui.events import ui_result, access_controls session = "privacy-auth-context" try: value = snapshot(2026) value["viewer"] = "previous-viewer" assert ui_result(value, session, False)[0]["viewer"] == "" prefs.remember(session, snapshot(2026, True)) result = access_controls("someone", 2026, gr.Request(session_hash=session), None, None) assert result["viewer"] == "" and not result["allow_private"] assert result["private"] is False assert not any(sum(values) for values in result["private_bins"].values()) finally: cleanup_ui(session) def test_private_scan_can_start_from_idle_after_login_without_a_public_scan(monkeypatch): from contrib.ui.events import hydrate, load_private_event import contrib.ui.events as ui session = "private-first" request = gr.Request(session_hash=session) profile, token = SimpleNamespace(username="someone"), SimpleNamespace(token="test") calls = [] def stream(query, *args): calls.append(query) yield snapshot(query.year, query.private) monkeypatch.setattr(ui.collection, "stream", stream) try: initial = hydrate("someone", 2025, request, profile, token) assert initial[1]["value"] == 2025 assert initial[3]["allow_private"] and initial[3]["phase"] == "idle" assert calls == [] result = next(load_private_event("someone", 2024, gr.EventData(None, {"revision":1}), request, profile, token))[0] assert len(calls) == 1 and calls[0].private and calls[0].year == 2024 assert result["private"] and result["viewer"] == "someone" finally: cleanup_ui(session) def test_year_selection_before_first_scan_is_only_form_preparation(monkeypatch): import contrib.ui.events as ui session = 'year-before-start' request = gr.Request(session_hash=session) monkeypatch.setattr(ui.collection, 'stream', lambda *a: (_ for _ in ()).throw(AssertionError('Unexpected fetch'))) try: ui.hydrate('someone', 2026, request, OWNER, TOKEN) value = next(ui.year_ui('someone', 2024, session, request, OWNER, TOKEN))[0] assert value['phase'] == 'idle' and value['year'] == 2024 assert len(value['bins']['space.commit']) == 366 assert not value['run_id'] prefs.remember(session, snapshot(2026)) value = next(ui.year_ui('another', 2025, session, request, OWNER, TOKEN))[0] assert value['phase'] == 'error' and value['username'] == 'someone' assert prefs.previous(session) is None finally: cleanup_ui(session) def test_returning_to_own_profile_allows_explicit_private_scan(monkeypatch): import contrib.ui.events as ui session = 'owner-after-another-profile' request = gr.Request(session_hash=session) profile, token = SimpleNamespace(username='someone'), SimpleNamespace(token='test') other = snapshot(2026); other['username'] = 'another' prefs.remember(session, other) calls = [] def stream(query, *args): calls.append(query) yield snapshot(query.year, query.private) monkeypatch.setattr(ui.collection, 'stream', stream) try: controls = ui.access_controls('someone', 2026, request, profile, token) assert controls['allow_private'] and controls['viewer'] == 'someone' assert calls == [] and controls['username'] == 'someone' list(ui.load_private_event('someone', 2026, gr.EventData(None, {'revision':1}), request, profile, token)) assert len(calls) == 1 and calls[0].private and calls[0].username == 'someone' finally: cleanup_ui(session)