diff --git a/docs/INTENT.md b/docs/INTENT.md index 97f365c16..b103c1a70 100644 --- a/docs/INTENT.md +++ b/docs/INTENT.md @@ -26,6 +26,23 @@ CPython interpreter inside a Kotlin Multiplatform application so that: written (`Greeter('Kotlin').greet(2)` in `sample/`). 3. **The same code runs on every target** a Kotlin Multiplatform app ships to: desktop JVM, Android, iOS, Android native, and the web (wasm, experimental). +4. **The project's own Python is statically typed by default** (TypedPython). The maintainer's goal: + + > "파이썬을 컴파일 언어 및 타입 안정화, 정적 타입 검사, 컴파일 타임 타입검사를 지원할 수 있도록" + > — *Support Python as a compiled language with type stability, static type checking and + > compile-time type checking*, so that Python stays usable as AI-friendly languages such as Rust + > and TypeScript gain ground. + + > "TypedPython을 별도 레포가 아니라 PythonMultiplatform의 기본 동작이 될거야" + > — *TypedPython is not a separate repository; it will be PythonMultiplatform's default behaviour.* + + The maintainer chose the scope "검사만 기본, 컴파일은 opt-in" (2026-10-02): every user Python source + is **checked** by default; **compiling** it to native code is opt-in. The compiled code still runs on + the embedded CPython, as an extension module (`docs/design/typedpython.md`). + + > The maintainer also asked to *explore* GPU kernels written in Python without a separate language + > ("GPU 관련 최적화까지도 별도 언어 없이 지원할 방안을 모색해보자"). That is an exploration, not yet + > a commitment; it is recorded in the design note only. The CPython version moved on from 3.13 (the build pins 3.14.x today and offers opt-in free-threaded 3.14t builds on desktop only); the intent — embed **CPython itself**, not a re-implementation — is unchanged. @@ -88,7 +105,7 @@ no per-platform Python → Kotlin binding code to maintain. | A replacement for the `pythonx` packages | §2.3 — they are real packages in their own repository | | Runtime-reflection or name-lookup based binding | §2.1 — impossible under Kotlin/Native and native image | | A hand-written wrapper per Kotlin function or class | Generic adaptation plus generated tables; per-declaration wrappers were tried in 2024 and rotted (`docs/design/pythonx-adapter-design.md`) | -| A Python re-implementation, transpiler, or JVM-hosted Python | It embeds CPython itself, so C extensions keep working | +| A Python re-implementation, transpiler, or JVM-hosted Python | It embeds CPython itself, so C extensions keep working. TypedPython (§1.4) does not change this: checked code is ordinary Python, and compiled code is a CPython extension module, not a translation to another runtime | | A build tool, dependency resolver or packager | That is pypackpack (the work) and toolchain (the Gradle vocabulary) | | A sub-interpreter parallelism model | Rejected: extensions that do not declare multi-interpreter support fail to import; parallelism comes from free-threaded CPython | | A vendored interpreter | The build downloads and verifies CPython per platform | diff --git a/docs/SPEC.md b/docs/SPEC.md index 539a84a03..c2a5fa92b 100644 --- a/docs/SPEC.md +++ b/docs/SPEC.md @@ -239,6 +239,20 @@ All in `PM/commonTest`, so they run wherever the interpreter loads. - **N-3** Binding Kotlin/Native klib declarations at run time (B-2). `Status: planned`. - **N-4** Linux and Windows desktop runs in CI (P-2). `Status: planned`. - **N-5** Native-image upcall verification as an automated test (U-6). `Status: planned`. +- **N-6** TypedPython soundness gate (INTENT §1.4; `docs/design/typedpython.md` §4.2): checking a set of + user Python files reports every Pyrefly type error, the Any-leak kinds (`implicit-any-parameter`, + `unannotated-return`, `unknown-variable-type`, `no-any-return-implicit`, `no-any-return-explicit`, + `explicit-any`) and any user-code expression of type `Any` outside `cast()` (`any-flow`) as errors; + the forbidden list (`eval`/`exec`, computed-name `getattr`/`setattr`/`delattr`, patching an imported + module or a module-level class, rebinding an unannotated name to a different type, and a list/set/dict + display whose inferred element type is a union) as warnings in `checked` mode and errors in + `compiled` mode. Nothing + inside dependencies is reported. `python -m typedpython check` exits 0 / 1 / 2 for clean / errors / + tool failure. `Status: partial` — the gate is asserted by + `python-multiplatform-ksp/src/test/python/` (run with that module's `.venv`); it is **not yet run by the + build** (design §6 step 1b). Rebinding and mixed containers are found by a second Pyrefly pass over a + probed copy (design §6.1), which runs only when a file has a name assigned twice or a container + display; `for` targets and comprehension variables are not probed. --- diff --git a/docs/design/typedpython.md b/docs/design/typedpython.md index fb126c89e..d80ea9f50 100644 --- a/docs/design/typedpython.md +++ b/docs/design/typedpython.md @@ -207,6 +207,30 @@ TypedPython 은 그래서 자체 텐서 런타임을 만들지 않고 **torch op 각 단계는 AGENTS.md §5·§14 의 TDD 와 측정 테스트 규정을 따른다. 특히 3 은 경계 비용 감소를, 4 는 upstream 대비 수치 오차를 기준선으로 박아 둔다. +### 6.1 단계 1 의 현재 상태 (2026-10-03, #16) + +- **위치:** `python-multiplatform-ksp/src/main/python/typedpython/`, 테스트는 + `python-multiplatform-ksp/src/test/python/`. 사용자 지시("ksp랑 통합")에 따라 KSP 모듈 안에 둔다. + KSP 프로세서는 Kotlin 심볼만 보고 파이썬 파일 변경으로는 다시 돌지 않으므로, **실행**은 1b 에서 + Gradle 태스크가 맡는다 — KSP 모듈은 소스와 배포(jar 리소스)의 자리다. +- **구현된 규칙:** SPEC N-6. Pyrefly 의 `Any` 유출 kind 6개를 에러로 올리고, Pyrefly 의 + per-expression 타입 리포트(`--report-pysa`)로 사용자 코드 안의 `Any` 식을 찾는다(`any-flow`). + `cast()` 인자, 버려지는 식 문장의 값, 함수 시그니처 자체는 제외한다. +- **재할당·혼합 컨테이너 (#22):** Pyrefly 1.3.2 는 둘 다 보고하지 않고, 타입 리포트에 대입 대상의 + 타입이 없다. 대신 이름을 *읽는* 위치의 타입은 기록한다. 그래서 대입문 바로 뒤에 같은 줄로 + `; x` 를 붙인 사본을 만들어(줄 번호 불변) 한 번 더 돌리고, 그 위치의 타입으로 판정한다. 두 번 + 이상 대입되는 이름이나 컨테이너 디스플레이가 있는 파일에서만 돈다. `for` 대상과 컴프리헨션 + 변수는 아직 탐침하지 않는다. +- **측정:** 1800줄 합성 모듈에서 plain `pyrefly check` 160–235 ms, 게이트 1.3–1.9 s — 두 번 잰 비율이 + 모두 약 8배다. 탐침 패스를 별도 실행으로 붙이면 약 15배(232 ms → 3.6 s, load 41)였다 — 합성 모듈의 + `total += ...` 가 탐침 대상이라 두 번째 실행이 돌았기 때문이다. 지금은 탐침 대상이 있으면 + **탐침 사본(과 같은 검사의 나머지 파일)에 대해 Pyrefly 를 한 번만** 돌리고, 에러·`any-flow`· + 타입을 모두 그 한 번에서 읽는다(에러 열은 삽입 길이만큼 왼쪽으로 되돌린다). 같은 합성 모듈에서 + 825 ms → 6.2 s, 약 7.6배(load 30 — 유효한 측정이 아니다. load 4 초과). 단독 측정이 아니다(두 번째는 load average 18.6). 절대값보다 비율을 기준으로 본다. 차이의 + 대부분은 타입 리포트가 typeshed 사본까지 디스크에 쓰는 비용으로 보인다(추정, 미분해). 1b 에서 + 빌드마다 돌리기 전에 줄일 대상이다. +- **고정:** Pyrefly `==1.3.2`. 타입 리포트는 Pyrefly 가 버전을 매기는 포맷이지 공개 API 가 아니다. + ## 7. 위험 - **Pyrefly 의 외부 API 안정성.** 라이브러리 임베드는 확인 전까지 미룬다 (§4.1). diff --git a/python-multiplatform-ksp/.gitignore b/python-multiplatform-ksp/.gitignore new file mode 100644 index 000000000..4dc423b60 --- /dev/null +++ b/python-multiplatform-ksp/.gitignore @@ -0,0 +1,5 @@ +.venv/ +__pycache__/ +.pytest_cache/ +*.egg-info/ +.pytest_tmp/ diff --git a/python-multiplatform-ksp/pyproject.toml b/python-multiplatform-ksp/pyproject.toml new file mode 100644 index 000000000..62f6993b2 --- /dev/null +++ b/python-multiplatform-ksp/pyproject.toml @@ -0,0 +1,29 @@ +# TypedPython soundness gate (docs/design/typedpython.md §4.2). +# The Kotlin processor in this module is built by Gradle; this file only describes the Python +# package that lives next to it in src/main/python. +[project] +name = "typedpython" +version = "0.1.0" +requires-python = ">=3.13" +# Pinned exactly: the gate reads Pyrefly's per-expression type report (--report-pysa), whose +# format is versioned by Pyrefly, not by a public API. +dependencies = ["pyrefly==1.3.2"] + +[project.scripts] +typedpython = "typedpython.__main__:cli" + +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[tool.hatch.build.targets.wheel] +packages = ["src/main/python/typedpython"] + +[project.optional-dependencies] +test = ["pytest"] + +[tool.pytest.ini_options] +testpaths = ["src/test/python"] +pythonpath = ["src/main/python"] +addopts = "--basetemp=.pytest_tmp" +markers = ["measurement: records cost; asserts only sanity bounds"] diff --git a/python-multiplatform-ksp/src/main/python/typedpython/__init__.py b/python-multiplatform-ksp/src/main/python/typedpython/__init__.py new file mode 100644 index 000000000..507809dab --- /dev/null +++ b/python-multiplatform-ksp/src/main/python/typedpython/__init__.py @@ -0,0 +1,4 @@ +"""TypedPython — static checking for the Python code of a PythonMultiplatform project. + +See docs/design/typedpython.md. This package is the soundness gate (§4.2). +""" diff --git a/python-multiplatform-ksp/src/main/python/typedpython/__main__.py b/python-multiplatform-ksp/src/main/python/typedpython/__main__.py new file mode 100644 index 000000000..78fbba395 --- /dev/null +++ b/python-multiplatform-ksp/src/main/python/typedpython/__main__.py @@ -0,0 +1,48 @@ +"""`python -m typedpython check [--mode checked|compiled] [--format text|json] PATHS...` + +Exit status: 0 no errors (warnings allowed), 1 errors, 2 the gate itself could not run. +""" +import argparse +import json +import sys +from pathlib import Path + +from typedpython import gate +from typedpython.pyrefly import PyreflyError + +EXIT_CLEAN, EXIT_ERRORS, EXIT_FAILURE = 0, 1, 2 + + +def main(argv: list[str]) -> int: + parser = argparse.ArgumentParser(prog="typedpython") + commands = parser.add_subparsers(dest="command", required=True) + check = commands.add_parser("check", help="type-check user Python code") + check.add_argument("paths", nargs="+", type=Path) + check.add_argument("--mode", choices=["checked", "compiled"], default="checked") + check.add_argument("--format", choices=["text", "json"], default="text") + check.add_argument("--search-path", action="append", type=Path, default=[], + help="directory to resolve imports from (repeatable)") + args = parser.parse_args(argv) + + try: + found = gate.check(args.paths, mode=args.mode, search_paths=args.search_path) + except PyreflyError as e: + print(f"typedpython: {e}", file=sys.stderr) + return EXIT_FAILURE + + if args.format == "json": + print(json.dumps([d.to_json() for d in found], indent=2)) + else: + for d in found: + print(f"{d.severity.upper()} {d.path}:{d.line}:{d.column}: {d.message} [{d.rule}]") + + return EXIT_ERRORS if any(d.severity == "error" for d in found) else EXIT_CLEAN + + +def cli() -> None: + """Console-script entry point (`typedpython check ...`).""" + sys.exit(main(sys.argv[1:])) + + +if __name__ == "__main__": + cli() diff --git a/python-multiplatform-ksp/src/main/python/typedpython/diagnostic.py b/python-multiplatform-ksp/src/main/python/typedpython/diagnostic.py new file mode 100644 index 000000000..2ec322edc --- /dev/null +++ b/python-multiplatform-ksp/src/main/python/typedpython/diagnostic.py @@ -0,0 +1,17 @@ +from dataclasses import dataclass, asdict +from typing import Literal + +Severity = Literal["error", "warning"] + + +@dataclass(frozen=True, order=True) +class Diagnostic: + path: str + line: int + column: int + rule: str + severity: Severity + message: str + + def to_json(self) -> dict[str, object]: + return asdict(self) diff --git a/python-multiplatform-ksp/src/main/python/typedpython/forbidden.py b/python-multiplatform-ksp/src/main/python/typedpython/forbidden.py new file mode 100644 index 000000000..cccc1651f --- /dev/null +++ b/python-multiplatform-ksp/src/main/python/typedpython/forbidden.py @@ -0,0 +1,110 @@ +"""The forbidden list (docs/design/typedpython.md §3.1), found by walking the AST. + +These are the dynamic features that make a static type unreliable at compile time. The gate +reports them as warnings for `checked` code and as errors for `compiled` code; this module only +finds them. + +Rebinding a name to a different type and containers that mix element types need inferred types +for assignment targets; they live in typedpython.rebinding. +""" +import ast +from dataclasses import dataclass + +DYNAMIC_ATTRIBUTE_CALLS = {"getattr", "setattr", "delattr"} +# Index of the attribute-name argument in each of the calls above. +NAME_ARGUMENT = 1 + + +@dataclass(frozen=True) +class Finding: + rule: str + line: int + column: int + message: str + + +def scan(source: str, path: str) -> list[Finding]: + tree = ast.parse(source, filename=path) + patchable = _module_level_modules_and_classes(tree) + findings: list[Finding] = [] + + for node in ast.walk(tree): + if isinstance(node, ast.Call) and isinstance(node.func, ast.Name): + findings += _check_call(node, node.func.id, patchable) + elif isinstance(node, (ast.Assign, ast.AugAssign, ast.AnnAssign, ast.Delete)): + for target in _targets(node): + if _is_patch_target(target, patchable): + assert isinstance(target, ast.Attribute) and isinstance(target.value, ast.Name) + findings.append(_finding( + "forbidden/monkeypatch", target, + f"`{target.value.id}.{target.attr}` changes a module or class from outside it", + )) + + return sorted(findings, key=lambda f: (f.line, f.column, f.rule)) + + +def _check_call(call: ast.Call, name: str, patchable: set[str]) -> list[Finding]: + if name in ("eval", "exec"): + return [_finding(f"forbidden/{name}", call, f"`{name}` runs code the checker cannot see")] + if name not in DYNAMIC_ATTRIBUTE_CALLS or len(call.args) <= NAME_ARGUMENT: + return [] + + attribute = call.args[NAME_ARGUMENT] + if not (isinstance(attribute, ast.Constant) and isinstance(attribute.value, str)): + return [_finding( + "forbidden/dynamic-attr", call, + f"`{name}` with a computed attribute name hides the attribute from the checker", + )] + + # A literal name is static, but setattr/delattr on a module or class is still a patch. + receiver = call.args[0] + if name != "getattr" and isinstance(receiver, ast.Name) and receiver.id in patchable: + return [_finding( + "forbidden/monkeypatch", call, + f"`{name}` on `{receiver.id}` changes a module or class from outside it", + )] + return [] + + +def _module_level_modules_and_classes(tree: ast.Module) -> set[str]: + """Names bound at module scope to an imported module/name or to a class definition.""" + names: set[str] = set() + for node in tree.body: + if isinstance(node, ast.Import): + names |= {(alias.asname or alias.name).split(".")[0] for alias in node.names} + elif isinstance(node, ast.ImportFrom): + names |= {alias.asname or alias.name for alias in node.names if alias.name != "*"} + elif isinstance(node, ast.ClassDef): + names.add(node.name) + return names + + +def _targets(node: ast.stmt) -> list[ast.expr]: + if isinstance(node, ast.Assign): + return [t for target in node.targets for t in _flatten(target)] + if isinstance(node, ast.Delete): + return [t for target in node.targets for t in _flatten(target)] + if isinstance(node, (ast.AugAssign, ast.AnnAssign)): + return [node.target] + return [] + + +def _flatten(target: ast.expr) -> list[ast.expr]: + if isinstance(target, (ast.Tuple, ast.List)): + return [t for element in target.elts for t in _flatten(element)] + if isinstance(target, ast.Starred): + return _flatten(target.value) + return [target] + + +def _is_patch_target(target: ast.expr, patchable: set[str]) -> bool: + return ( + isinstance(target, ast.Attribute) + and isinstance(target.value, ast.Name) + and target.value.id in patchable + ) + + +def _finding(rule: str, node: ast.expr, message: str) -> Finding: + # ast columns are 0-based; Pyrefly's (and every editor's) are 1-based. + return Finding(rule, node.lineno, node.col_offset + 1, message) diff --git a/python-multiplatform-ksp/src/main/python/typedpython/gate.py b/python-multiplatform-ksp/src/main/python/typedpython/gate.py new file mode 100644 index 000000000..1f34e2303 --- /dev/null +++ b/python-multiplatform-ksp/src/main/python/typedpython/gate.py @@ -0,0 +1,258 @@ +"""The soundness gate (docs/design/typedpython.md §4.2). + +A type checker may fall back to `Any` when it is unsure; a compiler cannot. The gate closes that +gap for the files it is given: + + Pyrefly errors, Any-leak kinds included -> error + an expression of type Any in user code -> error (`any-flow`), unless inside cast() + the §3.1 forbidden list -> warning (`checked`) / error (`compiled`) + +Only the given files are user code. Their dependencies are type-checked by Pyrefly only as far +as their stubs or sources say; nothing inside them is reported. +""" +import ast +import shutil +import tempfile +from collections.abc import Sequence +from dataclasses import dataclass +from pathlib import Path +from typing import Literal + +from typedpython import forbidden, pyrefly, rebinding +from typedpython.diagnostic import Diagnostic, Severity + +Mode = Literal["checked", "compiled"] + +Position = tuple[int, int] + + +@dataclass(frozen=True) +class _Run: + """One user file as Pyrefly sees it: the text it was given and where that text differs.""" + + original: str + text: str # the source Pyrefly reads: the original, or the probed copy of it + probes: list[rebinding.Probe] + remap: "_Remap" + + +class _Remap: + """Maps positions in probed text back to the original source. + + A probe inserts `; name` on the line of its assignment, so a column to the right of an + insertion is `len("; name")` further right than in the original; line numbers never change. + """ + + def __init__(self, probes: Sequence[rebinding.Probe]) -> None: + self._inserted: dict[int, list[tuple[int, int]]] = {} + for p in probes: + self._inserted.setdefault(p.line, []).append((p.column - 2, len(p.name) + 2)) + + def to_original(self, line: int, column: int) -> Position | None: + """The original position, or None when it lies inside an inserted probe load.""" + shift = 0 + for start, length in self._inserted.get(line, ()): + if column >= start + length: + shift += length + elif column >= start: + return None + return line, column - shift + + +def check( + paths: Sequence[Path], + mode: Mode = "checked", + search_paths: Sequence[Path] = (), +) -> list[Diagnostic]: + runs = _plan([str(Path(p).resolve()) for p in paths]) + forbidden_severity: Severity = "error" if mode == "compiled" else "warning" + diagnostics: list[Diagnostic] = [] + + # One Pyrefly run for everything. When any file has something to probe, it runs on a mirrored + # package tree holding the probed copies (and the other files unchanged); otherwise on the + # files themselves. + with tempfile.TemporaryDirectory(prefix="typedpython-probe-") as work: + probing = any(r.probes for r in runs.values()) + if probing: + copies = _mirror(Path(work), {o: r.text for o, r in runs.items()}) + roots = pyrefly.import_roots(list(runs)) + mirror = _Mirror(copies, [((Path(work) / str(i)).resolve(), r) + for i, r in enumerate(roots)]) + report = pyrefly.run( + [Path(c) for c in copies], + [*[Path(work) / str(i) for i in range(len(roots))], + *sorted({Path(c).parent for c in copies}), + *search_paths, *roots], + ) + else: + mirror = _Mirror({o: o for o in runs}, []) + report = pyrefly.run([Path(p) for p in runs], search_paths) + original_of = mirror.copies + run_of = {c: runs[o] for c, o in original_of.items()} + + reported: set[tuple[str, int, int]] = set() + for e in report.errors: + run = run_of.get(e.path) + if run is None: + diagnostics.append(Diagnostic( + mirror.original(e.path), e.line, e.column, f"pyrefly/{e.kind}", "error", e.message, + )) + continue + position = run.remap.to_original(e.line, e.column) + if position is None: + # An error inside an inserted `; name` means the probe itself is wrong. + diagnostics.append(Diagnostic( + run.original, e.line, e.column, "internal/probe-error", "error", + f"pyrefly/{e.kind} inside a probe load (`{e.message}`); this is a gate bug", + )) + continue + diagnostics.append(Diagnostic( + run.original, *position, f"pyrefly/{e.kind}", "error", e.message, + )) + reported.add((run.original, *position)) + + for copy, run in run_of.items(): + path = run.original + try: + tree = ast.parse(run.text, filename=path) + except SyntaxError: + continue # Pyrefly has already reported the parse error + + # Judged on the probed text: the probe loads are expression statements, so they are + # discarded values and never become any-flow findings. + cast_spans = _cast_argument_spans(tree) + discarded = _discarded_values(tree) + for expr in report.any_expressions: + if expr.path != copy: + continue + start = run.remap.to_original(expr.line, expr.column) + if start is None or (path, *start) in reported: + continue + if _is_signature(expr.type): + continue # a callable whose signature mentions Any; its result is judged on its own + if ((expr.line, expr.column), (expr.stop_line, expr.stop_column)) in discarded: + continue # an expression statement's value goes nowhere + if any(_inside((expr.line, expr.column), span) for span in cast_spans): + continue + diagnostics.append(Diagnostic( + path, *start, "any-flow", "error", + f"this expression has type `{expr.type}`; give it a type, or cast() it " + "where it leaves untyped code", + )) + reported.add((path, *start)) + + for finding in forbidden.scan(Path(path).read_text(), path): + diagnostics.append(Diagnostic( + path, finding.line, finding.column, finding.rule, forbidden_severity, + finding.message, + )) + + types = report.expression_types.get(copy, {}) + for f in rebinding.judge(run.probes, types): + diagnostics.append(Diagnostic(path, f.line, f.column, f.rule, forbidden_severity, + f.message)) + + return sorted(diagnostics) + + +def _plan(paths: Sequence[str]) -> dict[str, _Run]: + """Each file's text for Pyrefly: probed when it has rebinding/mixed-container candidates.""" + runs: dict[str, _Run] = {} + for path in paths: + source = Path(path).read_text() + probed, probes = source, [] + try: + tree = ast.parse(source, filename=path) + except SyntaxError: + tree = None + if tree is not None and (wanted := rebinding.candidates(rebinding.assignments(tree))): + probed, probes = rebinding.probe(source, wanted) + runs[path] = _Run(path, probed, probes, _Remap(probes)) + return runs + + +@dataclass(frozen=True) +class _Mirror: + copies: dict[str, str] # copy -> original + bases: list[tuple[Path, Path]] # (mirrored root, original root) + + def original(self, path: str) -> str: + """The original path of a file in the mirrored tree (a dependency's, for instance).""" + for base, root in self.bases: + try: + return str(root / Path(path).relative_to(base)) + except ValueError: + continue + return path + + +def _mirror(work: Path, probed: dict[str, str]) -> dict[str, str]: + """Copy each file's package tree under `work`, with the probed text in place of the file. + + The copy keeps the file's path relative to its import root, so relative imports and imports + of sibling modules inside the package resolve to the copied tree. Returns copy -> original. + """ + roots = pyrefly.import_roots(list(probed)) + copies: dict[str, str] = {} + copied: set[Path] = set() + for original, text in probed.items(): + root = pyrefly.import_roots([original])[0] + base = work / str(roots.index(root)) + relative = Path(original).relative_to(root) + if len(relative.parts) > 1: + package = root / relative.parts[0] + if package not in copied: + shutil.copytree( + package, base / relative.parts[0], dirs_exist_ok=True, + ignore=shutil.ignore_patterns("__pycache__", ".*"), + ) + copied.add(package) + target = base / relative + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text(text) + copies[str(target.resolve())] = original + return copies + + +def _cast_argument_spans(tree: ast.Module) -> list[tuple[Position, Position]]: + """Source spans of the value argument of every `cast(T, value)` / `typing.cast(T, value)`.""" + spans = [] + for node in ast.walk(tree): + if not isinstance(node, ast.Call) or len(node.args) != 2: + continue + func = node.func + is_cast = (isinstance(func, ast.Name) and func.id == "cast") or ( + isinstance(func, ast.Attribute) and func.attr == "cast" + and isinstance(func.value, ast.Name) and func.value.id == "typing" + ) + value = node.args[1] + if is_cast and value.end_lineno is not None and value.end_col_offset is not None: + spans.append(( + (value.lineno, value.col_offset + 1), + (value.end_lineno, value.end_col_offset + 1), + )) + return spans + + +def _discarded_values(tree: ast.Module) -> set[tuple[Position, Position]]: + """Spans of values computed and thrown away by an expression statement (`eval(s)` alone).""" + spans = set() + for node in ast.walk(tree): + if isinstance(node, ast.Expr): + value = node.value + if value.end_lineno is not None and value.end_col_offset is not None: + spans.add(( + (value.lineno, value.col_offset + 1), + (value.end_lineno, value.end_col_offset + 1), + )) + return spans + + +def _is_signature(spelled: str) -> bool: + # Pyrefly spells callables as `(x: int) -> str` and overload sets as `Overload[...]`. + return "->" in spelled or spelled.startswith("Overload[") + + +def _inside(position: Position, span: tuple[Position, Position]) -> bool: + start, stop = span + return start <= position <= stop diff --git a/python-multiplatform-ksp/src/main/python/typedpython/pyrefly.py b/python-multiplatform-ksp/src/main/python/typedpython/pyrefly.py new file mode 100644 index 000000000..b077ae942 --- /dev/null +++ b/python-multiplatform-ksp/src/main/python/typedpython/pyrefly.py @@ -0,0 +1,176 @@ +"""Running Pyrefly and reading what it reports (docs/design/typedpython.md §4.1). + +Pyrefly is used as a CLI, not embedded. Two outputs are read: + +- its JSON error list, with the Any-leak error kinds raised to `error` by a generated config; +- its per-expression type report (`--report-pysa`), which says which expressions in user code + have type `Any`. That report is a Pyrefly-versioned format, which is why pyproject.toml pins + Pyrefly exactly. +""" +import json +import os +import re +import subprocess +import sys +import tempfile +from collections.abc import Sequence +from dataclasses import dataclass +from pathlib import Path + +# Pyrefly error kinds that mean "an Any got in". Pyrefly leaves several of them off by default. +ANY_LEAK_KINDS = ( + "implicit-any-parameter", + "unannotated-return", + "unknown-variable-type", + "no-any-return-implicit", + "no-any-return-explicit", + "explicit-any", +) + +# How Pyrefly spells an Any in the type report: `typing.Any` for an explicit one and `Unknown` +# for an implicit one, alone or inside a larger type such as `list[typing.Any]`. +ANY_IN_TYPE = re.compile(r"(? (line, column) -> type, for every expression the type report records. + expression_types: dict[str, dict[tuple[int, int], str]] + + +def command() -> list[str]: + """The Pyrefly command line: `TYPEDPYTHON_PYREFLY`, or the copy installed next to this Python.""" + override = os.environ.get("TYPEDPYTHON_PYREFLY") + return [override] if override else [sys.executable, "-m", "pyrefly"] + + +def run(paths: Sequence[Path], search_paths: Sequence[Path] = ()) -> Report: + files = [str(Path(p).resolve()) for p in paths] + with tempfile.TemporaryDirectory(prefix="typedpython-") as work: + config = Path(work) / "pyrefly.toml" + config.write_text(_config([*search_paths, *import_roots(files)])) + report_dir = Path(work) / "pysa" + argv = [ + *command(), "check", *files, + "--config", str(config), + "--output-format", "json", + "--summary=none", + "--report-pysa", str(report_dir), + "--report-pysa-format", "json", + ] + try: + result = subprocess.run(argv, capture_output=True, text=True, check=False) + except OSError as e: + raise PyreflyError(f"cannot run {argv[0]}: {e}") from e + + # Exit code 1 only means "type errors found"; anything else is a failure of the tool. + if result.returncode not in (0, 1): + raise PyreflyError(f"pyrefly exited {result.returncode}: {result.stderr.strip()}") + try: + raw_errors = json.loads(result.stdout)["errors"] + except (json.JSONDecodeError, KeyError) as e: + raise PyreflyError(f"unreadable pyrefly output: {result.stdout[:200]!r}") from e + + errors = [ + Error(_resolve(e["path"]), e["line"], e["column"], e["name"], e["description"]) + for e in raw_errors + ] + types = _expression_types(report_dir, set(files)) + return Report(errors, _any_expressions(types), _types_by_start(types)) + + +def import_roots(files: Sequence[str]) -> list[Path]: + """The directory each file is imported from: above its outermost package, else its own. + + The config lives in a temporary directory, and Pyrefly would otherwise take that as the import + root, so a package's own imports (relative or absolute) would not resolve. + """ + roots: list[Path] = [] + for file in files: + root = Path(file).parent + while (root / "__init__.py").exists(): + root = root.parent + if root not in roots: + roots.append(root) + return roots + + +def _config(search_paths: Sequence[Path]) -> str: + lines = [] + if search_paths: + quoted = ", ".join(json.dumps(str(Path(p).resolve())) for p in search_paths) + lines.append(f"search-path = [{quoted}]") + lines.append("[errors]") + lines += [f'{kind} = "error"' for kind in ANY_LEAK_KINDS] + return "\n".join(lines) + "\n" + + +def _expression_types(report_dir: Path, files: set[str]) -> list[AnyExpression]: + """Every typed expression the report records in the given files (typeshed and deps skipped).""" + try: + index = json.loads((report_dir / "pyrefly.pysa.json").read_text()) + except (OSError, json.JSONDecodeError) as e: + raise PyreflyError(f"pyrefly wrote no readable type report: {e}") from e + + found: list[AnyExpression] = [] + for module in index["modules"].values(): + source = module.get("source_path", {}).get("FileSystem") + if source is None or _resolve(source) not in files: + continue # typeshed and dependencies: not user code + table_file = report_dir / "type_of_expressions" / module["info_filename"] + if not table_file.exists(): + continue + table = json.loads(table_file.read_text()) + for function in table["functions"].values(): + types = function["type_table"] + for location, type_index in function["locations"].items(): + found.append(_expression(_resolve(source), location, types[type_index]["string"])) + return found + + +def _any_expressions(expressions: list[AnyExpression]) -> list[AnyExpression]: + return [e for e in expressions if ANY_IN_TYPE.search(e.type)] + + +def _types_by_start(expressions: list[AnyExpression]) -> dict[str, dict[tuple[int, int], str]]: + table: dict[str, dict[tuple[int, int], str]] = {} + for e in expressions: + table.setdefault(e.path, {})[(e.line, e.column)] = e.type + return table + + +def _expression(path: str, location: str, spelled: str) -> AnyExpression: + # "6:9-6:15" -> start line/column, stop line/column (1-based, as in Pyrefly's errors). + start, stop = location.split("-") + line, column = map(int, start.split(":")) + stop_line, stop_column = map(int, stop.split(":")) + return AnyExpression(path, line, column, stop_line, stop_column, spelled) + + +def _resolve(path: str) -> str: + return str(Path(path).resolve()) diff --git a/python-multiplatform-ksp/src/main/python/typedpython/rebinding.py b/python-multiplatform-ksp/src/main/python/typedpython/rebinding.py new file mode 100644 index 000000000..8f26cef8e --- /dev/null +++ b/python-multiplatform-ksp/src/main/python/typedpython/rebinding.py @@ -0,0 +1,258 @@ +"""Rebinding a name to a different type, and mixed-element containers (design §3.1). + +Pyrefly 1.3.2 reports neither, and its per-expression type report has no entries for assignment +targets. It does record every *load* of a name, with the name's type at that point. So the gate +writes a probed copy of the file in which each assignment of interest is followed, on the same +line, by a bare load of each target (`x = 1` becomes `x = 1; x`), runs Pyrefly on the copy, and +reads the target types back at the probe positions. Line numbers do not change, and every column +before the insertion point stays where it was. +""" +import ast +import re +from collections.abc import Iterable +from dataclasses import dataclass, field + +# Container types whose element type should be one type, not a union. Tuples are excluded: a +# fixed-length tuple is typed per slot, so `tuple[int, str]` is not "mixed". +MIXABLE = re.compile(r"^(list|set|frozenset|dict)\[(.*)\]$", re.DOTALL) +LITERAL = re.compile(r"^Literal\[(.*)\]$", re.DOTALL) + + +@dataclass(frozen=True) +class Assignment: + """One assignment to a plain name, in the original source.""" + + scope: int + name: str + line: int + column: int + end_line: int + end_byte: int # ast end_col_offset of the whole statement (UTF-8 bytes) + augmented: bool + display: bool # the value is a list/set/dict display with more than one element + + +@dataclass(frozen=True) +class Probe: + name: str + line: int + column: int + assignment: Assignment | None = field(default=None, compare=False) + + +@dataclass(frozen=True) +class Finding: + rule: str + line: int + column: int + message: str + + +def assignments(tree: ast.Module) -> list[Assignment]: + """Assignments to unannotated names, scope by scope (module, class bodies, functions).""" + found: list[Assignment] = [] + for scope_id, scope in enumerate(_scopes(tree)): + declared = _declared_names(scope) + for stmt in _own_statements(scope): + found += [a for a in _assignments_in(stmt, scope_id) if a.name not in declared] + return found + + +def candidates(found: Iterable[Assignment]) -> list[Assignment]: + """The assignments worth probing: names assigned more than once, and container displays.""" + found = list(found) + counts: dict[tuple[int, str], int] = {} + for a in found: + counts[(a.scope, a.name)] = counts.get((a.scope, a.name), 0) + 1 + return [a for a in found if counts[(a.scope, a.name)] > 1 or a.display] + + +def probe(source: str, wanted: Iterable[Assignment] | None = None) -> tuple[str, list[Probe]]: + """Insert a load of each target after its statement; `wanted=None` probes every assignment.""" + if wanted is None: + wanted = assignments(ast.parse(source)) + lines = source.splitlines(keepends=True) + # Insert right-to-left within a line so earlier offsets stay valid. + by_position: dict[tuple[int, int], list[Assignment]] = {} + for a in wanted: + by_position.setdefault((a.end_line, a.end_byte), []).append(a) + + probes: list[Probe] = [] + for (line_no, end_byte) in sorted(by_position, reverse=True): + encoded = lines[line_no - 1].encode() + head, tail = encoded[:end_byte].decode(), encoded[end_byte:].decode() + inserted = "" + added: list[Probe] = [] + for a in by_position[(line_no, end_byte)]: + inserted += "; " + added.append(Probe(a.name, line_no, len(head) + len(inserted) + 1, a)) + inserted += a.name + lines[line_no - 1] = head + inserted + tail + # Probes already placed further right on this line moved by what was just inserted. + probes = [ + Probe(q.name, q.line, q.column + len(inserted), q.assignment) + if q.line == line_no and q.column > len(head) else q + for q in probes + ] + probes += added + + return "".join(lines), probes + + +def judge(probes: list[Probe], types: dict[tuple[int, int], str]) -> list[Finding]: + """Turn probed types into findings.""" + findings: list[Finding] = [] + history: dict[tuple[int, str], tuple[str, Assignment]] = {} + + for p in sorted(probes, key=lambda p: (p.line, p.column)): + a = p.assignment + spelled = types.get((p.line, p.column)) + if a is None or spelled is None or _is_unknown(spelled): + continue + + if a.display and (mixed := _mixed_element(spelled)): + findings.append(Finding( + "forbidden/mixed-container", a.line, a.column, + f"`{a.name}` is `{spelled}`: its elements mix `{mixed}`; annotate the union " + "on purpose or keep one element type", + )) + + key = (a.scope, a.name) + current = _base(spelled) + if key in history and history[key][0] != current: + before = history[key][0] + findings.append(Finding( + "forbidden/rebinding", a.line, a.column, + f"`{a.name}` was `{before}` and is rebound to `{current}`; annotate it " + f"(`{a.name}: {before} | {current}`) or use another name", + )) + history[key] = (current, a) + + return findings + + +# --- AST helpers ------------------------------------------------------------------------------- + +Scope = ast.Module | ast.ClassDef | ast.FunctionDef | ast.AsyncFunctionDef + + +def _scopes(tree: ast.Module) -> list[Scope]: + return [tree] + [ + n for n in ast.walk(tree) + if isinstance(n, (ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)) + ] + + +def _own_statements(scope: Scope) -> Iterable[ast.stmt]: + """Statements of this scope, descending into blocks but not into nested scopes.""" + stack = list(reversed(scope.body)) + while stack: + stmt = stack.pop() + yield stmt + if isinstance(stmt, (ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)): + continue + for name in ("body", "orelse", "finalbody"): + stack.extend(reversed(getattr(stmt, name, []))) + for handler in getattr(stmt, "handlers", []): + stack.extend(reversed(handler.body)) + for case in getattr(stmt, "cases", []): + stack.extend(reversed(case.body)) + + +def _declared_names(scope: Scope) -> set[str]: + """Names whose type is declared (annotated, parameters) or owned elsewhere (global).""" + names: set[str] = set() + if isinstance(scope, (ast.FunctionDef, ast.AsyncFunctionDef)): + args = scope.args + for arg in [*args.posonlyargs, *args.args, *args.kwonlyargs]: + names.add(arg.arg) + names |= {a.arg for a in (args.vararg, args.kwarg) if a is not None} + for stmt in _own_statements(scope): + if isinstance(stmt, ast.AnnAssign) and isinstance(stmt.target, ast.Name): + names.add(stmt.target.id) + elif isinstance(stmt, (ast.Global, ast.Nonlocal)): + names |= set(stmt.names) + return names + + +def _assignments_in(stmt: ast.stmt, scope_id: int) -> list[Assignment]: + if isinstance(stmt, ast.Assign): + targets = [t for target in stmt.targets for t in _names(target)] + display = len(stmt.targets) == 1 and isinstance(stmt.targets[0], ast.Name) \ + and _is_display(stmt.value) + augmented = False + elif isinstance(stmt, ast.AugAssign) and isinstance(stmt.target, ast.Name): + targets, display, augmented = [stmt.target], False, True + else: + return [] + assert stmt.end_lineno is not None and stmt.end_col_offset is not None + return [ + Assignment(scope_id, t.id, t.lineno, t.col_offset + 1, + stmt.end_lineno, stmt.end_col_offset, augmented, display) + for t in targets + ] + + +def _names(target: ast.expr) -> list[ast.Name]: + if isinstance(target, ast.Name): + return [target] + if isinstance(target, (ast.Tuple, ast.List)): + return [n for element in target.elts for n in _names(element)] + if isinstance(target, ast.Starred): + return _names(target.value) + return [] + + +def _is_display(value: ast.expr) -> bool: + if isinstance(value, (ast.List, ast.Set)): + return len(value.elts) > 1 + if isinstance(value, ast.Dict): + return len(value.keys) > 1 + return False + + +# --- type spelling ----------------------------------------------------------------------------- + +def _is_unknown(spelled: str) -> bool: + return spelled in ("Unknown", "typing.Any", "Any") + + +def _base(spelled: str) -> str: + """`Literal[1]` -> `int`, `Literal['a']` -> `str`, `Literal[True]` -> `bool`.""" + m = LITERAL.match(spelled) + if not m: + return spelled + value = m.group(1).strip() + if value in ("True", "False"): + return "bool" + if value[:1] in ("'", '"'): + return "str" + if value[:1] in ("b",) and value[1:2] in ("'", '"'): + return "bytes" + return "int" + + +def _mixed_element(spelled: str) -> str | None: + m = MIXABLE.match(spelled) + if not m: + return None + for argument in _split_top_level(m.group(2)): + if len(_split_top_level(argument, "|")) > 1: + return argument.strip() + return None + + +def _split_top_level(text: str, separator: str = ",") -> list[str]: + parts, depth, current = [], 0, "" + for ch in text: + if ch in "[(": + depth += 1 + elif ch in "])": + depth -= 1 + if ch == separator and depth == 0: + parts.append(current) + current = "" + else: + current += ch + parts.append(current) + return parts diff --git a/python-multiplatform-ksp/src/test/python/conftest.py b/python-multiplatform-ksp/src/test/python/conftest.py new file mode 100644 index 000000000..65b1e5f0a --- /dev/null +++ b/python-multiplatform-ksp/src/test/python/conftest.py @@ -0,0 +1,27 @@ +from pathlib import Path +import textwrap + +import pytest + +FIXTURES = Path(__file__).parent / "fixtures" + + +@pytest.fixture +def write(tmp_path): + """Write a dedented source file under tmp_path and return its path.""" + + def _write(name: str, source: str) -> Path: + path = tmp_path / name + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(textwrap.dedent(source)) + return path + + return _write + + +def rules(diagnostics): + return sorted({d.rule for d in diagnostics}) + + +def errors(diagnostics): + return [d for d in diagnostics if d.severity == "error"] diff --git a/python-multiplatform-ksp/src/test/python/fixtures/deps/untyped_dep.py b/python-multiplatform-ksp/src/test/python/fixtures/deps/untyped_dep.py new file mode 100644 index 000000000..af73da304 --- /dev/null +++ b/python-multiplatform-ksp/src/test/python/fixtures/deps/untyped_dep.py @@ -0,0 +1,7 @@ +# Stands in for a third-party package without stubs. It is deliberately unannotated: the gate +# must not report anything inside it, only where its untyped results flow into user code. +import json + + +def load(text): + return json.loads(text) diff --git a/python-multiplatform-ksp/src/test/python/test_any_flow.py b/python-multiplatform-ksp/src/test/python/test_any_flow.py new file mode 100644 index 000000000..fae430c44 --- /dev/null +++ b/python-multiplatform-ksp/src/test/python/test_any_flow.py @@ -0,0 +1,43 @@ +"""Any reaching user code from an untyped dependency (design §2.3, §4.2).""" +from conftest import FIXTURES, errors +from typedpython import gate + +DEPS = FIXTURES / "deps" + + +def test_untyped_result_flowing_into_user_code_is_an_error(write): + path = write("user.py", """ + from untyped_dep import load + + def total(text: str) -> None: + value = load(text) + print(value) + """) + found = errors(gate.check([path], search_paths=[DEPS])) + flow = [d for d in found if d.rule == "any-flow"] + assert flow, found + assert all(d.path == str(path) for d in flow) + + +def test_cast_at_the_boundary_resolves_it(write): + path = write("user.py", """ + from typing import cast + from untyped_dep import load + + def total(text: str) -> int: + value = cast(int, load(text)) + return value + 1 + """) + assert gate.check([path], search_paths=[DEPS]) == [] + + +def test_the_dependency_itself_is_not_checked(write): + path = write("user.py", """ + from typing import cast + from untyped_dep import load + + def total(text: str) -> int: + return cast(int, load(text)) + """) + found = gate.check([path], search_paths=[DEPS]) + assert not [d for d in found if d.path.endswith("untyped_dep.py")], found diff --git a/python-multiplatform-ksp/src/test/python/test_cli.py b/python-multiplatform-ksp/src/test/python/test_cli.py new file mode 100644 index 000000000..6b5d1f383 --- /dev/null +++ b/python-multiplatform-ksp/src/test/python/test_cli.py @@ -0,0 +1,48 @@ +"""`python -m typedpython check` exit codes: 0 clean, 1 errors, 2 tool failure.""" +import json +import os +import subprocess +import sys +from pathlib import Path + +SRC = Path(__file__).parents[2] / "main" / "python" + + +def run(*args, env_extra=None): + env = dict(os.environ, PYTHONPATH=str(SRC), **(env_extra or {})) + return subprocess.run( + [sys.executable, "-m", "typedpython", "check", *map(str, args)], + capture_output=True, text=True, env=env, + ) + + +def test_clean_exits_0(write): + path = write("user.py", "def f(x: int) -> int:\n return x\n") + result = run(path) + assert result.returncode == 0, result.stdout + result.stderr + + +def test_error_exits_1(write): + path = write("user.py", "def f(x):\n return x\n") + result = run(path) + assert result.returncode == 1, result.stdout + result.stderr + assert "implicit-any-parameter" in result.stdout + + +def test_warnings_alone_exit_0(write): + path = write("user.py", "def f(s: str) -> None:\n eval(s)\n") + assert run(path).returncode == 0 + assert run("--mode", "compiled", path).returncode == 1 + + +def test_json_output(write): + path = write("user.py", "def f(x):\n return x\n") + result = run("--format", "json", path) + data = json.loads(result.stdout) + assert {"path", "line", "column", "rule", "severity", "message"} <= set(data[0]) + + +def test_missing_pyrefly_exits_2(write): + path = write("user.py", "def f(x: int) -> int:\n return x\n") + result = run(path, env_extra={"TYPEDPYTHON_PYREFLY": "/nonexistent/pyrefly"}) + assert result.returncode == 2, result.stdout + result.stderr diff --git a/python-multiplatform-ksp/src/test/python/test_forbidden.py b/python-multiplatform-ksp/src/test/python/test_forbidden.py new file mode 100644 index 000000000..b672a22f8 --- /dev/null +++ b/python-multiplatform-ksp/src/test/python/test_forbidden.py @@ -0,0 +1,54 @@ +"""The §3.1 forbidden list: warning under `checked`, error under `compiled`.""" +import pytest + +from conftest import rules +from typedpython import forbidden, gate + +FLAGGED = { + "forbidden/eval": "def f(s: str) -> None:\n eval(s)\n", + "forbidden/exec": "def f(s: str) -> None:\n exec(s)\n", + "forbidden/dynamic-attr": "def f(o: object, n: str) -> None:\n getattr(o, n)\n", + "forbidden/monkeypatch": "import json\n\ndef f() -> None:\n json.dumps = print\n", +} + + +@pytest.mark.parametrize("rule", sorted(FLAGGED)) +def test_rule_is_found(rule): + assert rule in {f.rule for f in forbidden.scan(FLAGGED[rule], "user.py")} + + +@pytest.mark.parametrize("call", ["setattr(o, n, 1)", "delattr(o, n)"]) +def test_other_dynamic_attribute_calls(call): + source = f"def f(o: object, n: str) -> None:\n {call}\n" + assert {f.rule for f in forbidden.scan(source, "user.py")} == {"forbidden/dynamic-attr"} + + +def test_monkeypatching_a_module_level_class(): + source = ( + "class Greeter:\n pass\n\n" + "def patch() -> None:\n Greeter.hello = 1\n" + ) + assert {f.rule for f in forbidden.scan(source, "user.py")} == {"forbidden/monkeypatch"} + + +@pytest.mark.parametrize("source", [ + "def f(o: object) -> None:\n getattr(o, 'name')\n", + "class C:\n def __init__(self) -> None:\n self.x = 1\n", + "def f() -> None:\n class Local:\n pass\n obj = Local()\n obj.y = 2\n", +]) +def test_static_forms_are_not_flagged(source): + assert forbidden.scan(source, "user.py") == [] + + +def test_findings_carry_a_location(): + [finding] = forbidden.scan("x = 1\neval('x')\n", "user.py") + assert (finding.line, finding.column) == (2, 1) + + +def test_severity_follows_the_mode(write): + path = write("user.py", "def f(s: str) -> None:\n eval(s)\n") + checked = gate.check([path], mode="checked") + compiled = gate.check([path], mode="compiled") + assert [d.severity for d in checked if d.rule == "forbidden/eval"] == ["warning"] + assert [d.severity for d in compiled if d.rule == "forbidden/eval"] == ["error"] + assert rules(checked) == rules(compiled) diff --git a/python-multiplatform-ksp/src/test/python/test_imports.py b/python-multiplatform-ksp/src/test/python/test_imports.py new file mode 100644 index 000000000..4a9a9a876 --- /dev/null +++ b/python-multiplatform-ksp/src/test/python/test_imports.py @@ -0,0 +1,31 @@ +"""User code inside a package must resolve its own imports (defect found 2026-10-03). + +The gate writes its Pyrefly config into a temporary directory. Pyrefly infers the import root +from the config's location, so without help a package's relative and absolute self-imports did +not resolve: every value from them became `Unknown` and failed the gate as `any-flow`. +""" +from conftest import errors +from typedpython import gate + + +def package(write): + write("src/pkg/__init__.py", "") + write("src/pkg/b.py", "def make() -> int:\n return 1\n") + + +def test_relative_import_inside_a_package(write): + package(write) + path = write("src/pkg/a.py", "from .b import make\n\ndef f() -> int:\n x = make()\n return x\n") + assert errors(gate.check([path])) == [] + + +def test_absolute_import_of_the_own_package(write): + package(write) + path = write("src/pkg/c.py", "from pkg.b import make\n\ndef f() -> int:\n return make()\n") + assert errors(gate.check([path])) == [] + + +def test_sibling_module_without_a_package(write): + write("flat/helper.py", "def make() -> int:\n return 1\n") + path = write("flat/main.py", "from helper import make\n\ndef f() -> int:\n return make()\n") + assert errors(gate.check([path])) == [] diff --git a/python-multiplatform-ksp/src/test/python/test_measurement.py b/python-multiplatform-ksp/src/test/python/test_measurement.py new file mode 100644 index 000000000..45e5a15b2 --- /dev/null +++ b/python-multiplatform-ksp/src/test/python/test_measurement.py @@ -0,0 +1,50 @@ +"""What the gate costs on top of a plain `pyrefly check` (design §7, AGENTS.md §14). + +The extra work is the per-expression type report (`--report-pysa`) and the AST pass. The numbers +are printed so a run records them; only a loose sanity bound is asserted. +""" +import subprocess +import sys +import time + +import pytest + +from typedpython import gate + +FUNCTIONS = 300 + + +def synthetic_module() -> str: + lines = [] + for i in range(FUNCTIONS): + lines += [ + f"def f{i}(a: int, b: int) -> int:", + " total = 0", + " for k in range(a):", + " total += k * b", + " return total", + "", + ] + return "\n".join(lines) + + +@pytest.mark.measurement +def test_gate_cost_against_plain_pyrefly(write): + path = write("big.py", synthetic_module()) + lines = synthetic_module().count("\n") + 1 + + start = time.perf_counter() + subprocess.run( + [sys.executable, "-m", "pyrefly", "check", str(path), "--summary=none"], + capture_output=True, check=False, + ) + plain = time.perf_counter() - start + + start = time.perf_counter() + found = gate.check([path]) + gated = time.perf_counter() - start + + print(f"\n[measure] {lines} lines: plain pyrefly {plain * 1000:.0f} ms, " + f"gate {gated * 1000:.0f} ms, ratio {gated / plain:.2f}x") + assert found == [] + assert gated < 60 diff --git a/python-multiplatform-ksp/src/test/python/test_pyrefly_kinds.py b/python-multiplatform-ksp/src/test/python/test_pyrefly_kinds.py new file mode 100644 index 000000000..f7e864dfb --- /dev/null +++ b/python-multiplatform-ksp/src/test/python/test_pyrefly_kinds.py @@ -0,0 +1,69 @@ +"""Each Pyrefly Any-leak kind the gate raises to error (design §4.2).""" +import pytest + +from conftest import errors, rules +from typedpython import gate + +CASES = { + "implicit-any-parameter": """ + def f(x, y: int) -> int: + return y + """, + "unannotated-return": """ + def f(y: int): + return y + """, + "unknown-variable-type": """ + import json + def f() -> None: + d = json.loads("1") + z = d["k"] + """, + "no-any-return-implicit": """ + import json + def f() -> int: + d = json.loads("1") + return d["k"] + """, + "no-any-return-explicit": """ + from typing import Any + def f(v: Any) -> int: + return v + """, + "explicit-any": """ + from typing import Any + def f(v: Any) -> None: + pass + """, +} + + +@pytest.mark.parametrize("kind", sorted(CASES)) +def test_kind_is_an_error(write, kind): + path = write("user.py", CASES[kind]) + found = gate.check([path]) + assert f"pyrefly/{kind}" in rules(errors(found)), found + + +def test_ordinary_type_error_is_an_error(write): + path = write("user.py", """ + def f(y: int) -> int: + return y + "a" + """) + assert errors(gate.check([path])), "a plain type error must still fail the gate" + + +def test_fully_typed_code_is_clean(write): + path = write("user.py", """ + def area(w: int, h: int) -> int: + scale = 2 + return w * h * scale + + class Box: + def __init__(self, w: int) -> None: + self.w = w + + def double(self) -> int: + return self.w * 2 + """) + assert gate.check([path]) == [] diff --git a/python-multiplatform-ksp/src/test/python/test_rebinding.py b/python-multiplatform-ksp/src/test/python/test_rebinding.py new file mode 100644 index 000000000..731707041 --- /dev/null +++ b/python-multiplatform-ksp/src/test/python/test_rebinding.py @@ -0,0 +1,184 @@ +"""Rebinding a name to a different type, and mixed-element containers (design §3.1, #22). + +Both are forbidden-list items: warnings under `checked`, errors under `compiled`. They need the +inferred type of an assignment target, which Pyrefly's type report does not give directly; the +gate reads each target back through a probe load (see typedpython.rebinding). +""" +import pytest + +from typedpython import gate, rebinding + +REBIND = "forbidden/rebinding" +MIXED = "forbidden/mixed-container" + + +def found(write, source, mode="checked"): + return gate.check([write("user.py", source)], mode=mode) + + +def by_rule(diagnostics, rule): + return [d for d in diagnostics if d.rule == rule] + + +def test_rebinding_to_another_type_is_reported(write): + result = found(write, """ + def f() -> None: + x = 1 + x = "a" + print(x) + """) + [d] = by_rule(result, REBIND) + assert (d.line, d.severity) == (4, "warning") + assert "int" in d.message and "str" in d.message + + +def test_compiled_mode_makes_it_an_error(write): + result = found(write, """ + def f() -> None: + x = 1 + x = "a" + print(x) + """, mode="compiled") + assert [d.severity for d in by_rule(result, REBIND)] == ["error"] + + +def test_augmented_assignment_that_changes_the_type(write): + result = found(write, """ + def f() -> float: + total = 0 + total += 0.5 + return total + """) + assert by_rule(result, REBIND), result + + +def test_module_level_rebinding(write): + result = found(write, """ + limit = 10 + limit = "ten" + print(limit) + """) + assert by_rule(result, REBIND), result + + +@pytest.mark.parametrize("source", [ + # same type again + "def f() -> None:\n x = 1\n x = 2\n print(x)\n", + # a declared union governs; Pyrefly checks each assignment against it + "def f() -> None:\n x: int | str = 1\n x = 'a'\n print(x)\n", + # an annotated parameter reassigned within its type + "def f(n: int) -> int:\n n = n + 1\n return n\n", +]) +def test_not_rebinding(write, source): + assert by_rule(gate.check([write("user.py", source)]), REBIND) == [] + + +def test_mixed_list_is_reported(write): + result = found(write, """ + def f() -> None: + mixed = [1, "a"] + print(mixed) + """) + [d] = by_rule(result, MIXED) + assert d.line == 3 and "int | str" in d.message + + +def test_mixed_dict_values_are_reported(write): + result = found(write, """ + def f(a: int) -> None: + d = {"k": a, "j": "s"} + print(d) + """) + assert by_rule(result, MIXED), result + + +@pytest.mark.parametrize("source", [ + "def f() -> None:\n nums = [1, 2]\n print(nums)\n", + "def f() -> None:\n pair = (1, 'a')\n print(pair)\n", # tuples are typed per slot + "def f() -> None:\n xs: list[int | str] = [1, 'a']\n print(xs)\n", # declared on purpose +]) +def test_not_mixed(write, source): + assert by_rule(gate.check([write("user.py", source)]), MIXED) == [] + + +def test_probe_keeps_every_line_number(): + source = "def f() -> None:\n x = 1 # one\n x = 'a'\n y, z = 1, 2\n" + probed, probes = rebinding.probe(source) + assert probed.count("\n") == source.count("\n") + assert {p.name for p in probes} >= {"x", "y", "z"} + for p in probes: + line = probed.splitlines()[p.line - 1] + assert line[p.column - 1:].startswith(p.name), (p, line) + + +def test_no_candidates_means_no_second_pyrefly_run(write, monkeypatch): + calls = [] + real = gate.pyrefly.run + + def counting(*args, **kwargs): + calls.append(args) + return real(*args, **kwargs) + + monkeypatch.setattr(gate.pyrefly, "run", counting) + gate.check([write("user.py", "def f(a: int) -> int:\n b = a + 1\n return b\n")]) + assert len(calls) == 1 + + +def test_rebinding_inside_a_package_with_relative_imports(write): + write("pkg/__init__.py", "") + write("pkg/b.py", "def make() -> int:\n return 1\n") + path = write("pkg/a.py", """ + from .b import make + + def f() -> None: + x = make() + x = "s" + print(x) + """) + result = gate.check([path]) + assert by_rule(result, REBIND), result + assert not [d for d in result if d.severity == "error"], result + + +def test_pyrefly_runs_once_when_a_file_has_a_candidate(write, monkeypatch): + """The probed copy is the only Pyrefly run: errors, any-flow and types all come from it.""" + calls = [] + real = gate.pyrefly.run + + def counting(*args, **kwargs): + calls.append(args) + return real(*args, **kwargs) + + monkeypatch.setattr(gate.pyrefly, "run", counting) + result = found(write, """ + def f() -> None: + x = 1 + x = "a" + print(x) + """) + assert by_rule(result, REBIND) + assert len(calls) == 1 + + +def test_error_column_after_a_probe_points_at_the_original_source(write): + """`x = 1; y = undefined_name`: the probe goes after `x = 1`, left of the error.""" + source = "x = 1\nx = 2\nx = 1; y = undefined_name\n" + path = write("user.py", source) + result = gate.check([path]) + [d] = [d for d in result if d.rule == "pyrefly/unknown-name"] + original_column = source.splitlines()[2].index("undefined_name") + 1 + assert (d.line, d.column) == (3, original_column) + + +def test_error_left_of_a_probe_keeps_its_column(write): + source = "x = 1\nx = 2\nprint(undefined_name); x = 3\n" + path = write("user.py", source) + [d] = [d for d in gate.check([path]) if d.rule == "pyrefly/unknown-name"] + assert (d.line, d.column) == (3, source.splitlines()[2].index("undefined_name") + 1) + + +def test_probe_loads_are_not_any_flow_findings(write): + """The probe load `; x` has type Unknown here, but it is a discarded value, not a finding.""" + path = write("user.py", "x = eval('1')\nx = eval('2')\nprint(x)\n") + flows = {(d.line, d.column) for d in gate.check([path]) if d.rule == "any-flow"} + assert flows == {(3, 7)} # the user's own load of x, not the probe loads on lines 1 and 2