Spaces:
Running
Running
| 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) | |