We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
chore(deps): bump grpcio-status from 1.51.1 to 1.51.3
Bumps [grpcio-status](https://grpc.io) from 1.51.1 to 1.51.3.


[![Dependabot compatibility score](https://dependabot-badges.githubapp.com/badges/compatibility_score?dependency-name=grpcio-status&package-manager=pip&previous-version=1.51.1&new-version=1.51.3)](https://docs.github.com/en/github/managing-security-vulnerabilities/about-dependabot-security-updates#about-compatibility-scores)

You can trigger a rebase of this PR by commenting `@dependabot rebase`.

[//]: # (dependabot-automerge-start)
[//]: # (dependabot-automerge-end)

---

<details>
<summary>Dependabot commands and options</summary>
<br />

You can trigger Dependabot actions by commenting on this PR:
- `@dependabot rebase` will rebase this PR
- `@dependabot recreate` will recreate this PR, overwriting any edits that have been made to it
- `@dependabot merge` will merge this PR after your CI passes on it
- `@dependabot squash and merge` will squash and merge this PR after your CI passes on it
- `@dependabot cancel merge` will cancel a previously requested merge and block automerging
- `@dependabot reopen` will reopen this PR if it is closed
- `@dependabot close` will close this PR and stop Dependabot recreating it. You can achieve the same result by closing it manually
- `@dependabot ignore this major version` will close this PR and stop Dependabot creating any more for this major version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this minor version` will close this PR and stop Dependabot creating any more for this minor version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this dependency` will close this PR and stop Dependabot creating any more for this dependency (unless you reopen the PR or upgrade to it yourself)


</details>

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `warehouse/macaroons/caveats/_core.py`
Content:
```
1 # Licensed under the Apache License, Version 2.0 (the "License");
2 # you may not use this file except in compliance with the License.
3 # You may obtain a copy of the License at
4 #
5 # http://www.apache.org/licenses/LICENSE-2.0
6 #
7 # Unless required by applicable law or agreed to in writing, software
8 # distributed under the License is distributed on an "AS IS" BASIS,
9 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
10 # See the License for the specific language governing permissions and
11 # limitations under the License.
12 from __future__ import annotations
13 
14 import dataclasses
15 import json
16 import typing
17 
18 from collections.abc import Callable, Mapping, Sequence
19 from dataclasses import dataclass
20 from typing import Any, ClassVar, TypeVar
21 
22 from pydantic import ValidationError
23 from pydantic.dataclasses import dataclass as pydantic_dataclass
24 from pyramid.request import Request
25 
26 from warehouse.macaroons.caveats import _legacy
27 
28 T = TypeVar("T")
29 S = TypeVar("S")
30 
31 
32 class CaveatError(Exception):
33     pass
34 
35 
36 class CaveatDeserializationError(CaveatError):
37     pass
38 
39 
40 @dataclass(frozen=True, slots=True)
41 class Success:
42     def __bool__(self):
43         return True
44 
45 
46 @dataclass(frozen=True, slots=True)
47 class Failure:
48     reason: str
49 
50     def __bool__(self):
51         return False
52 
53 
54 Result = Success | Failure
55 
56 
57 @pydantic_dataclass(frozen=True)
58 class Caveat:
59     tag: ClassVar[int]
60 
61     def verify(self, request: Request, context: Any, permission: str) -> Result:
62         raise NotImplementedError
63 
64     def __serialize__(self) -> Sequence:
65         return (self.tag,) + dataclasses.astuple(self)
66 
67     @classmethod
68     def __deserialize__(cls: type[S], data: Sequence) -> S:
69         kwargs = {}
70         for i, field in enumerate(dataclasses.fields(cls)):
71             if len(data) > i:
72                 value = data[i]
73             elif field.default is not dataclasses.MISSING:
74                 value = field.default
75             elif field.default_factory is not dataclasses.MISSING:
76                 value = field.default_factory()
77             else:
78                 raise CaveatDeserializationError("Not enough values")
79 
80             kwargs[field.name] = value
81 
82         try:
83             obj = cls(**kwargs)
84         except ValidationError:
85             raise CaveatDeserializationError("invalid values for fields")
86 
87         return obj
88 
89 
90 class _CaveatRegistry:
91 
92     _tags: dict[int, type[Caveat]]
93 
94     def __init__(self, *args: Any, **kwargs: Any):
95         super().__init__(*args, **kwargs)
96         self._tags = {}
97 
98     def add(self, tag: int, cls: type[Caveat]):
99         if tag in self._tags:
100             raise TypeError(
101                 f"Cannot re-use tag: {tag}, already used by {self._tags[tag]}"
102             )
103 
104         self._tags[tag] = cls
105         cls.tag = tag
106 
107     def lookup(self, /, tag: int) -> type[Caveat] | None:
108         return self._tags.get(tag)
109 
110 
111 _caveat_registry = _CaveatRegistry()
112 
113 
114 def as_caveat(*, tag: int) -> Callable[[type[T]], type[T]]:
115     def deco(cls: type[T]) -> type[T]:
116         _caveat_registry.add(tag, typing.cast(type[Caveat], cls))
117         return cls
118 
119     return deco
120 
121 
122 def serialize(caveat: Caveat) -> bytes:
123     return json.dumps(
124         caveat.__serialize__(), sort_keys=True, separators=(",", ":")
125     ).encode("utf8")
126 
127 
128 def deserialize(data: bytes) -> Caveat:
129     loaded = json.loads(data)
130 
131     # Our original caveats were implemented as a mapping with arbitrary keys,
132     # so if we've gotten one of our those, we'll attempt to adapt it to our
133     # new format.
134     if isinstance(loaded, Mapping):
135         loaded = _legacy.adapt(loaded)
136         if loaded is None:
137             raise CaveatDeserializationError("caveat must be an array")
138 
139     if not isinstance(loaded, Sequence) or isinstance(loaded, str):
140         raise CaveatDeserializationError("caveat must be an array")
141 
142     if not len(loaded):
143         raise CaveatDeserializationError("caveat array cannot be empty")
144 
145     tag, *fields = loaded
146     cls = _caveat_registry.lookup(tag)
147 
148     if cls is None:
149         raise CaveatDeserializationError(f"caveat has unknown tag: {tag}")
150 
151     return cls.__deserialize__(fields)
152
```

Path: `warehouse/admin/views/checks.py`
Content:
```
1 # Licensed under the Apache License, Version 2.0 (the "License");
2 # you may not use this file except in compliance with the License.
3 # You may obtain a copy of the License at
4 #
5 # http://www.apache.org/licenses/LICENSE-2.0
6 #
7 # Unless required by applicable law or agreed to in writing, software
8 # distributed under the License is distributed on an "AS IS" BASIS,
9 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
10 # See the License for the specific language governing permissions and
11 # limitations under the License.
12 
13 from pyramid.httpexceptions import HTTPNotFound, HTTPSeeOther
14 from pyramid.view import view_config
15 from sqlalchemy.exc import NoResultFound
16 
17 from warehouse.malware.models import MalwareCheck, MalwareCheckState, MalwareCheckType
18 from warehouse.malware.tasks import backfill, remove_verdicts, run_scheduled_check
19 
20 EVALUATION_RUN_SIZE = 10000
21 
22 
23 @view_config(
24     route_name="admin.checks.list",
25     renderer="admin/malware/checks/index.html",
26     permission="moderator",
27     request_method="GET",
28     uses_session=True,
29 )
30 def get_checks(request):
31     all_checks = request.db.query(MalwareCheck)
32     active_checks = []
33     for check in all_checks:
34         if not check.is_stale:
35             active_checks.append(check)
36 
37     active_checks.sort(key=lambda check: check.created, reverse=True)
38 
39     return {"checks": active_checks}
40 
41 
42 @view_config(
43     route_name="admin.checks.detail",
44     renderer="admin/malware/checks/detail.html",
45     permission="moderator",
46     request_method="GET",
47     uses_session=True,
48 )
49 def get_check(request):
50     check = get_check_by_name(request.db, request.matchdict["check_name"])
51 
52     all_checks = (
53         request.db.query(MalwareCheck)
54         .filter(MalwareCheck.name == request.matchdict["check_name"])
55         .order_by(MalwareCheck.version.desc())
56         .all()
57     )
58 
59     return {
60         "check": check,
61         "checks": all_checks,
62         "states": MalwareCheckState,
63         "evaluation_run_size": EVALUATION_RUN_SIZE,
64     }
65 
66 
67 @view_config(
68     route_name="admin.checks.run_evaluation",
69     permission="admin",
70     request_method="POST",
71     uses_session=True,
72     require_methods=False,
73     require_csrf=True,
74 )
75 def run_evaluation(request):
76     check = get_check_by_name(request.db, request.matchdict["check_name"])
77 
78     if check.state not in (MalwareCheckState.Enabled, MalwareCheckState.Evaluation):
79         request.session.flash(
80             "Check must be in 'enabled' or 'evaluation' state to manually execute.",
81             queue="error",
82         )
83         return HTTPSeeOther(
84             request.route_path("admin.checks.detail", check_name=check.name)
85         )
86 
87     if check.check_type == MalwareCheckType.EventHook:
88         request.session.flash(
89             f"Running {check.name} on {EVALUATION_RUN_SIZE} {check.hooked_object.value}s\
90 !",
91             queue="success",
92         )
93         request.task(backfill).delay(check.name, EVALUATION_RUN_SIZE)
94 
95     else:
96         request.session.flash(f"Running {check.name} now!", queue="success")
97         request.task(run_scheduled_check).delay(check.name, manually_triggered=True)
98 
99     return HTTPSeeOther(
100         request.route_path("admin.checks.detail", check_name=check.name)
101     )
102 
103 
104 @view_config(
105     route_name="admin.checks.change_state",
106     permission="admin",
107     request_method="POST",
108     uses_session=True,
109     require_methods=False,
110     require_csrf=True,
111 )
112 def change_check_state(request):
113     check = get_check_by_name(request.db, request.matchdict["check_name"])
114 
115     try:
116         check_state = request.POST["check_state"]
117     except KeyError:
118         raise HTTPNotFound
119 
120     try:
121         check.state = MalwareCheckState(check_state)
122     except ValueError:
123         request.session.flash("Invalid check state provided.", queue="error")
124     else:
125         if check.state == MalwareCheckState.WipedOut:
126             request.task(remove_verdicts).delay(check.name)
127         request.session.flash(
128             f"Changed {check.name!r} check to {check.state.value!r}!", queue="success"
129         )
130     finally:
131         return HTTPSeeOther(
132             request.route_path("admin.checks.detail", check_name=check.name)
133         )
134 
135 
136 def get_check_by_name(db, check_name):
137     try:
138         # Throw an exception if and only if no results are returned.
139         newest = (
140             db.query(MalwareCheck)
141             .filter(MalwareCheck.name == check_name)
142             .order_by(MalwareCheck.version.desc())
143             .limit(1)
144             .one()
145         )
146     except NoResultFound:
147         raise HTTPNotFound
148 
149     return newest
150
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/warehouse/admin/views/checks.py b/warehouse/admin/views/checks.py
--- a/warehouse/admin/views/checks.py
+++ b/warehouse/admin/views/checks.py
@@ -86,8 +86,8 @@
 
     if check.check_type == MalwareCheckType.EventHook:
         request.session.flash(
-            f"Running {check.name} on {EVALUATION_RUN_SIZE} {check.hooked_object.value}s\
-!",
+            f"Running {check.name} on {EVALUATION_RUN_SIZE} "
+            f"{check.hooked_object.value}s!",
             queue="success",
         )
         request.task(backfill).delay(check.name, EVALUATION_RUN_SIZE)
diff --git a/warehouse/macaroons/caveats/_core.py b/warehouse/macaroons/caveats/_core.py
--- a/warehouse/macaroons/caveats/_core.py
+++ b/warehouse/macaroons/caveats/_core.py
@@ -15,7 +15,7 @@
 import json
 import typing
 
-from collections.abc import Callable, Mapping, Sequence
+from collections.abc import Mapping, Sequence
 from dataclasses import dataclass
 from typing import Any, ClassVar, TypeVar
 
@@ -111,7 +111,13 @@
 _caveat_registry = _CaveatRegistry()
 
 
-def as_caveat(*, tag: int) -> Callable[[type[T]], type[T]]:
+# TODO: The return signature detected is `"Union[Type[Dataclass], DataclassProxy]"`,
+# but the expectation is `Type[Dataclass]`.
+# See https://github.com/pydantic/pydantic/issues/4498 but not exactly the same.
+# This might not be corrected in pydantic until 2.0.
+# Original signature with type hints:
+# def as_caveat(*, tag: int) -> Callable[[type[T]], type[T]]:
+def as_caveat(*, tag: int):
     def deco(cls: type[T]) -> type[T]:
         _caveat_registry.add(tag, typing.cast(type[Caveat], cls))
         return cls

{"golden_diff": "diff --git a/warehouse/admin/views/checks.py b/warehouse/admin/views/checks.py\n--- a/warehouse/admin/views/checks.py\n+++ b/warehouse/admin/views/checks.py\n@@ -86,8 +86,8 @@\n \n     if check.check_type == MalwareCheckType.EventHook:\n         request.session.flash(\n-            f\"Running {check.name} on {EVALUATION_RUN_SIZE} {check.hooked_object.value}s\\\n-!\",\n+            f\"Running {check.name} on {EVALUATION_RUN_SIZE} \"\n+            f\"{check.hooked_object.value}s!\",\n             queue=\"success\",\n         )\n         request.task(backfill).delay(check.name, EVALUATION_RUN_SIZE)\ndiff --git a/warehouse/macaroons/caveats/_core.py b/warehouse/macaroons/caveats/_core.py\n--- a/warehouse/macaroons/caveats/_core.py\n+++ b/warehouse/macaroons/caveats/_core.py\n@@ -15,7 +15,7 @@\n import json\n import typing\n \n-from collections.abc import Callable, Mapping, Sequence\n+from collections.abc import Mapping, Sequence\n from dataclasses import dataclass\n from typing import Any, ClassVar, TypeVar\n \n@@ -111,7 +111,13 @@\n _caveat_registry = _CaveatRegistry()\n \n \n-def as_caveat(*, tag: int) -> Callable[[type[T]], type[T]]:\n+# TODO: The return signature detected is `\"Union[Type[Dataclass], DataclassProxy]\"`,\n+# but the expectation is `Type[Dataclass]`.\n+# See https://github.com/pydantic/pydantic/issues/4498 but not exactly the same.\n+# This might not be corrected in pydantic until 2.0.\n+# Original signature with type hints:\n+# def as_caveat(*, tag: int) -> Callable[[type[T]], type[T]]:\n+def as_caveat(*, tag: int):\n     def deco(cls: type[T]) -> type[T]:\n         _caveat_registry.add(tag, typing.cast(type[Caveat], cls))\n         return cls\n", "issue": "chore(deps): bump grpcio-status from 1.51.1 to 1.51.3\nBumps [grpcio-status](https://grpc.io) from 1.51.1 to 1.51.3.\n\n\n[![Dependabot compatibility score](https://dependabot-badges.githubapp.com/badges/compatibility_score?dependency-name=grpcio-status&package-manager=pip&previous-version=1.51.1&new-version=1.51.3)](https://docs.github.com/en/github/managing-security-vulnerabilities/about-dependabot-security-updates#about-compatibility-scores)\n\nYou can trigger a rebase of this PR by commenting `@dependabot rebase`.\n\n[//]: # (dependabot-automerge-start)\n[//]: # (dependabot-automerge-end)\n\n---\n\n<details>\n<summary>Dependabot commands and options</summary>\n<br />\n\nYou can trigger Dependabot actions by commenting on this PR:\n- `@dependabot rebase` will rebase this PR\n- `@dependabot recreate` will recreate this PR, overwriting any edits that have been made to it\n- `@dependabot merge` will merge this PR after your CI passes on it\n- `@dependabot squash and merge` will squash and merge this PR after your CI passes on it\n- `@dependabot cancel merge` will cancel a previously requested merge and block automerging\n- `@dependabot reopen` will reopen this PR if it is closed\n- `@dependabot close` will close this PR and stop Dependabot recreating it. You can achieve the same result by closing it manually\n- `@dependabot ignore this major version` will close this PR and stop Dependabot creating any more for this major version (unless you reopen the PR or upgrade to it yourself)\n- `@dependabot ignore this minor version` will close this PR and stop Dependabot creating any more for this minor version (unless you reopen the PR or upgrade to it yourself)\n- `@dependabot ignore this dependency` will close this PR and stop Dependabot creating any more for this dependency (unless you reopen the PR or upgrade to it yourself)\n\n\n</details>\n", "before_files": [{"content": "# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n# http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\nfrom __future__ import annotations\n\nimport dataclasses\nimport json\nimport typing\n\nfrom collections.abc import Callable, Mapping, Sequence\nfrom dataclasses import dataclass\nfrom typing import Any, ClassVar, TypeVar\n\nfrom pydantic import ValidationError\nfrom pydantic.dataclasses import dataclass as pydantic_dataclass\nfrom pyramid.request import Request\n\nfrom warehouse.macaroons.caveats import _legacy\n\nT = TypeVar(\"T\")\nS = TypeVar(\"S\")\n\n\nclass CaveatError(Exception):\n    pass\n\n\nclass CaveatDeserializationError(CaveatError):\n    pass\n\n\n@dataclass(frozen=True, slots=True)\nclass Success:\n    def __bool__(self):\n        return True\n\n\n@dataclass(frozen=True, slots=True)\nclass Failure:\n    reason: str\n\n    def __bool__(self):\n        return False\n\n\nResult = Success | Failure\n\n\n@pydantic_dataclass(frozen=True)\nclass Caveat:\n    tag: ClassVar[int]\n\n    def verify(self, request: Request, context: Any, permission: str) -> Result:\n        raise NotImplementedError\n\n    def __serialize__(self) -> Sequence:\n        return (self.tag,) + dataclasses.astuple(self)\n\n    @classmethod\n    def __deserialize__(cls: type[S], data: Sequence) -> S:\n        kwargs = {}\n        for i, field in enumerate(dataclasses.fields(cls)):\n            if len(data) > i:\n                value = data[i]\n            elif field.default is not dataclasses.MISSING:\n                value = field.default\n            elif field.default_factory is not dataclasses.MISSING:\n                value = field.default_factory()\n            else:\n                raise CaveatDeserializationError(\"Not enough values\")\n\n            kwargs[field.name] = value\n\n        try:\n            obj = cls(**kwargs)\n        except ValidationError:\n            raise CaveatDeserializationError(\"invalid values for fields\")\n\n        return obj\n\n\nclass _CaveatRegistry:\n\n    _tags: dict[int, type[Caveat]]\n\n    def __init__(self, *args: Any, **kwargs: Any):\n        super().__init__(*args, **kwargs)\n        self._tags = {}\n\n    def add(self, tag: int, cls: type[Caveat]):\n        if tag in self._tags:\n            raise TypeError(\n                f\"Cannot re-use tag: {tag}, already used by {self._tags[tag]}\"\n            )\n\n        self._tags[tag] = cls\n        cls.tag = tag\n\n    def lookup(self, /, tag: int) -> type[Caveat] | None:\n        return self._tags.get(tag)\n\n\n_caveat_registry = _CaveatRegistry()\n\n\ndef as_caveat(*, tag: int) -> Callable[[type[T]], type[T]]:\n    def deco(cls: type[T]) -> type[T]:\n        _caveat_registry.add(tag, typing.cast(type[Caveat], cls))\n        return cls\n\n    return deco\n\n\ndef serialize(caveat: Caveat) -> bytes:\n    return json.dumps(\n        caveat.__serialize__(), sort_keys=True, separators=(\",\", \":\")\n    ).encode(\"utf8\")\n\n\ndef deserialize(data: bytes) -> Caveat:\n    loaded = json.loads(data)\n\n    # Our original caveats were implemented as a mapping with arbitrary keys,\n    # so if we've gotten one of our those, we'll attempt to adapt it to our\n    # new format.\n    if isinstance(loaded, Mapping):\n        loaded = _legacy.adapt(loaded)\n        if loaded is None:\n            raise CaveatDeserializationError(\"caveat must be an array\")\n\n    if not isinstance(loaded, Sequence) or isinstance(loaded, str):\n        raise CaveatDeserializationError(\"caveat must be an array\")\n\n    if not len(loaded):\n        raise CaveatDeserializationError(\"caveat array cannot be empty\")\n\n    tag, *fields = loaded\n    cls = _caveat_registry.lookup(tag)\n\n    if cls is None:\n        raise CaveatDeserializationError(f\"caveat has unknown tag: {tag}\")\n\n    return cls.__deserialize__(fields)\n", "path": "warehouse/macaroons/caveats/_core.py"}, {"content": "# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n# http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nfrom pyramid.httpexceptions import HTTPNotFound, HTTPSeeOther\nfrom pyramid.view import view_config\nfrom sqlalchemy.exc import NoResultFound\n\nfrom warehouse.malware.models import MalwareCheck, MalwareCheckState, MalwareCheckType\nfrom warehouse.malware.tasks import backfill, remove_verdicts, run_scheduled_check\n\nEVALUATION_RUN_SIZE = 10000\n\n\n@view_config(\n    route_name=\"admin.checks.list\",\n    renderer=\"admin/malware/checks/index.html\",\n    permission=\"moderator\",\n    request_method=\"GET\",\n    uses_session=True,\n)\ndef get_checks(request):\n    all_checks = request.db.query(MalwareCheck)\n    active_checks = []\n    for check in all_checks:\n        if not check.is_stale:\n            active_checks.append(check)\n\n    active_checks.sort(key=lambda check: check.created, reverse=True)\n\n    return {\"checks\": active_checks}\n\n\n@view_config(\n    route_name=\"admin.checks.detail\",\n    renderer=\"admin/malware/checks/detail.html\",\n    permission=\"moderator\",\n    request_method=\"GET\",\n    uses_session=True,\n)\ndef get_check(request):\n    check = get_check_by_name(request.db, request.matchdict[\"check_name\"])\n\n    all_checks = (\n        request.db.query(MalwareCheck)\n        .filter(MalwareCheck.name == request.matchdict[\"check_name\"])\n        .order_by(MalwareCheck.version.desc())\n        .all()\n    )\n\n    return {\n        \"check\": check,\n        \"checks\": all_checks,\n        \"states\": MalwareCheckState,\n        \"evaluation_run_size\": EVALUATION_RUN_SIZE,\n    }\n\n\n@view_config(\n    route_name=\"admin.checks.run_evaluation\",\n    permission=\"admin\",\n    request_method=\"POST\",\n    uses_session=True,\n    require_methods=False,\n    require_csrf=True,\n)\ndef run_evaluation(request):\n    check = get_check_by_name(request.db, request.matchdict[\"check_name\"])\n\n    if check.state not in (MalwareCheckState.Enabled, MalwareCheckState.Evaluation):\n        request.session.flash(\n            \"Check must be in 'enabled' or 'evaluation' state to manually execute.\",\n            queue=\"error\",\n        )\n        return HTTPSeeOther(\n            request.route_path(\"admin.checks.detail\", check_name=check.name)\n        )\n\n    if check.check_type == MalwareCheckType.EventHook:\n        request.session.flash(\n            f\"Running {check.name} on {EVALUATION_RUN_SIZE} {check.hooked_object.value}s\\\n!\",\n            queue=\"success\",\n        )\n        request.task(backfill).delay(check.name, EVALUATION_RUN_SIZE)\n\n    else:\n        request.session.flash(f\"Running {check.name} now!\", queue=\"success\")\n        request.task(run_scheduled_check).delay(check.name, manually_triggered=True)\n\n    return HTTPSeeOther(\n        request.route_path(\"admin.checks.detail\", check_name=check.name)\n    )\n\n\n@view_config(\n    route_name=\"admin.checks.change_state\",\n    permission=\"admin\",\n    request_method=\"POST\",\n    uses_session=True,\n    require_methods=False,\n    require_csrf=True,\n)\ndef change_check_state(request):\n    check = get_check_by_name(request.db, request.matchdict[\"check_name\"])\n\n    try:\n        check_state = request.POST[\"check_state\"]\n    except KeyError:\n        raise HTTPNotFound\n\n    try:\n        check.state = MalwareCheckState(check_state)\n    except ValueError:\n        request.session.flash(\"Invalid check state provided.\", queue=\"error\")\n    else:\n        if check.state == MalwareCheckState.WipedOut:\n            request.task(remove_verdicts).delay(check.name)\n        request.session.flash(\n            f\"Changed {check.name!r} check to {check.state.value!r}!\", queue=\"success\"\n        )\n    finally:\n        return HTTPSeeOther(\n            request.route_path(\"admin.checks.detail\", check_name=check.name)\n        )\n\n\ndef get_check_by_name(db, check_name):\n    try:\n        # Throw an exception if and only if no results are returned.\n        newest = (\n            db.query(MalwareCheck)\n            .filter(MalwareCheck.name == check_name)\n            .order_by(MalwareCheck.version.desc())\n            .limit(1)\n            .one()\n        )\n    except NoResultFound:\n        raise HTTPNotFound\n\n    return newest\n", "path": "warehouse/admin/views/checks.py"}], "after_files": [{"content": "# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n# http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\nfrom __future__ import annotations\n\nimport dataclasses\nimport json\nimport typing\n\nfrom collections.abc import Mapping, Sequence\nfrom dataclasses import dataclass\nfrom typing import Any, ClassVar, TypeVar\n\nfrom pydantic import ValidationError\nfrom pydantic.dataclasses import dataclass as pydantic_dataclass\nfrom pyramid.request import Request\n\nfrom warehouse.macaroons.caveats import _legacy\n\nT = TypeVar(\"T\")\nS = TypeVar(\"S\")\n\n\nclass CaveatError(Exception):\n    pass\n\n\nclass CaveatDeserializationError(CaveatError):\n    pass\n\n\n@dataclass(frozen=True, slots=True)\nclass Success:\n    def __bool__(self):\n        return True\n\n\n@dataclass(frozen=True, slots=True)\nclass Failure:\n    reason: str\n\n    def __bool__(self):\n        return False\n\n\nResult = Success | Failure\n\n\n@pydantic_dataclass(frozen=True)\nclass Caveat:\n    tag: ClassVar[int]\n\n    def verify(self, request: Request, context: Any, permission: str) -> Result:\n        raise NotImplementedError\n\n    def __serialize__(self) -> Sequence:\n        return (self.tag,) + dataclasses.astuple(self)\n\n    @classmethod\n    def __deserialize__(cls: type[S], data: Sequence) -> S:\n        kwargs = {}\n        for i, field in enumerate(dataclasses.fields(cls)):\n            if len(data) > i:\n                value = data[i]\n            elif field.default is not dataclasses.MISSING:\n                value = field.default\n            elif field.default_factory is not dataclasses.MISSING:\n                value = field.default_factory()\n            else:\n                raise CaveatDeserializationError(\"Not enough values\")\n\n            kwargs[field.name] = value\n\n        try:\n            obj = cls(**kwargs)\n        except ValidationError:\n            raise CaveatDeserializationError(\"invalid values for fields\")\n\n        return obj\n\n\nclass _CaveatRegistry:\n\n    _tags: dict[int, type[Caveat]]\n\n    def __init__(self, *args: Any, **kwargs: Any):\n        super().__init__(*args, **kwargs)\n        self._tags = {}\n\n    def add(self, tag: int, cls: type[Caveat]):\n        if tag in self._tags:\n            raise TypeError(\n                f\"Cannot re-use tag: {tag}, already used by {self._tags[tag]}\"\n            )\n\n        self._tags[tag] = cls\n        cls.tag = tag\n\n    def lookup(self, /, tag: int) -> type[Caveat] | None:\n        return self._tags.get(tag)\n\n\n_caveat_registry = _CaveatRegistry()\n\n\n# TODO: The return signature detected is `\"Union[Type[Dataclass], DataclassProxy]\"`,\n# but the expectation is `Type[Dataclass]`.\n# See https://github.com/pydantic/pydantic/issues/4498 but not exactly the same.\n# This might not be corrected in pydantic until 2.0.\n# Original signature with type hints:\n# def as_caveat(*, tag: int) -> Callable[[type[T]], type[T]]:\ndef as_caveat(*, tag: int):\n    def deco(cls: type[T]) -> type[T]:\n        _caveat_registry.add(tag, typing.cast(type[Caveat], cls))\n        return cls\n\n    return deco\n\n\ndef serialize(caveat: Caveat) -> bytes:\n    return json.dumps(\n        caveat.__serialize__(), sort_keys=True, separators=(\",\", \":\")\n    ).encode(\"utf8\")\n\n\ndef deserialize(data: bytes) -> Caveat:\n    loaded = json.loads(data)\n\n    # Our original caveats were implemented as a mapping with arbitrary keys,\n    # so if we've gotten one of our those, we'll attempt to adapt it to our\n    # new format.\n    if isinstance(loaded, Mapping):\n        loaded = _legacy.adapt(loaded)\n        if loaded is None:\n            raise CaveatDeserializationError(\"caveat must be an array\")\n\n    if not isinstance(loaded, Sequence) or isinstance(loaded, str):\n        raise CaveatDeserializationError(\"caveat must be an array\")\n\n    if not len(loaded):\n        raise CaveatDeserializationError(\"caveat array cannot be empty\")\n\n    tag, *fields = loaded\n    cls = _caveat_registry.lookup(tag)\n\n    if cls is None:\n        raise CaveatDeserializationError(f\"caveat has unknown tag: {tag}\")\n\n    return cls.__deserialize__(fields)\n", "path": "warehouse/macaroons/caveats/_core.py"}, {"content": "# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n# http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nfrom pyramid.httpexceptions import HTTPNotFound, HTTPSeeOther\nfrom pyramid.view import view_config\nfrom sqlalchemy.exc import NoResultFound\n\nfrom warehouse.malware.models import MalwareCheck, MalwareCheckState, MalwareCheckType\nfrom warehouse.malware.tasks import backfill, remove_verdicts, run_scheduled_check\n\nEVALUATION_RUN_SIZE = 10000\n\n\n@view_config(\n    route_name=\"admin.checks.list\",\n    renderer=\"admin/malware/checks/index.html\",\n    permission=\"moderator\",\n    request_method=\"GET\",\n    uses_session=True,\n)\ndef get_checks(request):\n    all_checks = request.db.query(MalwareCheck)\n    active_checks = []\n    for check in all_checks:\n        if not check.is_stale:\n            active_checks.append(check)\n\n    active_checks.sort(key=lambda check: check.created, reverse=True)\n\n    return {\"checks\": active_checks}\n\n\n@view_config(\n    route_name=\"admin.checks.detail\",\n    renderer=\"admin/malware/checks/detail.html\",\n    permission=\"moderator\",\n    request_method=\"GET\",\n    uses_session=True,\n)\ndef get_check(request):\n    check = get_check_by_name(request.db, request.matchdict[\"check_name\"])\n\n    all_checks = (\n        request.db.query(MalwareCheck)\n        .filter(MalwareCheck.name == request.matchdict[\"check_name\"])\n        .order_by(MalwareCheck.version.desc())\n        .all()\n    )\n\n    return {\n        \"check\": check,\n        \"checks\": all_checks,\n        \"states\": MalwareCheckState,\n        \"evaluation_run_size\": EVALUATION_RUN_SIZE,\n    }\n\n\n@view_config(\n    route_name=\"admin.checks.run_evaluation\",\n    permission=\"admin\",\n    request_method=\"POST\",\n    uses_session=True,\n    require_methods=False,\n    require_csrf=True,\n)\ndef run_evaluation(request):\n    check = get_check_by_name(request.db, request.matchdict[\"check_name\"])\n\n    if check.state not in (MalwareCheckState.Enabled, MalwareCheckState.Evaluation):\n        request.session.flash(\n            \"Check must be in 'enabled' or 'evaluation' state to manually execute.\",\n            queue=\"error\",\n        )\n        return HTTPSeeOther(\n            request.route_path(\"admin.checks.detail\", check_name=check.name)\n        )\n\n    if check.check_type == MalwareCheckType.EventHook:\n        request.session.flash(\n            f\"Running {check.name} on {EVALUATION_RUN_SIZE} \"\n            f\"{check.hooked_object.value}s!\",\n            queue=\"success\",\n        )\n        request.task(backfill).delay(check.name, EVALUATION_RUN_SIZE)\n\n    else:\n        request.session.flash(f\"Running {check.name} now!\", queue=\"success\")\n        request.task(run_scheduled_check).delay(check.name, manually_triggered=True)\n\n    return HTTPSeeOther(\n        request.route_path(\"admin.checks.detail\", check_name=check.name)\n    )\n\n\n@view_config(\n    route_name=\"admin.checks.change_state\",\n    permission=\"admin\",\n    request_method=\"POST\",\n    uses_session=True,\n    require_methods=False,\n    require_csrf=True,\n)\ndef change_check_state(request):\n    check = get_check_by_name(request.db, request.matchdict[\"check_name\"])\n\n    try:\n        check_state = request.POST[\"check_state\"]\n    except KeyError:\n        raise HTTPNotFound\n\n    try:\n        check.state = MalwareCheckState(check_state)\n    except ValueError:\n        request.session.flash(\"Invalid check state provided.\", queue=\"error\")\n    else:\n        if check.state == MalwareCheckState.WipedOut:\n            request.task(remove_verdicts).delay(check.name)\n        request.session.flash(\n            f\"Changed {check.name!r} check to {check.state.value!r}!\", queue=\"success\"\n        )\n    finally:\n        return HTTPSeeOther(\n            request.route_path(\"admin.checks.detail\", check_name=check.name)\n        )\n\n\ndef get_check_by_name(db, check_name):\n    try:\n        # Throw an exception if and only if no results are returned.\n        newest = (\n            db.query(MalwareCheck)\n            .filter(MalwareCheck.name == check_name)\n            .order_by(MalwareCheck.version.desc())\n            .limit(1)\n            .one()\n        )\n    except NoResultFound:\n        raise HTTPNotFound\n\n    return newest\n", "path": "warehouse/admin/views/checks.py"}]}