tests/test_file_walker.py

"""M04 — File Tree Walker + Module Name Resolver. Red/green tests from the spec.

Success criteria under test:
- Walks a repo dir and returns one record per ``.py`` file with a resolved dotted
  module name (nested package ``pkg`` / ``pkg.sub`` / ``pkg.nested.deep``).
- A namespace dir (no ``__init__.py``) is treated as top-level (its ``.py`` is not
  prefixed by the dir name).
- A ``.py`` file that fails to decode (binary / null bytes) is skipped with a warning;
  a non-``.py`` binary file is ignored because only ``.py`` files are walked.
- Hidden dirs, ``__pycache__``, ``.venv`` and ``venv`` are skipped.
- SHA-256 is computed per file; one ``SourceFile`` record exists per valid ``.py`` file.
- ``repo_root`` resolving outside ``ALLOWED_BASE_DIR`` is rejected (path-traversal guard).

The fixture tree is created under the repo root via ``tempfile.TemporaryDirectory(dir=...)``
so that it is guaranteed to live inside ``ALLOWED_BASE_DIR`` (which defaults to the parent
of the workspace folder, resolved once at import time and not patchable).
"""

from __future__ import annotations

import hashlib
import logging
from collections.abc import Iterator
from pathlib import Path

import pytest
from loosecanvas.config import ALLOWED_BASE_DIR
from loosecanvas.contracts import SourceSnapshot
from loosecanvas.extractors.file_walker import walk_repo

# Repo root is one parent up from tests/ — guaranteed inside ALLOWED_BASE_DIR.
_REPO_ROOT = Path(__file__).resolve().parents[1]


@pytest.fixture
def repo_tree(tmp_path_factory: pytest.TempPathFactory) -> Iterator[Path]:
    """Build a small fixture tree *inside the repo root* so it stays in ALLOWED_BASE_DIR."""
    import tempfile

    with tempfile.TemporaryDirectory(dir=_REPO_ROOT) as raw:
        root = Path(raw)
        # Sanity: the chosen location must satisfy the guard we are about to exercise.
        assert root.resolve().is_relative_to(ALLOWED_BASE_DIR)

        # A real nested package.
        (root / "pkg").mkdir()
        (root / "pkg" / "__init__.py").write_text("", encoding="utf-8")
        # write_bytes to pin exact content (avoid Windows CRLF translation).
        (root / "pkg" / "sub.py").write_bytes(b"x = 1\n")
        (root / "pkg" / "nested").mkdir()
        (root / "pkg" / "nested" / "__init__.py").write_text("", encoding="utf-8")
        (root / "pkg" / "nested" / "deep.py").write_text("y = 2\n", encoding="utf-8")

        # A namespace dir (no __init__.py) — its module is top-level.
        (root / "nsdir").mkdir()
        (root / "nsdir" / "mod.py").write_text("z = 3\n", encoding="utf-8")

        # An undecodable .py file — must be skipped with a warning.
        (root / "broken.py").write_bytes(b"\x00\x01\x02\xff\xfe")

        # A non-.py binary file — ignored because only .py files are walked.
        (root / "data.bin").write_bytes(b"\x00\x01\x02")

        # Skipped directories.
        (root / ".hidden").mkdir()
        (root / ".hidden" / "secret.py").write_text("secret = 1\n", encoding="utf-8")
        (root / "__pycache__").mkdir()
        (root / "__pycache__" / "cached.py").write_text("c = 1\n", encoding="utf-8")
        (root / ".venv").mkdir()
        (root / ".venv" / "lib.py").write_text("v = 1\n", encoding="utf-8")
        (root / "venv").mkdir()
        (root / "venv" / "lib2.py").write_text("v = 2\n", encoding="utf-8")

        yield root


def test_returns_snapshot_and_module_map(repo_tree: Path) -> None:
    snapshot, module_map = walk_repo(repo_tree)
    assert isinstance(snapshot, SourceSnapshot)
    assert isinstance(module_map, dict)


def test_one_record_per_valid_py_file(repo_tree: Path) -> None:
    snapshot, _ = walk_repo(repo_tree)
    # Valid .py files: pkg/__init__.py, pkg/sub.py, pkg/nested/__init__.py,
    # pkg/nested/deep.py, nsdir/mod.py. (broken.py is skipped; skipped dirs excluded.)
    assert set(snapshot.files) == {
        "pkg/__init__.py",
        "pkg/sub.py",
        "pkg/nested/__init__.py",
        "pkg/nested/deep.py",
        "nsdir/mod.py",
    }


def test_nested_package_dotted_names(repo_tree: Path) -> None:
    _, module_map = walk_repo(repo_tree)
    assert module_map["pkg/__init__.py"] == "pkg"
    assert module_map["pkg/sub.py"] == "pkg.sub"
    assert module_map["pkg/nested/__init__.py"] == "pkg.nested"
    assert module_map["pkg/nested/deep.py"] == "pkg.nested.deep"


def test_namespace_dir_is_top_level(repo_tree: Path) -> None:
    _, module_map = walk_repo(repo_tree)
    # nsdir has no __init__.py, so mod.py is top-level, not "nsdir.mod".
    assert module_map["nsdir/mod.py"] == "mod"


def test_undecodable_py_is_skipped(repo_tree: Path) -> None:
    snapshot, module_map = walk_repo(repo_tree)
    assert "broken.py" not in snapshot.files
    assert "broken.py" not in module_map


def test_non_py_binary_ignored(repo_tree: Path) -> None:
    snapshot, _ = walk_repo(repo_tree)
    assert "data.bin" not in snapshot.files


def test_skipped_directories(repo_tree: Path) -> None:
    snapshot, _ = walk_repo(repo_tree)
    paths = set(snapshot.files)
    assert not any(p.startswith(".hidden/") for p in paths)
    assert not any("__pycache__" in p for p in paths)
    assert not any(p.startswith(".venv/") for p in paths)
    assert not any(p.startswith("venv/") for p in paths)


def test_sha256_and_content(repo_tree: Path) -> None:
    snapshot, _ = walk_repo(repo_tree)
    record = snapshot.files["pkg/sub.py"]
    raw = (repo_tree / "pkg" / "sub.py").read_bytes()
    assert record.sha256 == hashlib.sha256(raw).hexdigest()
    assert record.content == "x = 1\n"
    assert record.line_count == 1
    assert record.source_id == "pkg/sub.py"
    assert record.path == "pkg/sub.py"


def test_snapshot_metadata(repo_tree: Path) -> None:
    snapshot, _ = walk_repo(repo_tree)
    assert snapshot.root_path == str(repo_tree.resolve())
    assert snapshot.snapshot_id


def test_path_outside_allowed_base_raises(
    caplog: pytest.LogCaptureFixture,
) -> None:
    outside = ALLOWED_BASE_DIR.parent
    with (
        caplog.at_level(logging.WARNING, logger="loosecanvas.extractors.file_walker"),
        pytest.raises(ValueError, match="path not allowed") as excinfo,
    ):
        walk_repo(outside)
    # The client-visible message must be generic — no resolved path leaked.
    assert "ALLOWED_BASE_DIR" not in str(excinfo.value)
    assert str(outside.resolve()) not in str(excinfo.value)
    # The resolved path must appear in the server-side warning log, not in the exception.
    assert any(str(outside.resolve()) in rec.getMessage() for rec in caplog.records)