diff --git a/.github/ISSUE_TEMPLATE/import_failure.yml b/.github/ISSUE_TEMPLATE/import_failure.yml new file mode 100644 index 0000000..60e1e58 --- /dev/null +++ b/.github/ISSUE_TEMPLATE/import_failure.yml @@ -0,0 +1,133 @@ +name: Import failure +description: Report a problem fetching a LeetCode question or generating its local files +title: "[Import failure]: " +labels: + - bug +body: + - type: markdown + attributes: + value: | + Thanks for reporting an import problem. + + **Do not include browser cookies, `LEETCODE_SESSION`, CSRF tokens, personal solution code, or unredacted local paths.** Replace usernames and private directories with placeholders such as `` or ``. + + - type: dropdown + id: operation + attributes: + label: Operation + description: Which operation failed? + options: + - leet2git get + - leet2git import-all + - Source-file generation + - Test-file generation + - Other import-related operation + validations: + required: true + + - type: input + id: problems + attributes: + label: LeetCode problem + description: Provide the public problem number and slug or URL. List every affected problem if there is more than one. + placeholder: "2917 — find-the-k-or-of-an-array" + validations: + required: true + + - type: input + id: version + attributes: + label: leet2git version + description: Paste the package version from `leet2git --version`. + placeholder: "leet2git 0.3.0" + validations: + required: true + + - type: input + id: python + attributes: + label: Python version + description: Paste the output of `python --version` or `uv run python --version`. + placeholder: "Python 3.11.13" + validations: + required: true + + - type: dropdown + id: operating-system + attributes: + label: Operating system + options: + - Linux + - macOS + - Windows + - Other + validations: + required: true + + - type: dropdown + id: language + attributes: + label: LeetCode language + options: + - python3 + - python + - javascript + - java + - cpp + - Other + validations: + required: true + + - type: textarea + id: command + attributes: + label: Command + description: Show the exact command after redacting private paths. Do not include cookie or token values. + placeholder: "leet2git --source-repository get 2917" + render: shell + validations: + required: true + + - type: textarea + id: observed + attributes: + label: Observed behavior + description: Include the complete error message and what files, if any, were generated. Redact usernames and private paths. + validations: + required: true + + - type: textarea + id: expected + attributes: + label: Expected behavior + description: Describe the files or outcome you expected. + validations: + required: true + + - type: textarea + id: reproduction + attributes: + label: Reproduction steps + description: Provide the smallest reliable sequence that reproduces the problem, including whether it happens repeatedly. + placeholder: | + 1. Log in to LeetCode in a supported local browser. + 2. Run `leet2git ...`. + 3. Observe ... + validations: + required: true + + - type: textarea + id: additional-context + attributes: + label: Additional context + description: Add sanitized logs, screenshots, or relevant configuration flags. Do not attach generated personal solutions or authentication data. + + - type: checkboxes + id: safety-checks + attributes: + label: Safety checks + options: + - label: I removed cookies, session values, CSRF tokens, personal solution code, and unredacted private paths. + required: true + - label: I searched existing issues for this problem number or error message. + required: true diff --git a/.gitignore b/.gitignore index 3229eff..86809b7 100644 --- a/.gitignore +++ b/.gitignore @@ -129,4 +129,4 @@ dmypy.json .pyre/ # vscode -.vscode/ \ No newline at end of file +.vscode/ diff --git a/README.md b/README.md index 23a0d4e..4d007fd 100644 --- a/README.md +++ b/README.md @@ -222,6 +222,21 @@ The path to the code repository - generate_tests: If true, will try to generate local test files for the question. Currently only python3 is supported. +## Limitations + +leet2git imports the problem description and source template whenever those are available, even when it cannot safely generate a local test file. These test-generation limitations are reported as soft errors and do not cancel the source import. + +Local tests are currently skipped for: + +- Problems that depend on judge-provided objects such as `TreeNode`, `ListNode`, or `NestedInteger`. +- Interactive problems that call a hidden LeetCode oracle. +- Concurrency problems that require LeetCode's scheduler or callback harness. +- Problems whose examples use custom or in-place output validation, such as checking a mutated array prefix instead of only a return value. +- Problems whose example inputs or outputs cannot be parsed into a reliable generic assertion. + +If LeetCode does not provide a source snippet for the selected language, there is no source file to import. This is common for language-specific problems such as SQL or JavaScript exercises when `python3` is selected. + +Authentication failures, network or LeetCode API errors, and local source-file write errors can still prevent an import because no usable source file can be produced. ## Language Support diff --git a/pyproject.toml b/pyproject.toml index f5784a2..e4e1832 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -28,7 +28,6 @@ dependencies = [ "beautifulsoup4>=4.14.3", "browser-cookie3>=0.20.1", "click>=8.3.1", - "gitpython>=3.1.45", "httpx>=0.28.1", "platformdirs>=4.10.0", "pydantic>=2.13.4", diff --git a/scripts/sample_leetcode_imports.py b/scripts/sample_leetcode_imports.py new file mode 100644 index 0000000..87c7cd4 --- /dev/null +++ b/scripts/sample_leetcode_imports.py @@ -0,0 +1,630 @@ +"""Randomly smoke-test public LeetCode problem imports without browser cookies. + +The sampler is deliberately sequential and jittered to avoid request bursts. It +stops on rate limiting or access blocking instead of trying to work around it. +Generated files live in a temporary directory unless --keep-artifacts is used. +""" + +import ast +import json +import math +import multiprocessing +import queue +import random +import runpy +import tempfile +import time +from collections import Counter, defaultdict +from collections.abc import Callable, Sequence +from dataclasses import asdict, dataclass, field +from pathlib import Path +from typing import Protocol + +import click + +from leet2git.config_manager import AppConfig +from leet2git.file_handler import create_file_handler +from leet2git.leetcode_client import LeetcodeAPIError, LeetcodeClient +from leet2git.leetcode_models import ProblemListResponse +from leet2git.question_db import QuestionData +from leet2git.test_harness import get_local_test_limitation + +MAX_RUNTIME_SECONDS = 3595.0 +SHUTDOWN_RESERVE_SECONDS = 4.0 +DIFFICULTY_NAMES = {1: "Easy", 2: "Medium", 3: "Hard"} +ERA_COUNT = 6 + + +@dataclass(frozen=True) +class Candidate: + """One public problem eligible for sampling.""" + + question_id: int + slug: str + difficulty: str + era: int + + +@dataclass +class ProblemResult: + """Structured outcome for one attempted problem import.""" + + question_id: int + slug: str + difficulty: str + era: int + status: str + stage: str + error_type: str = "" + message: str = "" + input_count: int = 0 + output_count: int = 0 + function_names: list[str] = field(default_factory=list) + topics: list[str] = field(default_factory=list) + test_generated: bool = False + elapsed_seconds: float = 0.0 + + @classmethod + def failure( + cls, + candidate: Candidate, + stage: str, + error: BaseException, + elapsed_seconds: float = 0.0, + ) -> "ProblemResult": + """Build a concise failure result without response bodies or source code.""" + return cls( + question_id=candidate.question_id, + slug=candidate.slug, + difficulty=candidate.difficulty, + era=candidate.era, + status="failed", + stage=stage, + error_type=type(error).__name__, + message=_clean_message(str(error)), + elapsed_seconds=elapsed_seconds, + ) + + +@dataclass(frozen=True) +class SamplerSettings: + """Validated sampler limits and pacing.""" + + percentage: float = 1.0 + max_seconds: float = MAX_RUNTIME_SECONDS + min_delay: float = 2.0 + max_delay: float = 5.0 + request_timeout: float = 15.0 + item_timeout: float = 25.0 + language: str = "python3" + + def __post_init__(self) -> None: + if not 0 < self.percentage <= 100: + raise ValueError("percentage must be greater than 0 and at most 100") + if not 1 <= self.max_seconds <= MAX_RUNTIME_SECONDS: + raise ValueError(f"max_seconds must be between 1 and {MAX_RUNTIME_SECONDS:g}") + if self.min_delay < 1: + raise ValueError("min_delay must be at least 1 second") + if self.max_delay < self.min_delay: + raise ValueError("max_delay must be greater than or equal to min_delay") + if self.request_timeout <= 0 or self.item_timeout <= 0: + raise ValueError("timeouts must be positive") + if self.language not in {"python", "python3"}: + raise ValueError("the sampler currently supports python or python3") + + +@dataclass +class SamplerReport: + """Aggregate sampler results suitable for JSON output.""" + + seed: int + percentage: float + max_seconds: float + catalog_total: int = 0 + eligible_free: int = 0 + paid_excluded: int = 0 + target_imports: int = 0 + attempted: int = 0 + imported: int = 0 + passed: int = 0 + soft_errors: int = 0 + skipped: int = 0 + failed: int = 0 + duration_seconds: float = 0.0 + stop_reason: str = "" + difficulty_coverage: dict[str, int] = field(default_factory=dict) + era_coverage: dict[str, int] = field(default_factory=dict) + topic_coverage: list[str] = field(default_factory=list) + results: list[ProblemResult] = field(default_factory=list) + + def to_dict(self) -> dict[str, object]: + """Return a JSON-safe report dictionary.""" + return asdict(self) + + def exit_code(self) -> int: + """Return 0 for complete, 1 for hard failures, or 2 for incomplete runs.""" + if self.stop_reason.startswith("catalog:") or self.imported < self.target_imports: + return 2 + if self.failed: + return 1 + return 0 + + +ProblemRunner = Callable[[Candidate, Path, str, float], ProblemResult] + + +class QuestionDataClient(Protocol): + """Client behavior needed to inspect one public problem.""" + + def get_question_data( + self, + question_id: int, + title_slug: str, + language: str, + /, + ) -> QuestionData: ... + + +class ProblemCatalogClient(Protocol): + """Client behavior needed to build the public sampler population.""" + + def get_problem_list(self) -> ProblemListResponse: ... + + +def build_population(catalog: ProblemListResponse) -> tuple[list[Candidate], int]: + """Return free problems annotated with broad ID eras plus paid exclusion count.""" + free_rows = [row for row in catalog.stat_status_pairs if not row.paid_only] + paid_count = len(catalog.stat_status_pairs) - len(free_rows) + sorted_rows = sorted(free_rows, key=lambda row: row.stat.frontend_question_id) + population: list[Candidate] = [] + seen_ids: set[int] = set() + for index, row in enumerate(sorted_rows): + question_id = row.stat.frontend_question_id + slug = row.stat.question_title_slug + if question_id <= 0 or not slug or question_id in seen_ids: + continue + seen_ids.add(question_id) + era = min(ERA_COUNT - 1, index * ERA_COUNT // max(1, len(sorted_rows))) + population.append( + Candidate( + question_id=question_id, + slug=slug, + difficulty=DIFFICULTY_NAMES.get(row.difficulty.level, "Unknown"), + era=era, + ) + ) + return population, paid_count + + +def varied_candidate_order(population: Sequence[Candidate], rng: random.Random) -> list[Candidate]: + """Shuffle within difficulty/era strata, then round-robin across those strata.""" + buckets: dict[tuple[str, int], list[Candidate]] = defaultdict(list) + for candidate in population: + buckets[(candidate.difficulty, candidate.era)].append(candidate) + for candidates in buckets.values(): + rng.shuffle(candidates) + + keys = list(buckets) + rng.shuffle(keys) + ordered: list[Candidate] = [] + while keys: + next_keys: list[tuple[str, int]] = [] + for key in keys: + candidates = buckets[key] + if candidates: + ordered.append(candidates.pop()) + if candidates: + next_keys.append(key) + rng.shuffle(next_keys) + keys = next_keys + return ordered + + +def inspect_problem( + candidate: Candidate, + output_root: Path, + language: str, + client: QuestionDataClient, +) -> ProblemResult: + """Fetch one problem and exercise callable discovery, source, and test generation.""" + started = time.monotonic() + stage = "fetch" + try: + data = client.get_question_data(candidate.question_id, candidate.slug, language) + data.language = language + if data.title_slug != candidate.slug or data.internal_id <= 0: + raise ValueError("LeetCode returned mismatched question metadata") + + config = AppConfig(language=language, source_path=str(output_root)) + handler = create_file_handler(data, config) + + soft_error_stage = "test_generation" + soft_error_type = "JudgeHarnessUnsupported" + test_limitation = get_local_test_limitation(data) + if not test_limitation: + if not data.inputs or not data.outputs: + test_limitation = "Could not parse both example inputs and outputs" + soft_error_type = "MissingExamples" + elif len(data.inputs) != len(data.outputs): + test_limitation = f"Parsed {len(data.inputs)} inputs but {len(data.outputs)} outputs" + soft_error_type = "ExampleCountMismatch" + + stage = "callable_discovery" + try: + data.function_name = handler.get_function_name() + except Exception as error: + if not test_limitation: + test_limitation = f"Could not identify the callable: {_clean_message(str(error))}" + soft_error_stage = stage + soft_error_type = "CallableDiscoveryError" + + if test_limitation: + data.requires_custom_test_harness = True + + stage = "source_generation" + data.file_path = str(handler.generate_source()) + source_path = _safe_generated_path(output_root, data.file_path) + source_symbols = _validate_python_file(source_path, execute=True) + + result = ProblemResult( + question_id=candidate.question_id, + slug=candidate.slug, + difficulty=candidate.difficulty, + era=candidate.era, + status="passed", + stage="complete", + input_count=len(data.inputs), + output_count=len(data.outputs), + function_names=list(data.function_name), + topics=sorted(tag.slug for tag in data.categories if tag.slug), + ) + + stage = "test_generation" + if test_limitation: + result.status = "soft_error" + result.stage = soft_error_stage + result.error_type = soft_error_type + result.message = test_limitation + else: + try: + data.test_file_path = handler.generate_tests() + test_path = _safe_generated_path(output_root, data.test_file_path) + _validate_python_file(test_path, execute=True) + if len(data.function_name) > 1 and data.function_name[0] not in source_symbols: + raise ValueError( + f'Generated constructor "{data.function_name[0]}" is not module-level' + ) + result.test_generated = True + except Exception as error: + result.status = "soft_error" + result.stage = stage + result.error_type = type(error).__name__ + result.message = _clean_message(str(error)) + + result.elapsed_seconds = round(time.monotonic() - started, 3) + return result + except LeetcodeAPIError as error: + result = ProblemResult.failure(candidate, stage, error, round(time.monotonic() - started, 3)) + if f'"{language}" code snippet' in str(error): + result.status = "skipped" + result.stage = "snippet" + return result + except Exception as error: + return ProblemResult.failure(candidate, stage, error, round(time.monotonic() - started, 3)) + + +def run_problem_in_worker( + candidate: Candidate, + output_root: Path, + language: str, + timeout_seconds: float, +) -> ProblemResult: + """Run one import in a killable subprocess so the global deadline remains bounded.""" + context = multiprocessing.get_context("spawn") + result_queue = context.Queue(maxsize=1) + process = context.Process( + target=_problem_worker, + args=(candidate, output_root, language, timeout_seconds, result_queue), + ) + process.start() + process.join(timeout_seconds) + if process.is_alive(): + process.terminate() + process.join(2) + if process.is_alive(): + process.kill() + process.join(1) + result = ProblemResult.failure( + candidate, + "generation_timeout", + TimeoutError(f"Problem import exceeded {timeout_seconds:.1f} seconds"), + timeout_seconds, + ) + result_queue.close() + process.close() + return result + + try: + payload = result_queue.get(timeout=1) + except queue.Empty: + return ProblemResult.failure( + candidate, + "worker", + RuntimeError(f"Worker exited with code {process.exitcode} without a result"), + ) + finally: + result_queue.close() + process.close() + return ProblemResult(**payload) + + +def run_sampler( + settings: SamplerSettings, + *, + seed: int, + catalog_client: ProblemCatalogClient | None = None, + problem_runner: ProblemRunner = run_problem_in_worker, + clock: Callable[[], float] = time.monotonic, + sleeper: Callable[[float], None] = time.sleep, + progress: Callable[[str], None] = click.echo, + keep_artifacts: Path | None = None, +) -> SamplerReport: + """Run a bounded, sequential sample and return its structured report.""" + started = clock() + deadline = started + settings.max_seconds + rng = random.Random(seed) + report = SamplerReport( + seed=seed, + percentage=settings.percentage, + max_seconds=settings.max_seconds, + ) + client = catalog_client or LeetcodeClient( + timeout=min(settings.request_timeout, max(1.0, settings.max_seconds - 1)), + use_browser_cookies=False, + ) + + try: + catalog = client.get_problem_list() + except Exception as error: + report.stop_reason = f"catalog: {_clean_message(str(error))}" + report.duration_seconds = round(clock() - started, 3) + return report + + population, paid_count = build_population(catalog) + report.catalog_total = len(catalog.stat_status_pairs) + report.eligible_free = len(population) + report.paid_excluded = paid_count + report.target_imports = min( + len(population), + max(1, math.ceil(len(population) * settings.percentage / 100)), + ) + candidates = varied_candidate_order(population, rng) + progress( + f"seed={seed} free={report.eligible_free} paid_excluded={paid_count} " + f"target={report.target_imports}" + ) + + temporary_directory: tempfile.TemporaryDirectory[str] | None = None + if keep_artifacts is None: + temporary_directory = tempfile.TemporaryDirectory(prefix="leet2git-sampler-") + workspace = Path(temporary_directory.name) + else: + workspace = keep_artifacts.resolve() + workspace.mkdir(parents=True, exist_ok=True) + + last_failure_fingerprint: tuple[str, str, str] | None = None + repeated_failures = 0 + try: + for candidate in candidates: + if report.imported >= report.target_imports: + break + + delay = rng.uniform(settings.min_delay, settings.max_delay) + remaining = deadline - clock() + if remaining <= delay + SHUTDOWN_RESERVE_SECONDS: + report.stop_reason = "deadline reached before the next paced request" + break + sleeper(delay) + remaining = deadline - clock() + if remaining <= SHUTDOWN_RESERVE_SECONDS: + report.stop_reason = "deadline reached" + break + + item_timeout = min( + settings.item_timeout, + max(1.0, remaining - SHUTDOWN_RESERVE_SECONDS), + ) + problem_root = workspace / f"{candidate.question_id}_{candidate.slug}" + result = problem_runner(candidate, problem_root, settings.language, item_timeout) + report.results.append(result) + report.attempted += 1 + + if result.status == "passed": + report.passed += 1 + report.imported += 1 + elif result.status == "soft_error": + report.soft_errors += 1 + report.imported += 1 + elif result.status == "skipped": + report.skipped += 1 + else: + report.failed += 1 + + progress( + f"[{report.imported}/{report.target_imports}] {result.status.upper():10} " + f"#{candidate.question_id} {candidate.slug} ({result.stage})" + ) + + if _is_access_block(result): + report.stop_reason = "LeetCode rate-limited or blocked the sampler; stopped" + break + + if result.status == "failed": + fingerprint = (result.stage, result.error_type, result.message) + repeated_failures = ( + repeated_failures + 1 if fingerprint == last_failure_fingerprint else 1 + ) + last_failure_fingerprint = fingerprint + if repeated_failures >= 3: + report.stop_reason = "three identical failures in a row; circuit breaker opened" + break + else: + repeated_failures = 0 + last_failure_fingerprint = None + else: + report.stop_reason = "candidate catalog exhausted" + finally: + if temporary_directory is not None: + temporary_directory.cleanup() + + if report.imported < report.target_imports and not report.stop_reason: + report.stop_reason = "target was not reached" + _finalize_report(report, clock() - started) + return report + + +def _problem_worker( + candidate: Candidate, + output_root: Path, + language: str, + timeout_seconds: float, + result_queue: multiprocessing.Queue, +) -> None: + """Child-process entry point for one public problem import.""" + client = LeetcodeClient( + timeout=max(1.0, min(15.0, timeout_seconds - 1)), + use_browser_cookies=False, + ) + result = inspect_problem(candidate, output_root, language, client) + result_queue.put(asdict(result)) + + +def _safe_generated_path(root: Path, relative_path: str) -> Path: + """Resolve a generated path and reject writes outside the temporary workspace.""" + resolved_root = root.resolve() + path = (resolved_root / relative_path).resolve() + if path != resolved_root and resolved_root not in path.parents: + raise ValueError(f"Generated path escaped the sampler workspace: {relative_path}") + if not path.is_file(): + raise FileNotFoundError(f"Generated file does not exist: {relative_path}") + return path + + +def _validate_python_file(path: Path, *, execute: bool) -> set[str]: + """Parse, compile, and optionally import a generated Python file.""" + source = path.read_text(encoding="UTF8") + tree = ast.parse(source, filename=str(path)) + compile(tree, str(path), "exec") + if not execute: + return set() + return set(runpy.run_path(str(path), run_name="leet2git_sampler_module")) + + +def _clean_message(message: str) -> str: + """Keep reports concise and single-line.""" + return " ".join(message.split())[:500] + + +def _is_access_block(result: ProblemResult) -> bool: + message = result.message.casefold() + return result.status == "failed" and ( + "http 403" in message + or "http 429" in message + or "rate limit" in message + or "too many requests" in message + or "captcha" in message + ) + + +def _finalize_report(report: SamplerReport, duration_seconds: float) -> None: + imported_results = [ + result for result in report.results if result.status in {"passed", "soft_error"} + ] + report.duration_seconds = round(duration_seconds, 3) + report.difficulty_coverage = dict( + sorted(Counter(result.difficulty for result in imported_results).items()) + ) + report.era_coverage = dict( + sorted(Counter(str(result.era + 1) for result in imported_results).items()) + ) + report.topic_coverage = sorted({topic for result in imported_results for topic in result.topics}) + + +@click.command() +@click.option( + "--percentage", + type=click.FloatRange(min=0.01, max=100), + default=1.0, + show_default=True, + help="Percentage of the free public catalog to import.", +) +@click.option( + "--max-minutes", + type=click.FloatRange(min=0.5, max=60), + default=60.0, + show_default=True, + help="Wall-clock limit; a shutdown reserve keeps the process under 60 minutes.", +) +@click.option( + "--min-delay", + type=click.FloatRange(min=1, max=30), + default=2.0, + show_default=True, + help="Minimum delay between public problem requests.", +) +@click.option( + "--max-delay", + type=click.FloatRange(min=1, max=30), + default=5.0, + show_default=True, + help="Maximum randomized delay between public problem requests.", +) +@click.option("--seed", type=int, help="Seed to reproduce the candidate order and jitter.") +@click.option( + "--report", + "report_path", + type=click.Path(path_type=Path, dir_okay=False), + help="Optional path for the full JSON report.", +) +@click.option( + "--keep-artifacts", + type=click.Path(path_type=Path, file_okay=False), + help="Keep generated sample files here instead of deleting the temporary workspace.", +) +def main( + percentage: float, + max_minutes: float, + min_delay: float, + max_delay: float, + seed: int | None, + report_path: Path | None, + keep_artifacts: Path | None, +) -> None: + """Sample public Python problem scaffolds and report import defects.""" + selected_seed = seed if seed is not None else random.SystemRandom().randrange(2**63) + try: + settings = SamplerSettings( + percentage=percentage, + max_seconds=min(max_minutes * 60, MAX_RUNTIME_SECONDS), + min_delay=min_delay, + max_delay=max_delay, + ) + except ValueError as error: + raise click.UsageError(str(error)) from error + + report = run_sampler( + settings, + seed=selected_seed, + keep_artifacts=keep_artifacts, + ) + report_json = json.dumps(report.to_dict(), indent=2, sort_keys=True) + if report_path is not None: + report_path.parent.mkdir(parents=True, exist_ok=True) + report_path.write_text(report_json + "\n", encoding="UTF8") + click.echo(f"report={report_path}") + click.echo(report_json) + raise SystemExit(report.exit_code()) + + +if __name__ == "__main__": + main() diff --git a/src/leet2git/my_utils.py b/src/leet2git/cli_helpers.py similarity index 51% rename from src/leet2git/my_utils.py rename to src/leet2git/cli_helpers.py index e8f2766..35aac6d 100644 --- a/src/leet2git/my_utils.py +++ b/src/leet2git/cli_helpers.py @@ -1,25 +1,61 @@ """ -My util functions +CLI helper functions Authors: - Yuri Rocha (yurirocha15@gmail.com) """ import os import signal -from collections.abc import Mapping -from multiprocessing import Process +from collections.abc import Iterable, Mapping +from typing import Protocol -from leet2git.config_manager import ConfigManager -from leet2git.leetcode_client import LeetcodeClient -from leet2git.question_db import QuestionData, QuestionDB +from leet2git.question_db import IdTitleMap, QuestionData -def mgr_init(): +class _SourceConfig(Protocol): + source_path: str + + +class _ConfigManager(Protocol): + @property + def config(self) -> _SourceConfig: ... + + def reset_config(self, repo_path: str, language: str, /) -> None: ... + + +class _QuestionMap(Protocol): + def check_if_id_is_known(self, slug: str, /) -> bool: ... + + def set_id_title_map(self, id_title_map: IdTitleMap, /) -> None: ... + + def save(self) -> None: ... + + def get_id_from_title(self, slug: str, /) -> int | None: ... + + +class _IdTitleMapClient(Protocol): + def get_id_title_map(self) -> IdTitleMap: ... + + +class _Joinable(Protocol): + def join(self) -> object: ... + + +class _QuestionSink(Protocol): + def add_question(self, question: QuestionData, /) -> None: ... + + +def mgr_init() -> None: """initializer for SyncManager""" signal.signal(signal.SIGINT, signal.SIG_IGN) -def reset_config(cm: ConfigManager, source_repository: str, language: str, load_old: bool = True): +def reset_config( + cm: _ConfigManager, + source_repository: str, + language: str, + load_old: bool = True, +) -> None: """Reset the configuration file Args: @@ -35,8 +71,12 @@ def reset_config(cm: ConfigManager, source_repository: str, language: str, load_ cm.reset_config(source_repository, language) -def get_question_id(title_slug: str, qdb: QuestionDB, lc: LeetcodeClient) -> int: - """Get the question ID give the title slug +def get_question_id( + title_slug: str, + qdb: _QuestionMap, + lc: _IdTitleMapClient, +) -> int | None: + """Get the question id from the title Args: title_slug (str): the title slug @@ -44,20 +84,18 @@ def get_question_id(title_slug: str, qdb: QuestionDB, lc: LeetcodeClient) -> int lc (LeetcodeClient): the leetcode client Returns: - int: the question id + int | None: the question id, or None if not found """ - qid: int = -1 - if qdb.check_if_id_is_known(title_slug): - qid = qdb.get_id_from_title(title_slug) - else: + if not qdb.check_if_id_is_known(title_slug): qdb.set_id_title_map(lc.get_id_title_map()) qdb.save() - qid = qdb.get_id_from_title(title_slug) - return qid + return qdb.get_id_from_title(title_slug) def wait_to_finish_download( - jobs: list[Process], ret_dict: Mapping[object, QuestionData], qdb: QuestionDB + jobs: Iterable[_Joinable], + ret_dict: Mapping[object, QuestionData], + qdb: _QuestionSink, ) -> int: """Wait until every subprocess finishes diff --git a/src/leet2git/config_manager.py b/src/leet2git/config_manager.py index 5e4b82f..23c79fe 100644 --- a/src/leet2git/config_manager.py +++ b/src/leet2git/config_manager.py @@ -93,8 +93,11 @@ def load_config(self, override_config: ConfigOverrides | None = None): """ if not override_config: override_config = ConfigOverrides() - with open(self._config_file) as file: - config = AppConfig.model_validate_json(file.read()) + try: + with open(self._config_file) as file: + config = AppConfig.model_validate_json(file.read()) + except (OSError, ValueError) as e: + raise click.ClickException(f"Failed to load config: {e}") from e self._config = config.model_copy( update={ "legacy_data_path": self._legacy_data_path, @@ -102,7 +105,7 @@ def load_config(self, override_config: ConfigOverrides | None = None): } ) - def reset_config(self, repo_path: str, language: str = "python3", edit: bool = True): + def reset_config(self, repo_path: str, language: str = "python3", edit: bool = True) -> None: """Resets the config and open it on the default editor Args: diff --git a/src/leet2git/default_handler.py b/src/leet2git/default_handler.py index 1cca831..04152b4 100644 --- a/src/leet2git/default_handler.py +++ b/src/leet2git/default_handler.py @@ -5,6 +5,9 @@ """ import os +from pathlib import Path + +import click from leet2git.config_manager import AppConfig from leet2git.file_handler import FileHandler @@ -21,16 +24,6 @@ def __init__(self) -> None: self.question_data: QuestionData = QuestionData() self.config: AppConfig = AppConfig() - def set_data(self, question_data: QuestionData, config: AppConfig): - """Sets the data needed to generate the files - - Args: - question_data (QuestionData): the question data - config (Dict[str, Any]): the app configuration - """ - self.question_data = question_data - self.config = config - def get_function_name(self) -> list[str]: """Returns the function name @@ -39,47 +32,27 @@ def get_function_name(self) -> list[str]: """ return [] - def generate_source(self) -> str: + def generate_source(self) -> Path: """Generates the source file Returns: - str: the path to the test file + Path: the path to the generated source file """ - comment: str = self.conversions[self.question_data.language]["comment"] - extension: str = self.conversions[self.question_data.language]["extension"] - description = ( - [comment + " " + line + "\n" for line in self.question_data.description] - if self.config.source_code.add_description - else [] - ) - lines: list[str] = ( - [ - comment + f" @l2g {self.question_data.id} {self.question_data.language}\n", - comment + f" [{self.question_data.id}] {self.question_data.title}\n", - comment + f" Difficulty: {self.question_data.difficulty}\n", - comment + f" {self.question_data.url}\n", - comment + "\n", - ] - + description - + [ - "\n", - "\n", - ] - ) + comment, extension, lines = self._build_source_header() code = ( self.question_data.raw_code if self.question_data.raw_code else self.question_data.question_template ) - lines.extend(code) - self.question_data.file_path += extension - full_path = os.path.join(self.config.source_path, self.question_data.file_path) + lines.append(code) + file_path = self.question_data.file_path + extension + full_path = os.path.join(self.config.source_path, file_path) os.makedirs(os.path.dirname(full_path), exist_ok=True) with open(full_path, "w", encoding="UTF8") as f: f.writelines(lines) - return self.question_data.file_path + return Path(file_path) def generate_tests(self) -> str: """Not Implemented""" @@ -91,11 +64,9 @@ def generate_submission_file(self) -> str: Returns: str: a string containing the code """ - code: str = "" - with open( - os.path.join(self.config.source_path, self.question_data.file_path), encoding="UTF8" - ) as f: - for line in f: - code += line - - return code + file_path = os.path.join(self.config.source_path, self.question_data.file_path) + try: + with open(file_path, encoding="UTF8") as f: + return f.read() + except OSError as e: + raise click.ClickException(f"Failed to read source file: {e}") from e diff --git a/src/leet2git/file_handler.py b/src/leet2git/file_handler.py index 19e07f1..b56b65c 100644 --- a/src/leet2git/file_handler.py +++ b/src/leet2git/file_handler.py @@ -6,59 +6,55 @@ import os import signal +import subprocess from abc import ABC, abstractmethod from collections.abc import MutableMapping +from pathlib import Path +from typing import Protocol import click -from git import Repo from leet2git.config_manager import AppConfig -from leet2git.leetcode_client import LeetcodeAPIError, LeetcodeClient from leet2git.question_db import QuestionData +from leet2git.test_harness import get_local_test_limitation + +LANGUAGE_CONVERSIONS: dict[str, dict[str, str]] = { + "bash": {"extension": ".sh", "comment": "#"}, + "c": {"extension": ".c", "comment": "//"}, + "cpp": {"extension": ".cpp", "comment": "//"}, + "csharp": {"extension": ".cs", "comment": "//"}, + "golang": {"extension": ".go", "comment": "//"}, + "java": {"extension": ".java", "comment": "//"}, + "javascript": {"extension": ".js", "comment": "//"}, + "kotlin": {"extension": ".kt", "comment": "//"}, + "mysql": {"extension": ".sql", "comment": "--"}, + "php": {"extension": ".php", "comment": "//"}, + "python": {"extension": ".py", "comment": "#"}, + "python3": {"extension": ".py", "comment": "#"}, + "ruby": {"extension": ".rb", "comment": "#"}, + "rust": {"extension": ".rs", "comment": "//"}, + "scala": {"extension": ".scala", "comment": "//"}, + "swift": {"extension": ".swift", "comment": "//"}, +} + + +class QuestionDataClient(Protocol): + """Read-only client behavior needed during file generation.""" + + def get_question_data( + self, + question_id: int, + title_slug: str, + language: str, + code: str, + /, + ) -> QuestionData: ... class FileHandler(ABC): - """Abstract class for file handlers - - Attributes: - conversions (Dict[str, Dict[str, str]]): Variables that change with each language. - languages (List[str]): List of languages this handler generates files to - """ + """Abstract base class for file handlers.""" languages: list[str] = [] - conversions: dict[str, dict[str, str]] = { - "bash": {"extension": ".sh", "comment": "#"}, - "c": {"extension": ".c", "comment": "//"}, - "cpp": {"extension": ".cpp", "comment": "//"}, - "csharp": {"extension": ".cs", "comment": "//"}, - "golang": {"extension": ".go", "comment": "//"}, - "java": {"extension": ".java", "comment": "//"}, - "javascript": {"extension": ".js", "comment": "//"}, - "kotlin": {"extension": ".kt", "comment": "//"}, - "mysql": {"extension": ".sql", "comment": "--"}, - "php": {"extension": ".php", "comment": "//"}, - "python": {"extension": ".py", "comment": "#"}, - "python3": {"extension": ".py", "comment": "#"}, - "ruby": {"extension": ".rb", "comment": "#"}, - "rust": {"extension": ".rs", "comment": "//"}, - "scala": {"extension": ".scala", "comment": "//"}, - "swift": {"extension": ".swift", "comment": "//"}, - } - - @classmethod - def get_handler_type(cls, config: AppConfig) -> type["FileHandler"]: - """Returns the type of the correct subclass - - Args: - cls (Type[FileHandler]): this class type - config (Dict[str, Any]): the user config - - Returns: - Type[FileHandler]: the type of the correct subclass - """ - from leet2git.handler_registry import get_handler_type - - return get_handler_type(config.language) def check_if_exists(self, language: str) -> bool: """Check if there is a handler for a given language @@ -69,7 +65,7 @@ def check_if_exists(self, language: str) -> bool: Returns: bool: true if there is a handler for the language """ - return language.lower() in self.conversions + return language.lower() in LANGUAGE_CONVERSIONS def generate_repo(self, folder_path: str) -> None: """Generates a git repository @@ -77,59 +73,71 @@ def generate_repo(self, folder_path: str) -> None: Args: folder_path (str): the path to the repository folder """ - _ = Repo.init(folder_path) + try: + subprocess.run(["git", "init", folder_path], check=True) + except subprocess.CalledProcessError as e: + raise click.ClickException(f"Failed to initialize git repository: {e}") from e with open(os.path.join(folder_path, "README.md"), "w") as _: pass os.makedirs(os.path.join(folder_path, "src"), exist_ok=True) - @abstractmethod def set_data(self, question_data: QuestionData, config: AppConfig) -> None: - """Abstract method definition - - Raises: - NotImplementedError: should be implemented by child classes - """ - raise NotImplementedError + """Assign question data and configuration to the handler.""" + self.question_data = question_data + self.config = config @abstractmethod def get_function_name(self) -> list[str]: - """Abstract method definition - - Raises: - NotImplementedError: should be implemented by child classes - """ + """Return the list of function names found in the code template.""" raise NotImplementedError @abstractmethod - def generate_source(self) -> str: - """Abstract method definition - - Raises: - NotImplementedError: should be implemented by child classes - """ + def generate_source(self) -> Path: + """Generate and write the source file; return its relative path.""" raise NotImplementedError @abstractmethod def generate_tests(self) -> str: - """Abstract method definition - - Raises: - NotImplementedError: should be implemented by child classes - """ + """Generate and write the test file; return its relative path.""" raise NotImplementedError - def generete_tests(self) -> str: - """Backward-compatible misspelled alias for generate_tests.""" - return self.generate_tests() - @abstractmethod def generate_submission_file(self) -> str: - """Abstract method definition + """Generate the submission file content; return it as a string.""" + raise NotImplementedError - Raises: - NotImplementedError: should be implemented by child classes + def remove_test_entrypoint(self) -> None: + """Remove any source entrypoint that targets a failed local test generation.""" + return None + + def _build_source_header(self) -> tuple[str, str, list[str]]: + """Build the source file header block. + + Returns: + tuple[str, str, list[str]]: (comment_char, extension, header_lines) """ - raise NotImplementedError + comment: str = LANGUAGE_CONVERSIONS[self.question_data.language]["comment"] + extension: str = LANGUAGE_CONVERSIONS[self.question_data.language]["extension"] + description = ( + [comment + " " + line + "\n" for line in self.question_data.description] + if self.config.source_code.add_description + else [] + ) + lines: list[str] = ( + [ + comment + f" @l2g {self.question_data.id} {self.question_data.language}\n", + comment + f" [{self.question_data.id}] {self.question_data.title}\n", + comment + f" Difficulty: {self.question_data.difficulty}\n", + comment + f" {self.question_data.url}\n", + comment + "\n", + ] + + description + + [ + "\n", + "\n", + ] + ) + return comment, extension, lines # helper function @@ -140,13 +148,15 @@ def create_file_handler(data: QuestionData, config: AppConfig) -> FileHandler: """Create an instance of a File Handler Args: - data (QuestionData): the question data - config (Dict[str, Any]): the user configuration + data (QuestionData): the question data + config (AppConfig): the user configuration Returns: - FileHandler: [description] + FileHandler: the initialized file handler """ - handler_type: type[FileHandler] = FileHandler.get_handler_type(config) + from leet2git.handler_registry import get_handler_type + + handler_type: type[FileHandler] = get_handler_type(config.language) file_handler = super(FileHandler, handler_type).__new__(handler_type) file_handler.set_data(data, config) return file_handler @@ -156,40 +166,103 @@ def generate_files( args: MutableMapping[int, QuestionData], qid: int, title_slug: str, - lc: LeetcodeClient, + lc: QuestionDataClient, timestamp: float, config: AppConfig, - code: str | None = "", + code: str = "", ) -> None: """Auxiliar function to generate the question files Args: - args (Dict[int, QuestionData]): a dictionary managed by the subprocess manager - qid (int): the question id - title_slug (str): the question title-slug property - lc (LeetcodeClient): LeetcodeClient object - timestamp (float): the time the question was generated - config (Dict[str, Any]): the user config - code (Optional[str], optional): the question solution. Defaults to "". + args (Dict[int, QuestionData]): a dictionary managed by the subprocess manager + qid (int): the question id + title_slug (str): the question title-slug property + lc (LeetcodeClient): LeetcodeClient object + timestamp (float): the time the question was generated + config (AppConfig): the user config + code (Optional[str], optional): the question solution. Defaults to "". """ - s = signal.signal(signal.SIGINT, signal.SIG_IGN) + previous_signal_handler = signal.signal(signal.SIGINT, signal.SIG_IGN) try: - data, is_new = lc.get_question_data(qid, title_slug, config.language, code) - except (LeetcodeAPIError, ValueError) as e: - click.secho(str(e), fg="red") - signal.signal(signal.SIGINT, s) - return + try: + data = lc.get_question_data(qid, title_slug, config.language, code) + except Exception as error: + click.secho(str(error), fg="red") + return - if is_new: - # generate data.language = config.language data.creation_time = timestamp - file_handler = create_file_handler(data, config) - data.function_name = file_handler.get_function_name() - data.file_path = file_handler.generate_source() - if config.test_code.generate_tests and data.inputs and data.outputs: - data.test_file_path = file_handler.generate_tests() + try: + file_handler = create_file_handler(data, config) + except Exception as error: + click.secho(f"Error: Could not prepare import for {qid}: {error}", fg="red") + return + + test_limitation = get_local_test_limitation(data) + if config.test_code.generate_tests and not test_limitation: + if not data.inputs or not data.outputs: + test_limitation = "LeetCode did not provide parseable input and output examples" + elif len(data.inputs) != len(data.outputs): + test_limitation = f"parsed {len(data.inputs)} inputs but {len(data.outputs)} outputs" + + try: + data.function_name = file_handler.get_function_name() + except Exception as error: + if not test_limitation: + test_limitation = f"could not identify the callable: {error}" + + if test_limitation: + data.requires_custom_test_harness = True + + try: + data.file_path = str(file_handler.generate_source()) + except Exception as error: + click.secho(f"Error: Could not import source for {qid}: {error}", fg="red") + return + + if config.test_code.generate_tests: + if test_limitation: + _report_soft_test_error(qid, data.title, test_limitation) + else: + try: + data.test_file_path = file_handler.generate_tests() + except Exception as error: + data.requires_custom_test_harness = True + data.test_file_path = "" + try: + file_handler.remove_test_entrypoint() + except Exception as cleanup_error: + click.secho( + f"Could not remove the local test entrypoint: {cleanup_error}", + fg="yellow", + ) + _remove_partial_test_file(data, config) + _report_soft_test_error( + qid, + data.title, + f"test generation failed: {type(error).__name__}: {error}", + ) args[qid] = data click.secho(f"""The question "{qid}|{data.title}" was imported""") - signal.signal(signal.SIGINT, s) + finally: + signal.signal(signal.SIGINT, previous_signal_handler) + + +def _report_soft_test_error(qid: int, title: str, reason: str) -> None: + """Report a non-fatal local-test limitation after preserving the source import.""" + click.secho( + f'Soft error: imported source for "{qid}|{title}" without local tests: {reason}.', + fg="yellow", + ) + + +def _remove_partial_test_file(data: QuestionData, config: AppConfig) -> None: + """Remove a partially written Python test after non-fatal generation failure.""" + if data.language not in {"python", "python3"}: + return + test_path = Path(config.source_path, "tests", f"test_{data.id}.py") + try: + test_path.unlink(missing_ok=True) + except OSError as error: + click.secho(f"Could not remove incomplete test file {test_path}: {error}", fg="yellow") diff --git a/src/leet2git/leet2git.py b/src/leet2git/leet2git.py index d277648..fd877ca 100644 --- a/src/leet2git/leet2git.py +++ b/src/leet2git/leet2git.py @@ -7,6 +7,7 @@ import glob import os import time +from collections.abc import Mapping from multiprocessing import Process from multiprocessing.managers import SyncManager @@ -14,15 +15,15 @@ from click.core import Context from click.exceptions import Abort -from leet2git.config_manager import ConfigManager, ConfigOverrides -from leet2git.file_handler import create_file_handler, generate_files -from leet2git.leetcode_client import LeetcodeAPIError, LeetcodeAuthError, LeetcodeClient -from leet2git.my_utils import ( +from leet2git.cli_helpers import ( get_question_id, mgr_init, reset_config, wait_to_finish_download, ) +from leet2git.config_manager import ConfigManager, ConfigOverrides +from leet2git.file_handler import create_file_handler, generate_files +from leet2git.leetcode_client import LeetcodeAPIError, LeetcodeAuthError, LeetcodeClient from leet2git.question_db import QuestionData, QuestionDB from leet2git.readme_handler import ReadmeHandler from leet2git.version import version_info @@ -47,9 +48,9 @@ @click.pass_context def leet2git( ctx: Context, - source_repository: str | None = "", - language: str | None = "", -): + source_repository: str = "", + language: str = "", +) -> None: """Leet2Git App \f Args: @@ -70,7 +71,7 @@ def leet2git( @leet2git.command() @click.argument("question-id", type=int) @click.pass_obj -def get(cm: ConfigManager, question_id: int): +def get(cm: ConfigManager, question_id: int) -> None: """Generates all the files for a question Args: @@ -90,9 +91,8 @@ def get(cm: ConfigManager, question_id: int): # get question data args: dict[int, QuestionData] = {} - generate_files( - args, question_id, qdb.get_title_from_id(question_id), lc, time.time(), cm.config - ) + title_slug = qdb.get_title_from_id(question_id) or "" + generate_files(args, question_id, title_slug, lc, time.time(), cm.config) except (LeetcodeAPIError, LeetcodeAuthError) as e: click.secho(str(e), fg="red") return @@ -104,13 +104,13 @@ def get(cm: ConfigManager, question_id: int): # update readme rh = ReadmeHandler(cm.config) - rh.build_readme(qdb.get_sorted_list(sort_by="creation_time")) + rh.build_readme(qdb.get_questions_sorted_by_creation_time()) @leet2git.command() @click.argument("question-id", type=int) @click.pass_obj -def submit(cm: ConfigManager, question_id: int): +def submit(cm: ConfigManager, question_id: int) -> None: """Submit a question to Leetcode Args: @@ -129,9 +129,7 @@ def submit(cm: ConfigManager, question_id: int): try: lc = LeetcodeClient() - title_slug = ( - question_data.title_slug if question_data.title_slug else qdb.get_title_from_id(question_id) - ) + title_slug = question_data.title_slug or qdb.get_title_from_id(question_id) or "" lc.submit_question(code, question_data.internal_id, title_slug, cm.config.language) except (LeetcodeAPIError, LeetcodeAuthError) as e: click.secho(str(e), fg="red") @@ -140,7 +138,7 @@ def submit(cm: ConfigManager, question_id: int): @leet2git.command() @click.argument("question-id", type=int) @click.pass_obj -def run(cm: ConfigManager, question_id: int): +def run(cm: ConfigManager, question_id: int) -> None: """Run a question on Leetcode Servers Args: @@ -159,10 +157,8 @@ def run(cm: ConfigManager, question_id: int): try: lc = LeetcodeClient() - title_slug = ( - question_data.title_slug if question_data.title_slug else qdb.get_title_from_id(question_id) - ) - raw_inputs = "\n".join(["\n".join(i.split(", ")) for i in question_data.inputs]) + title_slug = question_data.title_slug or qdb.get_title_from_id(question_id) or "" + raw_inputs = question_data.to_wire_inputs() lc.submit_question( code, question_data.internal_id, @@ -177,7 +173,7 @@ def run(cm: ConfigManager, question_id: int): @leet2git.command() @click.pass_obj -def import_all(cm: ConfigManager): +def import_all(cm: ConfigManager) -> None: """Get all solutions and generate their files""" qdb: QuestionDB = QuestionDB(cm.config) qdb.load() @@ -186,19 +182,22 @@ def import_all(cm: ConfigManager): offset: int = 0 imported_cnt = 0 manager: SyncManager | None = None + jobs: list[Process] = [] + ret_dict: Mapping[object, QuestionData] | None = None try: lc = LeetcodeClient() while has_next: - jobs: list[Process] = [] + jobs = [] manager = SyncManager() manager.start(mgr_init) ret_dict = manager.dict() submissions = lc.get_submission_list(last_key, offset) for submission in submissions.submissions_dump: - qid: int = get_question_id(submission.title_slug, qdb, lc) + qid = get_question_id(submission.title_slug, qdb, lc) if ( - submission.status_display == "Accepted" + qid is not None + and submission.status_display == "Accepted" and submission.lang == cm.config.language and not qdb.check_if_exists(qid) ): @@ -230,7 +229,8 @@ def import_all(cm: ConfigManager): time.sleep(1) except KeyboardInterrupt: click.secho("Stopping the process...") - imported_cnt += wait_to_finish_download(jobs, ret_dict, qdb) + if ret_dict is not None: + imported_cnt += wait_to_finish_download(jobs, ret_dict, qdb) except (LeetcodeAPIError, LeetcodeAuthError, ValueError) as e: click.secho(str(e), fg="red") finally: @@ -240,7 +240,7 @@ def import_all(cm: ConfigManager): qdb.save() # update readme rh = ReadmeHandler(cm.config) - rh.build_readme(qdb.get_sorted_list(sort_by="creation_time")) + rh.build_readme(qdb.get_questions_sorted_by_creation_time()) click.secho(f"In total, {imported_cnt} questions were imported!") @@ -248,7 +248,7 @@ def import_all(cm: ConfigManager): @leet2git.command() @click.argument("question-id", type=int) @click.pass_obj -def delete(cm: ConfigManager, question_id: int): +def delete(cm: ConfigManager, question_id: int) -> None: """Delete a question and its files Args: @@ -263,12 +263,12 @@ def delete(cm: ConfigManager, question_id: int): if data.test_file_path: os.remove(os.path.join(cm.config.source_path, data.test_file_path)) except FileNotFoundError as e: - click.secho(e.args) + click.secho(str(e), fg="red") qdb.delete_question(question_id) qdb.save() # update readme rh = ReadmeHandler(cm.config) - rh.build_readme(qdb.get_sorted_list(sort_by="creation_time")) + rh.build_readme(qdb.get_questions_sorted_by_creation_time()) click.secho(f"The question {question_id} was removed.") else: click.secho(f"The question {question_id} could not be found!") @@ -281,7 +281,7 @@ def delete(cm: ConfigManager, question_id: int): @click.option("--language", "-l", default="python3", help="the default language") @click.option("--create-repo", "-c", is_flag=True, help="generates a git repository") @click.pass_obj -def init(cm: ConfigManager, source_repository: str, language: str, create_repo: bool): +def init(cm: ConfigManager, source_repository: str, language: str, create_repo: bool) -> None: """Creates a new configuration file and can generate a git repository. \f Args: @@ -309,7 +309,7 @@ def init(cm: ConfigManager, source_repository: str, language: str, create_repo: help="A soft reset only erases the database. A hard reset also erase the files.", ) @click.pass_obj -def reset(cm: ConfigManager, source_repository: str, language: str, soft: bool): +def reset(cm: ConfigManager, source_repository: str, language: str, soft: bool) -> None: """Reset the configuration file \f Args: @@ -337,7 +337,7 @@ def reset(cm: ConfigManager, source_repository: str, language: str, soft: bool): try: os.remove(file) except FileNotFoundError as e: - click.secho(e.args) + click.secho(str(e), fg="red") else: try: diff --git a/src/leet2git/leetcode_client.py b/src/leet2git/leetcode_client.py index 1eb290a..a404941 100644 --- a/src/leet2git/leetcode_client.py +++ b/src/leet2git/leetcode_client.py @@ -5,8 +5,9 @@ """ import asyncio +import json import os -import platform +import re import textwrap import time from collections.abc import Coroutine @@ -32,13 +33,58 @@ SubmitSolutionResponse, ) from leet2git.question_db import IdTitleMap, QuestionData +from leet2git.test_harness import get_local_test_limitation LEETCODE_COOKIE_DOMAINS = {"leetcode.com", ".leetcode.com"} USER_AGENT = ( "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/147.0.0.0 Safari/537.36" ) +_QUESTION_DATA_QUERY = """query questionData($titleSlug: String) { + question(titleSlug: $titleSlug) { + questionId + questionFrontendId + title + titleSlug + content + isPaidOnly + difficulty + likes + dislikes + exampleTestcases + topicTags { + name + slug + translatedName + __typename + } + codeSnippets { + lang + langSlug + code + __typename + } + stats + hints + solution { + id + canSeeDetail + paidOnly + hasVideoSolution + paidOnlyVideo + __typename + } + status + sampleTestCase + metaData + __typename + } +} +""" +_QUESTION_FETCH_RETRY_DELAY_SECONDS = 1.0 +_RETRYABLE_QUESTION_FETCH_STATUSES = frozenset({404, 500, 502, 503, 504}) ResponseModel = TypeVar("ResponseModel", bound=BaseModel) +T = TypeVar("T") class LeetcodeAuthError(RuntimeError): @@ -57,16 +103,11 @@ def __init__( *, transport: httpx.AsyncBaseTransport | None = None, timeout: float = 30.0, + use_browser_cookies: bool = True, ): - os_name = platform.system() - if os_name in ["Linux", "Darwin"]: - self.divider = "/" - elif os_name == "Windows": - self.divider = "\\" - self._transport = transport self._timeout = timeout - self.cookies, self.csrftoken = self.get_cookies() + self.cookies, self.csrftoken = self.get_cookies() if use_browser_cookies else ("", "") def get_cookies(self) -> tuple[str, str]: """Get the cookies from the browser @@ -83,9 +124,9 @@ def get_cookies(self) -> tuple[str, str]: click.secho(e.args, fg="red") continue - leetcode_cookies = self._get_leetcode_cookies(cookie_jar) - if self._has_cookie(leetcode_cookies, "LEETCODE_SESSION"): - return self._build_cookie_header(leetcode_cookies), self._get_cookie_value( + leetcode_cookies = _get_leetcode_cookies(cookie_jar) + if _has_cookie(leetcode_cookies, "LEETCODE_SESSION"): + return _build_cookie_header(leetcode_cookies), _get_cookie_value( leetcode_cookies, "csrftoken" ) @@ -111,23 +152,24 @@ def get_headers(self) -> dict[str, str]: "referer": "https://leetcode.com", "origin": "https://leetcode.com", "user-agent": USER_AGENT, - "cookie": self.cookies, } + if self.cookies: + headers["cookie"] = self.cookies if self.csrftoken: headers["x-csrftoken"] = self.csrftoken return headers def get_question_data( - self, question_id: int, title_slug: str, language: str, code: str | None = "" - ) -> tuple[QuestionData, bool]: + self, question_id: int, title_slug: str, language: str, code: str = "" + ) -> QuestionData: """Gets the data from a question Args: question_id (int): the question id title_slug (str): the question title - language str: the language to download the code - code (Optional[str]): the question solution + language (str): the language to download the code + code (str | None): the question solution Returns: QuestionData: The data needed to generate the question files @@ -144,35 +186,10 @@ def get_question_data( data.difficulty = question.difficulty data.question_template = self._get_code_snippet(question, language) data.categories = question.topic_tags + data.requires_custom_test_harness = _requires_custom_test_harness(question.meta_data) + data.requires_custom_test_harness = bool(get_local_test_limitation(data)) - soup = BeautifulSoup(question.content, features="html.parser") - for sup in soup.find_all("sup"): - sup.string = "^" + sup.get_text() - data.description = soup.get_text().replace("\r\n", "\n").split("\n") - sample_test_case = question.sample_test_case - example_test_cases = question.example_testcases - num_of_inputs = len(sample_test_case.split("\n")) - inputs = example_test_cases.split("\n") - data.inputs = [ - ", ".join(inputs[i : i + num_of_inputs]) for i in range(0, len(inputs), num_of_inputs) - ] - tmp_description = [] - example_started = False - for idx, line in enumerate(data.description): - stripped_line = line.strip() - if stripped_line.startswith("Example") and stripped_line.endswith(":"): - example_started = True - elif "Output: " in line and example_started: - data.outputs.append(line[8:]) - example_started = False - elif line == "Output" and example_started: - data.outputs.append(data.description[idx + 1].strip()) - example_started = False - if len(line) > 100: - tmp_description.extend(textwrap.wrap(line, width=100, break_long_words=False)) - else: - tmp_description.append(line) - data.description = tmp_description + self._parse_question_content(question, data) data.file_path = os.path.join( "src", @@ -182,7 +199,7 @@ def get_question_data( if code: data.raw_code = code - return data, True + return data def get_latest_submission(self, qid: str, language: str) -> str: """Get the latest stored submission code for a question/language.""" @@ -224,54 +241,25 @@ async def async_scrap_question_data(self, question_name: str) -> QuestionDataRes """Query a question information Args: - question_name (str): the question slug (which is inside the leetcode url) + question_name (str): the question slug (which is inside the leetcode url) Returns: - Dict[str, Dict[str, Any]]: the categories information + QuestionDataResponse: the question information """ url: str = "https://leetcode.com/graphql" payload = QuestionDataRequest( variables=QuestionDataVariables(titleSlug=question_name), - query="query questionData($titleSlug: String) {\n question(titleSlug: $titleSlug) {\ - \n questionId\ - \n questionFrontendId\ - \n title\ - \n titleSlug\ - \n content\ - \n isPaidOnly\ - \n difficulty\ - \n likes\ - \n dislikes\ - \n exampleTestcases\ - \n topicTags {\ - \n name\ - \n slug\ - \n translatedName\ - \n __typename\ - \n }\ - \n codeSnippets {\ - \n lang\ - \n langSlug\ - \n code\ - \n __typename\ - \n }\ - \n stats\ - \n hints\ - \n solution {\ - \n id\ - \n canSeeDetail\ - \n paidOnly\ - \n hasVideoSolution\ - \n paidOnlyVideo\ - \n __typename\ - \n }\ - \n status\ - \n sampleTestCase\ - \n metaData\ - \n __typename\n }\n}\n", + query=_QUESTION_DATA_QUERY, ) + try: + return await self._request_json("POST", url, QuestionDataResponse, json_body=payload) + except LeetcodeAPIError as error: + if not _is_retryable_question_fetch(error): + raise + + await asyncio.sleep(_QUESTION_FETCH_RETRY_DELAY_SECONDS) return await self._request_json("POST", url, QuestionDataResponse, json_body=payload) def submit_question( @@ -282,16 +270,16 @@ def submit_question( language: str, is_test: bool = False, test_input: str = "", - ): + ) -> None: """Submit question to Leetcode Args: - code (str): the code which will be submitted - internal_id (int): the question "questionId". (different from "frontend_id") - title_slug (str): the question title slug - language (str): the language of the code - is_test (bool): if true, do not submit, only test on leetcode servers - test_input (str): input to test. Only used if is_test is True + code (str): the code which will be submitted + internal_id (int): the question "questionId". (different from "frontend_id") + title_slug (str): the question title slug + language (str): the language of the code + is_test (bool): if true, do not submit, only test on leetcode servers + test_input (str): input to test. Only used if is_test is True """ self._run_async( self.async_submit_question(code, internal_id, title_slug, language, is_test, test_input) @@ -320,8 +308,6 @@ async def async_submit_question( if is_test: payload.data_input = test_input payload.judge_type = "large" - - if is_test: submission_response = await self._request_json( "POST", url, @@ -340,19 +326,24 @@ async def async_submit_question( click.secho("Waiting for submission results...") url = f"https://leetcode.com/submissions/detail/{submission_id}/check/" - submission_result: SubmissionResultResponse | None = None - status: str = "" + submission_result = await self._request_json("GET", url, SubmissionResultResponse) + status = submission_result.state while status != "SUCCESS": + await asyncio.sleep(1) submission_result = await self._request_json("GET", url, SubmissionResultResponse) status = submission_result.state - await asyncio.sleep(1) - if submission_result is None: - raise LeetcodeAPIError("LeetCode did not return a submission result.") + self._display_submission_result(submission_result, submission_result.status_code, is_test) + def _display_submission_result( + self, submission_result: SubmissionResultResponse, status_code: int | None, is_test: bool + ) -> None: + """Display the formatted submission result to the user.""" click.clear() click.secho(f"Result: {submission_result.status_msg or 'Unknown'}") - status_code = submission_result.status_code + if status_code is None: + click.secho("Submission status code unavailable.", fg="yellow") + return if status_code == 10: click.secho( f"Total Runtime: {submission_result.status_runtime or 'unknown'} " @@ -389,11 +380,11 @@ async def async_get_submission_list( """Get a list with 20 submissions Args: - last_key (str, optional): the key of the last query. Defaults to "". - offset (int, optional): the offset (used to query older values). Defaults to 0. + last_key (str, optional): the key of the last query. Defaults to "". + offset (int, optional): the offset (used to query older values). Defaults to 0. Returns: - Dict[str, Any]: the query response + SubmissionListResponse: the query response """ url: str = f"https://leetcode.com/api/submissions/?offset={offset}&limit=20&lastkey={last_key}" @@ -411,16 +402,23 @@ def get_id_title_map(self) -> IdTitleMap: """Get id/title mappings using the async HTTP implementation.""" return self._run_async(self.async_get_id_title_map()) + def get_problem_list(self) -> ProblemListResponse: + """Get the public LeetCode problem catalog.""" + return self._run_async(self.async_get_problem_list()) + + async def async_get_problem_list(self) -> ProblemListResponse: + """Get the public LeetCode problem catalog asynchronously.""" + url = "https://leetcode.com/api/problems/all/" + return await self._request_json("GET", url, ProblemListResponse) + async def async_get_id_title_map(self) -> IdTitleMap: """Get a dictionary that maps the id to the question title slug Returns: IdTitleMap: maps the id to the title slug """ - url: str = "https://leetcode.com/api/problems/all/" - id_title_map: IdTitleMap = IdTitleMap() - response = await self._request_json("GET", url, ProblemListResponse) + response = await self.async_get_problem_list() for pair in response.stat_status_pairs: id_title_map.id_to_title[pair.stat.frontend_question_id] = pair.stat.question_title_slug id_title_map.title_to_id[pair.stat.question_title_slug] = pair.stat.frontend_question_id @@ -460,12 +458,14 @@ async def _request_json( except ValueError as e: raise LeetcodeAPIError(f"LeetCode returned a non-JSON response: {url}") from e + _raise_for_leetcode_error(payload, url) + try: return model_type.model_validate(payload) except ValidationError as e: raise LeetcodeAPIError(f"LeetCode returned unexpected JSON for {url}: {e}") from e - def _run_async(self, coroutine: Coroutine[Any, Any, Any]) -> Any: + def _run_async(self, coroutine: Coroutine[Any, Any, T]) -> T: """Run an async client method from synchronous Click commands.""" try: asyncio.get_running_loop() @@ -492,21 +492,193 @@ def _get_code_snippet(self, question: QuestionPayload, language: str) -> str: return snippet.code raise LeetcodeAPIError(f'LeetCode did not return a "{language}" code snippet.') - def _get_leetcode_cookies(self, cookie_jar: CookieJar) -> list[Cookie]: - """Return browser cookies scoped to LeetCode.""" - return [cookie for cookie in cookie_jar if cookie.domain in LEETCODE_COOKIE_DOMAINS] + def _parse_question_content(self, question: QuestionPayload, data: QuestionData) -> None: + """Parse HTML content and extract description and test outputs from question payload.""" + soup = BeautifulSoup(question.content, features="html.parser") + for sup in soup.find_all("sup"): + sup.string = "^" + sup.get_text() + data.description = soup.get_text().replace("\r\n", "\n").split("\n") + sample_test_case = question.sample_test_case + example_test_cases = question.example_testcases + num_of_inputs = len(sample_test_case.split("\n")) + inputs = example_test_cases.split("\n") + data.inputs = [ + ", ".join(inputs[i : i + num_of_inputs]) for i in range(0, len(inputs), num_of_inputs) + ] + data.outputs = _extract_example_outputs(data.description) + description_inputs = _extract_example_inputs(data.description) + if ( + data.outputs + and len(data.inputs) != len(data.outputs) + and len(description_inputs) == len(data.outputs) + ): + data.inputs = description_inputs + tmp_description = [] + for line in data.description: + if len(line) > 100: + tmp_description.extend(textwrap.wrap(line, width=100, break_long_words=False)) + else: + tmp_description.append(line) + data.description = tmp_description + + +# ============================================================================ +# Module-level helper functions for cookie extraction +# ============================================================================ + + +_EXAMPLE_HEADER = re.compile(r"^Example(?:\s+\d+)?\s*:", re.IGNORECASE) +_INPUT_LABEL = re.compile(r"^Input\s*:?\s*(.*)$", re.IGNORECASE) +_INPUT_BOUNDARY = re.compile(r"^Output\s*:?.*$", re.IGNORECASE) +_OUTPUT_LABEL = re.compile(r"^Output\s*:?\s*(.*)$", re.IGNORECASE) +_OUTPUT_BOUNDARY = re.compile( + r"^(?:Input|Explanation|Explantion|Constraints)\s*:?", + re.IGNORECASE, +) + + +def _extract_example_inputs(description: list[str]) -> list[str]: + """Extract statement inputs for APIs that omit later examples from exampleTestcases.""" + inputs: list[str] = [] + input_lines: list[str] | None = None + inline_value = False + example_started = False + + def finish_input() -> None: + nonlocal input_lines, inline_value + if input_lines is None: + return + separator = " " if inline_value or len(input_lines) <= 1 else ", " + value = separator.join(input_lines).strip() + if value: + inputs.append(value) + input_lines = None + inline_value = False + + for line in description: + stripped_line = line.strip() + if _EXAMPLE_HEADER.match(stripped_line): + finish_input() + example_started = True + continue + if not example_started: + continue + + input_match = _INPUT_LABEL.match(stripped_line) + if input_match: + finish_input() + remainder = input_match.group(1).strip() + inline_value = bool(remainder) + input_lines = [remainder] if remainder else [] + continue + + if input_lines is not None: + if _INPUT_BOUNDARY.match(stripped_line): + finish_input() + elif stripped_line: + input_lines.append(stripped_line) + + finish_input() + return inputs + + +def _extract_example_outputs(description: list[str]) -> list[str]: + """Extract complete outputs from LeetCode example text, including multiline values.""" + outputs: list[str] = [] + output_lines: list[str] | None = None + example_started = False + + def finish_output() -> None: + nonlocal output_lines + if output_lines is None: + return + output = " ".join(output_lines).strip() + if output: + outputs.append(output) + output_lines = None + + for line in description: + stripped_line = line.strip() + if _EXAMPLE_HEADER.match(stripped_line): + finish_output() + example_started = True + continue + if not example_started: + continue + + output_match = _OUTPUT_LABEL.match(stripped_line) + if output_match: + finish_output() + remainder = output_match.group(1).strip() + output_lines = [remainder] if remainder else [] + continue + + if output_lines is not None: + if _OUTPUT_BOUNDARY.match(stripped_line): + finish_output() + elif stripped_line: + output_lines.append(stripped_line) + + finish_output() + return outputs + + +def _requires_custom_test_harness(meta_data: str | None) -> bool: + """Return whether LeetCode validates output through an in-place/custom judge contract.""" + if not meta_data: + return False + try: + metadata = json.loads(meta_data) + except (TypeError, ValueError): + return True + return not isinstance(metadata, dict) or "output" in metadata + + +def _is_retryable_question_fetch(error: LeetcodeAPIError) -> bool: + """Recognize transient endpoint responses for the idempotent questionData query.""" + cause = error.__cause__ + return ( + isinstance(cause, httpx.HTTPStatusError) + and cause.response.status_code in _RETRYABLE_QUESTION_FETCH_STATUSES + ) + + +def _raise_for_leetcode_error(payload: Any, url: str) -> None: + """Turn LeetCode's successful-HTTP error payloads into actionable exceptions.""" + if not isinstance(payload, dict): + return + error = payload.get("error") + if not isinstance(error, str) or not error.strip(): + return + + message = error.strip() + normalized_message = message.casefold() + if "not authenticated" in normalized_message or "unauthenticated" in normalized_message: + raise LeetcodeAuthError( + "LeetCode rejected the saved browser session. Log in to https://leetcode.com " + "in Chrome or Firefox and try again." + ) + raise LeetcodeAPIError(f"LeetCode returned an error for {url}: {message}") + + +def _get_leetcode_cookies(cookie_jar: CookieJar) -> list[Cookie]: + """Return browser cookies scoped to LeetCode.""" + return [cookie for cookie in cookie_jar if cookie.domain in LEETCODE_COOKIE_DOMAINS] + + +def _has_cookie(cookies: list[Cookie], name: str) -> bool: + """Check whether a LeetCode cookie exists.""" + return any(cookie.name == name and bool(cookie.value) for cookie in cookies) + - def _has_cookie(self, cookies: list[Cookie], name: str) -> bool: - """Check whether a LeetCode cookie exists.""" - return any(cookie.name == name and bool(cookie.value) for cookie in cookies) +def _get_cookie_value(cookies: list[Cookie], name: str) -> str: + """Return a LeetCode cookie value without logging it.""" + for cookie in cookies: + if cookie.name == name: + return cookie.value or "" + return "" - def _get_cookie_value(self, cookies: list[Cookie], name: str) -> str: - """Return a LeetCode cookie value without logging it.""" - for cookie in cookies: - if cookie.name == name: - return cookie.value or "" - return "" - def _build_cookie_header(self, cookies: list[Cookie]) -> str: - """Build a Cookie header from browser cookies without a profile-page request.""" - return "; ".join(f"{cookie.name}={cookie.value}" for cookie in cookies if cookie.value) +def _build_cookie_header(cookies: list[Cookie]) -> str: + """Build a Cookie header from browser cookies without a profile-page request.""" + return "; ".join(f"{cookie.name}={cookie.value}" for cookie in cookies if cookie.value) diff --git a/src/leet2git/leetcode_models.py b/src/leet2git/leetcode_models.py index 139356a..6f2d187 100644 --- a/src/leet2git/leetcode_models.py +++ b/src/leet2git/leetcode_models.py @@ -4,7 +4,16 @@ from pydantic import BaseModel, ConfigDict, Field -from leet2git.question_db import TopicTag + +class TopicTag(BaseModel): + """LeetCode topic metadata attached to a question.""" + + model_config = ConfigDict(populate_by_name=True, validate_assignment=True) + + name: str = "" + slug: str = "" + translated_name: str | None = Field(default=None, alias="translatedName") + typename: str | None = Field(default=None, alias="__typename") class QuestionDataVariables(BaseModel): @@ -48,6 +57,7 @@ class QuestionPayload(BaseModel): sample_test_case: str = Field(alias="sampleTestCase") topic_tags: list[TopicTag] = Field(alias="topicTags") code_snippets: list[CodeSnippet] = Field(alias="codeSnippets") + meta_data: str | None = Field(default=None, alias="metaData") class QuestionDataBody(BaseModel): @@ -91,10 +101,18 @@ class ProblemStat(BaseModel): question_title_slug: str = Field(alias="question__title_slug") +class ProblemDifficulty(BaseModel): + """Numeric difficulty metadata in the public problem catalog.""" + + level: int = 0 + + class ProblemStatusPair(BaseModel): """Problem list row.""" stat: ProblemStat + difficulty: ProblemDifficulty = Field(default_factory=ProblemDifficulty) + paid_only: bool = False class ProblemListResponse(BaseModel): diff --git a/src/leet2git/python_handler.py b/src/leet2git/python_handler.py index d71a875..0ea3db8 100644 --- a/src/leet2git/python_handler.py +++ b/src/leet2git/python_handler.py @@ -10,6 +10,7 @@ import shutil import subprocess import tokenize +from pathlib import Path import click from autoimport import fix_files @@ -29,16 +30,6 @@ def __init__(self) -> None: self.question_data: QuestionData = QuestionData() self.config: AppConfig = AppConfig() - def set_data(self, question_data: QuestionData, config: AppConfig): - """Sets the data needed to generate the files - - Args: - question_data (QuestionData): the question data - config (Dict[str, Any]): the app configuration - """ - self.question_data = question_data - self.config = config - def get_function_name(self) -> list[str]: """Returns the function name @@ -48,46 +39,27 @@ def get_function_name(self) -> list[str]: functions = self._get_template_callables(self.question_data.question_template) if not functions: raise ValueError("Could not find a Python function in the LeetCode code template.") - self.question_data.function_name = functions return functions - def generate_source(self) -> str: + def generate_source(self) -> Path: """Generates the source file Returns: - str: the path to the test file + Path: the path to the generated source file """ - comment: str = self.conversions[self.question_data.language]["comment"] - extension: str = self.conversions[self.question_data.language]["extension"] - description = ( - [comment + " " + line + "\n" for line in self.question_data.description] - if self.config.source_code.add_description - else [] - ) - lines: list[str] = ( - [ - comment + f" @l2g {self.question_data.id} {self.question_data.language}\n", - comment + f" [{self.question_data.id}] {self.question_data.title}\n", - comment + f" Difficulty: {self.question_data.difficulty}\n", - comment + f" {self.question_data.url}\n", - comment + "\n", - ] - + description - + [ - "\n", - "\n", - ] - ) + comment, extension, lines = self._build_source_header() code, is_solution = ( (self.question_data.raw_code, True) if self.question_data.raw_code else (self.question_data.question_template, False) ) code_lines = self.parse_raw_code(code, is_solution) + if self.question_data.language == "python3": + code_lines = self._ensure_future_annotations(code_lines) lines.extend(code_lines) - self.question_data.file_path += extension + file_path = self.question_data.file_path + extension - full_path: str = os.path.join(self.config.source_path, self.question_data.file_path) + full_path: str = os.path.join(self.config.source_path, file_path) os.makedirs(os.path.dirname(full_path), exist_ok=True) with open(full_path, "w", encoding="UTF8") as f: @@ -97,7 +69,7 @@ def generate_source(self) -> str: with open(full_path, "r+", encoding="UTF8") as f: fix_files((f,)) - if self.config.test_code.generate_tests: + if self.config.test_code.generate_tests and not self.question_data.requires_custom_test_harness: with open(full_path, "a", encoding="UTF8") as f: f.write("\n") f.write("\n") @@ -111,7 +83,22 @@ def generate_source(self) -> str: self.run_formatter(full_path) - return self.question_data.file_path + return Path(file_path) + + def remove_test_entrypoint(self) -> None: + """Remove the generated pytest launcher after local test generation fails.""" + full_path = os.path.join(self.config.source_path, self.question_data.file_path) + try: + with open(full_path, encoding="UTF8") as file: + source = file.read() + tree = ast.parse(source) + main_line = self._find_main_block_line(tree) + if main_line is None: + return + with open(full_path, "w", encoding="UTF8") as file: + file.write("".join(source.splitlines(keepends=True)[: main_line - 1])) + except (OSError, SyntaxError) as error: + click.secho(f"Could not remove the local test entrypoint: {error}", fg="yellow") def generate_tests(self) -> str: """Generates the test file @@ -119,28 +106,21 @@ def generate_tests(self) -> str: Returns: str: the path to the test file """ - extension: str = self.conversions[self.question_data.language]["extension"] - self.question_data.inputs = [ + from leet2git.file_handler import LANGUAGE_CONVERSIONS + + extension: str = LANGUAGE_CONVERSIONS[self.question_data.language]["extension"] + inputs = [ s.replace("null", "None").replace("true", "True").replace("false", "False") for s in self.question_data.inputs ] - self.question_data.outputs = [ + outputs = [ s.replace("null", "None").replace("true", "True").replace("false", "False") for s in self.question_data.outputs ] - inputs = self.question_data.inputs - outputs = self.question_data.outputs + design_cases: list[tuple[list[str], list[list[object]], list[object]]] = [] if len(self.question_data.function_name) > 1: - inputs = [] - outputs = [] - for q_input, q_output in zip( - self.question_data.inputs, self.question_data.outputs, strict=True - ): - tmp_inputs = q_input.split(", ") - inputs.append([]) - for tmp_input in tmp_inputs: - inputs[-1].append(ast.literal_eval(tmp_input)) - outputs.append(ast.literal_eval(q_output)) + for q_input, q_output in zip(inputs, outputs, strict=True): + design_cases.append(self._parse_design_case(q_input, q_output)) elif not self.question_data.function_name: raise ValueError("No function name") full_path: str = os.path.join( @@ -167,31 +147,25 @@ def generate_tests(self) -> str: if len(self.question_data.function_name) == 1: f.write(f" from src.{self.question_data.file_path[4:-3]} import Solution\n") f.write(" solution = Solution()\n") + f.write("\n") + f.write(f" def _init_variables_{self.question_data.id}():\n") + f.write(" return solution\n") else: - try: - f.write( - f" from src.{self.question_data.file_path[4:-3]} \ - import {self.question_data.function_name[0]}\n" - ) - f.write( - f" solution = {self.question_data.function_name[0]}\ - ({str(inputs[0][1][0])[1:-1]})\n" - ) - # if we meet a question with some wild inputs - except ValueError as e: - print(e.args) - print(self.question_data) - f.write("\n") - f.write(f" def _init_variables_{self.question_data.id}():\n") - f.write(" return solution\n") + constructor = self.question_data.function_name[0] + f.write(f" from src.{self.question_data.file_path[4:-3]} import {constructor}\n") + f.write("\n") + f.write(f" def _init_variables_{self.question_data.id}(*args):\n") + f.write(f" return {constructor}(*args)\n") f.write("\n") f.write(f" yield _init_variables_{self.question_data.id}\n") f.write("\n") f.write(f"class TestClass{self.question_data.id}:") - for i, (q_input, q_output) in enumerate(zip(inputs, outputs, strict=True)): - f.write("\n") - f.write(f" def test_solution_{i}(self, init_variables_{self.question_data.id}):\n") - if len(self.question_data.function_name) == 1: + if len(self.question_data.function_name) == 1: + for i, (q_input, q_output) in enumerate(zip(inputs, outputs, strict=True)): + f.write("\n") + f.write( + f" def test_solution_{i}(self, init_variables_{self.question_data.id}):\n" + ) f.write( " assert" + (" not" if q_output == "False" else "") @@ -200,32 +174,80 @@ def generate_tests(self) -> str: + (f" == {q_output}" if q_output not in ["True", "False"] else "") + "\n" ) - else: + else: + for i, (method_names, method_inputs, expected_outputs) in enumerate(design_cases): + f.write("\n") + f.write( + f" def test_solution_{i}(self, init_variables_{self.question_data.id}):\n" + ) + constructor_args = ", ".join(repr(value) for value in method_inputs[0]) + f.write( + f" solution = init_variables_{self.question_data.id}" + f"({constructor_args})\n" + ) for input_func, input_val, output in zip( - q_input[0][1:], q_input[1][1:], q_output[1:], strict=True + method_names[1:], + method_inputs[1:], + expected_outputs[1:], + strict=True, ): - f.write( - " assert" - + (" not" if output == "False" else "") - + f" init_variables_{self.question_data.id}().\ - {input_func}({str(input_val)[1:-1]})" - + (f" == {output}" if output not in ["True", "False"] else "") - + "\n" - ) + arguments = ", ".join(repr(value) for value in input_val) + call = f"solution.{input_func}({arguments})" + f.write(" " + self._build_assertion(call, output) + "\n") self.run_formatter(full_path) return os.path.join("tests", f"test_{self.question_data.id}{extension}") + def _parse_design_case( + self, raw_input: str, raw_output: str + ) -> tuple[list[str], list[list[object]], list[object]]: + """Parse one LeetCode design-problem example into methods, arguments, and outputs.""" + try: + parsed_input = ast.literal_eval(f"({raw_input})") + parsed_output = ast.literal_eval(raw_output) + except (SyntaxError, ValueError) as e: + raise ValueError("Could not parse the design-problem example data.") from e + + if not isinstance(parsed_input, tuple) or len(parsed_input) != 2: + raise ValueError("Design-problem input must contain method and argument lists.") + method_names, method_inputs = parsed_input + if ( + not isinstance(method_names, list) + or not all(isinstance(name, str) for name in method_names) + or not isinstance(method_inputs, list) + or not all(isinstance(arguments, list) for arguments in method_inputs) + or not isinstance(parsed_output, list) + ): + raise ValueError("Design-problem example data has an unexpected shape.") + if not method_names or not (len(method_names) == len(method_inputs) == len(parsed_output)): + raise ValueError("Design-problem method, argument, and output counts must match.") + + return method_names, method_inputs, parsed_output + + @staticmethod + def _build_assertion(call: str, expected: object) -> str: + """Build a valid pytest assertion for a generated design-problem method call.""" + if expected is True: + return f"assert {call}" + if expected is False: + return f"assert not {call}" + if expected is None: + return f"assert {call} is None" + return f"assert {call} == {expected!r}" + def generate_submission_file(self) -> str: """Generates the submission file Returns: str: a string containing the code """ - full_path: str = os.path.join(self.config.source_path, self.question_data.file_path) - with open(full_path, encoding="UTF8") as f: - source = f.read() + full_path = os.path.join(self.config.source_path, self.question_data.file_path) + try: + with open(full_path, encoding="UTF8") as f: + source = f.read() + except OSError as e: + raise click.ClickException(f"Failed to read source file: {e}") from e try: tree = ast.parse(source) @@ -238,7 +260,7 @@ def generate_submission_file(self) -> str: return "".join(source.splitlines(keepends=True)[: main_line - 1]) - def generate_repo(self, folder_path: str): + def generate_repo(self, folder_path: str) -> None: """Generates a git repository Args: @@ -272,6 +294,43 @@ def parse_raw_code(self, raw_code: str, is_solution: bool) -> list[str]: return lines + @staticmethod + def _ensure_future_annotations(code_lines: list[str]) -> list[str]: + """Defer Python 3 annotations while preserving docstrings and existing future imports.""" + try: + tree = ast.parse("".join(code_lines)) + except SyntaxError: + return code_lines + + if any( + isinstance(node, ast.ImportFrom) + and node.module == "__future__" + and any(name.name == "annotations" for name in node.names) + for node in tree.body + ): + return code_lines + + insertion_line = 0 + body_index = 0 + if ( + tree.body + and isinstance(tree.body[0], ast.Expr) + and isinstance(tree.body[0].value, ast.Constant) + and isinstance(tree.body[0].value.value, str) + ): + insertion_line = tree.body[0].end_lineno or tree.body[0].lineno + body_index = 1 + + for node in tree.body[body_index:]: + if not isinstance(node, ast.ImportFrom) or node.module != "__future__": + break + insertion_line = node.end_lineno or node.lineno + + future_lines = ["from __future__ import annotations\n"] + if insertion_line >= len(code_lines) or code_lines[insertion_line].strip(): + future_lines.append("\n") + return code_lines[:insertion_line] + future_lines + code_lines[insertion_line:] + def _get_template_callables(self, source: str) -> list[str]: """Return callable names from a Python LeetCode template.""" functions: list[str] = [] diff --git a/src/leet2git/question_db.py b/src/leet2git/question_db.py index 713482c..64cbd5e 100644 --- a/src/leet2git/question_db.py +++ b/src/leet2git/question_db.py @@ -4,14 +4,17 @@ - Yuri Rocha (yurirocha15@gmail.com) """ +import json import operator import os import pickle +from pickle import UnpicklingError import click -from pydantic import BaseModel, ConfigDict, Field +from pydantic import BaseModel, ConfigDict, Field, ValidationError from leet2git.config_manager import AppConfig +from leet2git.leetcode_models import TopicTag DB_DIR_NAME = ".leet2git" DB_FILE_NAME = "database.json" @@ -20,17 +23,6 @@ DB_VERSION = 1 -class TopicTag(BaseModel): - """LeetCode topic metadata attached to a question.""" - - model_config = ConfigDict(populate_by_name=True, validate_assignment=True) - - name: str = "" - slug: str = "" - translated_name: str | None = Field(default=None, alias="translatedName") - typename: str | None = Field(default=None, alias="__typename") - - class QuestionData(BaseModel): """Stores all the data related to a question""" @@ -53,6 +45,11 @@ class QuestionData(BaseModel): inputs: list[str] = Field(default_factory=list) outputs: list[str] = Field(default_factory=list) categories: list[TopicTag] = Field(default_factory=list) + requires_custom_test_harness: bool = False + + def to_wire_inputs(self) -> str: + """Return inputs formatted for LeetCode's test/run wire protocol.""" + return "\n".join("\n".join(i.split(", ")) for i in self.inputs) def __setstate__(self, state: dict[str, object]) -> None: """Support unpickling state written by the former dataclass model.""" @@ -113,12 +110,18 @@ def __init__(self, config: AppConfig): self.id_title_map: IdTitleMap = IdTitleMap() self.migrated_from_legacy = False - def load(self): + def load(self) -> None: """Load the question data from disk""" if os.path.isfile(self.db_file): - with open(self.db_file, encoding="UTF8") as f: - self._load_state(DatabaseState.model_validate_json(f.read())) - return + try: + with open(self.db_file, encoding="UTF8") as f: + self._load_state(DatabaseState.model_validate_json(f.read())) + except (ValidationError, json.JSONDecodeError, OSError) as e: + click.secho( + f"Warning: Failed to load database: {e}. Starting with empty database.", + fg="yellow", + ) + raise if os.path.isfile(self.legacy_db_file) or os.path.isfile(self.legacy_id_title_map_file): self._load_legacy_pickles() @@ -129,16 +132,20 @@ def load(self): fg="yellow", ) - def save(self): + def save(self) -> None: """Save the question data to disk""" - os.makedirs(self.db_dir, exist_ok=True) - state = DatabaseState( - version=DB_VERSION, - questions=self.question_data_dict, - id_title_map=self.id_title_map, - ) - with open(self.db_file, "w", encoding="UTF8") as f: - f.write(state.model_dump_json(indent=2, by_alias=True)) + try: + os.makedirs(self.db_dir, exist_ok=True) + state = DatabaseState( + version=DB_VERSION, + questions=self.question_data_dict, + id_title_map=self.id_title_map, + ) + with open(self.db_file, "w", encoding="UTF8") as f: + f.write(state.model_dump_json(indent=2, by_alias=True)) + except OSError as e: + click.secho(f"Error: Failed to save database: {e}", fg="red") + raise def get_data(self) -> dict[int, QuestionData]: """Returns the question data @@ -149,46 +156,42 @@ def get_data(self) -> dict[int, QuestionData]: return self.question_data_dict def get_question(self, question_id: int) -> QuestionData | None: - """get a question data if it exists + """Get a question data if it exists. Args: question_id (int): the question id Returns: - Optional[QuestionData]: the question data + QuestionData | None: the question data if found, None if not found """ if self.check_if_exists(question_id): return self.question_data_dict[question_id] return None - def add_question(self, qd: QuestionData): + def add_question(self, qd: QuestionData) -> None: """Add a question to the dictionary Args: - qd (QuestionData): The question data + qd (QuestionData): The question data """ self.question_data_dict[qd.id] = qd - def delete_question(self, question_id: int): + def delete_question(self, question_id: int) -> None: """Removes a question from the dictionary Args: - question_id (int): the question id + question_id (int): the question id """ if question_id in self.question_data_dict: self.question_data_dict.pop(question_id) - def get_sorted_list(self, sort_by: str) -> list[QuestionData]: - """Returns a sorted list with all the questions - - Args: - sort_by (str): the attribute used to sort the list. - Can be any QuestionData attribute. + def get_questions_sorted_by_creation_time(self) -> list[QuestionData]: + """Returns a sorted list with all the questions sorted by creation time. Returns: - List[QuestionData]: [description] + List[QuestionData]: questions sorted by creation_time """ - return sorted(self.question_data_dict.values(), key=operator.attrgetter(sort_by)) + return sorted(self.question_data_dict.values(), key=operator.attrgetter("creation_time")) def check_if_exists(self, question_id: int) -> bool: """Checks if a question exists in the database @@ -201,18 +204,18 @@ def check_if_exists(self, question_id: int) -> bool: """ return question_id in self.question_data_dict - def get_title_from_id(self, question_id: int) -> str: + def get_title_from_id(self, question_id: int) -> str | None: """Get the question title slug from its id Args: question_id (int): the question id Returns: - str: the question title slug + str | None: the question title slug, or None if not cached """ if self.check_if_slug_is_known(question_id): return self.id_title_map.id_to_title[question_id] - return "" + return None def check_if_slug_is_known(self, question_id: int) -> bool: """Checks if the title slug is cached locally @@ -225,40 +228,40 @@ def check_if_slug_is_known(self, question_id: int) -> bool: """ return question_id in self.id_title_map.id_to_title - def get_id_from_title(self, slug: str) -> int: + def get_id_from_title(self, slug: str) -> int | None: """Get the question id from its title slug Args: - str: the question title slug + slug (str): the question title slug Returns: - id (int): the question id + int | None: the question id, or None if not cached """ if self.check_if_id_is_known(slug): return self.id_title_map.title_to_id[slug] - return -1 + return None def check_if_id_is_known(self, slug: str) -> bool: """Checks if the id is cached locally Args: - str: the question title slug + slug (str): the question title slug Returns: - bool: true if the id sis cached locally + bool: true if the id sis cached locally """ return slug in self.id_title_map.title_to_id - def set_id_title_map(self, id_title_map: IdTitleMap): + def set_id_title_map(self, id_title_map: IdTitleMap) -> None: """Sets the id to slug dict Args: - id_title_map (IdTitleMap): - a dictionary mapping the question id to the title slug and vice-versa + id_title_map (IdTitleMap): + a dictionary mapping the question id to the title slug and vice-versa """ self.id_title_map = id_title_map - def reset(self): + def reset(self) -> None: """Delete database""" self.question_data_dict: dict[int, QuestionData] = {} self.id_title_map: IdTitleMap = IdTitleMap() @@ -294,8 +297,20 @@ def _load_state(self, state: DatabaseState) -> None: def _load_legacy_pickles(self) -> None: """Load legacy platform-path pickle files for one-way migration.""" if os.path.isfile(self.legacy_db_file): - with open(self.legacy_db_file, "rb") as f: - self.question_data_dict = self._load_question_data(pickle.load(f)) + try: + with open(self.legacy_db_file, "rb") as f: + self.question_data_dict = self._load_question_data(pickle.load(f)) + except (UnpicklingError, EOFError) as e: + click.secho( + f"Warning: Failed to load legacy question database: {e}", + fg="yellow", + ) if os.path.isfile(self.legacy_id_title_map_file): - with open(self.legacy_id_title_map_file, "rb") as f: - self.id_title_map = self._load_id_title_map(pickle.load(f)) + try: + with open(self.legacy_id_title_map_file, "rb") as f: + self.id_title_map = self._load_id_title_map(pickle.load(f)) + except (UnpicklingError, EOFError) as e: + click.secho( + f"Warning: Failed to load legacy id_title_map: {e}", + fg="yellow", + ) diff --git a/src/leet2git/readme_handler.py b/src/leet2git/readme_handler.py index 6bc2b58..86bdb0d 100644 --- a/src/leet2git/readme_handler.py +++ b/src/leet2git/readme_handler.py @@ -28,7 +28,7 @@ def __init__(self, config: AppConfig): self.print_categories: bool = config.readme.show_category self.print_difficulty: bool = config.readme.show_difficulty - def build_readme(self, question_list: list[QuestionData]): + def build_readme(self, question_list: list[QuestionData]) -> None: """Updates the README file Args: @@ -55,7 +55,6 @@ def build_readme(self, question_list: list[QuestionData]): fields=["ID", "Problem", "Leetcode ID", "Categories", "Difficulty"], ) for question in question_list: - difficulty_str = "" if self.print_difficulty: difficulty_str = f"[{question.difficulty}](#{question.difficulty})" else: @@ -81,11 +80,10 @@ def build_readme(self, question_list: list[QuestionData]): categories_str += c.name + ", " categories_str = categories_str[:-2] - if not question.difficulty: - question.difficulty = "Easy" - difficulty_tables[question.difficulty].values.append( + effective_difficulty = question.difficulty or "Easy" + difficulty_tables[effective_difficulty].values.append( [ - str(len(difficulty_tables[question.difficulty].values) + 1), + str(len(difficulty_tables[effective_difficulty].values) + 1), f"[{question.title}]({question.file_path})", f"[{question.id}]({question.url})", categories_str, @@ -176,4 +174,4 @@ def dump_tables( qdb = QuestionDB(config) qdb.load() rh = ReadmeHandler(config) - rh.build_readme(qdb.get_sorted_list(sort_by="creation_time")) + rh.build_readme(qdb.get_questions_sorted_by_creation_time()) diff --git a/src/leet2git/test_harness.py b/src/leet2git/test_harness.py new file mode 100644 index 0000000..c7f33a0 --- /dev/null +++ b/src/leet2git/test_harness.py @@ -0,0 +1,25 @@ +"""Classify LeetCode examples that need judge-specific local test support.""" + +import re + +from leet2git.question_db import QuestionData + +_COMMENTED_CLASS = re.compile(r"^\s*#\s*class\s+([A-Za-z_]\w*)\b", re.MULTILINE) +_UNSUPPORTED_HARNESS_TOPICS = frozenset({"concurrency", "interactive"}) + + +def get_local_test_limitation(data: QuestionData) -> str: + """Return why generic local tests are unsafe, or an empty string when supported.""" + judge_classes = sorted(set(_COMMENTED_CLASS.findall(data.question_template))) + if judge_classes: + return "LeetCode supplies judge-only class definitions: " + ", ".join(judge_classes) + + topics = {tag.slug for tag in data.categories if tag.slug} + special_topics = sorted(_UNSUPPORTED_HARNESS_TOPICS.intersection(topics)) + if special_topics: + return "LeetCode requires a judge-specific harness for: " + ", ".join(special_topics) + + if data.requires_custom_test_harness: + return "LeetCode uses custom or in-place output validation" + + return "" diff --git a/src/leet2git/version.py b/src/leet2git/version.py index bd08a69..ab6002d 100755 --- a/src/leet2git/version.py +++ b/src/leet2git/version.py @@ -9,6 +9,7 @@ import platform import sys +import click from pydantic import BaseModel, ValidationError, field_validator __version__ = "0.3.0" @@ -40,7 +41,7 @@ def version_info() -> str: return "\n".join(f"{k + ':':>30} {v}" for k, v in info.items()) -def update_version_string(new_version: str): +def update_version_string(new_version: str) -> None: """Updates the version string Args: @@ -49,7 +50,7 @@ def update_version_string(new_version: str): try: version = VersionString(value=new_version).value except ValidationError: - print(f"Version {new_version} is not valid") + click.secho(f"Version {new_version} is not valid", fg="red") return file_path = os.path.abspath(__file__) @@ -57,7 +58,7 @@ def update_version_string(new_version: str): content = f.read() version_line = _find_version_line(content) if version_line is None: - print("Could not find __version__") + click.secho("Could not find __version__", fg="red") return lines = content.splitlines(keepends=True) diff --git a/tests/test_cli.py b/tests/test_cli.py index c6d9404..a8ebbee 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -95,7 +95,7 @@ def add_question(self, question): def save(self): self.save_count += 1 - def get_sorted_list(self, sort_by): + def get_questions_sorted_by_creation_time(self): return sorted(self.questions.values(), key=lambda question: question.creation_time) class FakeClient: @@ -205,7 +205,7 @@ def delete_question(self, question_id): def save(self): pass - def get_sorted_list(self, sort_by): + def get_questions_sorted_by_creation_time(self): return [] class FakeReadmeHandler: @@ -319,7 +319,7 @@ def check_if_exists(self, question_id): def save(self): self.save_count += 1 - def get_sorted_list(self, sort_by): + def get_questions_sorted_by_creation_time(self): return sorted(self.questions.values(), key=lambda question: question.id) class FakeClient: diff --git a/tests/test_my_utils.py b/tests/test_cli_helpers.py similarity index 89% rename from tests/test_my_utils.py rename to tests/test_cli_helpers.py index 5ba7855..999ddbd 100644 --- a/tests/test_my_utils.py +++ b/tests/test_cli_helpers.py @@ -1,5 +1,5 @@ -from leet2git.my_utils import get_question_id, reset_config, wait_to_finish_download -from leet2git.question_db import QuestionData +from leet2git.cli_helpers import get_question_id, reset_config, wait_to_finish_download +from leet2git.question_db import IdTitleMap, QuestionData class FakeConfigManager: @@ -39,7 +39,7 @@ def get_id_from_title(self, title_slug): class FakeClient: def get_id_title_map(self): - return type("IdTitleMap", (), {"title_to_id": {"two-sum": 1}})() + return IdTitleMap(title_to_id={"two-sum": 1}) qdb = FakeQuestionDB() diff --git a/tests/test_file_handler.py b/tests/test_file_handler.py index 9d2e2a4..4c29400 100644 --- a/tests/test_file_handler.py +++ b/tests/test_file_handler.py @@ -7,18 +7,15 @@ def test_generate_files_builds_source_and_tests(monkeypatch): class FakeClient: def get_question_data(self, qid, title_slug, language, code): - return ( - QuestionData( - id=qid, - title="Two Sum", - title_slug=title_slug, - file_path="src/leetcode_1_two_sum", - language=language, - inputs=["[2,7,11,15], 9"], - outputs=["[0,1]"], - raw_code=code, - ), - True, + return QuestionData( + id=qid, + title="Two Sum", + title_slug=title_slug, + file_path="src/leetcode_1_two_sum", + language=language, + inputs=["[2,7,11,15], 9"], + outputs=["[0,1]"], + raw_code=code, ) class FakeHandler: @@ -53,3 +50,177 @@ def get_question_data(self, *args): assert generated == {} assert "api changed" in capsys.readouterr().out + + +def test_generate_files_skips_generic_tests_for_custom_output_metadata(monkeypatch, capsys): + class FakeClient: + def get_question_data(self, qid, title_slug, language, code): + return QuestionData( + id=qid, + title="Remove Element", + title_slug=title_slug, + language=language, + inputs=["[3,2,2,3], 3"], + outputs=["2, nums = [2,2,_,_]"], + requires_custom_test_harness=True, + ) + + class FakeHandler: + def get_function_name(self): + return ["removeElement"] + + def generate_source(self): + return "src/leetcode_27_remove_element.py" + + def generate_tests(self): + raise AssertionError("generic tests must not be generated") + + monkeypatch.setattr("leet2git.file_handler.create_file_handler", lambda *_: FakeHandler()) + generated = {} + + generate_files(generated, 27, "remove-element", FakeClient(), 123.0, AppConfig()) + + assert generated[27].file_path == "src/leetcode_27_remove_element.py" + assert generated[27].test_file_path == "" + output = capsys.readouterr().out + assert "Soft error" in output + assert "custom or in-place output validation" in output + + +def test_generate_files_treats_judge_objects_as_soft_test_limitations(monkeypatch, capsys): + class FakeClient: + def get_question_data(self, qid, title_slug, language, code): + return QuestionData( + id=qid, + title="Vertical Traversal", + title_slug=title_slug, + language=language, + question_template="# class TreeNode:\n# pass\nclass Solution:\n pass\n", + inputs=["[3,9,20]"], + outputs=["[[9],[3],[20]]"], + ) + + class FakeHandler: + def get_function_name(self): + return ["verticalTraversal"] + + def generate_source(self): + return "src/leetcode_987_vertical_traversal.py" + + def generate_tests(self): + raise AssertionError("generic tests must not be generated") + + monkeypatch.setattr("leet2git.file_handler.create_file_handler", lambda *_: FakeHandler()) + generated = {} + + generate_files(generated, 987, "vertical-traversal", FakeClient(), 123.0, AppConfig()) + + assert generated[987].file_path == "src/leetcode_987_vertical_traversal.py" + assert generated[987].test_file_path == "" + output = capsys.readouterr().out + assert "Soft error" in output + assert "TreeNode" in output + + +def test_generate_files_preserves_source_when_callable_discovery_fails(monkeypatch, capsys): + class FakeClient: + def get_question_data(self, qid, title_slug, language, code): + return QuestionData( + id=qid, + title="Unusual Template", + title_slug=title_slug, + language=language, + inputs=["1"], + outputs=["1"], + ) + + class FakeHandler: + def get_function_name(self): + raise ValueError("no callable found") + + def generate_source(self): + return "src/leetcode_999_unusual_template.py" + + def generate_tests(self): + raise AssertionError("tests must be skipped without a callable") + + monkeypatch.setattr("leet2git.file_handler.create_file_handler", lambda *_: FakeHandler()) + generated = {} + + generate_files(generated, 999, "unusual-template", FakeClient(), 123.0, AppConfig()) + + assert generated[999].file_path == "src/leetcode_999_unusual_template.py" + assert generated[999].function_name == [] + assert "no callable found" in capsys.readouterr().out + + +def test_generate_files_cleans_partial_tests_after_soft_generation_error(tmp_path, monkeypatch, capsys): + class FakeClient: + def get_question_data(self, qid, title_slug, language, code): + return QuestionData( + id=qid, + title="Broken Example", + title_slug=title_slug, + language=language, + inputs=["1"], + outputs=["1"], + ) + + class FakeHandler: + entrypoint_removed = False + + def get_function_name(self): + return ["solve"] + + def generate_source(self): + return "src/leetcode_998_broken_example.py" + + def generate_tests(self): + test_path = tmp_path / "tests" / "test_998.py" + test_path.parent.mkdir(parents=True) + test_path.write_text("invalid test", encoding="UTF8") + raise SyntaxError("invalid generated assertion") + + def remove_test_entrypoint(self): + self.entrypoint_removed = True + + handler = FakeHandler() + monkeypatch.setattr("leet2git.file_handler.create_file_handler", lambda *_: handler) + generated = {} + config = AppConfig(source_path=str(tmp_path)) + + generate_files(generated, 998, "broken-example", FakeClient(), 123.0, config) + + assert generated[998].file_path == "src/leetcode_998_broken_example.py" + assert generated[998].test_file_path == "" + assert handler.entrypoint_removed is True + assert not (tmp_path / "tests" / "test_998.py").exists() + assert "SyntaxError" in capsys.readouterr().out + + +def test_generate_files_contains_source_generation_failures(monkeypatch, capsys): + class FakeClient: + def get_question_data(self, qid, title_slug, language, code): + return QuestionData( + id=qid, + title="No Source", + title_slug=title_slug, + language=language, + inputs=["1"], + outputs=["1"], + ) + + class FakeHandler: + def get_function_name(self): + return ["solve"] + + def generate_source(self): + raise OSError("disk full") + + monkeypatch.setattr("leet2git.file_handler.create_file_handler", lambda *_: FakeHandler()) + generated = {} + + generate_files(generated, 997, "no-source", FakeClient(), 123.0, AppConfig()) + + assert generated == {} + assert "Could not import source" in capsys.readouterr().out diff --git a/tests/test_leetcode_client.py b/tests/test_leetcode_client.py index 7817fd9..3a90ff4 100644 --- a/tests/test_leetcode_client.py +++ b/tests/test_leetcode_client.py @@ -44,7 +44,6 @@ def cookie_jar(*cookies: Cookie) -> CookieJar: def make_client(handler): client = LeetcodeClient.__new__(LeetcodeClient) - client.divider = "/" client.cookies = "" client.csrftoken = "csrf" client._transport = httpx.MockTransport(handler) @@ -57,6 +56,7 @@ def question_response( content="

desc

", sample_test_case="[1]\n1", example_testcases="[1]\n1", + meta_data=None, ): return { "data": { @@ -68,6 +68,7 @@ def question_response( "difficulty": "Easy", "exampleTestcases": example_testcases, "sampleTestCase": sample_test_case, + "metaData": meta_data, "topicTags": [{"name": "Array", "slug": "array"}], "codeSnippets": [ { @@ -123,20 +124,6 @@ def test_get_cookies_continues_after_browser_cookie_error(monkeypatch, capsys): assert "chrome unavailable" in capsys.readouterr().out -def test_init_sets_windows_path_divider(monkeypatch): - chrome_cookies = cookie_jar(cookie("LEETCODE_SESSION", "session", ".leetcode.com")) - monkeypatch.setattr("leet2git.leetcode_client.platform.system", lambda: "Windows") - monkeypatch.setattr("leet2git.leetcode_client.browser_cookie3.chrome", lambda: chrome_cookies) - monkeypatch.setattr( - "leet2git.leetcode_client.browser_cookie3.firefox", - lambda: pytest.fail("Firefox should not be used when Chrome has a session"), - ) - - client = LeetcodeClient() - - assert client.divider == "\\" - - def test_get_cookies_requires_leetcode_session(monkeypatch): chrome_cookies = cookie_jar(cookie("csrftoken", "csrf", "leetcode.com")) firefox_cookies = cookie_jar() @@ -191,6 +178,24 @@ def test_get_headers_include_cookie_and_conditional_csrf(): assert headers["user-agent"] +def test_init_supports_public_requests_without_browser_cookies(monkeypatch): + monkeypatch.setattr( + "leet2git.leetcode_client.browser_cookie3.chrome", + lambda: pytest.fail("Public clients must not inspect Chrome cookies"), + ) + monkeypatch.setattr( + "leet2git.leetcode_client.browser_cookie3.firefox", + lambda: pytest.fail("Public clients must not inspect Firefox cookies"), + ) + + client = LeetcodeClient(use_browser_cookies=False) + + assert client.cookies == "" + assert client.csrftoken == "" + assert "cookie" not in client.get_headers() + assert "x-csrftoken" not in client.get_headers() + + def test_get_question_data_parses_description_examples_categories_and_raw_code(): long_line = " ".join(["word"] * 30) content = ( @@ -218,9 +223,8 @@ def handler(request: httpx.Request) -> httpx.Response: client = make_client(handler) - data, is_new = client.get_question_data(1, "two-sum", "python3", "accepted code") + data = client.get_question_data(1, "two-sum", "python3", "accepted code") - assert is_new is True assert data.internal_id == 1 assert data.title == "Two Sum" assert data.url == "https://leetcode.com/problems/two-sum" @@ -234,6 +238,111 @@ def handler(request: httpx.Request) -> httpx.Response: assert all(len(line) <= 100 for line in data.description if line.startswith("word")) +def test_get_question_data_parses_complete_multiline_outputs(): + content = ( + '

Example 1:

\n' + "
\n"
+        "Input: nums = [1,1,2]\n"
+        "Output:\n"
+        "[[1,1,2],\n"
+        " [1,2,1],\n"
+        " [2,1,1]]\n"
+        "Explantion: The permutations may be returned in any order.\n"
+        "
\n" + '

Example 2:

\n' + "
\n"
+        "Input: nums = [1,2,3]\n"
+        "Output: [[1,2,3],[1,3,2]]\n"
+        "
" + ) + + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response( + 200, + json=question_response( + content=content, + sample_test_case="[1,1,2]", + example_testcases="[1,1,2]\n[1,2,3]", + ), + ) + + data = make_client(handler).get_question_data(47, "permutations-ii", "python3") + + assert data.inputs == ["[1,1,2]", "[1,2,3]"] + assert data.outputs == [ + "[[1,1,2], [1,2,1], [2,1,1]]", + "[[1,2,3],[1,3,2]]", + ] + + +def test_get_question_data_falls_back_to_statement_inputs_when_graphql_omits_examples(): + content = """ +

Example 1:

+
Input: nums = [7,12,9,8,9,15], k = 4
+Output: 9
+

Example 2:

+
Input: nums = [2,12,1,11,4,5], k = 6
+Output: 0
+

Example 3:

+
Input: nums = [10,8,5,9,11,6,8], k = 1
+Output: 15
+""" + + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response( + 200, + json=question_response( + content=content, + sample_test_case="[7,12,9,8,9,15]\n4", + example_testcases="[7,12,9,8,9,15]\n4", + ), + ) + + data = make_client(handler).get_question_data(2917, "find-the-k-or-of-an-array", "python3") + + assert data.inputs == [ + "nums = [7,12,9,8,9,15], k = 4", + "nums = [2,12,1,11,4,5], k = 6", + "nums = [10,8,5,9,11,6,8], k = 1", + ] + assert data.outputs == ["9", "0", "15"] + + +def test_get_question_data_marks_custom_output_metadata_as_unsupported(): + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response( + 200, + json=question_response( + meta_data=json.dumps( + { + "name": "removeElement", + "output": {"paramindex": 0, "size": "ret"}, + } + ) + ), + ) + + data = make_client(handler).get_question_data(27, "remove-element", "python3") + + assert data.requires_custom_test_harness is True + + +def test_get_question_data_accepts_ordinary_metadata_and_fails_closed_on_malformed_metadata(): + responses = iter( + [ + question_response(meta_data=json.dumps({"name": "twoSum"})), + question_response(meta_data="not-json"), + ] + ) + client = make_client(lambda _: httpx.Response(200, json=next(responses))) + + ordinary = client.get_question_data(1, "two-sum", "python3") + malformed = client.get_question_data(1, "two-sum", "python3") + + assert ordinary.requires_custom_test_harness is False + assert malformed.requires_custom_test_harness is True + + def test_mutating_requests_require_csrf(): client = make_client(lambda _: httpx.Response(500)) client.cookies = "LEETCODE_SESSION=session" @@ -252,6 +361,29 @@ def test_mutating_requests_require_csrf(): ) +def test_async_submit_question_reports_rejected_browser_session(): + requests = [] + + def handler(request: httpx.Request) -> httpx.Response: + requests.append((request.method, str(request.url))) + return httpx.Response(200, json={"error": "User is not authenticated"}) + + client = make_client(handler) + + with pytest.raises(LeetcodeAuthError, match="Log in") as exc_info: + asyncio.run( + client.async_submit_question( + "class Solution: ...", + 1, + "two-sum", + "python3", + ) + ) + + assert "submission_id" not in str(exc_info.value) + assert requests == [("POST", "https://leetcode.com/problems/two-sum/submit/")] + + def test_async_scrap_question_data_posts_current_graphql_shape(): def handler(request: httpx.Request) -> httpx.Response: payload = json.loads(request.content) @@ -293,6 +425,89 @@ def handler(request: httpx.Request) -> httpx.Response: assert response.data.question.question_id == "1" +@pytest.mark.parametrize("status_code", [404, 500, 502, 503, 504]) +def test_async_scrap_question_data_retries_one_transient_endpoint_error(status_code, monkeypatch): + request_count = 0 + sleep_delays = [] + + async def fake_sleep(delay: float) -> None: + sleep_delays.append(delay) + + def handler(request: httpx.Request) -> httpx.Response: + nonlocal request_count + request_count += 1 + if request_count == 1: + return httpx.Response(status_code) + return httpx.Response(200, json=question_response()) + + monkeypatch.setattr("leet2git.leetcode_client.asyncio.sleep", fake_sleep) + response = asyncio.run(make_client(handler).async_scrap_question_data("two-sum")) + + assert response.data.question is not None + assert request_count == 2 + assert sleep_delays == [1.0] + + +def test_async_scrap_question_data_stops_after_one_failed_retry(monkeypatch): + request_count = 0 + sleep_delays = [] + + async def fake_sleep(delay: float) -> None: + sleep_delays.append(delay) + + def handler(request: httpx.Request) -> httpx.Response: + nonlocal request_count + request_count += 1 + return httpx.Response(503) + + monkeypatch.setattr("leet2git.leetcode_client.asyncio.sleep", fake_sleep) + + with pytest.raises(LeetcodeAPIError, match="HTTP 503"): + asyncio.run(make_client(handler).async_scrap_question_data("two-sum")) + + assert request_count == 2 + assert sleep_delays == [1.0] + + +@pytest.mark.parametrize("status_code", [400, 403, 429, 501]) +def test_async_scrap_question_data_does_not_retry_blocking_responses(status_code, monkeypatch): + request_count = 0 + + async def fail_if_slept(_delay: float) -> None: + pytest.fail("blocking responses must not be retried") + + def handler(request: httpx.Request) -> httpx.Response: + nonlocal request_count + request_count += 1 + return httpx.Response(status_code) + + monkeypatch.setattr("leet2git.leetcode_client.asyncio.sleep", fail_if_slept) + + with pytest.raises(LeetcodeAPIError, match=f"HTTP {status_code}"): + asyncio.run(make_client(handler).async_scrap_question_data("two-sum")) + + assert request_count == 1 + + +def test_get_question_data_does_not_retry_missing_question_payload(monkeypatch): + request_count = 0 + + async def fail_if_slept(_delay: float) -> None: + pytest.fail("a valid not-found payload must not be retried") + + def handler(request: httpx.Request) -> httpx.Response: + nonlocal request_count + request_count += 1 + return httpx.Response(200, json={"data": {"question": None}}) + + monkeypatch.setattr("leet2git.leetcode_client.asyncio.sleep", fail_if_slept) + + with pytest.raises(LeetcodeAPIError, match='could not find question "missing"'): + make_client(handler).get_question_data(9999, "missing", "python3") + + assert request_count == 1 + + def test_async_get_id_title_map_parses_problem_list(): def handler(request: httpx.Request) -> httpx.Response: assert request.method == "GET" diff --git a/tests/test_python_handler.py b/tests/test_python_handler.py index 93daeea..bdfd661 100644 --- a/tests/test_python_handler.py +++ b/tests/test_python_handler.py @@ -1,3 +1,6 @@ +import ast +import runpy + import pytest from leet2git.config_manager import AppConfig @@ -97,6 +100,136 @@ def test_generate_source_adds_test_entrypoint_when_enabled(tmp_path, monkeypatch assert "pytest.main([os.path.join('tests', 'test_1.py')])" in content +def test_generate_source_omits_test_entrypoint_for_soft_error_import(tmp_path, monkeypatch): + question = QuestionData( + id=987, + title="Vertical Traversal", + file_path="src/leetcode_987_vertical_traversal", + language="python3", + question_template=( + "# class TreeNode:\n" + "# pass\n" + "class Solution:\n" + " def verticalTraversal(self, root: TreeNode | None):\n" + ), + requires_custom_test_harness=True, + ) + handler = make_handler(question, AppConfig(source_path=str(tmp_path))) + monkeypatch.setattr("leet2git.python_handler.fix_files", lambda _: None) + monkeypatch.setattr(handler, "run_formatter", lambda _: None) + + file_path = handler.generate_source() + + content = (tmp_path / file_path).read_text(encoding="UTF8") + assert "class Solution" in content + assert 'if __name__ == "__main__"' not in content + + +def test_remove_test_entrypoint_preserves_generated_solution(tmp_path, monkeypatch): + question = QuestionData( + id=1, + title="Two Sum", + file_path="src/leetcode_1_two_sum", + language="python3", + question_template="class Solution:\n def twoSum(self, nums, target):\n", + ) + handler = make_handler(question, AppConfig(source_path=str(tmp_path))) + monkeypatch.setattr("leet2git.python_handler.fix_files", lambda _: None) + monkeypatch.setattr(handler, "run_formatter", lambda _: None) + question.file_path = str(handler.generate_source()) + + handler.remove_test_entrypoint() + + content = (tmp_path / question.file_path).read_text(encoding="UTF8") + assert "class Solution" in content + assert 'if __name__ == "__main__"' not in content + + +@pytest.mark.parametrize("judge_type", ["TreeNode", "ListNode"]) +def test_generate_source_defers_judge_type_annotations(judge_type, tmp_path, monkeypatch): + question = QuestionData( + id=1, + title="Judge Type", + file_path="src/leetcode_1_judge_type", + language="python3", + question_template=( + f"# class {judge_type}:\n" + "# pass\n" + "class Solution:\n" + f" def visit(self, node: {judge_type} | None) -> {judge_type} | None:\n" + ), + ) + handler = make_handler( + question, + AppConfig( + source_path=str(tmp_path), + test_code=LeetTestCodeConfig(generate_tests=False), + ), + ) + monkeypatch.setattr("leet2git.python_handler.fix_files", lambda _: None) + monkeypatch.setattr(handler, "run_formatter", lambda _: None) + + source_path = tmp_path / handler.generate_source() + content = source_path.read_text(encoding="UTF8") + + assert content.count("from __future__ import annotations") == 1 + namespace = runpy.run_path(str(source_path), run_name="generated_solution") + assert "Solution" in namespace + + +def test_generate_source_keeps_module_docstring_before_future_import(tmp_path, monkeypatch): + question = QuestionData( + id=1, + title="Documented", + file_path="src/leetcode_1_documented", + language="python3", + question_template=( + '"""Starter module documentation."""\n\nclass Solution:\n def solve(self):\n' + ), + ) + handler = make_handler( + question, + AppConfig( + source_path=str(tmp_path), + test_code=LeetTestCodeConfig(generate_tests=False), + ), + ) + monkeypatch.setattr("leet2git.python_handler.fix_files", lambda _: None) + monkeypatch.setattr(handler, "run_formatter", lambda _: None) + + source_path = tmp_path / handler.generate_source() + content = source_path.read_text(encoding="UTF8") + + assert content.index('"""Starter module documentation."""') < content.index( + "from __future__ import annotations" + ) + namespace = runpy.run_path(str(source_path), run_name="generated_solution") + assert namespace["__doc__"] == "Starter module documentation." + + +def test_generate_source_does_not_duplicate_existing_annotations_future(tmp_path, monkeypatch): + question = QuestionData( + id=1, + title="Future", + file_path="src/leetcode_1_future", + language="python3", + raw_code=("from __future__ import annotations\n\nclass Solution:\n pass\n"), + ) + handler = make_handler( + question, + AppConfig( + source_path=str(tmp_path), + test_code=LeetTestCodeConfig(generate_tests=False), + ), + ) + monkeypatch.setattr("leet2git.python_handler.fix_files", lambda _: None) + monkeypatch.setattr(handler, "run_formatter", lambda _: None) + + source_path = tmp_path / handler.generate_source() + + assert source_path.read_text(encoding="UTF8").count("from __future__ import annotations") == 1 + + def test_generate_tests_creates_single_function_pytest_file(tmp_path, monkeypatch): question = QuestionData( id=1, @@ -119,6 +252,71 @@ def test_generate_tests_creates_single_function_pytest_file(tmp_path, monkeypatc assert "twoSum([3,2,4], 6) == [1, 2]" in content +def test_generate_tests_creates_constructor_style_pytest_file(tmp_path, monkeypatch): + question = QuestionData( + id=2013, + title="Detect Squares", + file_path="src/leetcode_2013_detect_squares.py", + language="python3", + question_template=( + "class DetectSquares:\n" + " def __init__(self):\n" + " pass\n" + " def add(self, point):\n" + " pass\n" + " def count(self, point):\n" + " pass\n" + ), + inputs=[ + '["DetectSquares","add","add","add","count","count","add","count"], ' + "[[],[[3,10]],[[11,2]],[[3,2]],[[11,10]],[[14,8]],[[11,2]],[[11,10]]]" + ], + outputs=["[null,null,null,null,1,0,null,2]"], + ) + handler = make_handler(question, AppConfig(source_path=str(tmp_path))) + question.function_name = handler.get_function_name() + monkeypatch.setattr(handler, "run_formatter", lambda _: None) + + test_file_path = handler.generate_tests() + + content = (tmp_path / test_file_path).read_text(encoding="UTF8") + assert question.function_name == ["DetectSquares", "add", "count"] + assert "from src.leetcode_2013_detect_squares import DetectSquares" in content + assert "return DetectSquares(*args)" in content + assert "solution = init_variables_2013()" in content + assert "solution = add()" not in content + assert "add([3, 10]) is None" in content + assert "count([11, 10]) == 1" in content + assert "count([11, 10]) == 2" in content + ast.parse(content) + + +def test_generate_tests_uses_each_design_cases_constructor_arguments(tmp_path, monkeypatch): + question = QuestionData( + id=1000, + title="Accumulator", + file_path="src/leetcode_1000_accumulator.py", + language="python3", + function_name=["Accumulator", "add"], + inputs=[ + '["Accumulator","add"], [[1],[2]]', + '["Accumulator","add"], [[10],[2]]', + ], + outputs=["[null,3]", "[null,12]"], + ) + handler = make_handler(question, AppConfig(source_path=str(tmp_path))) + monkeypatch.setattr(handler, "run_formatter", lambda _: None) + + test_file_path = handler.generate_tests() + + content = (tmp_path / test_file_path).read_text(encoding="UTF8") + assert "solution = init_variables_1000(1)" in content + assert "solution = init_variables_1000(10)" in content + assert "solution.add(2) == 3" in content + assert "solution.add(2) == 12" in content + ast.parse(content) + + def test_generate_tests_requires_function_name(tmp_path): handler = make_handler( QuestionData( diff --git a/tests/test_question_db.py b/tests/test_question_db.py index c7f4982..f6752a3 100644 --- a/tests/test_question_db.py +++ b/tests/test_question_db.py @@ -38,14 +38,18 @@ def test_question_data_validates_and_coerces_fields(): def test_question_db_round_trips_pydantic_models(tmp_path): config = make_config(tmp_path) question_db = QuestionDB(config) - question_db.add_question(QuestionData(id=1, title="Two Sum")) + question_db.add_question(QuestionData(id=1, title="Two Sum", requires_custom_test_harness=True)) question_db.set_id_title_map(IdTitleMap(id_to_title={1: "two-sum"}, title_to_id={"two-sum": 1})) question_db.save() loaded_db = QuestionDB(config) loaded_db.load() - assert loaded_db.get_question(1) == QuestionData(id=1, title="Two Sum") + assert loaded_db.get_question(1) == QuestionData( + id=1, + title="Two Sum", + requires_custom_test_harness=True, + ) assert loaded_db.get_title_from_id(1) == "two-sum" assert loaded_db.get_id_from_title("two-sum") == 1 @@ -63,7 +67,10 @@ def test_question_db_loads_legacy_dict_payloads(tmp_path): question_db.load() - assert question_db.get_question(1) == QuestionData(id=1, title="Two Sum") + question = question_db.get_question(1) + assert question == QuestionData(id=1, title="Two Sum") + assert question is not None + assert question.requires_custom_test_harness is False assert question_db.get_title_from_id(1) == "two-sum" assert question_db.get_id_from_title("two-sum") == 1 assert question_db.migrated_from_legacy is True diff --git a/tests/test_sample_leetcode_imports.py b/tests/test_sample_leetcode_imports.py new file mode 100644 index 0000000..5f5e40f --- /dev/null +++ b/tests/test_sample_leetcode_imports.py @@ -0,0 +1,450 @@ +import random + +import pytest +from scripts.sample_leetcode_imports import ( + Candidate, + ProblemResult, + SamplerSettings, + build_population, + inspect_problem, + run_sampler, + varied_candidate_order, +) + +from leet2git.leetcode_models import ProblemListResponse +from leet2git.question_db import QuestionData + + +def make_catalog(*rows: tuple[int, str, int, bool]) -> ProblemListResponse: + return ProblemListResponse.model_validate( + { + "stat_status_pairs": [ + { + "stat": { + "frontend_question_id": question_id, + "question__title_slug": slug, + }, + "difficulty": {"level": difficulty}, + "paid_only": paid, + } + for question_id, slug, difficulty, paid in rows + ] + } + ) + + +def test_build_population_excludes_paid_and_assigns_id_eras(): + catalog = make_catalog( + (1, "one", 1, False), + (2, "two", 2, True), + (100, "hundred", 3, False), + ) + + population, paid_count = build_population(catalog) + + assert paid_count == 1 + assert [(candidate.question_id, candidate.difficulty) for candidate in population] == [ + (1, "Easy"), + (100, "Hard"), + ] + assert population[0].era == 0 + assert population[-1].era == 3 + + +def test_varied_candidate_order_is_seeded_and_round_robins_strata(): + population = [ + Candidate( + question_id=era * 10 + difficulty, + slug=f"problem-{era}-{difficulty}", + difficulty={1: "Easy", 2: "Medium", 3: "Hard"}[difficulty], + era=era, + ) + for era in range(6) + for difficulty in range(1, 4) + ] + + first = varied_candidate_order(population, random.Random(42)) + second = varied_candidate_order(population, random.Random(42)) + + assert first == second + assert len({(candidate.difficulty, candidate.era) for candidate in first[:18]}) == 18 + + +@pytest.mark.parametrize( + "settings", + [ + {"percentage": 0}, + {"max_seconds": 3596}, + {"min_delay": 0.5}, + {"min_delay": 5, "max_delay": 4}, + {"language": "javascript"}, + ], +) +def test_sampler_settings_reject_unsafe_or_unsupported_values(settings): + with pytest.raises(ValueError): + SamplerSettings(**settings) + + +def test_inspect_problem_generates_and_validates_python_files(tmp_path, monkeypatch): + class FakeClient: + def get_question_data(self, question_id, slug, language): + return QuestionData( + id=question_id, + internal_id=1, + title="Two Sum", + title_slug=slug, + difficulty="Easy", + file_path="src/leetcode_1_two_sum", + language=language, + question_template=("class Solution:\n def twoSum(self, nums, target):\n"), + inputs=["[2,7,11,15], 9"], + outputs=["[0,1]"], + ) + + monkeypatch.setattr("leet2git.python_handler.fix_files", lambda _: None) + monkeypatch.setattr("leet2git.python_handler.PythonHandler.run_formatter", lambda *_: None) + candidate = Candidate(1, "two-sum", "Easy", 0) + + result = inspect_problem(candidate, tmp_path, "python3", FakeClient()) + + assert result.status == "passed" + assert result.stage == "complete" + assert result.function_names == ["twoSum"] + assert result.test_generated is True + assert (tmp_path / "src" / "leetcode_1_two_sum.py").is_file() + assert (tmp_path / "tests" / "test_1.py").is_file() + + +def test_inspect_problem_reports_non_module_constructor_as_soft_error(tmp_path, monkeypatch): + class FakeClient: + def get_question_data(self, question_id, slug, language): + return QuestionData( + id=question_id, + internal_id=297, + title="Serialize and Deserialize Binary Tree", + title_slug=slug, + difficulty="Hard", + file_path="src/leetcode_297_serialize_and_deserialize_binary_tree", + language=language, + question_template=( + "class Codec:\n" + " def serialize(self, root):\n" + " pass\n" + " def deserialize(self, data):\n" + " pass\n" + ), + inputs=['["Codec","serialize","deserialize"], [[],[None],[""]]'], + outputs=['[null,"",null]'], + ) + + monkeypatch.setattr("leet2git.python_handler.fix_files", lambda _: None) + monkeypatch.setattr("leet2git.python_handler.PythonHandler.run_formatter", lambda *_: None) + candidate = Candidate(297, "serialize-and-deserialize-binary-tree", "Hard", 1) + + result = inspect_problem(candidate, tmp_path, "python3", FakeClient()) + + assert result.status == "soft_error" + assert result.stage == "test_generation" + assert "not module-level" in result.message + + +def test_inspect_problem_classifies_judge_supplied_types_as_unsupported_tests(tmp_path, monkeypatch): + class FakeClient: + def get_question_data(self, question_id, slug, language): + return QuestionData( + id=question_id, + internal_id=987, + title="Vertical Traversal", + title_slug=slug, + difficulty="Hard", + file_path="src/leetcode_987_vertical_traversal", + language=language, + question_template=( + "# class TreeNode:\n" + "# pass\n" + "class Solution:\n" + " def verticalTraversal(self, root: TreeNode | None):\n" + ), + inputs=["root = [3,9,20,None,None,15,7]"], + outputs=["[[9],[3,15],[20],[7]]"], + ) + + monkeypatch.setattr("leet2git.python_handler.fix_files", lambda _: None) + monkeypatch.setattr("leet2git.python_handler.PythonHandler.run_formatter", lambda *_: None) + candidate = Candidate(987, "vertical-order-traversal-of-a-binary-tree", "Hard", 1) + + result = inspect_problem(candidate, tmp_path, "python3", FakeClient()) + + assert result.status == "soft_error" + assert result.stage == "test_generation" + assert result.error_type == "JudgeHarnessUnsupported" + assert "TreeNode" in result.message + assert (tmp_path / "src" / "leetcode_987_vertical_traversal.py").is_file() + assert not (tmp_path / "tests" / "test_987.py").exists() + + +def test_inspect_problem_classifies_custom_output_metadata_as_unsupported(tmp_path, monkeypatch): + class FakeClient: + def get_question_data(self, question_id, slug, language): + return QuestionData( + id=question_id, + internal_id=27, + title="Remove Element", + title_slug=slug, + difficulty="Easy", + file_path="src/leetcode_27_remove_element", + language=language, + question_template=( + "class Solution:\n def removeElement(self, nums, val):\n pass\n" + ), + inputs=["[3,2,2,3], 3"], + outputs=["2, nums = [2,2,_,_]"], + requires_custom_test_harness=True, + ) + + monkeypatch.setattr("leet2git.python_handler.fix_files", lambda _: None) + monkeypatch.setattr("leet2git.python_handler.PythonHandler.run_formatter", lambda *_: None) + candidate = Candidate(27, "remove-element", "Easy", 0) + + result = inspect_problem(candidate, tmp_path, "python3", FakeClient()) + + assert result.status == "soft_error" + assert result.stage == "test_generation" + assert result.error_type == "JudgeHarnessUnsupported" + assert "custom or in-place output validation" in result.message + assert (tmp_path / "src" / "leetcode_27_remove_element.py").is_file() + assert not (tmp_path / "tests" / "test_27.py").exists() + + +class FakeClock: + def __init__(self) -> None: + self.now = 0.0 + + def __call__(self) -> float: + return self.now + + def sleep(self, seconds: float) -> None: + self.now += seconds + + +class FakeCatalogClient: + def __init__(self, catalog: ProblemListResponse): + self.catalog = catalog + + def get_problem_list(self) -> ProblemListResponse: + return self.catalog + + +def passing_result(candidate: Candidate, *_args) -> ProblemResult: + return ProblemResult( + question_id=candidate.question_id, + slug=candidate.slug, + difficulty=candidate.difficulty, + era=candidate.era, + status="passed", + stage="complete", + topics=["array"], + ) + + +def soft_error_result(candidate: Candidate, *_args) -> ProblemResult: + return ProblemResult( + question_id=candidate.question_id, + slug=candidate.slug, + difficulty=candidate.difficulty, + era=candidate.era, + status="soft_error", + stage="test_generation", + error_type="JudgeHarnessUnsupported", + message="LeetCode supplies judge-only class definitions: TreeNode", + topics=["binary-tree"], + ) + + +def test_run_sampler_reaches_percentage_target_with_seeded_pacing(tmp_path): + catalog = make_catalog( + (1, "one", 1, False), + (2, "two", 2, False), + (3, "three", 3, False), + (4, "four", 1, False), + ) + clock = FakeClock() + progress: list[str] = [] + + report = run_sampler( + SamplerSettings( + percentage=50, + max_seconds=30, + min_delay=1, + max_delay=1, + item_timeout=5, + ), + seed=7, + catalog_client=FakeCatalogClient(catalog), + problem_runner=passing_result, + clock=clock, + sleeper=clock.sleep, + progress=progress.append, + keep_artifacts=tmp_path, + ) + + assert report.target_imports == 2 + assert report.attempted == 2 + assert report.imported == 2 + assert report.duration_seconds == 2 + assert report.exit_code() == 0 + assert report.topic_coverage == ["array"] + assert progress[0].startswith("seed=7") + + +def test_run_sampler_counts_soft_errors_as_imported_and_exits_successfully(tmp_path): + catalog = make_catalog((1, "one", 1, False)) + clock = FakeClock() + + report = run_sampler( + SamplerSettings( + percentage=100, + max_seconds=10, + min_delay=1, + max_delay=1, + item_timeout=5, + ), + seed=3, + catalog_client=FakeCatalogClient(catalog), + problem_runner=soft_error_result, + clock=clock, + sleeper=clock.sleep, + progress=lambda _: None, + keep_artifacts=tmp_path, + ) + + assert report.imported == 1 + assert report.soft_errors == 1 + assert report.passed == 0 + assert report.failed == 0 + assert report.exit_code() == 0 + assert report.topic_coverage == ["binary-tree"] + + +def test_sampler_catalog_failure_is_incomplete_not_success(tmp_path): + class FailingCatalogClient: + def get_problem_list(self): + raise RuntimeError("catalog unavailable") + + report = run_sampler( + SamplerSettings(max_seconds=10, min_delay=1, max_delay=1), + seed=4, + catalog_client=FailingCatalogClient(), + progress=lambda _: None, + keep_artifacts=tmp_path, + ) + + assert report.stop_reason == "catalog: catalog unavailable" + assert report.exit_code() == 2 + + +def test_completed_sampler_reports_hard_failures_even_after_reaching_target(tmp_path): + catalog = make_catalog((1, "one", 1, False), (2, "two", 2, False)) + clock = FakeClock() + calls = 0 + + def fail_then_pass(candidate: Candidate, *_args) -> ProblemResult: + nonlocal calls + calls += 1 + if calls == 1: + return ProblemResult.failure(candidate, "source_generation", OSError("disk error")) + return passing_result(candidate) + + report = run_sampler( + SamplerSettings( + percentage=50, + max_seconds=20, + min_delay=1, + max_delay=1, + item_timeout=5, + ), + seed=5, + catalog_client=FakeCatalogClient(catalog), + problem_runner=fail_then_pass, + clock=clock, + sleeper=clock.sleep, + progress=lambda _: None, + keep_artifacts=tmp_path, + ) + + assert report.imported == report.target_imports == 1 + assert report.failed == 1 + assert report.passed == 1 + assert report.exit_code() == 1 + + +def test_run_sampler_stops_before_deadline_without_shortening_delay(tmp_path): + catalog = make_catalog( + (1, "one", 1, False), + (2, "two", 2, False), + (3, "three", 3, False), + (4, "four", 1, False), + ) + clock = FakeClock() + + report = run_sampler( + SamplerSettings( + percentage=50, + max_seconds=7, + min_delay=2, + max_delay=2, + item_timeout=1, + ), + seed=9, + catalog_client=FakeCatalogClient(catalog), + problem_runner=passing_result, + clock=clock, + sleeper=clock.sleep, + progress=lambda _: None, + keep_artifacts=tmp_path, + ) + + assert report.attempted == 1 + assert report.imported == 1 + assert report.duration_seconds <= 7 + assert "deadline" in report.stop_reason + assert report.exit_code() == 2 + + +def test_run_sampler_stops_immediately_on_rate_limit(tmp_path): + catalog = make_catalog((1, "one", 1, False), (2, "two", 2, False)) + clock = FakeClock() + + def rate_limited(candidate: Candidate, *_args) -> ProblemResult: + return ProblemResult( + question_id=candidate.question_id, + slug=candidate.slug, + difficulty=candidate.difficulty, + era=candidate.era, + status="failed", + stage="fetch", + error_type="LeetcodeAPIError", + message="LeetCode request failed with HTTP 429", + ) + + report = run_sampler( + SamplerSettings( + percentage=100, + max_seconds=30, + min_delay=1, + max_delay=1, + item_timeout=5, + ), + seed=2, + catalog_client=FakeCatalogClient(catalog), + problem_runner=rate_limited, + clock=clock, + sleeper=clock.sleep, + progress=lambda _: None, + keep_artifacts=tmp_path, + ) + + assert report.attempted == 1 + assert report.failed == 1 + assert "rate-limited" in report.stop_reason + assert report.exit_code() == 2 diff --git a/tests/test_test_harness.py b/tests/test_test_harness.py new file mode 100644 index 0000000..9868a88 --- /dev/null +++ b/tests/test_test_harness.py @@ -0,0 +1,27 @@ +import pytest + +from leet2git.leetcode_models import TopicTag +from leet2git.question_db import QuestionData +from leet2git.test_harness import get_local_test_limitation + + +@pytest.mark.parametrize("judge_type", ["TreeNode", "ListNode", "NestedInteger"]) +def test_local_test_limitation_detects_judge_provided_classes(judge_type): + data = QuestionData(question_template=f"# class {judge_type}:\n# pass\n") + + assert judge_type in get_local_test_limitation(data) + + +@pytest.mark.parametrize("topic", ["interactive", "concurrency"]) +def test_local_test_limitation_detects_judge_harness_topics(topic): + data = QuestionData(categories=[TopicTag(name=topic.title(), slug=topic)]) + + assert topic in get_local_test_limitation(data) + + +def test_local_test_limitation_detects_custom_output_and_accepts_ordinary_problem(): + custom = QuestionData(requires_custom_test_harness=True) + ordinary = QuestionData(question_template="class Solution:\n pass\n") + + assert "custom or in-place" in get_local_test_limitation(custom) + assert get_local_test_limitation(ordinary) == "" diff --git a/uv.lock b/uv.lock index 6096ab0..237e10e 100644 --- a/uv.lock +++ b/uv.lock @@ -260,30 +260,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/13/37/a065dc3bd6e49423a6532c642ca7378d3f467b1ef44c2800c937af7f9739/filelock-3.29.4-py3-none-any.whl", hash = "sha256:dac1648087d5115554850d113e7dd8c83ab2d38e3435dde2d4f163847e57b767", size = 42757, upload-time = "2026-06-13T16:11:59.582Z" }, ] -[[package]] -name = "gitdb" -version = "4.0.12" -source = { registry = "https://pypi.org/simple" } -dependencies = [ - { name = "smmap" }, -] -sdist = { url = "https://files.pythonhosted.org/packages/72/94/63b0fc47eb32792c7ba1fe1b694daec9a63620db1e313033d18140c2320a/gitdb-4.0.12.tar.gz", hash = "sha256:5ef71f855d191a3326fcfbc0d5da835f26b13fbcba60c32c21091c349ffdb571", size = 394684, upload-time = "2025-01-02T07:20:46.413Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/a0/61/5c78b91c3143ed5c14207f463aecfc8f9dbb5092fb2869baf37c273b2705/gitdb-4.0.12-py3-none-any.whl", hash = "sha256:67073e15955400952c6565cc3e707c554a4eea2e428946f7a4c162fab9bd9bcf", size = 62794, upload-time = "2025-01-02T07:20:43.624Z" }, -] - -[[package]] -name = "gitpython" -version = "3.1.50" -source = { registry = "https://pypi.org/simple" } -dependencies = [ - { name = "gitdb" }, -] -sdist = { url = "https://files.pythonhosted.org/packages/33/f6/354ae6491228b5eb40e10d89c4d13c651fe1cf7556e35ebdded50cff57ce/gitpython-3.1.50.tar.gz", hash = "sha256:80da2d12504d52e1f998772dc5baf6e553f8d2fcfe1fcc226c9d9a2ee3372dcc", size = 219798, upload-time = "2026-05-06T04:01:26.571Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/20/7a/1c6e3562dfd8950adbb11ffbc65d21e7c89d01a6e4f137fa981056de25c5/gitpython-3.1.50-py3-none-any.whl", hash = "sha256:d352abe2908d07355014abdd21ddf798c2a961469239afec4962e9da884858f9", size = 212507, upload-time = "2026-05-06T04:01:23.799Z" }, -] - [[package]] name = "h11" version = "0.16.0" @@ -365,7 +341,6 @@ dependencies = [ { name = "beautifulsoup4" }, { name = "browser-cookie3" }, { name = "click" }, - { name = "gitpython" }, { name = "httpx" }, { name = "platformdirs" }, { name = "pydantic" }, @@ -389,7 +364,6 @@ requires-dist = [ { name = "browser-cookie3", specifier = ">=0.20.1" }, { name = "build", marker = "extra == 'dev'", specifier = ">=1.3.0" }, { name = "click", specifier = ">=8.3.1" }, - { name = "gitpython", specifier = ">=3.1.45" }, { name = "httpx", specifier = ">=0.28.1" }, { name = "platformdirs", specifier = ">=4.10.0" }, { name = "pre-commit", marker = "extra == 'dev'", specifier = ">=4.5.0" }, @@ -880,15 +854,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/ad/32/fdea8e7f7b2b8bae13ee6ab7df1a10d28a24dbeb03a62ff033563f95d77c/shadowcopy-0.0.4-py3-none-any.whl", hash = "sha256:fc51e59a639dc6a5a3a7a9b4e3ecadc71989e339f2d995d90aaa491acd4ba4eb", size = 4212, upload-time = "2023-07-08T00:01:34.042Z" }, ] -[[package]] -name = "smmap" -version = "5.0.3" -source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/1f/ea/49c993d6dfdd7338c9b1000a0f36817ed7ec84577ae2e52f890d1a4ff909/smmap-5.0.3.tar.gz", hash = "sha256:4d9debb8b99007ae47165abc08670bd74cb74b5227dda7f643eccc4e9eb5642c", size = 22506, upload-time = "2026-03-09T03:43:26.1Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/c1/d4/59e74daffcb57a07668852eeeb6035af9f32cbfd7a1d2511f17d2fe6a738/smmap-5.0.3-py3-none-any.whl", hash = "sha256:c106e05d5a61449cf6ba9a1e650227ecfb141590d2a98412103ff35d89fc7b2f", size = 24390, upload-time = "2026-03-09T03:43:24.361Z" }, -] - [[package]] name = "soupsieve" version = "2.8.4"