A refusal's detail now stands where it is sent (HTTPException's detail, a JSONResponse's "detail") or is returned by a factory that only ever is, with any spread before its code and reason, so no spread can replace them. The scan follows no variable and no kwargs any more; what it cannot read in a reason (a lambda's or comprehension's name, a nonlocal or global, a local that reaches itself, a constant or map changed elsewhere, an app import it cannot resolve) is reported. The detail guard reads the same slots and fails on anything but a code dict or a factory's call.
771 lines
28 KiB
Python
771 lines
28 KiB
Python
"""The refusal scan (`refusal_sites.py`) on small sources: each shape on its closed list is followed to its
|
|
values, and each exit from that list is reported, unread or as an unread code. A scan that silently drops a
|
|
shape is a guard green over nothing, so every exit is pinned here, not only the paths the app takes. Pure:
|
|
no app import, no database."""
|
|
|
|
import textwrap
|
|
|
|
from tests.refusal_sites import scan, scan_app
|
|
|
|
|
|
def _scan(**files: str) -> dict:
|
|
return scan(
|
|
{f"{name.replace('__', '/')}.py": textwrap.dedent(src) for name, src in files.items()}
|
|
)
|
|
|
|
|
|
def _code(result: dict, code: str) -> dict:
|
|
return result["codes"][code]
|
|
|
|
|
|
def _unread(result: dict, code: str) -> list[str]:
|
|
return [u.split(": ", 1)[1] for u in _code(result, code)["unread"]]
|
|
|
|
|
|
REFUSAL = """
|
|
from fastapi import HTTPException
|
|
|
|
def refused(status: int, reason: str, **params: object) -> HTTPException:
|
|
return HTTPException(status_code=status, detail={**params, "code": "refused", "reason": reason})
|
|
"""
|
|
|
|
|
|
def test_a_literal_wherever_it_sits_in_the_dict():
|
|
r = _scan(
|
|
a="""
|
|
def f(x):
|
|
raise HTTPException(409, {"code": "clip", "limit_s": 5, "reason": "too_long"})
|
|
def g(x):
|
|
raise HTTPException(409, {"reason": "before", "code": "clip"})
|
|
def h(x):
|
|
raise HTTPException(409, {
|
|
"code": "clip", # a comment between
|
|
"reason": "after_comment",
|
|
})
|
|
def i(x):
|
|
raise HTTPException(409, detail=dict(code="clip", reason="by_call"))
|
|
"""
|
|
)
|
|
assert _code(r, "clip")["reasons"] == ["after_comment", "before", "by_call", "too_long"]
|
|
assert _code(r, "clip")["unread"] == []
|
|
|
|
|
|
def test_both_arms_of_a_conditional_and_a_none_arm_makes_the_site_reasonless():
|
|
r = _scan(
|
|
a="""
|
|
def f(x):
|
|
raise HTTPException(409, {"code": "edit", "reason": "a" if x else "b"})
|
|
def g(x):
|
|
raise HTTPException(409, {"code": "playlist", "reason": "lit" if x else None})
|
|
def h(x):
|
|
raise HTTPException(409, {"code": "busy", "reason": None})
|
|
"""
|
|
)
|
|
assert _code(r, "edit")["reasons"] == ["a", "b"] and _code(r, "edit")["reasonless"] == []
|
|
assert _code(r, "playlist")["reasons"] == ["lit"]
|
|
assert _code(r, "playlist")["reasonless"] == ["a.py:5"]
|
|
assert _code(r, "busy")["reasonless"] == ["a.py:7"]
|
|
|
|
|
|
def test_what_the_list_does_not_hold_is_unread():
|
|
r = _scan(
|
|
a="""
|
|
def f(x, svc):
|
|
raise HTTPException(409, {"code": "edit", "reason": x.why or "c"})
|
|
def g(svc):
|
|
raise HTTPException(409, {"code": "edit", "reason": svc.reason()})
|
|
def h():
|
|
raise HTTPException(409, {"code": "edit", "reason": 5})
|
|
def i(e):
|
|
raise HTTPException(409, {"code": "edit", "reason": f"x{e}"})
|
|
"""
|
|
)
|
|
assert _unread(r, "edit") == ["x.why or 'c'", "svc.reason()", "5", "f'x{e}'"]
|
|
assert _code(r, "edit")["reasons"] == []
|
|
|
|
|
|
def test_a_helper_is_followed_to_every_call_however_it_is_written():
|
|
r = _scan(
|
|
refusal=REFUSAL,
|
|
routes__a="""
|
|
from http import HTTPStatus
|
|
from app import refusal
|
|
from app.refusal import refused
|
|
|
|
def f(status):
|
|
raise refused(status, "by_name_status")
|
|
def g():
|
|
raise refused(HTTPStatus.CONFLICT, reason="by_keyword")
|
|
def h():
|
|
raise refusal.refused(409, "qualified")
|
|
def i(x):
|
|
raise refused(409, "a" if x else "second_arm")
|
|
""",
|
|
routes__b="""
|
|
from other.lib import refused
|
|
|
|
def f():
|
|
refused("an outside library's, not the app's")
|
|
""",
|
|
)
|
|
assert _code(r, "refused")["reasons"] == [
|
|
"a",
|
|
"by_keyword",
|
|
"by_name_status",
|
|
"qualified",
|
|
"second_arm",
|
|
]
|
|
assert _code(r, "refused")["unread"] == []
|
|
|
|
|
|
def test_a_call_the_helper_cannot_be_followed_through_is_unread():
|
|
r = _scan(
|
|
refusal=REFUSAL,
|
|
routes__a="""
|
|
from app.refusal import refused
|
|
|
|
def f(args, kw):
|
|
pick = refused
|
|
raise refused(*args)
|
|
def g(kw):
|
|
raise refused(409, **kw)
|
|
def h():
|
|
raise refused(409)
|
|
""",
|
|
routes__b="""
|
|
import app.refusal
|
|
|
|
def f():
|
|
raise app.refusal.refused(409, "dotted")
|
|
""",
|
|
routes__c="""
|
|
from ..refusal import refused
|
|
|
|
def f():
|
|
raise refused(409, "relative")
|
|
""",
|
|
)
|
|
unread = _unread(r, "refused")
|
|
assert len(unread) == 6
|
|
assert sum("the helper used as a value" in u for u in unread) == 1
|
|
assert sum("a reference the scan cannot resolve" in u for u in unread) == 2
|
|
assert sum("does not pass reason" in u for u in unread) == 2 # `**kw`, and nothing at all
|
|
assert any(u == "refused(*args)" for u in unread)
|
|
assert _code(r, "refused")["reasons"] == []
|
|
|
|
|
|
def test_a_default_is_not_followed_a_call_must_pass_the_reason():
|
|
r = _scan(
|
|
routes___common="""
|
|
def not_found(what: str = "generic"):
|
|
return HTTPException(404, {"code": "not_found", "reason": what})
|
|
|
|
def get_or_404(db, id_, *, what: str = "generic"):
|
|
raise not_found(what)
|
|
""",
|
|
routes__a="""
|
|
from app.routes._common import get_or_404, not_found
|
|
|
|
def f(db):
|
|
get_or_404(db, 1, what="user")
|
|
get_or_404(db, 2)
|
|
raise not_found("debug_session")
|
|
def g():
|
|
raise not_found()
|
|
""",
|
|
)
|
|
assert _code(r, "not_found")["reasons"] == ["debug_session", "user"]
|
|
assert [u.endswith("(does not pass what)") for u in _unread(r, "not_found")] == [True, True]
|
|
|
|
|
|
def test_a_positional_only_reason_passed_by_name_is_not_bound_to_it():
|
|
r = _scan(
|
|
a="""
|
|
def deny(reason: str, /, **params):
|
|
return HTTPException(409, {"code": "playlist", "reason": reason})
|
|
|
|
def f():
|
|
raise deny("by_position")
|
|
def g():
|
|
raise deny(reason="lands_in_params")
|
|
"""
|
|
)
|
|
assert _code(r, "playlist")["reasons"] == ["by_position"]
|
|
assert _unread(r, "playlist") == ["deny(reason='lands_in_params') (does not pass reason)"]
|
|
|
|
|
|
def test_a_helper_reaching_two_sites_gives_both_its_reasons_and_one_reached_again_is_unread():
|
|
r = _scan(
|
|
a="""
|
|
def deny(reason: str, admin):
|
|
if admin:
|
|
return HTTPException(409, {"code": "activate", "reason": reason})
|
|
return HTTPException(409, {"code": "playlist", "reason": reason})
|
|
|
|
def again(reason: str, n):
|
|
if n:
|
|
return again(reason, n - 1)
|
|
return HTTPException(409, {"code": "edit", "reason": reason})
|
|
|
|
def f():
|
|
raise deny("foo", True)
|
|
def g():
|
|
raise again("bar", 3)
|
|
"""
|
|
)
|
|
assert _code(r, "activate")["reasons"] == ["foo"]
|
|
assert _code(r, "playlist")["reasons"] == ["foo"]
|
|
assert _code(r, "edit")["reasons"] == ["bar"]
|
|
assert any("reached again while it is followed" in u for u in _unread(r, "edit"))
|
|
|
|
|
|
def test_a_parameter_its_body_binds_again_is_unread():
|
|
r = _scan(
|
|
a="""
|
|
from typing import Literal
|
|
|
|
def refused(status, reason: str = ""):
|
|
reason = reason or "fallback"
|
|
return HTTPException(status, {"code": "refused", "reason": reason})
|
|
|
|
def typed(reason: Literal["a"]):
|
|
if not reason:
|
|
reason = "b"
|
|
return HTTPException(409, {"code": "edit", "reason": reason})
|
|
|
|
def f():
|
|
raise refused(409, "x")
|
|
"""
|
|
)
|
|
assert _code(r, "refused")["reasons"] == []
|
|
assert _unread(r, "refused") == ["reason (a parameter its body binds again)"]
|
|
assert _unread(r, "edit") == ["reason (a parameter its body binds again)"]
|
|
|
|
|
|
def test_a_local_is_followed_only_through_plain_assignments():
|
|
r = _scan(
|
|
a="""
|
|
from typing import Literal
|
|
|
|
def f(x):
|
|
why = "first"
|
|
if x:
|
|
why = "second" if x > 1 else "third"
|
|
raise HTTPException(409, {"code": "edit", "reason": why})
|
|
def g(svc):
|
|
hit, why = svc.fetch()
|
|
raise HTTPException(409, {"code": "playlist", "reason": why})
|
|
def h(items):
|
|
for why in items:
|
|
pass
|
|
raise HTTPException(409, {"code": "playlist", "reason": why})
|
|
def i():
|
|
why: Literal["typed"] = pick()
|
|
raise HTTPException(409, {"code": "playlist", "reason": why})
|
|
def j(items):
|
|
LABELS[why] = "NotAReason"
|
|
raise HTTPException(409, {"code": "playlist", "reason": why})
|
|
def k(*why):
|
|
raise HTTPException(409, {"code": "playlist", "reason": why})
|
|
"""
|
|
)
|
|
assert _code(r, "edit")["reasons"] == ["first", "second", "third"]
|
|
assert _code(r, "playlist")["reasons"] == []
|
|
assert _unread(r, "playlist") == [
|
|
"why",
|
|
"why",
|
|
"why",
|
|
"why",
|
|
"why",
|
|
] # unpacked, a loop, annotated, unbound, `*args`
|
|
|
|
|
|
def test_a_call_answers_with_its_literal_return_type():
|
|
r = _scan(
|
|
a="""
|
|
from typing import Literal
|
|
|
|
def why(x) -> Literal["quota", "gone"]:
|
|
return "quota"
|
|
def how(x) -> str:
|
|
return "x"
|
|
|
|
def f(x):
|
|
raise HTTPException(409, {"code": "youtube", "reason": why(x)})
|
|
def g(x):
|
|
raise HTTPException(409, {"code": "youtube", "reason": how(x)})
|
|
"""
|
|
)
|
|
assert _code(r, "youtube")["reasons"] == ["gone", "quota"]
|
|
assert _unread(r, "youtube") == ["how(x)"]
|
|
|
|
|
|
def test_a_module_constant_is_read_only_as_mod_name_and_only_when_bound_once():
|
|
r = _scan(
|
|
urlguard="""
|
|
REASON_MALFORMED = "malformed"
|
|
TWICE = "a"
|
|
TWICE = "b"
|
|
if FLAG:
|
|
LATER = "c"
|
|
""",
|
|
routes__a="""
|
|
from app import urlguard
|
|
|
|
BARE = "bare"
|
|
|
|
def f():
|
|
raise HTTPException(409, {"code": "url_blocked", "reason": urlguard.REASON_MALFORMED})
|
|
def g():
|
|
raise HTTPException(409, {"code": "url_blocked", "reason": urlguard.TWICE})
|
|
def h():
|
|
raise HTTPException(409, {"code": "url_blocked", "reason": urlguard.LATER})
|
|
def i():
|
|
raise HTTPException(409, {"code": "url_blocked", "reason": BARE})
|
|
""",
|
|
)
|
|
assert _code(r, "url_blocked")["reasons"] == ["malformed"]
|
|
assert _unread(r, "url_blocked") == ["urlguard.TWICE", "urlguard.LATER", "BARE"]
|
|
|
|
|
|
def test_a_name_imported_from_two_places_or_under_an_if_is_not_looked_up():
|
|
r = _scan(
|
|
m1='R = Literal["one"]',
|
|
m2='R = Literal["two"]',
|
|
a="""
|
|
def f():
|
|
from app.m1 import R
|
|
def g():
|
|
from app.m2 import R
|
|
def h(reason: R):
|
|
return HTTPException(409, {"code": "edit", "reason": reason})
|
|
""",
|
|
b="""
|
|
if FLAG:
|
|
from app.m1 import R
|
|
def h(reason: R):
|
|
return HTTPException(409, {"code": "playlist", "reason": reason})
|
|
""",
|
|
c="""
|
|
from app.m1 import R
|
|
def h(reason: R):
|
|
return HTTPException(409, {"code": "clip", "reason": reason})
|
|
""",
|
|
)
|
|
assert _code(r, "edit")["reasons"] == [] and _unread(r, "edit") == ["reason"]
|
|
assert _code(r, "playlist")["reasons"] == [] and _unread(r, "playlist") == ["reason"]
|
|
assert _code(r, "clip")["reasons"] == ["one"]
|
|
|
|
|
|
def test_a_caught_exception_answers_with_the_literal_its_class_declares_on_self():
|
|
r = _scan(
|
|
errors="""
|
|
from typing import Literal
|
|
ClipReason = Literal["bad_range", "too_short"]
|
|
EditReason = Literal["not_media"]
|
|
|
|
class ClipError(Exception):
|
|
def __init__(self, reason: ClipReason, **params: float):
|
|
self.reason: ClipReason = reason
|
|
self.params = params
|
|
|
|
class EditError(Exception):
|
|
reason: EditReason
|
|
|
|
class Untyped(Exception):
|
|
def __init__(self, reason: str):
|
|
self.reason = reason
|
|
|
|
class Narrower(ClipError):
|
|
pass
|
|
""",
|
|
routes__a="""
|
|
from app import errors
|
|
|
|
def f():
|
|
try:
|
|
pass
|
|
except errors.ClipError as e:
|
|
raise HTTPException(422, {"code": "clip", "reason": e.reason})
|
|
except (errors.ClipError, errors.EditError) as e:
|
|
raise HTTPException(422, {"code": "clip", "reason": e.reason})
|
|
except errors.EditError as e:
|
|
raise HTTPException(422, {"code": "clip", "reason": e.reason})
|
|
except errors.Untyped as e:
|
|
raise HTTPException(422, {"code": "clip", "reason": e.reason})
|
|
except errors.Narrower as e:
|
|
raise HTTPException(422, {"code": "clip", "reason": e.reason})
|
|
except errors.ClipError as e:
|
|
raise HTTPException(422, {"code": "clip", "reason": e})
|
|
""",
|
|
)
|
|
assert _code(r, "clip")["reasons"] == ["bad_range", "too_short"]
|
|
assert _unread(r, "clip") == [
|
|
"e.reason", # a tuple of classes
|
|
"e.reason (EditError.reason has no Literal type)", # declared on the class body only
|
|
"e.reason (Untyped.reason has no Literal type)",
|
|
"e.reason (Narrower.reason has no Literal type)", # a subclass: not looked up the bases
|
|
"e", # the exception itself
|
|
]
|
|
|
|
|
|
def test_a_parameter_answers_with_its_literal_or_its_class_and_any_other_type_is_unread():
|
|
r = _scan(
|
|
a="""
|
|
from typing import Literal
|
|
Why = Literal["one", "two"]
|
|
|
|
class Err(Exception):
|
|
def __init__(self, reason: Literal["three"]):
|
|
self.reason: Literal["three"] = reason
|
|
|
|
def typed(reason: Why):
|
|
return HTTPException(409, {"code": "playlist", "reason": reason})
|
|
def inline(reason: Literal["four"]):
|
|
return HTTPException(409, {"code": "playlist", "reason": reason})
|
|
def by_class(exc: Err):
|
|
return HTTPException(409, {"code": "playlist", "reason": exc.reason})
|
|
def rebound(exc: Err):
|
|
exc = other()
|
|
return HTTPException(409, {"code": "playlist", "reason": exc.reason})
|
|
def optional(reason: Literal["five"] | None = None):
|
|
return HTTPException(409, {"code": "playlist", "reason": reason})
|
|
def quoted(reason: "Why"):
|
|
return HTTPException(409, {"code": "playlist", "reason": reason})
|
|
def number(reason: int):
|
|
return HTTPException(409, {"code": "playlist", "reason": reason})
|
|
|
|
class C:
|
|
def method(self, reason: str):
|
|
return HTTPException(409, {"code": "playlist", "reason": reason})
|
|
"""
|
|
)
|
|
assert _code(r, "playlist")["reasons"] == ["four", "one", "three", "two"]
|
|
unread = _unread(r, "playlist")
|
|
# a parameter bound again, `| None`, a quoted annotation, `int`
|
|
assert unread[:4] == ["exc.reason", "reason", "reason", "reason"]
|
|
assert "method(reason) is not a module-level helper" in unread[4]
|
|
assert len(unread) == 5
|
|
|
|
|
|
def test_a_str_helper_nobody_calls_sends_nothing():
|
|
r = _scan(
|
|
a="""
|
|
def helper(reason: str):
|
|
return HTTPException(409, {"code": "playlist", "reason": reason})
|
|
"""
|
|
)
|
|
assert _code(r, "playlist") == {"reasons": [], "unread": [], "sites": 1, "reasonless": []}
|
|
|
|
|
|
def test_a_map_lookup_is_its_values_and_its_keys_must_be_exactly_what_the_index_can_be():
|
|
src = """
|
|
from typing import Literal
|
|
SettingReason = Literal["too_low", "too_high"]
|
|
|
|
class InvalidSetting(ValueError):
|
|
def __init__(self, key: str, reason: SettingReason, **params: object):
|
|
self.reason: SettingReason = reason
|
|
self.params = params
|
|
|
|
_MAP: dict[SettingReason, str] = {{{keys}}}
|
|
|
|
def refused(status: int, reason: str, **params: object):
|
|
return HTTPException(status, {{**params, "code": "refused", "reason": reason}})
|
|
|
|
def setting_refused(exc: InvalidSetting):
|
|
return refused(400, _MAP[{index}], **exc.params)
|
|
"""
|
|
both = '"too_low": "setting_too_low", "too_high": "setting_too_high"'
|
|
whole = _scan(a=src.format(keys=both, index="exc.reason"))
|
|
assert _code(whole, "refused")["reasons"] == ["setting_too_high", "setting_too_low"]
|
|
assert whole["errors"] == [] and whole["unreadCodes"] == []
|
|
for keys, index in [
|
|
('"too_low": "setting_too_low"', "exc.reason"), # a key short
|
|
(both + ', "gone": "setting_gone"', "exc.reason"), # a key no index can be
|
|
(both, "exc.key"), # an index the scan cannot read
|
|
]:
|
|
r = _scan(a=src.format(keys=keys, index=index))
|
|
assert len(r["errors"]) == 1 and "not what its index can be" in r["errors"][0]
|
|
|
|
|
|
def test_a_lookup_in_anything_but_a_module_dict_of_literal_keys_is_unread():
|
|
r = _scan(
|
|
a="""
|
|
BUILT = make()
|
|
NUMBERED = {1: "one"}
|
|
|
|
def f(table, k):
|
|
raise HTTPException(409, {"code": "edit", "reason": table[k]})
|
|
def g(k):
|
|
raise HTTPException(409, {"code": "edit", "reason": BUILT[k]})
|
|
def h(k):
|
|
raise HTTPException(409, {"code": "edit", "reason": NUMBERED[k]})
|
|
"""
|
|
)
|
|
assert _unread(r, "edit") == ["table[k]", "BUILT[k]", "NUMBERED[k]"]
|
|
assert r["errors"] == []
|
|
|
|
|
|
def test_a_code_must_be_one_literal():
|
|
r = _scan(
|
|
a="""
|
|
BUSY = "busy"
|
|
def f(code):
|
|
raise HTTPException(409, {"code": "busy"})
|
|
def g(code):
|
|
raise HTTPException(409, {"code": code})
|
|
def h():
|
|
raise HTTPException(409, {"code": BUSY})
|
|
def i(x):
|
|
raise HTTPException(409, {"code": "a" if x else "b"})
|
|
def j(x):
|
|
raise HTTPException(409, {"code": "a" if x else None})
|
|
def k(key):
|
|
raise HTTPException(409, {"code": "edit", key: "late"})
|
|
"""
|
|
)
|
|
assert r["codes"]["busy"]["reasonless"] == ["a.py:4"]
|
|
assert set(r["codes"]) == {"busy"}
|
|
assert len(r["unreadCodes"]) == 5
|
|
assert r["unreadCodes"][-1].endswith(
|
|
"(a spread or a computed key the code and the reason do not follow)"
|
|
)
|
|
|
|
|
|
def test_a_spread_or_a_computed_key_stands_before_both_the_code_and_the_reason():
|
|
def unread_codes(detail: str) -> list[str]:
|
|
return _scan(a=f"def f(x, more, key):\n raise HTTPException(409, {detail})\n")[
|
|
"unreadCodes"
|
|
]
|
|
|
|
assert unread_codes('{**more, "code": "refused", "reason": "a"}') == []
|
|
assert unread_codes('{key: 1, **more, "reason": "a", "code": "refused"}') == []
|
|
for late in [
|
|
'{"code": "refused", "reason": "a", **more}', # the spread may replace both
|
|
'{"code": "refused", **more, "reason": "a"}', # … the code
|
|
'{"reason": "a", **more, "code": "refused"}', # … the reason
|
|
'{**more, "code": "refused"}', # no reason after it: the spread may carry one
|
|
'{"code": "refused", "reason": "a", key: "x"}', # a key that may be either
|
|
'dict(**more, code="refused", reason="a")', # dict() raises on a repeated key
|
|
'dict(more, code="refused", reason="a")',
|
|
]:
|
|
assert unread_codes(late), late
|
|
|
|
|
|
def test_a_detail_stands_only_where_it_is_sent():
|
|
def result(body: str, extra: str = "") -> dict:
|
|
return _scan(a=f"def f(x, cached, fix):\n{textwrap.indent(body, ' ')}\n{extra}")
|
|
|
|
for body in [
|
|
'detail = {"code": "login", "reason": "a"}\nraise HTTPException(401, detail)',
|
|
'if (detail := {"code": "login", "reason": "a"}):\n raise HTTPException(401, detail)',
|
|
'raise HTTPException(401, {"code": "login", "reason": "a"}.copy())',
|
|
'raise HTTPException(401, fix({"code": "login", "reason": "a"}))',
|
|
'raise HTTPException(401, cached or {"code": "login", "reason": "a"})',
|
|
'return JSONResponse({"code": "login", "reason": "a"}, status_code=401)', # the whole body
|
|
'raise HTTPException(401, None, {"code": "login", "reason": "a"})', # the headers
|
|
'raise HTTPException(401, headers={"code": "login", "reason": "a"})',
|
|
'raise HTTPException(401, "a" if {"code": "login", "reason": "a"} else "b")',
|
|
]:
|
|
r = result(body)
|
|
assert r["unreadCodes"] and "login" not in r["codes"], body
|
|
assert "(a detail put where the scan cannot follow it)" in r["unreadCodes"][0], body
|
|
held = _scan(a='class C:\n detail = {"code": "login"}\n')
|
|
assert "(a detail put where the scan cannot follow it)" in held["unreadCodes"][0]
|
|
|
|
sent = result(
|
|
'raise HTTPException(401, {"code": "login", "reason": "a"} if x else {"code": "login"})\n'
|
|
'raise HTTPException(401, detail=dict(code="login", reason="b"))\n'
|
|
'return JSONResponse(content={"detail": {"code": "login", "reason": "c"}})'
|
|
)
|
|
assert sent["unreadCodes"] == [] and _code(sent, "login")["reasons"] == ["a", "b", "c"]
|
|
|
|
|
|
def test_a_factory_is_a_detail_only_when_each_return_is_one_and_each_call_is_sent():
|
|
factory = """
|
|
def quota(e) -> dict:
|
|
return {"code": "quota", "reason": "files"}
|
|
"""
|
|
sent = "def f(e):\n raise HTTPException(422, detail=quota(e))\n"
|
|
assert _scan(a=factory + sent)["codes"]["quota"]["reasons"] == ["files"]
|
|
for change in [
|
|
"def g(e):\n d = quota(e)\n d['code'] = 'other'\n raise HTTPException(422, d)\n",
|
|
"def g(e):\n raise HTTPException(422, detail=quota(e).copy())\n",
|
|
"pick = quota\n",
|
|
]:
|
|
r = _scan(a=factory + sent + change)
|
|
assert r["unreadCodes"] and "quota" not in r["codes"], change
|
|
twice = _scan(a="quota = None\n" + factory + sent) # the def is the last of two bindings
|
|
assert twice["unreadCodes"] and "quota" not in twice["codes"]
|
|
for other in [
|
|
" if e:\n return None\n",
|
|
" if e:\n return cached\n",
|
|
" if e:\n return {'code': 'quota', 'reason': 'x'} if e > 1 else None\n",
|
|
" log({'code': 'quota', 'reason': 'logged'})\n", # a detail it does not return
|
|
]:
|
|
bent = factory.replace("-> dict:\n", "-> dict:\n" + other)
|
|
assert _scan(a=bent + sent)["unreadCodes"], other
|
|
method = "class C:\n def quota(self):\n return {'code': 'quota', 'reason': 'm'}\n"
|
|
assert _scan(a=method)["unreadCodes"]
|
|
dotted = _scan(
|
|
a=factory,
|
|
routes__b="import app.a\n\ndef g(e):\n raise HTTPException(422, detail=app.a.quota(e))\n",
|
|
)
|
|
assert dotted["unreadCodes"] and "quota" not in dotted["codes"]
|
|
|
|
|
|
def test_a_name_a_lambda_a_comprehension_a_nonlocal_or_a_global_binds_is_unread():
|
|
r = _scan(
|
|
a="""
|
|
def f(svc):
|
|
why = "invalid"
|
|
make = lambda why: HTTPException(401, {"code": "login", "reason": why})
|
|
raise make(svc.state)
|
|
def g(svc):
|
|
why = "invalid"
|
|
raise [HTTPException(401, {"code": "login", "reason": why}) for why in svc.all][0]
|
|
def h(svc):
|
|
why = "invalid"
|
|
def inner():
|
|
nonlocal why
|
|
why = svc.reason
|
|
inner()
|
|
raise HTTPException(401, {"code": "login", "reason": why})
|
|
def i(svc):
|
|
global why
|
|
why = "invalid"
|
|
raise HTTPException(401, {"code": "login", "reason": why})
|
|
"""
|
|
)
|
|
assert _code(r, "login")["reasons"] == []
|
|
assert sorted(_unread(r, "login")) == [
|
|
"why (bound by a lambda or a comprehension)",
|
|
"why (bound by a lambda or a comprehension)",
|
|
"why (named by a nonlocal or a global)",
|
|
"why (named by a nonlocal or a global)",
|
|
]
|
|
|
|
|
|
def test_a_local_that_reaches_itself_is_unread():
|
|
r = _scan(
|
|
a="""
|
|
def f(x):
|
|
why = "a"
|
|
if x:
|
|
why = why if x > 1 else "b"
|
|
raise HTTPException(409, {"code": "edit", "reason": why})
|
|
def g(x):
|
|
a = "z"
|
|
b = a
|
|
a = b
|
|
raise HTTPException(409, {"code": "playlist", "reason": a})
|
|
def h(x):
|
|
why = "c"
|
|
if x:
|
|
raise HTTPException(409, {"code": "busy", "reason": why})
|
|
raise HTTPException(409, {"code": "busy", "reason": why})
|
|
"""
|
|
)
|
|
assert _code(r, "busy")["reasons"] == ["c"] and _code(r, "busy")["unread"] == []
|
|
assert _code(r, "edit")["reasons"] == ["a", "b"]
|
|
assert _unread(r, "edit") == ["why (reached again while it is resolved)"]
|
|
assert _code(r, "playlist")["reasons"] == ["z"]
|
|
assert any("reached again while it is resolved" in u for u in _unread(r, "playlist"))
|
|
|
|
|
|
def test_a_module_constant_or_a_map_changed_elsewhere_is_unread():
|
|
def reasons(urlguard: str, route: str = "", **more: str) -> tuple[list[str], list[str]]:
|
|
r = _scan(
|
|
**more,
|
|
urlguard="REASON = 'malformed'\n_MAP = {'a': 'map_a'}\n" + urlguard,
|
|
routes__a="from app import urlguard\n"
|
|
+ route
|
|
+ "\ndef f(k: Literal['a']):\n"
|
|
+ " raise HTTPException(409, {'code': 'url_blocked', 'reason': urlguard.REASON})\n"
|
|
+ " raise HTTPException(409, {'code': 'url_blocked', 'reason': urlguard._MAP[k]})\n",
|
|
)
|
|
return _code(r, "url_blocked")["reasons"], _unread(r, "url_blocked")
|
|
|
|
assert reasons("def get(k):\n return _MAP.get(k)\n") == (["malformed", "map_a"], [])
|
|
other = "from app import other\ndef g():\n other.REASON = 'x'\n" # another module's REASON
|
|
assert reasons("", other, other="REASON = 'y'\n") == (["malformed", "map_a"], [])
|
|
shadow = (
|
|
"def h(*urlguard):\n"
|
|
" raise HTTPException(409, {'code': 'url_blocked', 'reason': urlguard.REASON})\n"
|
|
)
|
|
assert reasons("", shadow)[1] == ["urlguard.REASON"]
|
|
for urlguard, route in [
|
|
("def reset():\n global REASON\n REASON = 'x'\n", ""),
|
|
("", "def g():\n urlguard.REASON = 'x'\n"),
|
|
("", "def g():\n del urlguard.REASON\n"),
|
|
]:
|
|
assert reasons(urlguard, route)[0] == ["map_a"], (urlguard, route)
|
|
for urlguard in [
|
|
"_MAP['a'] = compute()\n",
|
|
"def init(x):\n _MAP['a'] = x\n",
|
|
"_MAP.update(a='other')\n",
|
|
"def drop():\n del _MAP['a']\n",
|
|
"register(_MAP)\n",
|
|
]:
|
|
got, unread = reasons(urlguard)
|
|
assert got == ["malformed"] and "(the map is changed after it is built)" in unread[0]
|
|
|
|
|
|
def test_a_helper_bound_twice_is_not_followed():
|
|
r = _scan(
|
|
refusal="refused = None\n" + textwrap.dedent(REFUSAL),
|
|
routes__a="from app.refusal import refused\n\ndef f():\n raise refused(409, 'lost')\n",
|
|
)
|
|
assert _code(r, "refused")["reasons"] == []
|
|
assert _unread(r, "refused")[0].endswith("refused(reason) is not a module-level helper")
|
|
|
|
|
|
def test_a_dotted_path_through_an_app_module_that_goes_nowhere_is_unread():
|
|
r = scan(
|
|
{
|
|
"__init__.py": "",
|
|
"refusal.py": textwrap.dedent(REFUSAL),
|
|
"routes/a.py": "import app as a\n\ndef f():\n raise a.refusal.refused(409, 'dotted')\n",
|
|
}
|
|
)
|
|
assert _unread(r, "refused") == ["a.refusal.refused (a reference the scan cannot resolve)"]
|
|
|
|
|
|
def test_a_helper_reached_through_an_app_module_that_does_not_bind_it_is_unread():
|
|
for shim in [
|
|
"from app.refusal import *\n",
|
|
"try:\n from app.refusal import refused\nexcept ImportError:\n refused = None\n",
|
|
]:
|
|
r = _scan(
|
|
refusal=REFUSAL,
|
|
shim=shim,
|
|
routes__a="from app.shim import refused\n\ndef f():\n raise refused(409, 'hidden')\n",
|
|
)
|
|
assert _unread(r, "refused") == ["refused (a reference the scan cannot resolve)"], shim
|
|
package = scan(
|
|
{
|
|
"pkg/refusal.py": textwrap.dedent(REFUSAL),
|
|
"pkg/__init__.py": "from .refusal import refused\n",
|
|
"routes/a.py": "from app.pkg import refused\n\ndef f():\n raise refused(409, 'hidden')\n",
|
|
}
|
|
)
|
|
assert _unread(package, "refused") == ["refused (a reference the scan cannot resolve)"]
|
|
|
|
|
|
def test_the_data_files_are_excluded_by_name_only():
|
|
r = _scan(
|
|
plex__tmdb='def f():\n return {"code": "HU", "name": "Hungary"}',
|
|
plex__other='def f():\n return {"code": "HU"}',
|
|
)
|
|
assert r["excluded"] == {"plex/tmdb.py": 1}
|
|
assert "HU" in r["codes"]
|
|
|
|
|
|
def test_the_app_scans():
|
|
"""The real tree, on the interpreter the backend runs: every refusal it holds is read (the
|
|
frontend's contract judges what is read)."""
|
|
result = scan_app()
|
|
assert {"refused", "not_found", "login", "clip"} <= set(result["codes"])
|
|
assert result["errors"] == [] and result["unreadCodes"] == []
|