Merge pull request 'Make My Work authorization constant-request and deadline-safe' (#172) from timmy/171-constant-request-work-auth into main

src/gitea_proxy.py

@@ -544,25 +544,29 @@ async def issue_detail(repository: str, number: int) -> dict:
     }
 
 
-async def _work_contains(stream: str, repository: str, number: int) -> bool:
+async def _current_login_and_target(path: str) -> tuple[str, dict]:
-    page = 1
+    user, target = await asyncio.gather(current_user(), fetch(path))
-    while page <= 100:
+    login = user.get("login") if isinstance(user, dict) else None
-        result = await work_page(stream, page)
+    if not isinstance(login, str) or not login or not isinstance(target, dict):
-        if any(
+        return "", {}
-            item.get("number") == number
+    return login, target
-            and isinstance(item.get("repository"), dict)
+
-            and item["repository"].get("full_name") == repository
+
-            for item in result["items"]
+def _login_in_users(login: str, value: object) -> bool:
-        ):
+    return isinstance(value, list) and any(
-            return True
+        isinstance(user, dict) and user.get("login") == login for user in value
-        if not result["has_more"]:
+    )
-            return False
-        page += 1
-    return False
 
 
 async def is_assigned_issue(repository: str, number: int) -> bool:
-    return await _work_contains("issue", repository, number)
+    login, issue = await _current_login_and_target(
+        f"repos/{repository}/issues/{number}"
+    )
+    return (
+        issue.get("state") == "open"
+        and not isinstance(issue.get("pull_request"), dict)
+        and _login_in_users(login, issue.get("assignees"))
+    )
 
 
 async def pull_requests() -> WorkItems:
@@ -592,11 +596,23 @@ async def pull_requests() -> WorkItems:
 
 
 async def is_requested_review(repository: str, number: int) -> bool:
-    return await _work_contains("review", repository, number)
+    login, pull = await _current_login_and_target(
+        f"repos/{repository}/pulls/{number}"
+    )
+    return (
+        pull.get("state") == "open"
+        and _login_in_users(login, pull.get("requested_reviewers"))
+    )
 
 
 async def is_assigned_pull(repository: str, number: int) -> bool:
-    return await _work_contains("pull", repository, number)
+    login, pull = await _current_login_and_target(
+        f"repos/{repository}/pulls/{number}"
+    )
+    return (
+        pull.get("state") == "open"
+        and _login_in_users(login, pull.get("assignees"))
+    )
 
 
 async def pull_completion_detail(repository: str, number: int) -> dict:

tests/test_gitea_work_search.py

@@ -142,53 +142,75 @@ async def test_pull_request_searches_start_concurrently(monkeypatch):
 
 
 @pytest.mark.anyio
-async def test_requested_review_guard_uses_only_dedicated_review_search(monkeypatch):
+async def test_requested_review_guard_uses_direct_pull_and_current_user(monkeypatch):
     requested = []
 
-    async def fake_page(stream, page=1, limit=50):
+    async def fake_fetch(path):
-        requested.append((stream, page))
+        requested.append(path)
-        items = [] if page == 1 else [{
+        if path == "user":
-            "number": 77,
+            return {"login": "timmy"}
-            "repository": {"full_name": "stackchain/api"},
+        assert path == "repos/stackchain/api/pulls/77"
-        }]
         return {
-            "stream": stream, "items": items, "page": page,
+            "state": "open",
-            "total": 51, "has_more": page == 1,
+            "requested_reviewers": [{"login": "timmy"}],
         }
 
-    monkeypatch.setattr(gitea_proxy, "work_page", fake_page)
+    async def reject_search(*args, **kwargs):
+        raise AssertionError("authorization must not scan review search pages")
+
+    monkeypatch.setattr(gitea_proxy, "fetch", fake_fetch)
+    monkeypatch.setattr(gitea_proxy, "work_page", reject_search)
 
     assert await gitea_proxy.is_requested_review("stackchain/api", 77) is True
-    assert await gitea_proxy.is_requested_review("stackchain/api", 9) is False
+    assert sorted(requested) == ["repos/stackchain/api/pulls/77", "user"]
-    assert requested == [
-        ("review", 1), ("review", 2),
-        ("review", 1), ("review", 2),
-    ]
 
 
 @pytest.mark.anyio
-async def test_assignment_guards_find_eligible_items_beyond_first_page(monkeypatch):
+async def test_assigned_issue_guard_uses_direct_target_and_current_user(monkeypatch):
     requested = []
 
-    async def fake_page(stream, page=1, limit=50):
+    async def fake_fetch(path):
-        requested.append((stream, page))
+        requested.append(path)
+        if path == "user":
+            return {"login": "timmy"}
+        assert path == "repos/stackchain/api/issues/77"
         return {
-            "stream": stream,
+            "state": "open",
-            "items": [] if page == 1 else [{
+            "assignees": [{"login": "timmy"}],
-                "number": 77,
-                "repository": {"full_name": "stackchain/api"},
-            }],
-            "page": page, "total": 51, "has_more": page == 1,
         }
 
-    monkeypatch.setattr(gitea_proxy, "work_page", fake_page)
+    async def reject_search(*args, **kwargs):
+        raise AssertionError("authorization must not scan work search pages")
+
+    monkeypatch.setattr(gitea_proxy, "fetch", fake_fetch)
+    monkeypatch.setattr(gitea_proxy, "work_page", reject_search)
 
     assert await gitea_proxy.is_assigned_issue("stackchain/api", 77) is True
+    assert sorted(requested) == ["repos/stackchain/api/issues/77", "user"]
+
+
+@pytest.mark.anyio
+async def test_assigned_pull_guard_uses_direct_target_and_current_user(monkeypatch):
+    requested = []
+
+    async def fake_fetch(path):
+        requested.append(path)
+        if path == "user":
+            return {"login": "timmy"}
+        assert path == "repos/stackchain/api/pulls/77"
+        return {
+            "state": "open",
+            "assignees": [{"login": "timmy"}],
+        }
+
+    async def reject_search(*args, **kwargs):
+        raise AssertionError("authorization must not scan work search pages")
+
+    monkeypatch.setattr(gitea_proxy, "fetch", fake_fetch)
+    monkeypatch.setattr(gitea_proxy, "work_page", reject_search)
+
     assert await gitea_proxy.is_assigned_pull("stackchain/api", 77) is True
-    assert requested == [
+    assert sorted(requested) == ["repos/stackchain/api/pulls/77", "user"]
-        ("issue", 1), ("issue", 2),
-        ("pull", 1), ("pull", 2),
-    ]
 
 
 @pytest.mark.anyio