mirror of
https://github.com/langgenius/dify.git
synced 2026-11-10 21:35:45 +08:00
ci: add new rules + some better code guide (#44828)
Co-authored-by: Codex <codex@openai.com> Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
This commit is contained in:
10 files changed
+752
No files matched your search
@@ -67,6 +67,10 @@ jobs:
|
||||
- 'scripts/check_no_new_getattr.py'
|
||||
- 'scripts/check_no_spec_mock.py'
|
||||
- 'scripts/check_no_new_base_model.py'
|
||||
- 'scripts/check_no_new_constructor_bypass.py'
|
||||
- 'scripts/check_no_new_cast.py'
|
||||
- 'scripts/ast_grep_rules/no_new_cast.yml'
|
||||
- 'scripts/ast_grep_rules/no_new_constructor_bypass.yml'
|
||||
- 'scripts/ast_grep_rules/no_new_base_model.yml'
|
||||
- 'scripts/check_no_new_test_doubles.py'
|
||||
- 'scripts/check_no_new_controller_sqlalchemy.py'
|
||||
@@ -172,6 +176,10 @@ jobs:
|
||||
- 'scripts/check_no_new_getattr.py'
|
||||
- 'scripts/check_no_spec_mock.py'
|
||||
- 'scripts/check_no_new_base_model.py'
|
||||
- 'scripts/check_no_new_constructor_bypass.py'
|
||||
- 'scripts/check_no_new_cast.py'
|
||||
- 'scripts/ast_grep_rules/no_new_cast.yml'
|
||||
- 'scripts/ast_grep_rules/no_new_constructor_bypass.yml'
|
||||
- 'scripts/ast_grep_rules/no_new_base_model.yml'
|
||||
- 'scripts/check_no_new_test_doubles.py'
|
||||
- 'scripts/check_no_new_controller_sqlalchemy.py'
|
||||
|
||||
@@ -107,6 +107,18 @@ jobs:
|
||||
GUARD_BASE_REV: ${{ github.event_name == 'pull_request' && 'HEAD^1' || inputs.base-rev }}
|
||||
run: uv run --project api python scripts/check_no_new_base_model.py --base-rev "$GUARD_BASE_REV"
|
||||
|
||||
- name: Run No New Cast Guard
|
||||
if: inputs.run-python-style
|
||||
env:
|
||||
GUARD_BASE_REV: ${{ github.event_name == 'pull_request' && 'HEAD^1' || inputs.base-rev }}
|
||||
run: uv run --project api python scripts/check_no_new_cast.py --base-rev "$GUARD_BASE_REV"
|
||||
|
||||
- name: Run No New Constructor Bypass Guard
|
||||
if: inputs.run-python-style
|
||||
env:
|
||||
GUARD_BASE_REV: ${{ github.event_name == 'pull_request' && 'HEAD^1' || inputs.base-rev }}
|
||||
run: uv run --project api python scripts/check_no_new_constructor_bypass.py --base-rev "$GUARD_BASE_REV"
|
||||
|
||||
- name: Run No New Test Doubles Guard
|
||||
if: inputs.run-python-style || inputs.run-dify-agent-style
|
||||
# A PR event can retain an older base SHA after its target branch advances.
|
||||
|
||||
@@ -82,6 +82,8 @@ lint:
|
||||
@$(MAKE) api-contract-lint
|
||||
@uv run --project api python scripts/check_no_spec_mock.py
|
||||
@uv run --project api python scripts/check_no_new_base_model.py
|
||||
@uv run --project api python scripts/check_no_new_constructor_bypass.py
|
||||
@uv run --project api python scripts/check_no_new_cast.py
|
||||
@uv run --directory api --dev lint-imports
|
||||
@uv run --project api --dev dotenv-linter ./api/.env.example ./web/.env.example
|
||||
@echo "✅ Linting complete"
|
||||
|
||||
@@ -47,6 +47,28 @@ changes, and CI checks the PR against its target revision.
|
||||
- Treat nullable types (`T | None`) strictly as union types; document the semantic meaning for each branch.
|
||||
- Treat default parameter values as introducing distinct function overloads; verify that every added default has demonstrable practical utility and avoids masking unsafe runtime states.
|
||||
|
||||
## Invariants And Construction
|
||||
|
||||
- Start each design from explicit principles and domain invariants. An invariant is a condition that all valid states preserve.
|
||||
- Derive implementation rules and memory documents from these principles. Explain the invariant that each rule protects.
|
||||
- Document each invariant in the docstring of the module, class, or function that owns it. Use a comment for a local invariant.
|
||||
- Use abstract interpretation to reason about possible values and state transitions. Check that each operation preserves the relevant invariants.
|
||||
- Use dedicated types to distinguish domain values with different meanings. Use `NewType` for static distinctions between values with identical representations.
|
||||
- `NewType` does not validate values at runtime. Use validated value objects when an invariant requires runtime checks.
|
||||
- Model mutually exclusive alternatives (an OR relationship) with union types. Use a tagged union when variants need an explicit discriminator.
|
||||
- Give each tagged variant a distinct `Literal` tag. Include only the fields that are valid for that variant.
|
||||
- Prefer safe-by-construction functional idioms. Use immutable values, explicit result variants, and pure functions where practical.
|
||||
- Validate external values at boundaries with Pydantic v2 models or `TypeAdapter`. Construct trusted domain values only after validation succeeds.
|
||||
|
||||
## Control Flow And Contracts
|
||||
|
||||
- Use structured control flow. Do not add `continue` or `break` statements.
|
||||
- Express iteration with explicit conditions, comprehensions, or small functions that return a result.
|
||||
- State preconditions and postconditions for each new or changed function in the code you produce.
|
||||
- Put these contracts in the function docstring. Include accepted inputs, required state, results, side effects, and failure behavior as applicable.
|
||||
- Specify which invariants remain true after success and failure. Use types and executable checks to enforce contracts where practical.
|
||||
- Include preconditions and postconditions in the change report. State which checks support the claimed postconditions.
|
||||
|
||||
[architecture]: ../docs/BACKEND_ARCHITECTURE.md
|
||||
[repositories]: ../docs/BACKEND_REPOSITORIES.md
|
||||
[sqlalchemy]: ../docs/BACKEND_SQLALCHEMY.md
|
||||
|
||||
@@ -0,0 +1,191 @@
|
||||
"""Check the cast guard with real Git revisions and ast-grep scans."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[5]
|
||||
SCRIPT = REPO_ROOT / "scripts" / "check_no_new_cast.py"
|
||||
SOURCE_PATH = "api/services/example.py"
|
||||
|
||||
|
||||
def git(repo: Path, *args: str) -> str:
|
||||
"""Precondition: repo supports the command. Postcondition: return stdout or raise on failure."""
|
||||
return subprocess.run(["git", *args], cwd=repo, text=True, capture_output=True, check=True).stdout.strip()
|
||||
|
||||
|
||||
def write_source(repo: Path, source: str, path: str) -> None:
|
||||
"""Precondition: repo is writable. Postcondition: path contains source and its parent directories exist."""
|
||||
target = repo / path
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
target.write_text(source)
|
||||
|
||||
|
||||
def commit(repo: Path) -> str:
|
||||
"""Precondition: repo contains changes. Postcondition: commit all changes and return the revision."""
|
||||
git(repo, "add", ".")
|
||||
git(repo, "commit", "-m", "guard fixture")
|
||||
return git(repo, "rev-parse", "HEAD")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def repo(tmp_path: Path) -> Path:
|
||||
"""Precondition: tmp_path is writable. Postcondition: return a repository with a clean baseline commit."""
|
||||
git(tmp_path, "init", "-b", "main")
|
||||
git(tmp_path, "config", "user.name", "Guard Tests")
|
||||
git(tmp_path, "config", "user.email", "guard@example.com")
|
||||
write_source(tmp_path, "value = 1\n", SOURCE_PATH)
|
||||
commit(tmp_path)
|
||||
return tmp_path
|
||||
|
||||
|
||||
def run_guard(repo: Path, *args: str) -> subprocess.CompletedProcess[str]:
|
||||
"""Precondition: Git and ast-grep are available. Postcondition: return the guard's exit code and output."""
|
||||
return subprocess.run([sys.executable, str(SCRIPT), *args], cwd=repo, text=True, capture_output=True, check=False)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"source",
|
||||
[
|
||||
"import typing\nvalue = typing.cast(str, value)\n",
|
||||
"from typing import cast\nvalue = cast(str, value)\n",
|
||||
"from typing import (Any, cast)\nvalue = cast(str, value)\n",
|
||||
"import typing as types\nvalue = types.cast(str, value)\n",
|
||||
"from typing import cast as force\nvalue = force(str, value)\n",
|
||||
"import typing_extensions\nvalue = typing_extensions.cast(str, value)\n",
|
||||
"import typing_extensions as types\nvalue = types.cast(str, value)\n",
|
||||
"from typing_extensions import cast\nvalue = cast(str, value)\n",
|
||||
"from typing_extensions import cast as force\nvalue = force(str, value)\n",
|
||||
"from typing import cast\nvalue = cast(\n str,\n value,\n)\n",
|
||||
"from typing import cast\nforce = cast\nvalue = force(str, value)\n",
|
||||
"from typing import cast as original\nforce = original\nvalue = force(str, value)\n",
|
||||
"import typing as types\nforce = types.cast\nvalue = force(str, value)\n",
|
||||
"import typing_extensions\nforce = typing_extensions.cast\nvalue = force(str, value)\n",
|
||||
"import typing_extensions as types\nforce = types.cast\nvalue = force(str, value)\n",
|
||||
"def f(value):\n from typing import cast as force\n return force(str, value)\n",
|
||||
"import typing\nvalue = typing.cast(typ=str, val=value)\n",
|
||||
],
|
||||
)
|
||||
@pytest.mark.parametrize("path", [SOURCE_PATH, "api/tests/unit_tests/test_example.py", "api/services/example.pyi"])
|
||||
def test_rejects_new_cast_calls(repo: Path, source: str, path: str) -> None:
|
||||
"""Precondition: source calls typing.cast. Postcondition: staged inspection reports the file and rule."""
|
||||
write_source(repo, source, path)
|
||||
git(repo, "add", ".")
|
||||
result = run_guard(repo, "--staged")
|
||||
assert result.returncode == 1, result.stderr
|
||||
assert f"{path}:" in result.stderr
|
||||
assert "no-new-cast" in result.stderr
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"source",
|
||||
[
|
||||
"from sqlalchemy import cast\nvalue = cast(column, String)\n",
|
||||
"from sqlalchemy import cast as convert\nvalue = convert(column, String)\n",
|
||||
"import sqlalchemy as sa\nvalue = sa.cast(column, String)\n",
|
||||
"value = converter.cast(value)\n",
|
||||
"def cast(value):\n return str(value)\nvalue = cast(value)\n",
|
||||
"# typing.cast(str, value)\nvalue = 'cast(str, value)'\n",
|
||||
"from typing import cast\nvalue: str = 'value'\n",
|
||||
"from typing import cast as force\nfrom sqlalchemy import cast\nvalue = cast(column, String)\n",
|
||||
"from external import cast as force\nvalue = force(str, value)\n",
|
||||
"from sqlalchemy import cast\nforce = cast\nvalue = force(column, String)\n",
|
||||
],
|
||||
)
|
||||
def test_allows_unrelated_casts_and_non_call_syntax(repo: Path, source: str) -> None:
|
||||
"""Precondition: source contains no typing.cast call. Postcondition: staged inspection succeeds."""
|
||||
write_source(repo, source, SOURCE_PATH)
|
||||
git(repo, "add", ".")
|
||||
result = run_guard(repo, "--staged")
|
||||
assert result.returncode == 0, result.stderr
|
||||
|
||||
|
||||
@pytest.mark.parametrize("mode", ["staged", "base"])
|
||||
@pytest.mark.parametrize("change", ["unchanged", "copy", "duplicate", "different", "move"])
|
||||
def test_preserves_only_existing_occurrences(repo: Path, mode: str, change: str) -> None:
|
||||
"""Precondition: baseline contains one cast. Postcondition: unchanged calls and identical moves pass."""
|
||||
source = "import typing\nvalue = typing.cast(str, value)\n"
|
||||
write_source(repo, source, SOURCE_PATH)
|
||||
baseline = commit(repo)
|
||||
changes = {
|
||||
"unchanged": source + "answer = 42\n",
|
||||
"copy": source,
|
||||
"duplicate": source + "other = typing.cast(str, value)\n",
|
||||
"different": source.replace("cast(str", "cast(int"),
|
||||
"move": "value = 1\n",
|
||||
}
|
||||
write_source(repo, changes[change], SOURCE_PATH)
|
||||
if change in {"copy", "move"}:
|
||||
write_source(repo, source, "api/services/moved.py")
|
||||
git(repo, "add", ".")
|
||||
if mode == "base":
|
||||
commit(repo)
|
||||
args = ("--base-rev", baseline) if mode == "base" else ("--staged",)
|
||||
result = run_guard(repo, *args)
|
||||
assert result.returncode == int(change in {"copy", "duplicate", "different"}), result.stderr
|
||||
|
||||
|
||||
def test_changed_import_exposes_a_new_cast_with_an_unchanged_call(repo: Path) -> None:
|
||||
"""Precondition: baseline calls an unrelated function. Postcondition: changing its import to typing fails."""
|
||||
source = "from external import cast as force\n\nvalue = force(str, value)\n"
|
||||
write_source(repo, source, SOURCE_PATH)
|
||||
baseline = commit(repo)
|
||||
write_source(repo, source.replace("from external", "from typing"), SOURCE_PATH)
|
||||
commit(repo)
|
||||
result = run_guard(repo, "--base-rev", baseline)
|
||||
assert result.returncode == 1, result.stderr
|
||||
|
||||
|
||||
def test_staged_mode_uses_the_index(repo: Path) -> None:
|
||||
"""Precondition: index has no cast and local source adds one. Postcondition: staged inspection succeeds."""
|
||||
write_source(repo, "value: str = 'value'\n", SOURCE_PATH)
|
||||
git(repo, "add", ".")
|
||||
write_source(repo, "import typing\nvalue = typing.cast(str, value)\n", SOURCE_PATH)
|
||||
result = run_guard(repo, "--staged")
|
||||
assert result.returncode == 0, result.stderr
|
||||
|
||||
|
||||
def test_local_mode_checks_unstaged_and_untracked_sources(repo: Path) -> None:
|
||||
"""Precondition: local sources add casts. Postcondition: local inspection reports both sources."""
|
||||
source = "import typing\nvalue = typing.cast(str, value)\n"
|
||||
write_source(repo, source, SOURCE_PATH)
|
||||
write_source(repo, source, "api/services/untracked.py")
|
||||
result = run_guard(repo)
|
||||
assert result.returncode == 1, result.stderr
|
||||
assert SOURCE_PATH in result.stderr
|
||||
assert "api/services/untracked.py" in result.stderr
|
||||
|
||||
|
||||
def test_renames_preserve_the_baseline(repo: Path) -> None:
|
||||
"""Precondition: an existing cast moves with its file. Postcondition: staged inspection preserves its baseline."""
|
||||
write_source(repo, "import typing\nvalue = typing.cast(str, value)\n", SOURCE_PATH)
|
||||
commit(repo)
|
||||
git(repo, "mv", SOURCE_PATH, "api/services/renamed file.py")
|
||||
result = run_guard(repo, "--staged")
|
||||
assert result.returncode == 0, result.stderr
|
||||
|
||||
|
||||
def test_invalid_baseline_fails(repo: Path) -> None:
|
||||
"""Precondition: the baseline does not exist. Postcondition: the guard returns a check failure."""
|
||||
result = run_guard(repo, "--base-rev", "missing-revision")
|
||||
assert result.returncode == 2
|
||||
assert result.stderr
|
||||
|
||||
|
||||
def test_lint_and_ci_run_the_guard() -> None:
|
||||
"""Precondition: repository configuration is readable. Postcondition: lint and CI include the guard and rule."""
|
||||
command = "uv run --project api python scripts/check_no_new_cast.py"
|
||||
assert command in (REPO_ROOT / "Makefile").read_text()
|
||||
style = (REPO_ROOT / ".github/workflows/style.yml").read_text()
|
||||
assert f'{command} --base-rev "$GUARD_BASE_REV"' in style
|
||||
workflow = (REPO_ROOT / ".github/workflows/main-ci.yml").read_text()
|
||||
for name in ("api", "python-style"):
|
||||
path_filter = re.search(rf"(?m)^ {name}:\n((?: - '[^']+'\n)+)", workflow)
|
||||
assert path_filter is not None
|
||||
assert "scripts/check_no_new_cast.py" in path_filter.group(1)
|
||||
assert "scripts/ast_grep_rules/no_new_cast.yml" in path_filter.group(1)
|
||||
@@ -0,0 +1,202 @@
|
||||
"""Check constructor rules with real Git revisions and ast-grep scans."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[5]
|
||||
SCRIPT = REPO_ROOT / "scripts" / "check_no_new_constructor_bypass.py"
|
||||
SOURCE_PATH = "api/services/example.py"
|
||||
|
||||
|
||||
def git(repo: Path, *args: str) -> str:
|
||||
"""Precondition: repo supports the Git command. Postcondition: return stdout or raise on failure."""
|
||||
return subprocess.run(["git", *args], cwd=repo, text=True, capture_output=True, check=True).stdout.strip()
|
||||
|
||||
|
||||
def write_source(repo: Path, source: str, path: str) -> None:
|
||||
"""Precondition: repo is writable. Postcondition: path contains source and its parent directories exist."""
|
||||
target = repo / path
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
target.write_text(source)
|
||||
|
||||
|
||||
def commit(repo: Path) -> str:
|
||||
"""Precondition: repo contains changes. Postcondition: commit all changes and return the revision."""
|
||||
git(repo, "add", ".")
|
||||
git(repo, "commit", "-m", "guard fixture")
|
||||
return git(repo, "rev-parse", "HEAD")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def repo(tmp_path: Path) -> Path:
|
||||
"""Precondition: tmp_path is writable. Postcondition: return a repository with a clean baseline commit."""
|
||||
git(tmp_path, "init", "-b", "main")
|
||||
git(tmp_path, "config", "user.name", "Guard Tests")
|
||||
git(tmp_path, "config", "user.email", "guard@example.com")
|
||||
write_source(tmp_path, "value = 1\n", SOURCE_PATH)
|
||||
commit(tmp_path)
|
||||
return tmp_path
|
||||
|
||||
|
||||
def run_guard(repo: Path, *args: str) -> subprocess.CompletedProcess[str]:
|
||||
"""Precondition: Git and ast-grep are available. Postcondition: return the guard's exit code and output."""
|
||||
return subprocess.run(
|
||||
[sys.executable, str(SCRIPT), *args],
|
||||
cwd=repo,
|
||||
text=True,
|
||||
capture_output=True,
|
||||
check=False,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"source",
|
||||
[
|
||||
"value = object.__new__(Model)\n",
|
||||
"value = Model.__new__(Model)\n",
|
||||
"value = package.Model.__new__(package.Model)\n",
|
||||
"from builtins import object as Allocator\nvalue = Allocator.__new__(Model)\n",
|
||||
"value = Model.__new__(\n Model,\n)\n",
|
||||
"allocate = Model.__new__\nvalue = allocate(Model)\n",
|
||||
"value = build(Model.__new__)\n",
|
||||
"class Model:\n def build(cls):\n return object.__new__(cls)\n",
|
||||
"class Model:\n def __new__(cls):\n return Other.__new__(Other)\n",
|
||||
"class Model:\n def __new__(cls):\n allocate = object.__new__\n return allocate(cls)\n",
|
||||
"class Model:\n def __new__(cls):\n def helper():\n return object.__new__(cls)\n",
|
||||
"class Model:\n def __new__(cls):\n return lambda: object.__new__(cls)\n",
|
||||
"def __new__(cls):\n return object.__new__(cls)\n",
|
||||
"class Model:\n def __new__(cls):\n class Inner:\n value = object.__new__(cls)\n",
|
||||
],
|
||||
)
|
||||
@pytest.mark.parametrize(
|
||||
"path",
|
||||
[SOURCE_PATH, "api/tests/unit_tests/test_example.py", "api/services/example.pyi"],
|
||||
)
|
||||
def test_rejects_constructor_bypasses(repo: Path, source: str, path: str) -> None:
|
||||
"""Precondition: source bypasses construction. Postcondition: staged inspection reports its file and rule."""
|
||||
write_source(repo, source, path)
|
||||
git(repo, "add", ".")
|
||||
result = run_guard(repo, "--staged")
|
||||
assert result.returncode == 1, result.stderr
|
||||
assert f"{path}:" in result.stderr
|
||||
assert "no-new-constructor-bypass" in result.stderr
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"source",
|
||||
[
|
||||
"value = Model(config)\n",
|
||||
"# object.__new__(Model)\nvalue = 'Model.__new__(Model)'\n",
|
||||
"class Model:\n def __new__(cls):\n return super().__new__(cls)\n",
|
||||
"class Model:\n def __new__(cls):\n return object.__new__(cls)\n",
|
||||
"class Model:\n def __new__(klass):\n return Base.__new__(klass)\n",
|
||||
"class Model(str):\n def __new__(cls, value: str) -> Model:\n return str.__new__(cls, value)\n",
|
||||
"class Model:\n def __new__(cls: type[Model]) -> Model:\n return object.__new__(cls)\n",
|
||||
"class Model:\n def __new__(cls, /):\n return super().__new__(cls)\n",
|
||||
"class Model:\n def __new__(cls):\n return super(Model, cls).__new__(cls)\n",
|
||||
"class Outer:\n class Model:\n def __new__(cls):\n return object.__new__(cls)\n",
|
||||
],
|
||||
)
|
||||
def test_allows_constructors_and_allocation_in_new(repo: Path, source: str) -> None:
|
||||
"""Precondition: source uses permitted construction. Postcondition: staged inspection succeeds."""
|
||||
write_source(repo, source, SOURCE_PATH)
|
||||
git(repo, "add", ".")
|
||||
result = run_guard(repo, "--staged")
|
||||
assert result.returncode == 0, result.stderr
|
||||
|
||||
|
||||
@pytest.mark.parametrize("change", ["unchanged", "copy", "duplicate", "different", "move"])
|
||||
@pytest.mark.parametrize("mode", ["staged", "base"])
|
||||
def test_preserves_only_existing_occurrences(repo: Path, change: str, mode: str) -> None:
|
||||
"""Precondition: baseline contains one bypass.
|
||||
|
||||
Postcondition: only unchanged occurrences and identical moves pass.
|
||||
"""
|
||||
source = "def build():\n return object.__new__(Model)\n"
|
||||
write_source(repo, source, SOURCE_PATH)
|
||||
baseline = commit(repo)
|
||||
changed = {
|
||||
"unchanged": source + "answer = 42\n",
|
||||
"copy": source,
|
||||
"duplicate": source + "other = object.__new__(Model)\n",
|
||||
"different": source.replace("Model", "Other"),
|
||||
"move": "value = 1\n",
|
||||
}
|
||||
write_source(repo, changed[change], SOURCE_PATH)
|
||||
if change in {"copy", "move"}:
|
||||
write_source(repo, source, "api/services/moved.py")
|
||||
git(repo, "add", ".")
|
||||
if mode == "base":
|
||||
commit(repo)
|
||||
args = ("--base-rev", baseline) if mode == "base" else ("--staged",)
|
||||
result = run_guard(repo, *args)
|
||||
assert result.returncode == int(change in {"copy", "duplicate", "different"}), result.stderr
|
||||
|
||||
|
||||
def test_detects_a_scope_change_without_a_call_change(repo: Path) -> None:
|
||||
"""Precondition: baseline allocation is permitted. Postcondition: renaming its owning method causes rejection."""
|
||||
source = "class Model:\n def __new__(cls):\n return object.__new__(cls)\n"
|
||||
write_source(repo, source, SOURCE_PATH)
|
||||
baseline = commit(repo)
|
||||
write_source(repo, source.replace("def __new__", "def build"), SOURCE_PATH)
|
||||
commit(repo)
|
||||
result = run_guard(repo, "--base-rev", baseline)
|
||||
assert result.returncode == 1, result.stderr
|
||||
|
||||
|
||||
def test_staged_mode_uses_the_index(repo: Path) -> None:
|
||||
"""Precondition: index is valid and working source bypasses construction.
|
||||
|
||||
Postcondition: staged inspection succeeds.
|
||||
"""
|
||||
write_source(repo, "value = Model()\n", SOURCE_PATH)
|
||||
git(repo, "add", ".")
|
||||
write_source(repo, "value = object.__new__(Model)\n", SOURCE_PATH)
|
||||
result = run_guard(repo, "--staged")
|
||||
assert result.returncode == 0, result.stderr
|
||||
|
||||
|
||||
def test_local_mode_checks_unstaged_and_untracked_sources(repo: Path) -> None:
|
||||
"""Precondition: local sources introduce bypasses. Postcondition: local inspection reports both sources."""
|
||||
write_source(repo, "value = object.__new__(Model)\n", SOURCE_PATH)
|
||||
write_source(repo, "value = Model.__new__(Model)\n", "api/services/untracked.py")
|
||||
result = run_guard(repo)
|
||||
assert result.returncode == 1, result.stderr
|
||||
assert SOURCE_PATH in result.stderr
|
||||
assert "api/services/untracked.py" in result.stderr
|
||||
|
||||
|
||||
def test_renames_preserve_the_baseline(repo: Path) -> None:
|
||||
"""Precondition: an existing bypass moves with its file. Postcondition: staged inspection preserves the baseline."""
|
||||
write_source(repo, "value = object.__new__(Model)\n", SOURCE_PATH)
|
||||
commit(repo)
|
||||
git(repo, "mv", SOURCE_PATH, "api/services/renamed file.py")
|
||||
result = run_guard(repo, "--staged")
|
||||
assert result.returncode == 0, result.stderr
|
||||
|
||||
|
||||
def test_invalid_baseline_fails(repo: Path) -> None:
|
||||
"""Precondition: the baseline does not exist. Postcondition: the guard returns a check failure."""
|
||||
result = run_guard(repo, "--base-rev", "missing-revision")
|
||||
assert result.returncode == 2
|
||||
assert result.stderr
|
||||
|
||||
|
||||
def test_lint_and_ci_run_the_guard() -> None:
|
||||
"""Precondition: repository configuration is readable. Postcondition: lint and CI include the guard and its rule."""
|
||||
command = "uv run --project api python scripts/check_no_new_constructor_bypass.py"
|
||||
assert command in (REPO_ROOT / "Makefile").read_text()
|
||||
style = (REPO_ROOT / ".github/workflows/style.yml").read_text()
|
||||
assert f'{command} --base-rev "$GUARD_BASE_REV"' in style
|
||||
workflow = (REPO_ROOT / ".github/workflows/main-ci.yml").read_text()
|
||||
for name in ("api", "python-style"):
|
||||
path_filter = re.search(rf"(?m)^ {name}:\n((?: - '[^']+'\n)+)", workflow)
|
||||
assert path_filter is not None
|
||||
assert "scripts/check_no_new_constructor_bypass.py" in path_filter.group(1)
|
||||
assert "scripts/ast_grep_rules/no_new_constructor_bypass.yml" in path_filter.group(1)
|
||||
@@ -0,0 +1,84 @@
|
||||
id: no-new-cast
|
||||
language: python
|
||||
severity: error
|
||||
message: Use type narrowing or validate values instead of typing.cast().
|
||||
utils:
|
||||
typing-import:
|
||||
kind: import_from_statement
|
||||
has:
|
||||
field: module_name
|
||||
regex: '^(typing|typing_extensions)$'
|
||||
cast-reference:
|
||||
any:
|
||||
- pattern: typing.cast
|
||||
- pattern: typing_extensions.cast
|
||||
- all:
|
||||
- pattern: $MODULE.cast
|
||||
- inside:
|
||||
kind: module
|
||||
stopBy: end
|
||||
has:
|
||||
stopBy: end
|
||||
all:
|
||||
- any:
|
||||
- pattern:
|
||||
context: import typing as $MODULE
|
||||
selector: aliased_import
|
||||
- pattern:
|
||||
context: import typing_extensions as $MODULE
|
||||
selector: aliased_import
|
||||
- has:
|
||||
field: name
|
||||
regex: '^(typing|typing_extensions)$'
|
||||
- all:
|
||||
- kind: identifier
|
||||
- pattern: cast
|
||||
- inside:
|
||||
kind: module
|
||||
stopBy: end
|
||||
has:
|
||||
stopBy: end
|
||||
all:
|
||||
- matches: typing-import
|
||||
- has:
|
||||
stopBy: neighbor
|
||||
all:
|
||||
- kind: dotted_name
|
||||
- regex: '^cast$'
|
||||
- all:
|
||||
- kind: identifier
|
||||
- pattern: $IMPORTED
|
||||
- inside:
|
||||
kind: module
|
||||
stopBy: end
|
||||
has:
|
||||
stopBy: end
|
||||
all:
|
||||
- pattern:
|
||||
context: from typing import cast as $IMPORTED
|
||||
selector: aliased_import
|
||||
- inside:
|
||||
matches: typing-import
|
||||
stopBy: neighbor
|
||||
rule:
|
||||
all:
|
||||
- kind: call
|
||||
- any:
|
||||
- has:
|
||||
field: function
|
||||
matches: cast-reference
|
||||
- all:
|
||||
- has:
|
||||
field: function
|
||||
kind: identifier
|
||||
pattern: $ALIAS
|
||||
- inside:
|
||||
kind: module
|
||||
stopBy: end
|
||||
has:
|
||||
stopBy: end
|
||||
all:
|
||||
- pattern: $ALIAS = $ORIGINAL
|
||||
- has:
|
||||
field: right
|
||||
matches: cast-reference
|
||||
@@ -0,0 +1,56 @@
|
||||
id: no-new-constructor-bypass
|
||||
language: python
|
||||
severity: error
|
||||
message: Use the class constructor to preserve initialization and validation.
|
||||
utils:
|
||||
allocation-in-new:
|
||||
all:
|
||||
- any:
|
||||
- pattern: $OWNER.__new__($CLS)
|
||||
- pattern: $OWNER.__new__($CLS, $$$TAIL)
|
||||
- inside:
|
||||
all:
|
||||
- kind: function_definition
|
||||
- has:
|
||||
field: name
|
||||
regex: '^__new__$'
|
||||
- has:
|
||||
field: parameters
|
||||
has:
|
||||
all:
|
||||
- nthChild: 1
|
||||
- any:
|
||||
- all:
|
||||
- kind: typed_parameter
|
||||
- has:
|
||||
pattern: $CLS
|
||||
stopBy: neighbor
|
||||
- pattern: $CLS
|
||||
stopBy: neighbor
|
||||
- inside:
|
||||
kind: class_definition
|
||||
stopBy:
|
||||
any:
|
||||
- kind: function_definition
|
||||
- kind: lambda
|
||||
- kind: class_definition
|
||||
stopBy:
|
||||
any:
|
||||
- kind: function_definition
|
||||
- kind: lambda
|
||||
- kind: class_definition
|
||||
rule:
|
||||
any:
|
||||
- all:
|
||||
- pattern: $OWNER.__new__($$$ARGS)
|
||||
- not:
|
||||
matches: allocation-in-new
|
||||
- all:
|
||||
- pattern: $OWNER.__new__
|
||||
- not:
|
||||
inside:
|
||||
kind: call
|
||||
has:
|
||||
field: function
|
||||
pattern: $OWNER.__new__
|
||||
stopBy: neighbor
|
||||
@@ -0,0 +1,87 @@
|
||||
"""Reject new typing.cast and typing_extensions.cast calls in API Python files.
|
||||
|
||||
Preconditions: Run from the repository root with Git and ast-grep available.
|
||||
Use --staged for index contents or --base-rev for committed revisions.
|
||||
Without either option, inspect local API changes against the main merge base.
|
||||
|
||||
Postconditions: Identical existing calls retain their Git baseline.
|
||||
Copies and changed calls fail. Import aliases and direct assignment aliases
|
||||
are included. Other libraries' cast functions are outside this rule's scope.
|
||||
This syntactic rule does not resolve arbitrary assignments or shadowed names.
|
||||
Exit codes are 0 for success, 1 for violations, and 2 for check failures.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import sys
|
||||
from collections import Counter
|
||||
from pathlib import Path
|
||||
|
||||
from ast_grep_guard import (
|
||||
Violation,
|
||||
filter_moved_matches,
|
||||
match_key,
|
||||
print_violations,
|
||||
rule_path,
|
||||
run_ast_grep,
|
||||
)
|
||||
from check_no_new_base_model import working_file_versions
|
||||
from check_no_new_test_doubles import changed_file_versions
|
||||
|
||||
|
||||
def find_new_cast_calls(
|
||||
versions: dict[str, tuple[str, str]],
|
||||
) -> list[Violation]:
|
||||
"""Compare cast calls in complete source versions.
|
||||
|
||||
Preconditions: Each key is a repository-relative path with old and new source.
|
||||
Postconditions: Return API violations without changing files or source versions.
|
||||
Each removed occurrence offsets one identical addition. Scan failures raise.
|
||||
"""
|
||||
rule = rule_path("no_new_cast.yml")
|
||||
removed: Counter[str] = Counter()
|
||||
added = []
|
||||
for path, (old_source, new_source) in versions.items():
|
||||
if Path(path).is_relative_to("api"):
|
||||
existing = Counter(match_key(match) for match in run_ast_grep(old_source, rule=rule))
|
||||
for match in run_ast_grep(new_source, rule=rule):
|
||||
key = match_key(match)
|
||||
if existing[key]:
|
||||
existing[key] -= 1
|
||||
else:
|
||||
added.append((path, match))
|
||||
removed.update(existing)
|
||||
return [
|
||||
Violation(
|
||||
path,
|
||||
match.line_number,
|
||||
"no-new-cast: use type narrowing or validate values instead of typing.cast().",
|
||||
)
|
||||
for path, match in filter_moved_matches(added, removed)
|
||||
]
|
||||
|
||||
|
||||
def main() -> int:
|
||||
"""Precondition: Git and ast-grep are available in the repository root.
|
||||
|
||||
Postconditions: Report violations without changing repository files.
|
||||
Return 0 for success, 1 for violations, or 2 for check failures.
|
||||
"""
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
source = parser.add_mutually_exclusive_group()
|
||||
source.add_argument("--base-rev", help="Compare this revision with committed HEAD.")
|
||||
source.add_argument("--staged", action="store_true", help="Compare staged changes with HEAD.")
|
||||
args = parser.parse_args()
|
||||
try:
|
||||
versions = changed_file_versions(args) if args.base_rev or args.staged else working_file_versions()
|
||||
violations = find_new_cast_calls(versions)
|
||||
except (OSError, RuntimeError) as error:
|
||||
sys.stderr.write(f"{error}\n")
|
||||
return 2
|
||||
print_violations(sorted(violations, key=lambda item: (item.path, item.line_number)))
|
||||
return int(bool(violations))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,88 @@
|
||||
"""Reject new __new__ calls and references that bypass constructor checks.
|
||||
|
||||
Preconditions: Run from the repository root with Git and ast-grep available.
|
||||
Use --staged for index contents or --base-rev for committed revisions.
|
||||
Without either option, inspect local API changes against the main merge base.
|
||||
|
||||
Postconditions: Existing matching occurrences retain their Git baseline.
|
||||
Identical moves retain that baseline. Copies and changed calls fail.
|
||||
Allocation for the first parameter of a class's __new__ method is allowed.
|
||||
Nested functions and lambdas do not inherit that allocation permission.
|
||||
This syntactic rule does not resolve dynamic attribute access.
|
||||
Exit codes are 0 for success, 1 for violations, and 2 for check failures.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import sys
|
||||
from collections import Counter
|
||||
from pathlib import Path
|
||||
|
||||
from ast_grep_guard import (
|
||||
Violation,
|
||||
filter_moved_matches,
|
||||
match_key,
|
||||
print_violations,
|
||||
rule_path,
|
||||
run_ast_grep,
|
||||
)
|
||||
from check_no_new_base_model import working_file_versions
|
||||
from check_no_new_test_doubles import changed_file_versions
|
||||
|
||||
|
||||
def find_new_constructor_bypasses(
|
||||
versions: dict[str, tuple[str, str]],
|
||||
) -> list[Violation]:
|
||||
"""Compare constructor access in complete source versions.
|
||||
|
||||
Preconditions: Each key is a repository-relative path with old and new source.
|
||||
Postconditions: Return API violations without changing files or source versions.
|
||||
Each removed occurrence offsets one identical addition. Scan failures raise.
|
||||
"""
|
||||
rule = rule_path("no_new_constructor_bypass.yml")
|
||||
removed: Counter[str] = Counter()
|
||||
added = []
|
||||
for path, (old_source, new_source) in versions.items():
|
||||
if Path(path).is_relative_to("api"):
|
||||
existing = Counter(match_key(match) for match in run_ast_grep(old_source, rule=rule))
|
||||
for match in run_ast_grep(new_source, rule=rule):
|
||||
key = match_key(match)
|
||||
if existing[key]:
|
||||
existing[key] -= 1
|
||||
else:
|
||||
added.append((path, match))
|
||||
removed.update(existing)
|
||||
return [
|
||||
Violation(
|
||||
path,
|
||||
match.line_number,
|
||||
"no-new-constructor-bypass: use the class constructor to preserve initialization and validation.",
|
||||
)
|
||||
for path, match in filter_moved_matches(added, removed)
|
||||
]
|
||||
|
||||
|
||||
def main() -> int:
|
||||
"""Precondition: Git and ast-grep are available in the repository root.
|
||||
|
||||
Postconditions: Report violations without changing repository files.
|
||||
Return 0 for success, 1 for violations, or 2 for check failures.
|
||||
"""
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
source = parser.add_mutually_exclusive_group()
|
||||
source.add_argument("--base-rev", help="Compare this revision with committed HEAD.")
|
||||
source.add_argument("--staged", action="store_true", help="Compare staged changes with HEAD.")
|
||||
args = parser.parse_args()
|
||||
try:
|
||||
versions = changed_file_versions(args) if args.base_rev or args.staged else working_file_versions()
|
||||
violations = find_new_constructor_bypasses(versions)
|
||||
except (OSError, RuntimeError) as error:
|
||||
sys.stderr.write(f"{error}\n")
|
||||
return 2
|
||||
print_violations(sorted(violations, key=lambda item: (item.path, item.line_number)))
|
||||
return int(bool(violations))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in new issue
Block a user