from threading import Event import pytest from contrib import cache, collection from contrib.aggregation import Accumulator from contrib.domain import ActivityEvent, CollectionRequest, Update, timestamp from contrib.config import SOURCES from contrib.collectors.classifiers import classify from contrib.collectors import activity from contrib.hub import Cancelled, HubError, validate_username def event(key="model.commit", identity="repo:commit", at="2024-02-29T12:00:00Z"): return ActivityEvent(key, identity, timestamp(at)) def test_deduplication_keeps_creation_separate_from_commit_and_filters_year(): acc = Accumulator(CollectionRequest("someone", 2024)) acc.apply(Update("repositories", [event(), event(), event("model.created"), event(at="2023-12-31T12:00:00Z")])) assert len(acc.snapshot["bins"]["model.commit"]) == 366 assert sum(acc.snapshot["bins"]["model.commit"]) == 1 assert sum(acc.snapshot["bins"]["model.created"]) == 1 assert acc.snapshot["bins"]["model.commit"][59] == 1 def test_utc_date_boundary(): assert timestamp("2024-01-01T00:30:00+02:00").year == 2023 assert timestamp("invalid") is None def test_partial_and_cancelled_are_never_complete(): acc = Accumulator(CollectionRequest("someone", 2024)) for source in SOURCES: acc.apply(Update(source, status="complete")) acc.finish() assert acc.snapshot["phase"] == "complete" acc.apply(Update("likes", status="partial", warnings=["Feed unavailable"])) acc.finish() assert acc.snapshot["phase"] == "partial" acc.finish(cancelled=True) assert acc.snapshot["phase"] == "cancelled" def test_snapshots_do_not_mutate_previously_emitted_data(): acc = Accumulator(CollectionRequest("someone", 2024)) first = acc.copy() acc.apply(Update("repositories", [event()])) assert sum(first["bins"]["model.commit"]) == 0 def discussion(author, time="2024-03-01T12:00:00Z"): return {"time": time, "eventId": "comment-1", "repoType": "model", "repoId": "other/model", "discussionData": {"num": 2, "createdAt": "2024-02-01T12:00:00Z", "author": {"name": author}, "isPullRequest": True}} def test_own_pr_opening_and_later_followup_are_distinct(): records = classify("community", discussion("someone"), "someone") assert [r.key for r in records] == ["pr.opened", "pr.followup"] assert records[0].at.month == 2 assert records[1].at.month == 3 opening = classify("community", discussion("someone", "2024-02-01T12:00:03Z"), "someone") assert [r.key for r in opening] == ["pr.opened"] def test_external_pr_followup_and_like_subtype(): assert [r.key for r in classify("community", discussion("other"), "someone")] == ["pr.followup"] records = classify("likes", {"time": "2024-01-02T00:00:00Z", "repoType": "space", "repoId": "other/app"}, "someone") assert records[0].key == "like.space" def test_collection_upvotes_match_current_hub_feed(): records = classify("upvotes", {"time": "2026-10-28T09:23:27Z", "targetType": "collection", "targetUrl": "/collections/org/name-id", "target": {"id": "id"}}, "someone") assert records[0].key == "upvote.collection" @pytest.mark.parametrize("payload", [[{"time": "x"}], {"recentActivity": [{"time": "x"}], "cursor": "next"}, {"items": [{"time": "x"}]}]) def test_benign_feed_envelopes(payload): assert len(activity.normalize(payload)[0]) == 1 def test_unknown_feed_schema_is_not_zero_activity(): with pytest.raises(HubError): activity.normalize({"unexpected": []}) def test_network_failure_preserves_first_batch(monkeypatch): class FakeHub: count = 0 def __init__(self, cancel): pass def close(self): pass def get(self, *args, **kwargs): self.count += 1 if self.count > 1: raise HubError("Temporary failure") return {"recentActivity": [{"time": "2024-03-01T00:00:00Z", "repoType": "space", "repoId": "other/app"}], "cursor": "next"}, None monkeypatch.setattr(activity, "HubClient", FakeHub) updates = [] activity.collect_activity(CollectionRequest("someone", 2024), "likes", Event(), updates.append) assert updates[-1].status == "partial" assert any(update.events for update in updates) assert not any(update.status == "complete" for update in updates) def test_repeated_feed_page_stops_with_partial_result(monkeypatch): class FakeHub: def __init__(self, cancel): pass def close(self): pass def get(self, *args, **kwargs): return {"recentActivity": [{"time": "2024-03-01T00:00:00Z", "repoType": "space", "repoId": "other/app"}]}, None monkeypatch.setattr(activity, "HubClient", FakeHub) updates = [] activity.collect_activity(CollectionRequest("someone", 2024), "likes", Event(), updates.append) assert updates[-1].status == "partial" assert updates[-1].pages == 2 def test_cache_expiry_and_private_isolation(monkeypatch): monkeypatch.setattr(cache, "current_year", lambda: 2024) cache._items.clear() now = [10.0] monkeypatch.setattr(cache, "monotonic", lambda: now[0]) public = CollectionRequest("someone", 2024) value = Accumulator(public).copy() value["phase"] = "complete" cache.put(public, value) found = cache.get(public) assert found["cached"] found["username"] = "modified" assert cache.get(public)["username"] == "someone" private = CollectionRequest("someone", 2024, private=True, token="test-only") assert cache.get(private) is None cache.put(private, {**value, "username": "private-result"}) assert cache.get(public)["username"] == "someone" now[0] += 601 assert cache.get(public) is None def test_cancel_does_not_touch_other_sessions(): a, b = Event(), Event() collection._active.update(test_a=a, test_b=b) try: collection.stop("test_a") assert a.is_set() and not b.is_set() finally: collection._active.pop("test_a") collection._active.pop("test_b") def test_stream_cleans_up_after_generator_close(monkeypatch): request = CollectionRequest("someone", 2024) generator = collection.stream(request, "test-close", refresh=True) assert next(generator)["phase"] == "queued" assert "test-close" in collection._active generator.close() assert "test-close" not in collection._active def test_closing_live_stream_joins_all_collection_workers(monkeypatch): finished = [] class FakeHub: def __init__(self, cancel): pass def verify_profile(self, username): return {} def close(self): pass def worker(request, source, cancel, emit): emit(Update(source, detail="Reading")) cancel.wait(5) finished.append(source) monkeypatch.setattr(collection, "HubClient", FakeHub) monkeypatch.setattr(collection, "collect_activity", worker) monkeypatch.setattr(collection, "collect_repositories", lambda request, cancel, emit: worker(request, "repositories", cancel, emit)) generator = collection.stream(CollectionRequest("someone", 2024), "test-workers", refresh=True) assert next(generator)["phase"] == "queued" assert next(generator)["phase"] == "loading" assert next(generator)["phase"] == "loading" generator.close() assert set(finished) == set(SOURCES) assert "test-workers" not in collection._active @pytest.mark.parametrize("value", ["https://huggingface.co/person", "@person", "