Spaces:
Running
Running
| 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" | |
| 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 | |
| def test_invalid_username_rejected(value): | |
| with pytest.raises(HubError): | |
| validate_username(value) | |