first commit
This commit is contained in:
eeymoo
@@ -0,0 +1,128 @@
|
||||
"""Dataset builder: apply mutation rules to samples and persist Ground Truth."""
|
||||
|
||||
import random
|
||||
from collections import Counter
|
||||
from pathlib import Path
|
||||
from typing import List, Optional
|
||||
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.dataset.diff_extractor import _detect_language, extract_diff_bundle, bundle_to_sample_dict
|
||||
from app.dataset.git_parser import select_candidates
|
||||
from app.dataset.rules.registry import get_registry
|
||||
from app.models import Sample, Defect
|
||||
|
||||
|
||||
def build_dataset_from_repo(
|
||||
db: Session,
|
||||
repo_path: str | Path,
|
||||
count: int = 12,
|
||||
languages: Optional[List[str]] = None,
|
||||
) -> List[Sample]:
|
||||
"""Build a dataset from a Git repository with injected defects.
|
||||
|
||||
Only candidates that actually receive a mutation are kept: every sample
|
||||
in the dataset must carry Ground Truth, so candidates whose changed files
|
||||
match no rule are skipped and the scan continues until ``count`` mutated
|
||||
samples have been collected.
|
||||
"""
|
||||
languages = languages or ["python", "java", "javascript"]
|
||||
registry = get_registry()
|
||||
candidates = select_candidates(repo_path, count=max(count * 5, 30), languages=languages)
|
||||
|
||||
samples = []
|
||||
type_counts: Counter = Counter()
|
||||
for candidate in candidates:
|
||||
if len(samples) >= count:
|
||||
break
|
||||
bundle = extract_diff_bundle(repo_path, candidate.sha)
|
||||
primary_lang = _infer_primary_language(bundle.after_files or bundle.before_files, languages)
|
||||
sample_dict = bundle_to_sample_dict(bundle, primary_language=primary_lang)
|
||||
sample = Sample(**sample_dict)
|
||||
|
||||
# Collect every (file, rule) mutation opportunity for this commit.
|
||||
# Rules parse and mutate a single source file, so each changed file
|
||||
# of the primary language is tried individually; line numbers are
|
||||
# shifted into the coordinate space of the joined source stored on
|
||||
# the sample. Among all opportunities, prefer the defect type that is
|
||||
# currently least represented in this build, so frequent patterns
|
||||
# (e.g. `&&` swaps) do not dominate the dataset.
|
||||
after_source = _join_source(bundle.after_files)
|
||||
matches = []
|
||||
if primary_lang in registry.all_rules():
|
||||
rules = list(registry.rules_for(primary_lang))
|
||||
random.Random(candidate.sha).shuffle(rules)
|
||||
line_offset = 0
|
||||
for path, content in (bundle.after_files or {}).items():
|
||||
if _detect_language(path) != primary_lang:
|
||||
line_offset += len(content.splitlines()) + 2
|
||||
continue
|
||||
for rule in rules:
|
||||
m = rule.detect_and_mutate(content, filename=path)
|
||||
if m:
|
||||
m.line_start += line_offset
|
||||
m.line_end += line_offset
|
||||
matches.append((path, m))
|
||||
line_offset += len(content.splitlines()) + 2
|
||||
|
||||
if not matches:
|
||||
# Skip candidates whose files match no rule: samples without
|
||||
# Ground Truth are useless for the experiment.
|
||||
continue
|
||||
|
||||
rng = random.Random(candidate.sha)
|
||||
min_count = min(type_counts[m.defect_type] for _, m in matches)
|
||||
best = [(p, m) for p, m in matches if type_counts[m.defect_type] == min_count]
|
||||
path, mutation = rng.choice(best)
|
||||
type_counts[mutation.defect_type] += 1
|
||||
|
||||
files = dict(bundle.after_files)
|
||||
files[path] = mutation.mutated_source
|
||||
mutated_joined = _join_source(files)
|
||||
mutation.description = f"{mutation.description} (file: {path})"
|
||||
|
||||
sample.diff = _compute_diff_from_mutated(after_source, mutated_joined)
|
||||
sample.after_context = {"mutated": mutated_joined}
|
||||
defect = Defect(
|
||||
sample=sample,
|
||||
defect_type=mutation.defect_type,
|
||||
language=mutation.language,
|
||||
line_start=mutation.line_start,
|
||||
line_end=mutation.line_end,
|
||||
description=mutation.description,
|
||||
reference_fix=mutation.reference_fix,
|
||||
)
|
||||
sample.defects.append(defect)
|
||||
|
||||
db.add(sample)
|
||||
samples.append(sample)
|
||||
|
||||
db.commit()
|
||||
for sample in samples:
|
||||
db.refresh(sample)
|
||||
return samples
|
||||
|
||||
|
||||
def _infer_primary_language(files: dict, languages: List[str]) -> str:
|
||||
from app.dataset.diff_extractor import _detect_language
|
||||
counts = {}
|
||||
for path in files:
|
||||
lang = _detect_language(path)
|
||||
if lang in languages:
|
||||
counts[lang] = counts.get(lang, 0) + 1
|
||||
if counts:
|
||||
return max(counts, key=counts.get)
|
||||
return languages[0]
|
||||
|
||||
|
||||
def _join_source(files: dict) -> str:
|
||||
return "\n\n".join(files.values())
|
||||
|
||||
|
||||
def _compute_diff_from_mutated(original: str, mutated: str) -> str:
|
||||
"""Produce a simple unified-diff-like string from original and mutated."""
|
||||
import difflib
|
||||
|
||||
orig_lines = original.splitlines(keepends=True)
|
||||
mut_lines = mutated.splitlines(keepends=True)
|
||||
return "".join(difflib.unified_diff(orig_lines, mut_lines, lineterm=""))
|
||||
@@ -0,0 +1,94 @@
|
||||
"""Unidiff extraction with before/after context."""
|
||||
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
import git
|
||||
from git import Repo
|
||||
|
||||
|
||||
@dataclass
|
||||
class DiffBundle:
|
||||
repo: str
|
||||
commit_sha: str
|
||||
diff: str
|
||||
before_files: dict
|
||||
after_files: dict
|
||||
|
||||
|
||||
def _detect_language(filename: str) -> str:
|
||||
ext = Path(filename).suffix.lower()
|
||||
mapping = {
|
||||
".py": "python",
|
||||
".java": "java",
|
||||
".js": "javascript",
|
||||
".ts": "javascript",
|
||||
".jsx": "javascript",
|
||||
".tsx": "javascript",
|
||||
}
|
||||
return mapping.get(ext, "unknown")
|
||||
|
||||
|
||||
def extract_diff_bundle(
|
||||
repo_path: str,
|
||||
commit_sha: str,
|
||||
parent_index: int = 0,
|
||||
) -> DiffBundle:
|
||||
"""Extract unified diff and before/after file snapshots for a commit."""
|
||||
repo = Repo(str(repo_path))
|
||||
commit = repo.commit(commit_sha)
|
||||
parents = commit.parents
|
||||
if parents:
|
||||
base = parents[parent_index]
|
||||
diff = base.diff(commit, create_patch=True, unified=3)
|
||||
else:
|
||||
# Initial commit: diff against empty tree
|
||||
diff = commit.diff(git.Git(repo).hash_object("/dev/null", t=None), create_patch=True, unified=3)
|
||||
|
||||
diff_text = "\n".join(d.diff.decode("utf-8", errors="replace") for d in diff if d.diff)
|
||||
|
||||
before_files = {}
|
||||
after_files = {}
|
||||
for d in diff:
|
||||
a_path = d.a_path or d.b_path
|
||||
b_path = d.b_path or d.a_path
|
||||
if a_path and d.a_blob:
|
||||
try:
|
||||
before_files[a_path] = d.a_blob.data_stream.read().decode("utf-8", errors="replace")
|
||||
except Exception:
|
||||
before_files[a_path] = ""
|
||||
if b_path and d.b_blob:
|
||||
try:
|
||||
after_files[b_path] = d.b_blob.data_stream.read().decode("utf-8", errors="replace")
|
||||
except Exception:
|
||||
after_files[b_path] = ""
|
||||
|
||||
return DiffBundle(
|
||||
repo=Path(repo_path).name,
|
||||
commit_sha=commit_sha,
|
||||
diff=diff_text,
|
||||
before_files=before_files,
|
||||
after_files=after_files,
|
||||
)
|
||||
|
||||
|
||||
def bundle_to_sample_dict(bundle: DiffBundle, primary_language: Optional[str] = None) -> dict:
|
||||
"""Convert a DiffBundle into a dict matching the Sample schema."""
|
||||
if not primary_language:
|
||||
# Infer from changed files
|
||||
exts = {Path(p).suffix.lower() for p in bundle.after_files or bundle.before_files}
|
||||
for ext, lang in {".py": "python", ".java": "java", ".js": "javascript"}.items():
|
||||
if ext in exts:
|
||||
primary_language = lang
|
||||
break
|
||||
primary_language = primary_language or "unknown"
|
||||
|
||||
return {
|
||||
"repo": bundle.repo,
|
||||
"commit_sha": bundle.commit_sha,
|
||||
"language": primary_language,
|
||||
"diff": bundle.diff,
|
||||
"before_context": bundle.before_files,
|
||||
"after_context": bundle.after_files,
|
||||
}
|
||||
@@ -0,0 +1,121 @@
|
||||
"""Git repository parsing and commit candidate selection."""
|
||||
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import List, Optional
|
||||
|
||||
from git import Repo
|
||||
|
||||
|
||||
@dataclass
|
||||
class CommitCandidate:
|
||||
repo: str
|
||||
sha: str
|
||||
message: str
|
||||
author: str
|
||||
date: str
|
||||
stats: dict
|
||||
files: List[dict]
|
||||
|
||||
|
||||
def list_commits(
|
||||
repo_path: str | Path,
|
||||
max_count: Optional[int] = None,
|
||||
reverse: bool = True,
|
||||
) -> List[CommitCandidate]:
|
||||
"""List commits from a Git repository."""
|
||||
repo = Repo(str(repo_path))
|
||||
repo_name = Path(repo_path).name
|
||||
commits = []
|
||||
iterator = list(repo.iter_commits())
|
||||
if reverse:
|
||||
iterator = reversed(iterator)
|
||||
for commit in iterator:
|
||||
if max_count and len(commits) >= max_count:
|
||||
break
|
||||
stats = commit.stats.total
|
||||
files = []
|
||||
for item in commit.stats.files.items():
|
||||
filename, file_stats = item
|
||||
files.append(
|
||||
{
|
||||
"path": filename,
|
||||
"insertions": file_stats["insertions"],
|
||||
"deletions": file_stats["deletions"],
|
||||
"lines": file_stats["lines"],
|
||||
}
|
||||
)
|
||||
commits.append(
|
||||
CommitCandidate(
|
||||
repo=repo_name,
|
||||
sha=commit.hexsha,
|
||||
message=commit.message.strip(),
|
||||
author=str(commit.author),
|
||||
date=commit.committed_datetime.isoformat(),
|
||||
stats=stats,
|
||||
files=files,
|
||||
)
|
||||
)
|
||||
return commits
|
||||
|
||||
|
||||
def score_commit(commit: CommitCandidate) -> float:
|
||||
"""Score a commit by size, message quality, and language diversity."""
|
||||
total_lines = commit.stats.get("lines", 0)
|
||||
# Prefer moderate size: ~50-200 lines ideal
|
||||
size_score = 1.0 - abs(total_lines - 125) / 200.0
|
||||
size_score = max(0.0, min(1.0, size_score))
|
||||
|
||||
# Message quality: length and presence of verb/noun clues
|
||||
msg = commit.message.lower()
|
||||
msg_score = min(1.0, len(commit.message) / 40.0)
|
||||
if any(k in msg for k in ("fix", "bug", "refactor", "feature", "add", "update")):
|
||||
msg_score = min(1.0, msg_score + 0.2)
|
||||
|
||||
# Language diversity bonus based on file extensions
|
||||
exts = {Path(f["path"]).suffix.lower() for f in commit.files if Path(f["path"]).suffix}
|
||||
diversity_score = min(1.0, len(exts) / 3.0)
|
||||
|
||||
return size_score * 0.5 + msg_score * 0.3 + diversity_score * 0.2
|
||||
|
||||
|
||||
def select_candidates(
|
||||
repo_path: str | Path,
|
||||
count: int = 12,
|
||||
languages: Optional[List[str]] = None,
|
||||
scan_limit: int = 300,
|
||||
) -> List[CommitCandidate]:
|
||||
"""Select top-scoring commits, optionally balanced by language.
|
||||
|
||||
Only the most recent ``scan_limit`` commits are scanned: computing
|
||||
per-commit stats spawns a git subprocess each time, so scanning the full
|
||||
history of a large repository is prohibitively slow.
|
||||
"""
|
||||
languages = languages or ["python", "java", "javascript"]
|
||||
commits = list_commits(repo_path, max_count=scan_limit, reverse=False)
|
||||
scored = [(c, score_commit(c)) for c in commits]
|
||||
scored.sort(key=lambda x: x[1], reverse=True)
|
||||
|
||||
# Simple balancing: prefer at least one commit per target language when detectable
|
||||
by_lang = {lang: [] for lang in languages}
|
||||
others = []
|
||||
for commit, score in scored:
|
||||
ext_set = {Path(f["path"]).suffix.lower() for f in commit.files}
|
||||
placed = False
|
||||
for lang in languages:
|
||||
hint = ".py" if lang == "python" else ".java" if lang == "java" else ".js"
|
||||
if hint in ext_set:
|
||||
by_lang[lang].append((commit, score))
|
||||
placed = True
|
||||
break
|
||||
if not placed:
|
||||
others.append((commit, score))
|
||||
|
||||
result = []
|
||||
per_lang = max(1, count // len(languages))
|
||||
for lang in languages:
|
||||
result.extend(by_lang[lang][:per_lang])
|
||||
result.extend(others)
|
||||
result = result[:count]
|
||||
result.sort(key=lambda x: x[1], reverse=True)
|
||||
return [c for c, _ in result]
|
||||
@@ -0,0 +1,30 @@
|
||||
"""Base class for pluggable defect mutation rules."""
|
||||
|
||||
from abc import ABC, abstractmethod
|
||||
from dataclasses import dataclass
|
||||
from typing import List, Optional
|
||||
|
||||
|
||||
@dataclass
|
||||
class Mutation:
|
||||
defect_type: str
|
||||
language: str
|
||||
line_start: int
|
||||
line_end: int
|
||||
mutated_source: str
|
||||
reference_fix: str
|
||||
description: str
|
||||
|
||||
|
||||
class MutationRule(ABC):
|
||||
name: str = ""
|
||||
language: str = ""
|
||||
defect_type: str = ""
|
||||
|
||||
@abstractmethod
|
||||
def detect_and_mutate(self, source: str, filename: str = "") -> Optional[Mutation]:
|
||||
"""Return a mutation if the rule can introduce a defect, else None."""
|
||||
...
|
||||
|
||||
def _line_for_position(self, source: str, position: int) -> int:
|
||||
return source[:position].count("\n") + 1
|
||||
@@ -0,0 +1,57 @@
|
||||
"""AST-level boundary error injection for Java using javalang."""
|
||||
|
||||
from typing import Optional
|
||||
|
||||
import javalang
|
||||
|
||||
from app.dataset.rules.base import Mutation, MutationRule
|
||||
|
||||
|
||||
class JavaBoundaryErrorRule(MutationRule):
|
||||
"""Mutate array length boundary from `<` to `<=`.
|
||||
|
||||
Uses `javalang` to locate a binary comparison against `.length` and flips
|
||||
the operator to introduce an off-by-one access.
|
||||
"""
|
||||
|
||||
name = "java_boundary_error"
|
||||
language = "java"
|
||||
defect_type = "boundary_condition_error"
|
||||
|
||||
def detect_and_mutate(self, source: str, filename: str = "") -> Optional[Mutation]:
|
||||
try:
|
||||
tree = javalang.parse.parse(source)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
for path, node in tree:
|
||||
if not isinstance(node, javalang.tree.BinaryOperation):
|
||||
continue
|
||||
if node.operator != "<":
|
||||
continue
|
||||
right = node.operandr
|
||||
if isinstance(right, javalang.tree.MemberReference) and right.member == "length":
|
||||
# BinaryOperation itself may lack position; use enclosing statement
|
||||
if_statement = next((n for n in path if isinstance(n, javalang.tree.IfStatement)), None)
|
||||
pos = if_statement.position if if_statement else node.position
|
||||
if not pos:
|
||||
continue
|
||||
lines = source.splitlines(keepends=True)
|
||||
line_no = pos.line
|
||||
line = lines[line_no - 1]
|
||||
mutated_line = line.replace("< ", "<= ", 1)
|
||||
if mutated_line == line:
|
||||
mutated_line = line.replace("<", "<=", 1)
|
||||
if mutated_line == line:
|
||||
continue
|
||||
mutated = "".join(lines[:line_no - 1] + [mutated_line] + lines[line_no:])
|
||||
return Mutation(
|
||||
defect_type=self.defect_type,
|
||||
language=self.language,
|
||||
line_start=line_no,
|
||||
line_end=line_no,
|
||||
mutated_source=mutated,
|
||||
reference_fix="Use strict `< length` to avoid ArrayIndexOutOfBoundsException.",
|
||||
description="Changed array boundary check to off-by-one (<= length).",
|
||||
)
|
||||
return None
|
||||
@@ -0,0 +1,69 @@
|
||||
"""AST-level concurrency issue injection for Java using javalang."""
|
||||
|
||||
from typing import Optional
|
||||
|
||||
import javalang
|
||||
|
||||
from app.dataset.rules.base import Mutation, MutationRule
|
||||
|
||||
|
||||
class JavaConcurrencyRule(MutationRule):
|
||||
"""Remove a synchronized block to expose a race condition.
|
||||
|
||||
Uses `javalang` to locate `synchronized (lock) { ... }` and replaces it
|
||||
with the bare block body.
|
||||
"""
|
||||
|
||||
name = "java_concurrency"
|
||||
language = "java"
|
||||
defect_type = "concurrency_issue"
|
||||
|
||||
def detect_and_mutate(self, source: str, filename: str = "") -> Optional[Mutation]:
|
||||
try:
|
||||
tree = javalang.parse.parse(source)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
for path, node in tree:
|
||||
if not isinstance(node, javalang.tree.SynchronizedStatement):
|
||||
continue
|
||||
pos = node.position
|
||||
if not pos:
|
||||
continue
|
||||
lines = source.splitlines(keepends=True)
|
||||
start = pos.line
|
||||
# Estimate end by braces of the synchronized block
|
||||
end = start
|
||||
depth = 0
|
||||
for idx in range(start - 1, len(lines)):
|
||||
depth += lines[idx].count("{") - lines[idx].count("}")
|
||||
if depth > 0:
|
||||
end = idx + 1
|
||||
if depth <= 0 and idx > start - 1:
|
||||
end = idx + 1
|
||||
break
|
||||
body_lines = lines[start - 1:end]
|
||||
# drop header line and closing brace line, keep body; body is at same
|
||||
# indentation as the synchronized header minus one level
|
||||
inner = body_lines[1:-1] if len(body_lines) > 2 else []
|
||||
dedented = []
|
||||
for line in inner:
|
||||
if line.startswith(" "):
|
||||
dedented.append(" " + line[12:])
|
||||
elif line.startswith(" "):
|
||||
dedented.append(" " + line[8:])
|
||||
elif line.startswith(" "):
|
||||
dedented.append(line[4:])
|
||||
else:
|
||||
dedented.append(line)
|
||||
mutated = "".join(lines[: start - 1] + dedented + lines[end:])
|
||||
return Mutation(
|
||||
defect_type=self.defect_type,
|
||||
language=self.language,
|
||||
line_start=start,
|
||||
line_end=end,
|
||||
mutated_source=mutated,
|
||||
reference_fix="Restore `synchronized (lock)` to protect the critical section.",
|
||||
description="Removed synchronized block, exposing a race condition.",
|
||||
)
|
||||
return None
|
||||
@@ -0,0 +1,52 @@
|
||||
"""AST-level logical operator misuse injection for Java using javalang."""
|
||||
|
||||
from typing import Optional
|
||||
|
||||
import javalang
|
||||
|
||||
from app.dataset.rules.base import Mutation, MutationRule
|
||||
|
||||
|
||||
class JavaLogicOperatorRule(MutationRule):
|
||||
"""Swap `&&` with `||` in a boolean expression.
|
||||
|
||||
Uses `javalang` to locate a binary operation with `&&` and replaces the
|
||||
operator with `||`.
|
||||
"""
|
||||
|
||||
name = "java_logic_operator"
|
||||
language = "java"
|
||||
defect_type = "logic_operator_misuse"
|
||||
|
||||
def detect_and_mutate(self, source: str, filename: str = "") -> Optional[Mutation]:
|
||||
try:
|
||||
tree = javalang.parse.parse(source)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
for path, node in tree:
|
||||
if not isinstance(node, javalang.tree.BinaryOperation):
|
||||
continue
|
||||
if node.operator != "&&":
|
||||
continue
|
||||
return_statement = next((n for n in path if isinstance(n, javalang.tree.ReturnStatement)), None)
|
||||
pos = return_statement.position if return_statement else node.position
|
||||
if not pos:
|
||||
continue
|
||||
lines = source.splitlines(keepends=True)
|
||||
line_no = pos.line
|
||||
line = lines[line_no - 1]
|
||||
mutated_line = line.replace("&&", "||", 1)
|
||||
if mutated_line == line:
|
||||
continue
|
||||
mutated = "".join(lines[:line_no - 1] + [mutated_line] + lines[line_no:])
|
||||
return Mutation(
|
||||
defect_type=self.defect_type,
|
||||
language=self.language,
|
||||
line_start=line_no,
|
||||
line_end=line_no,
|
||||
mutated_source=mutated,
|
||||
reference_fix="Restore `&&` for correct conjunction semantics.",
|
||||
description="Replaced boolean `&&` with `||`.",
|
||||
)
|
||||
return None
|
||||
@@ -0,0 +1,75 @@
|
||||
"""AST-level null-pointer injection for Java using javalang."""
|
||||
|
||||
from typing import Optional
|
||||
|
||||
import javalang
|
||||
|
||||
from app.dataset.rules.base import Mutation, MutationRule
|
||||
|
||||
|
||||
class JavaNoneReferenceRule(MutationRule):
|
||||
"""Remove a null-check guard in Java source.
|
||||
|
||||
Uses the pure-Python `javalang` parser to locate an `if (x != null)` guard
|
||||
and remove it, leaving the dereference unprotected. This keeps mutation
|
||||
semantics precise without regex/text replacement.
|
||||
"""
|
||||
|
||||
name = "java_none_reference"
|
||||
language = "java"
|
||||
defect_type = "null_pointer"
|
||||
|
||||
def detect_and_mutate(self, source: str, filename: str = "") -> Optional[Mutation]:
|
||||
try:
|
||||
tree = javalang.parse.parse(source)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
for path, node in tree:
|
||||
if not isinstance(node, javalang.tree.IfStatement):
|
||||
continue
|
||||
cond = node.condition
|
||||
# Match: x != null
|
||||
if (
|
||||
isinstance(cond, javalang.tree.BinaryOperation)
|
||||
and cond.operator == "!="
|
||||
and isinstance(cond.operandr, javalang.tree.Literal)
|
||||
and cond.operandr.value == "null"
|
||||
):
|
||||
var_name = getattr(cond.operandl, "member", str(cond.operandl))
|
||||
lines = source.splitlines(keepends=True)
|
||||
start = node.position.line if node.position else 1
|
||||
# Estimate end line by finding matching brace (simplistic)
|
||||
end = start
|
||||
depth = 0
|
||||
for idx in range(start - 1, len(lines)):
|
||||
depth += lines[idx].count("{") - lines[idx].count("}")
|
||||
if depth > 0:
|
||||
end = idx + 1
|
||||
if depth <= 0 and idx > start - 1:
|
||||
end = idx + 1
|
||||
break
|
||||
body_lines = lines[start - 1:end]
|
||||
# keep body lines between header and closing brace, dedent one level
|
||||
inner = body_lines[1:-1] if len(body_lines) > 2 else []
|
||||
dedented = []
|
||||
for line in inner:
|
||||
if line.startswith(" "):
|
||||
dedented.append(" " + line[12:])
|
||||
elif line.startswith(" "):
|
||||
dedented.append(" " + line[8:])
|
||||
elif line.startswith(" "):
|
||||
dedented.append(line[4:])
|
||||
else:
|
||||
dedented.append(line)
|
||||
mutated = "".join(lines[: start - 1] + dedented + lines[end:])
|
||||
return Mutation(
|
||||
defect_type=self.defect_type,
|
||||
language=self.language,
|
||||
line_start=start,
|
||||
line_end=end,
|
||||
mutated_source=mutated,
|
||||
reference_fix=f"Add `if ({var_name} != null)` guard before dereferencing.",
|
||||
description=f"Removed null-check guard for '{var_name}'.",
|
||||
)
|
||||
return None
|
||||
@@ -0,0 +1,50 @@
|
||||
"""AST-level resource leak injection for Java using javalang."""
|
||||
|
||||
from typing import Optional
|
||||
|
||||
import javalang
|
||||
|
||||
from app.dataset.rules.base import Mutation, MutationRule
|
||||
|
||||
|
||||
class JavaResourceLeakRule(MutationRule):
|
||||
"""Replace try-with-resources with a plain try block, leaking the resource.
|
||||
|
||||
Uses `javalang` to locate a try-with-resources statement and removes the
|
||||
resource specification, leaving the stream unclosed.
|
||||
"""
|
||||
|
||||
name = "java_resource_leak"
|
||||
language = "java"
|
||||
defect_type = "resource_not_closed"
|
||||
|
||||
def detect_and_mutate(self, source: str, filename: str = "") -> Optional[Mutation]:
|
||||
try:
|
||||
tree = javalang.parse.parse(source)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
for path, node in tree:
|
||||
if not isinstance(node, javalang.tree.TryStatement):
|
||||
continue
|
||||
if not node.resources:
|
||||
continue
|
||||
pos = node.position
|
||||
if not pos:
|
||||
continue
|
||||
lines = source.splitlines(keepends=True)
|
||||
start = pos.line
|
||||
# Find the resource clause line e.g. try (BufferedReader br = ...)
|
||||
resource_line = lines[start - 1]
|
||||
new_header = resource_line.split("(", 1)[0].rstrip() + " {\n"
|
||||
mutated = "".join(lines[: start - 1] + [new_header] + lines[start:])
|
||||
return Mutation(
|
||||
defect_type=self.defect_type,
|
||||
language=self.language,
|
||||
line_start=start,
|
||||
line_end=start,
|
||||
mutated_source=mutated,
|
||||
reference_fix="Use try-with-resources or explicitly close the stream in finally.",
|
||||
description="Removed try-with-resources, leaking the acquired resource.",
|
||||
)
|
||||
return None
|
||||
@@ -0,0 +1,63 @@
|
||||
"""AST-level boundary error injection for JavaScript using esprima."""
|
||||
|
||||
from typing import Optional
|
||||
|
||||
import esprima
|
||||
|
||||
from app.dataset.rules.base import Mutation, MutationRule
|
||||
|
||||
|
||||
class JSBoundaryErrorRule(MutationRule):
|
||||
"""Mutate array length boundary from `<` to `<=`.
|
||||
|
||||
Uses `esprima` to locate a binary expression comparing against `.length`
|
||||
and flips the operator.
|
||||
"""
|
||||
|
||||
name = "js_boundary_error"
|
||||
language = "javascript"
|
||||
defect_type = "boundary_condition_error"
|
||||
|
||||
def detect_and_mutate(self, source: str, filename: str = "") -> Optional[Mutation]:
|
||||
# Modern JS is usually ESM: try module grammar first, then script.
|
||||
try:
|
||||
tree = esprima.parseModule(source, loc=True)
|
||||
except Exception:
|
||||
try:
|
||||
tree = esprima.parseScript(source, loc=True)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
for node in self._walk(tree):
|
||||
if node.type != "BinaryExpression" or node.operator != "<":
|
||||
continue
|
||||
right = node.right
|
||||
if right.type == "MemberExpression" and getattr(right.property, "name", None) == "length":
|
||||
line_no = node.loc.start.line
|
||||
lines = source.splitlines(keepends=True)
|
||||
line = lines[line_no - 1]
|
||||
mutated_line = line.replace("<", "<=", 1)
|
||||
if mutated_line == line:
|
||||
continue
|
||||
mutated = "".join(lines[:line_no - 1] + [mutated_line] + lines[line_no:])
|
||||
return Mutation(
|
||||
defect_type=self.defect_type,
|
||||
language=self.language,
|
||||
line_start=line_no,
|
||||
line_end=line_no,
|
||||
mutated_source=mutated,
|
||||
reference_fix="Use strict `< length` to avoid out-of-bounds access.",
|
||||
description="Changed array boundary check to off-by-one (<= length).",
|
||||
)
|
||||
return None
|
||||
|
||||
def _walk(self, node):
|
||||
yield node
|
||||
for key in getattr(node, "__dict__", {}):
|
||||
child = getattr(node, key)
|
||||
if isinstance(child, list):
|
||||
for item in child:
|
||||
if hasattr(item, "type"):
|
||||
yield from self._walk(item)
|
||||
elif hasattr(child, "type"):
|
||||
yield from self._walk(child)
|
||||
@@ -0,0 +1,60 @@
|
||||
"""AST-level concurrency issue injection for JavaScript using esprima."""
|
||||
|
||||
from typing import Optional
|
||||
|
||||
import esprima
|
||||
|
||||
from app.dataset.rules.base import Mutation, MutationRule
|
||||
|
||||
|
||||
class JSConcurrencyRule(MutationRule):
|
||||
"""Remove an `await mutex.acquire()` / `mutex.release()` pair.
|
||||
|
||||
Uses `esprima` to locate a try block followed by a finally that releases a
|
||||
mutex and removes the finally/release, exposing a race.
|
||||
"""
|
||||
|
||||
name = "js_concurrency"
|
||||
language = "javascript"
|
||||
defect_type = "concurrency_issue"
|
||||
|
||||
def detect_and_mutate(self, source: str, filename: str = "") -> Optional[Mutation]:
|
||||
# Modern JS is usually ESM: try module grammar first, then script.
|
||||
try:
|
||||
tree = esprima.parseModule(source, loc=True)
|
||||
except Exception:
|
||||
try:
|
||||
tree = esprima.parseScript(source, loc=True)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
def walk(node):
|
||||
yield node
|
||||
for key in node.__dict__:
|
||||
child = getattr(node, key)
|
||||
if isinstance(child, list):
|
||||
for item in child:
|
||||
if hasattr(item, "type"):
|
||||
yield from walk(item)
|
||||
elif hasattr(child, "type"):
|
||||
yield from walk(child)
|
||||
|
||||
for node in walk(tree):
|
||||
if node.type != "TryStatement" or not node.finalizer:
|
||||
continue
|
||||
start = node.loc.start.line
|
||||
end = node.finalizer.loc.end.line
|
||||
lines = source.splitlines(keepends=True)
|
||||
# Drop the entire finally block; the try block ends on the same line as finally starts
|
||||
finally_start = node.finalizer.loc.start.line - 1
|
||||
mutated = "".join(lines[:finally_start] + [" }\n"] + lines[end:])
|
||||
return Mutation(
|
||||
defect_type=self.defect_type,
|
||||
language=self.language,
|
||||
line_start=start,
|
||||
line_end=end,
|
||||
mutated_source=mutated,
|
||||
reference_fix="Restore mutex release in finally to protect the critical section.",
|
||||
description="Removed mutex release, exposing a race condition.",
|
||||
)
|
||||
return None
|
||||
@@ -0,0 +1,61 @@
|
||||
"""AST-level logical operator misuse injection for JavaScript using esprima."""
|
||||
|
||||
from typing import Optional
|
||||
|
||||
import esprima
|
||||
|
||||
from app.dataset.rules.base import Mutation, MutationRule
|
||||
|
||||
|
||||
class JSLogicOperatorRule(MutationRule):
|
||||
"""Swap `&&` with `||` in a boolean expression.
|
||||
|
||||
Uses `esprima` to locate a LogicalExpression using `&&` and replaces the
|
||||
operator with `||`.
|
||||
"""
|
||||
|
||||
name = "js_logic_operator"
|
||||
language = "javascript"
|
||||
defect_type = "logic_operator_misuse"
|
||||
|
||||
def detect_and_mutate(self, source: str, filename: str = "") -> Optional[Mutation]:
|
||||
# Modern JS is usually ESM: try module grammar first, then script.
|
||||
try:
|
||||
tree = esprima.parseModule(source, loc=True)
|
||||
except Exception:
|
||||
try:
|
||||
tree = esprima.parseScript(source, loc=True)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
for node in self._walk(tree):
|
||||
if node.type != "LogicalExpression" or node.operator != "&&":
|
||||
continue
|
||||
line_no = node.loc.start.line
|
||||
lines = source.splitlines(keepends=True)
|
||||
line = lines[line_no - 1]
|
||||
mutated_line = line.replace("&&", "||", 1)
|
||||
if mutated_line == line:
|
||||
continue
|
||||
mutated = "".join(lines[:line_no - 1] + [mutated_line] + lines[line_no:])
|
||||
return Mutation(
|
||||
defect_type=self.defect_type,
|
||||
language=self.language,
|
||||
line_start=line_no,
|
||||
line_end=line_no,
|
||||
mutated_source=mutated,
|
||||
reference_fix="Restore `&&` for correct short-circuit logic.",
|
||||
description="Replaced boolean `&&` with `||`.",
|
||||
)
|
||||
return None
|
||||
|
||||
def _walk(self, node):
|
||||
yield node
|
||||
for key in getattr(node, "__dict__", {}):
|
||||
child = getattr(node, key)
|
||||
if isinstance(child, list):
|
||||
for item in child:
|
||||
if hasattr(item, "type"):
|
||||
yield from self._walk(item)
|
||||
elif hasattr(child, "type"):
|
||||
yield from self._walk(child)
|
||||
@@ -0,0 +1,76 @@
|
||||
"""AST-level null-pointer injection for JavaScript using esprima."""
|
||||
|
||||
from typing import Optional
|
||||
|
||||
import esprima
|
||||
|
||||
from app.dataset.rules.base import Mutation, MutationRule
|
||||
|
||||
|
||||
class JSNoneReferenceRule(MutationRule):
|
||||
"""Remove a `if (x !== null)` guard in JavaScript.
|
||||
|
||||
Uses the Python port of `esprima` to locate the guard statement and
|
||||
replaces it with the body, leaving a potential null dereference.
|
||||
"""
|
||||
|
||||
name = "js_none_reference"
|
||||
language = "javascript"
|
||||
defect_type = "null_pointer"
|
||||
|
||||
def detect_and_mutate(self, source: str, filename: str = "") -> Optional[Mutation]:
|
||||
# Modern JS is usually ESM: try module grammar first, then script.
|
||||
try:
|
||||
tree = esprima.parseModule(source, loc=True)
|
||||
except Exception:
|
||||
try:
|
||||
tree = esprima.parseScript(source, loc=True)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
def walk(node):
|
||||
yield node
|
||||
for key in node.__dict__:
|
||||
child = getattr(node, key)
|
||||
if isinstance(child, list):
|
||||
for item in child:
|
||||
if hasattr(item, "type"):
|
||||
yield from walk(item)
|
||||
elif hasattr(child, "type"):
|
||||
yield from walk(child)
|
||||
|
||||
for node in walk(tree):
|
||||
if node.type != "IfStatement":
|
||||
continue
|
||||
cond = node.test
|
||||
if (
|
||||
cond.type == "BinaryExpression"
|
||||
and cond.operator == "!=="
|
||||
and cond.right.type == "Literal"
|
||||
and cond.right.value is None
|
||||
):
|
||||
var_name = getattr(cond.left, "name", str(cond.left))
|
||||
start = node.loc.start.line
|
||||
end = node.consequent.loc.end.line
|
||||
lines = source.splitlines(keepends=True)
|
||||
# Drop guard header and closing brace, keep body (1-based -> 0-based)
|
||||
body_start = node.consequent.loc.start.line
|
||||
body_end = node.consequent.loc.end.line - 1
|
||||
body_lines = lines[body_start:body_end]
|
||||
dedented = []
|
||||
for line in body_lines:
|
||||
if line.startswith(" "):
|
||||
dedented.append(line[4:])
|
||||
else:
|
||||
dedented.append(line)
|
||||
mutated = "".join(lines[: start - 1] + dedented + lines[end:])
|
||||
return Mutation(
|
||||
defect_type=self.defect_type,
|
||||
language=self.language,
|
||||
line_start=start,
|
||||
line_end=end,
|
||||
mutated_source=mutated,
|
||||
reference_fix=f"Add `if ({var_name} !== null)` guard before dereferencing.",
|
||||
description=f"Removed null-check guard for '{var_name}'.",
|
||||
)
|
||||
return None
|
||||
@@ -0,0 +1,60 @@
|
||||
"""AST-level resource leak injection for JavaScript using esprima."""
|
||||
|
||||
from typing import Optional
|
||||
|
||||
import esprima
|
||||
|
||||
from app.dataset.rules.base import Mutation, MutationRule
|
||||
|
||||
|
||||
class JSResourceLeakRule(MutationRule):
|
||||
"""Remove a fetch Response body close/usage, leaking the reader.
|
||||
|
||||
Uses `esprima` to locate a `try/finally` that closes a reader and removes
|
||||
the finally block.
|
||||
"""
|
||||
|
||||
name = "js_resource_leak"
|
||||
language = "javascript"
|
||||
defect_type = "resource_not_closed"
|
||||
|
||||
def detect_and_mutate(self, source: str, filename: str = "") -> Optional[Mutation]:
|
||||
# Modern JS is usually ESM: try module grammar first, then script.
|
||||
try:
|
||||
tree = esprima.parseModule(source, loc=True)
|
||||
except Exception:
|
||||
try:
|
||||
tree = esprima.parseScript(source, loc=True)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
def walk(node):
|
||||
yield node
|
||||
for key in node.__dict__:
|
||||
child = getattr(node, key)
|
||||
if isinstance(child, list):
|
||||
for item in child:
|
||||
if hasattr(item, "type"):
|
||||
yield from walk(item)
|
||||
elif hasattr(child, "type"):
|
||||
yield from walk(child)
|
||||
|
||||
for node in walk(tree):
|
||||
if node.type != "TryStatement" or not node.finalizer:
|
||||
continue
|
||||
start = node.loc.start.line
|
||||
end = node.finalizer.loc.end.line
|
||||
lines = source.splitlines(keepends=True)
|
||||
# Drop the finally block entirely, close the try block
|
||||
finally_start = node.finalizer.loc.start.line - 1
|
||||
mutated = "".join(lines[:finally_start] + [" }\n"] + lines[end:])
|
||||
return Mutation(
|
||||
defect_type=self.defect_type,
|
||||
language=self.language,
|
||||
line_start=start,
|
||||
line_end=end,
|
||||
mutated_source=mutated,
|
||||
reference_fix="Restore finally block to close/release resources.",
|
||||
description="Removed finally block, leaving resource unreleased.",
|
||||
)
|
||||
return None
|
||||
@@ -0,0 +1,47 @@
|
||||
"""Custom rule demonstrating pluggable extensibility (A1b)."""
|
||||
|
||||
import ast
|
||||
from typing import Optional
|
||||
|
||||
from app.dataset.rules.base import Mutation, MutationRule
|
||||
|
||||
|
||||
class UnusedVariableRule(MutationRule):
|
||||
"""A custom rule: replace a variable read with an undefined name.
|
||||
|
||||
This is intentionally simple and demonstrates that adding a new file under
|
||||
app/dataset/rules/<language>/ is enough to register a rule.
|
||||
"""
|
||||
|
||||
name = "unused_variable_demo"
|
||||
language = "python"
|
||||
defect_type = "custom_demo"
|
||||
|
||||
def detect_and_mutate(self, source: str, filename: str = "") -> Optional[Mutation]:
|
||||
try:
|
||||
tree = ast.parse(source)
|
||||
except SyntaxError:
|
||||
return None
|
||||
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, ast.FunctionDef) and node.body:
|
||||
first = node.body[0]
|
||||
if isinstance(first, ast.Assign) and isinstance(first.targets[0], ast.Name):
|
||||
var_name = first.targets[0].id
|
||||
line_no = first.lineno
|
||||
lines = source.splitlines(keepends=True)
|
||||
line = lines[line_no - 1]
|
||||
mutated_line = line.replace(var_name, "__undefined_" + var_name, 1)
|
||||
if mutated_line == line:
|
||||
continue
|
||||
mutated = "".join(lines[:line_no - 1] + [mutated_line] + lines[line_no:])
|
||||
return Mutation(
|
||||
defect_type=self.defect_type,
|
||||
language=self.language,
|
||||
line_start=line_no,
|
||||
line_end=line_no,
|
||||
mutated_source=mutated,
|
||||
reference_fix=f"Use the original variable name '{var_name}'.",
|
||||
description=f"Custom rule: replaced '{var_name}' with an undefined placeholder.",
|
||||
)
|
||||
return None
|
||||
@@ -0,0 +1,50 @@
|
||||
"""AST-level boundary condition injection for Python."""
|
||||
|
||||
import ast
|
||||
from typing import Optional
|
||||
|
||||
from app.dataset.rules.base import Mutation, MutationRule
|
||||
|
||||
|
||||
class BoundaryErrorRule(MutationRule):
|
||||
"""Mutate a list-index boundary check from `< len(seq)` to `<= len(seq)`.
|
||||
|
||||
Uses `ast` to find comparisons guarding index access and flips the operator
|
||||
so the boundary becomes off-by-one.
|
||||
"""
|
||||
|
||||
name = "boundary_error"
|
||||
language = "python"
|
||||
defect_type = "boundary_condition_error"
|
||||
|
||||
def detect_and_mutate(self, source: str, filename: str = "") -> Optional[Mutation]:
|
||||
try:
|
||||
tree = ast.parse(source)
|
||||
except SyntaxError:
|
||||
return None
|
||||
|
||||
for node in ast.walk(tree):
|
||||
if not isinstance(node, ast.If):
|
||||
continue
|
||||
test = node.test
|
||||
if isinstance(test, ast.Compare) and isinstance(test.left, ast.Name):
|
||||
if len(test.ops) == 1 and isinstance(test.ops[0], ast.Lt):
|
||||
# i < len(x) -> i <= len(x)
|
||||
comparator = test.comparators[0]
|
||||
if isinstance(comparator, ast.Call) and isinstance(comparator.func, ast.Name) and comparator.func.id == "len":
|
||||
lines = source.splitlines(keepends=True)
|
||||
line = lines[node.lineno - 1]
|
||||
mutated_line = line.replace("< len(", "<= len(", 1)
|
||||
if mutated_line == line:
|
||||
continue
|
||||
mutated = "".join(lines[: node.lineno - 1] + [mutated_line] + lines[node.lineno:])
|
||||
return Mutation(
|
||||
defect_type=self.defect_type,
|
||||
language=self.language,
|
||||
line_start=node.lineno,
|
||||
line_end=getattr(node, "end_lineno", node.lineno) or node.lineno,
|
||||
mutated_source=mutated,
|
||||
reference_fix="Use strict `< len(seq)` to avoid index-out-of-range.",
|
||||
description="Changed index boundary check to off-by-one (<= len).",
|
||||
)
|
||||
return None
|
||||
@@ -0,0 +1,52 @@
|
||||
"""AST-level concurrency safety injection for Python."""
|
||||
|
||||
import ast
|
||||
from typing import Optional
|
||||
|
||||
from app.dataset.rules.base import Mutation, MutationRule
|
||||
|
||||
|
||||
class ConcurrencyRule(MutationRule):
|
||||
"""Remove a threading.Lock.acquire/release pair to introduce race condition.
|
||||
|
||||
Uses `ast` to find a with-statement using a lock and replaces it with the
|
||||
bare body, removing synchronization.
|
||||
"""
|
||||
|
||||
name = "concurrency"
|
||||
language = "python"
|
||||
defect_type = "concurrency_issue"
|
||||
|
||||
def detect_and_mutate(self, source: str, filename: str = "") -> Optional[Mutation]:
|
||||
try:
|
||||
tree = ast.parse(source)
|
||||
except SyntaxError:
|
||||
return None
|
||||
|
||||
for node in ast.walk(tree):
|
||||
if not isinstance(node, ast.With):
|
||||
continue
|
||||
first_item = node.items[0]
|
||||
ctx = first_item.context_expr
|
||||
if isinstance(ctx, ast.Call) and isinstance(ctx.func, ast.Attribute) and ctx.func.attr == "acquire":
|
||||
start = node.lineno
|
||||
end = getattr(node, "end_lineno", node.lineno) or node.lineno
|
||||
lines = source.splitlines(keepends=True)
|
||||
body = lines[start:end]
|
||||
dedented = []
|
||||
for line in body[1:]:
|
||||
if line.startswith(" "):
|
||||
dedented.append(line[4:])
|
||||
else:
|
||||
dedented.append(line)
|
||||
mutated = "".join(lines[: start - 1] + dedented + lines[end:])
|
||||
return Mutation(
|
||||
defect_type=self.defect_type,
|
||||
language=self.language,
|
||||
line_start=start,
|
||||
line_end=end,
|
||||
mutated_source=mutated,
|
||||
reference_fix="Restore `with lock:` to protect the critical section.",
|
||||
description="Removed lock acquisition, exposing a race condition.",
|
||||
)
|
||||
return None
|
||||
@@ -0,0 +1,44 @@
|
||||
"""AST-level logical operator misuse injection for Python."""
|
||||
|
||||
import ast
|
||||
from typing import Optional
|
||||
|
||||
from app.dataset.rules.base import Mutation, MutationRule
|
||||
|
||||
|
||||
class LogicOperatorRule(MutationRule):
|
||||
"""Swap `and` with `or` in a boolean expression.
|
||||
|
||||
Uses `ast` to locate a BoolOp using `And` and replaces it with `Or`,
|
||||
preserving exact source position via line replacement.
|
||||
"""
|
||||
|
||||
name = "logic_operator"
|
||||
language = "python"
|
||||
defect_type = "logic_operator_misuse"
|
||||
|
||||
def detect_and_mutate(self, source: str, filename: str = "") -> Optional[Mutation]:
|
||||
try:
|
||||
tree = ast.parse(source)
|
||||
except SyntaxError:
|
||||
return None
|
||||
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, ast.BoolOp) and isinstance(node.op, ast.And):
|
||||
line_no = node.lineno
|
||||
lines = source.splitlines(keepends=True)
|
||||
line = lines[line_no - 1]
|
||||
mutated_line = line.replace(" and ", " or ", 1)
|
||||
if mutated_line == line:
|
||||
continue
|
||||
mutated = "".join(lines[:line_no - 1] + [mutated_line] + lines[line_no:])
|
||||
return Mutation(
|
||||
defect_type=self.defect_type,
|
||||
language=self.language,
|
||||
line_start=line_no,
|
||||
line_end=getattr(node, "end_lineno", line_no) or line_no,
|
||||
mutated_source=mutated,
|
||||
reference_fix="Restore the original `and` operator for correct short-circuit logic.",
|
||||
description="Replaced boolean `and` with `or`.",
|
||||
)
|
||||
return None
|
||||
@@ -0,0 +1,65 @@
|
||||
"""AST-level null-pointer / None-reference injection for Python."""
|
||||
|
||||
import ast
|
||||
from typing import Optional
|
||||
|
||||
from app.dataset.rules.base import Mutation, MutationRule
|
||||
|
||||
|
||||
class NoneReferenceRule(MutationRule):
|
||||
"""Replace a checked variable access with an unchecked None dereference.
|
||||
|
||||
This rule uses the standard library `ast` module to precisely locate a
|
||||
variable that is used after an `if x is not None:` guard, then removes the
|
||||
guard. The mutation position is derived from AST line numbers so it is
|
||||
exact and reproducible.
|
||||
"""
|
||||
|
||||
name = "none_reference"
|
||||
language = "python"
|
||||
defect_type = "null_pointer"
|
||||
|
||||
def detect_and_mutate(self, source: str, filename: str = "") -> Optional[Mutation]:
|
||||
try:
|
||||
tree = ast.parse(source)
|
||||
except SyntaxError:
|
||||
return None
|
||||
|
||||
for node in ast.walk(tree):
|
||||
if not isinstance(node, ast.If):
|
||||
continue
|
||||
test = node.test
|
||||
# Match: if x is not None:
|
||||
if (
|
||||
isinstance(test, ast.Compare)
|
||||
and isinstance(test.left, ast.Name)
|
||||
and len(test.ops) == 1
|
||||
and isinstance(test.ops[0], ast.IsNot)
|
||||
and len(test.comparators) == 1
|
||||
and isinstance(test.comparators[0], ast.Constant)
|
||||
and test.comparators[0].value is None
|
||||
):
|
||||
var_name = test.left.id
|
||||
lines = source.splitlines(keepends=True)
|
||||
start = node.lineno
|
||||
end = getattr(node, "end_lineno", node.lineno) or node.lineno
|
||||
body_lines = lines[start:end]
|
||||
dedented = []
|
||||
for line in body_lines[1:]:
|
||||
if line.startswith(" "):
|
||||
dedented.append(line[4:])
|
||||
elif line.startswith("\t"):
|
||||
dedented.append(line[1:])
|
||||
else:
|
||||
dedented.append(line)
|
||||
mutated = "".join(lines[: start - 1] + dedented + lines[end:])
|
||||
return Mutation(
|
||||
defect_type=self.defect_type,
|
||||
language=self.language,
|
||||
line_start=start,
|
||||
line_end=end,
|
||||
mutated_source=mutated,
|
||||
reference_fix=f"Add `if {var_name} is not None:` guard before use.",
|
||||
description=f"Removed None-check guard for variable '{var_name}'.",
|
||||
)
|
||||
return None
|
||||
@@ -0,0 +1,57 @@
|
||||
"""AST-level resource leak injection for Python."""
|
||||
|
||||
import ast
|
||||
from typing import Optional
|
||||
|
||||
from app.dataset.rules.base import Mutation, MutationRule
|
||||
|
||||
|
||||
class ResourceLeakRule(MutationRule):
|
||||
"""Convert a `with open(...)` block into an unclosed `open(...).read()`.
|
||||
|
||||
Uses `ast` to locate a with-statement managing a file resource and replaces
|
||||
it with a direct call chain that leaks the file handle.
|
||||
"""
|
||||
|
||||
name = "resource_leak"
|
||||
language = "python"
|
||||
defect_type = "resource_not_closed"
|
||||
|
||||
def detect_and_mutate(self, source: str, filename: str = "") -> Optional[Mutation]:
|
||||
try:
|
||||
tree = ast.parse(source)
|
||||
except SyntaxError:
|
||||
return None
|
||||
|
||||
for node in ast.walk(tree):
|
||||
if not isinstance(node, ast.With):
|
||||
continue
|
||||
first_item = node.items[0]
|
||||
ctx = first_item.context_expr
|
||||
if isinstance(ctx, ast.Call) and isinstance(ctx.func, ast.Name) and ctx.func.id == "open":
|
||||
start = node.lineno
|
||||
end = getattr(node, "end_lineno", node.lineno) or node.lineno
|
||||
lines = source.splitlines(keepends=True)
|
||||
# Keep the with header expression but replace 'with open(...)' by 'f = open(...)'
|
||||
header = lines[start - 1]
|
||||
header_expr = header.split("with ", 1)[1].split(" as ", 1)[0].strip().rstrip(":\n")
|
||||
var = header.split(" as ", 1)[1].strip().rstrip(":\n") if " as " in header else "f"
|
||||
body = lines[start:end]
|
||||
dedented = []
|
||||
for line in body[1:]:
|
||||
if line.startswith(" "):
|
||||
dedented.append(line[4:])
|
||||
else:
|
||||
dedented.append(line)
|
||||
replacement = [f"{var} = {header_expr}\n"] + dedented
|
||||
mutated = "".join(lines[: start - 1] + replacement + lines[end:])
|
||||
return Mutation(
|
||||
defect_type=self.defect_type,
|
||||
language=self.language,
|
||||
line_start=start,
|
||||
line_end=end,
|
||||
mutated_source=mutated,
|
||||
reference_fix="Use `with open(...) as f:` to ensure the file is closed.",
|
||||
description="Replaced context-managed open() with an unclosed file handle.",
|
||||
)
|
||||
return None
|
||||
@@ -0,0 +1,49 @@
|
||||
"""Registry that auto-discovers mutation rules from the rules directory."""
|
||||
|
||||
import importlib
|
||||
import inspect
|
||||
from pathlib import Path
|
||||
from typing import Dict, List, Type
|
||||
|
||||
from app.dataset.rules.base import MutationRule
|
||||
|
||||
|
||||
class RuleRegistry:
|
||||
def __init__(self, rules_dir: Path):
|
||||
self.rules_dir = rules_dir
|
||||
self._rules: Dict[str, List[MutationRule]] = {}
|
||||
|
||||
def discover(self) -> None:
|
||||
"""Scan rules directory and register all MutationRule subclasses."""
|
||||
self._rules.clear()
|
||||
for lang_dir in self.rules_dir.iterdir():
|
||||
if not lang_dir.is_dir():
|
||||
continue
|
||||
for py_file in lang_dir.glob("*.py"):
|
||||
if py_file.name.startswith("_"):
|
||||
continue
|
||||
module_name = f"app.dataset.rules.{lang_dir.name}.{py_file.stem}"
|
||||
try:
|
||||
module = importlib.import_module(module_name)
|
||||
except Exception:
|
||||
continue
|
||||
for _, obj in inspect.getmembers(module, inspect.isclass):
|
||||
if (
|
||||
issubclass(obj, MutationRule)
|
||||
and obj is not MutationRule
|
||||
and not getattr(obj, "__abstractmethods__", False)
|
||||
):
|
||||
rule = obj()
|
||||
self._rules.setdefault(rule.language, []).append(rule)
|
||||
|
||||
def rules_for(self, language: str) -> List[MutationRule]:
|
||||
return self._rules.get(language, [])
|
||||
|
||||
def all_rules(self) -> Dict[str, List[MutationRule]]:
|
||||
return self._rules.copy()
|
||||
|
||||
|
||||
def get_registry() -> RuleRegistry:
|
||||
registry = RuleRegistry(Path(__file__).parent)
|
||||
registry.discover()
|
||||
return registry
|
||||
Reference in New Issue
Block a user