"""Tests for the v12.4 Python pivot — async/decorator extraction,
generated-dir exclusion, Python idiom detectors.

Each test corresponds to a finding from the dogfood pass on
``a Python research repo`` / ``an agent-eval workspace`` so the fix doesn't regress.
"""

from __future__ import annotations

import pytest
from tree_sitter_language_pack import get_parser

from roam.languages.python_lang import PythonExtractor
from roam.output.file_role_hints import header_indicates_generated, is_excluded_path

# ---------------------------------------------------------------------------
# is_async + decorators on Python symbols
# ---------------------------------------------------------------------------


def _extract(source: bytes) -> list[dict]:
    parser = get_parser("python")
    tree = parser.parse(source)
    return PythonExtractor().extract_symbols(tree, source, "test.py")


class TestAsyncDetection:
    """``async def`` should set ``is_async=True`` on the symbol."""

    def test_async_function_marked_async(self):
        symbols = _extract(b"async def cook(x): return x\n")
        cook = next(s for s in symbols if s["name"] == "cook")
        assert cook["is_async"] is True

    def test_sync_function_not_marked_async(self):
        symbols = _extract(b"def sync_fn(x): return x\n")
        sym = next(s for s in symbols if s["name"] == "sync_fn")
        assert sym["is_async"] is False

    def test_async_method_in_class_marked_async(self):
        source = b"class A:\n    async def run(self): pass\n"
        symbols = _extract(source)
        run = next(s for s in symbols if s["name"] == "run")
        assert run["kind"] == "method"
        assert run["is_async"] is True

    def test_decorated_async_method_marked_async(self):
        """A decorated ``async def`` is wrapped in decorated_definition;
        the async detection must drill into the inner function."""
        source = b"import functools\n@functools.wraps(None)\nasync def cached(): pass\n"
        symbols = _extract(source)
        cached = next(s for s in symbols if s["name"] == "cached")
        assert cached["is_async"] is True
        assert "@functools.wraps" in cached.get("decorators", "")


class TestDecoratorExtraction:
    """Decorators above ``def`` / ``class`` get joined into the
    ``decorators`` field comma-separated."""

    def test_property_decorator_captured(self):
        source = b"class A:\n    @property\n    def x(self): return 1\n"
        symbols = _extract(source)
        x = next(s for s in symbols if s["name"] == "x")
        assert "@property" in x.get("decorators", "")

    def test_pytest_fixture_decorator_captured(self):
        source = b"import pytest\n@pytest.fixture\ndef client(): pass\n"
        symbols = _extract(source)
        client = next(s for s in symbols if s["name"] == "client")
        assert "@pytest.fixture" in client.get("decorators", "")

    def test_undecorated_has_empty_decorators(self):
        source = b"def plain(): pass\n"
        symbols = _extract(source)
        plain = next(s for s in symbols if s["name"] == "plain")
        assert plain.get("decorators", "") == ""


# ---------------------------------------------------------------------------
# Generated-dir / file-role hint detection
# ---------------------------------------------------------------------------


class TestFileRoleHints:
    """``output.file_role_hints.is_excluded_path`` covers the canonical
    excluded-from-headline paths."""

    def test_examples_dir_excluded(self):
        assert is_excluded_path("examples/quickstart.py") is True

    def test_workspaces_dir_excluded(self):
        """Surfaced 2026-05-02 R-pivot — ``workspaces/`` are agent-
        generated benchmark artifacts, not source code."""
        assert is_excluded_path("workspaces/agent/foo/main.py") is True

    def test_dev_dir_excluded(self):
        assert is_excluded_path("dev/repo_hygiene.py") is True

    def test_github_dir_excluded(self):
        assert is_excluded_path(".github/scripts/pr-comment.js") is True

    def test_benchmarks_dir_excluded(self):
        assert is_excluded_path("benchmarks/agent-eval/compare.py") is True

    def test_node_modules_excluded(self):
        assert is_excluded_path("node_modules/foo/index.js") is True

    def test_vendor_dir_excluded(self):
        assert is_excluded_path("vendor/third_party/lib.py") is True

    def test_normal_source_not_excluded(self):
        assert is_excluded_path("src/roam/cli.py") is False
        assert is_excluded_path("tests/test_foo.py") is False

    def test_empty_path_not_excluded(self):
        assert is_excluded_path("") is False

    def test_windows_separators_normalised(self):
        assert is_excluded_path("workspaces\\agent\\foo.py") is True


class TestGeneratedHeaderDetection:
    def test_protobuf_marker(self):
        text = "// Code generated by protoc. DO NOT EDIT.\npackage foo\n"
        assert header_indicates_generated(text) is True

    def test_at_generated_tag(self):
        text = '"""@generated by some-tool"""\nimport foo\n'
        assert header_indicates_generated(text) is True

    def test_normal_file_not_generated(self):
        text = '"""A normal Python module."""\nimport foo\n'
        assert header_indicates_generated(text) is False

    def test_empty_text(self):
        assert header_indicates_generated("") is False


# ---------------------------------------------------------------------------
# Python idiom detectors (regex-based)
# ---------------------------------------------------------------------------


class TestPythonIdiomRegexes:
    """Sanity-check the regex patterns directly without spinning up a DB."""

    def test_mutable_default_list_matches(self):
        from roam.catalog.python_idioms import _MUTABLE_DEFAULT_RE

        assert _MUTABLE_DEFAULT_RE.search("def f(x=[])") is not None
        assert _MUTABLE_DEFAULT_RE.search("def g(d={})") is not None
        assert _MUTABLE_DEFAULT_RE.search("def h(x=list())") is not None
        assert _MUTABLE_DEFAULT_RE.search("def k(s=set())") is not None

    def test_immutable_default_does_not_match(self):
        from roam.catalog.python_idioms import _MUTABLE_DEFAULT_RE

        assert _MUTABLE_DEFAULT_RE.search("def f(x=None)") is None
        assert _MUTABLE_DEFAULT_RE.search('def g(s="")') is None
        assert _MUTABLE_DEFAULT_RE.search("def h(n=0)") is None
        assert _MUTABLE_DEFAULT_RE.search("def k(t=())") is None

    def test_bare_except_matches(self):
        from roam.catalog.python_idioms import _BARE_EXCEPT_RE

        assert _BARE_EXCEPT_RE.search("try:\n    x()\nexcept:\n    pass\n") is not None
        # Indented bare except inside a function
        assert _BARE_EXCEPT_RE.search("def f():\n    try:\n        x()\n    except:\n        pass\n") is not None

    def test_typed_except_does_not_match(self):
        from roam.catalog.python_idioms import _BARE_EXCEPT_RE

        assert _BARE_EXCEPT_RE.search("try:\n    x()\nexcept ValueError:\n    pass\n") is None
        assert _BARE_EXCEPT_RE.search("try:\n    x()\nexcept (A, B) as e:\n    pass\n") is None

    def test_none_eq_matches(self):
        from roam.catalog.python_idioms import _NONE_EQ_RE

        assert _NONE_EQ_RE.search("if x == None:") is not None
        assert _NONE_EQ_RE.search("if x != None:") is not None

    def test_none_is_does_not_match(self):
        from roam.catalog.python_idioms import _NONE_EQ_RE

        assert _NONE_EQ_RE.search("if x is None:") is None

    def test_logger_fstring_matches(self):
        from roam.catalog.python_idioms import _LOGGER_FSTRING_RE

        assert _LOGGER_FSTRING_RE.search('logger.info(f"x={x}")') is not None
        assert _LOGGER_FSTRING_RE.search('log.warning(f"oops")') is not None
        assert _LOGGER_FSTRING_RE.search("logging.debug(f'fmt')") is not None
        assert _LOGGER_FSTRING_RE.search('self.logger.error(f"x")') is not None

    def test_logger_lazy_format_does_not_match(self):
        from roam.catalog.python_idioms import _LOGGER_FSTRING_RE

        assert _LOGGER_FSTRING_RE.search('logger.info("x=%s", x)') is None
        assert _LOGGER_FSTRING_RE.search('logger.info("plain")') is None


# ---------------------------------------------------------------------------
# v12.4-iter additions
# ---------------------------------------------------------------------------


class TestStringStripper:
    """``_strip_strings_and_comments`` replaces docstring/comment content
    with same-length whitespace so detectors don't false-match inside
    strings."""

    def test_triple_quoted_docstring_blanked(self):
        from roam.catalog.python_idioms import _strip_strings_and_comments

        src = 'def f():\n    """example: x == None"""\n    return 1\n'
        out = _strip_strings_and_comments(src)
        assert "x == None" not in out
        assert len(out) == len(src)
        assert out.count("\n") == src.count("\n")

    def test_single_quoted_string_blanked(self):
        from roam.catalog.python_idioms import _strip_strings_and_comments

        out = _strip_strings_and_comments('x = "open(foo)"\n')
        assert "open(foo)" not in out
        assert "open(" not in out

    def test_comment_blanked(self):
        from roam.catalog.python_idioms import _strip_strings_and_comments

        out = _strip_strings_and_comments("x = 1  # foo == None\n")
        assert "== None" not in out

    def test_real_code_preserved(self):
        from roam.catalog.python_idioms import _strip_strings_and_comments

        src = "if x == None:\n    pass\n"
        out = _strip_strings_and_comments(src)
        assert "x == None" in out


class TestSyncInAsyncRegex:
    @pytest.mark.parametrize(
        "snippet",
        [
            "    response = requests.get(url)",
            "    requests.post(url, data=d)",
            "    time.sleep(1)",
            "    subprocess.run(['ls'])",
            "    urllib.request.urlopen(url)",
        ],
    )
    def test_blocking_patterns_match(self, snippet):
        from roam.catalog.python_idioms import _SYNC_IN_ASYNC_PATTERNS

        assert any(p.search(snippet) for p, _ in _SYNC_IN_ASYNC_PATTERNS)

    def test_async_alternatives_do_not_match(self):
        from roam.catalog.python_idioms import _SYNC_IN_ASYNC_PATTERNS

        for snippet in ("await asyncio.sleep(1)", "await httpx.AsyncClient().get(url)"):
            assert not any(p.search(snippet) for p, _ in _SYNC_IN_ASYNC_PATTERNS)


class TestOpenWithoutWithRegex:
    def test_assignment_open_matches(self):
        from roam.catalog.python_idioms import _OPEN_WITHOUT_WITH_RE

        assert _OPEN_WITHOUT_WITH_RE.search("    f = open(path)") is not None

    def test_os_open_skipped(self):
        from roam.catalog.python_idioms import _OPEN_WITHOUT_WITH_RE

        assert _OPEN_WITHOUT_WITH_RE.search("os.open(path)") is None


class TestStarImportRegex:
    def test_star_import_matches(self):
        from roam.catalog.python_idioms import _STAR_IMPORT_RE

        assert _STAR_IMPORT_RE.search("from foo import *") is not None
        assert _STAR_IMPORT_RE.search("from foo.bar import *") is not None

    def test_explicit_import_does_not_match(self):
        from roam.catalog.python_idioms import _STAR_IMPORT_RE

        assert _STAR_IMPORT_RE.search("from foo import a, b") is None


class TestDictKeysIterRegex:
    def test_keys_iter_matches(self):
        from roam.catalog.python_idioms import _DICT_KEYS_ITER_RE

        assert _DICT_KEYS_ITER_RE.search("for k in d.keys():") is not None

    def test_direct_iter_does_not_match(self):
        from roam.catalog.python_idioms import _DICT_KEYS_ITER_RE

        assert _DICT_KEYS_ITER_RE.search("for k in d:") is None
        assert _DICT_KEYS_ITER_RE.search("for k, v in d.items():") is None


class TestHasDecorator:
    def test_exact(self):
        from roam.catalog.python_idioms import has_decorator

        assert has_decorator("@pytest.fixture", "pytest.fixture") is True

    def test_with_args(self):
        from roam.catalog.python_idioms import has_decorator

        assert has_decorator("@pytest.fixture(scope='module')", "pytest.fixture") is True

    def test_substring_in_other_decorator_does_not_match(self):
        from roam.catalog.python_idioms import has_decorator

        deco = '@click.option("--help", help="like pytest.fixture but inline")'
        assert has_decorator(deco, "pytest.fixture") is False

    def test_in_list(self):
        from roam.catalog.python_idioms import has_decorator

        deco = "@click.command(),@pytest.fixture,@click.argument('p')"
        assert has_decorator(deco, "pytest.fixture") is True

    def test_empty(self):
        from roam.catalog.python_idioms import has_decorator

        assert has_decorator("", "pytest.fixture") is False
        assert has_decorator(None, "pytest.fixture") is False


class TestFixtureKind:
    def test_pytest_fixture(self):
        from roam.catalog.python_idioms import fixture_kind

        assert fixture_kind("@pytest.fixture") == "pytest fixture"

    def test_pytest_fixture_with_args(self):
        from roam.catalog.python_idioms import fixture_kind

        assert fixture_kind("@pytest.fixture(scope='session')") == "pytest fixture"

    def test_parametrize(self):
        from roam.catalog.python_idioms import fixture_kind

        assert fixture_kind("@pytest.mark.parametrize('x', [1, 2])") == "parametrize"

    def test_async_test(self):
        from roam.catalog.python_idioms import fixture_kind

        assert fixture_kind("@pytest.mark.asyncio") == "async test"

    def test_no_pytest(self):
        from roam.catalog.python_idioms import fixture_kind

        assert fixture_kind("@property") is None
        assert fixture_kind("") is None


class TestIsModelClass:
    def test_pydantic(self):
        from roam.catalog.python_idioms import is_model_class

        is_m, label = is_model_class("class User(BaseModel)", "")
        assert is_m is True
        assert label == "pydantic"

    def test_dataclass(self):
        from roam.catalog.python_idioms import is_model_class

        is_m, label = is_model_class("class Point", "@dataclass")
        assert is_m is True
        assert label == "dataclass"

    def test_attrs(self):
        from roam.catalog.python_idioms import is_model_class

        is_m, label = is_model_class("class Point", "@attrs.define")
        assert is_m is True
        assert label == "attrs"

    def test_enum_subclass(self):
        from roam.catalog.python_idioms import is_model_class

        is_m, label = is_model_class("class Status(Enum)", "")
        assert is_m is True
        assert label == "enum"

    def test_plain_class(self):
        from roam.catalog.python_idioms import is_model_class

        is_m, label = is_model_class("class Foo", "")
        assert is_m is False
        assert label is None


class TestPyTypesSignatureHealth:
    def test_fully_typed(self):
        from roam.commands.cmd_py_types import _signature_health

        h = _signature_health("def f(x: int, y: str) -> bool")
        assert h["has_return"] is True
        assert h["params_typed"] == 2
        assert h["params_untyped"] == 0
        assert h["uses_any"] is False
        assert h["old_typing"] is False

    def test_no_return(self):
        from roam.commands.cmd_py_types import _signature_health

        h = _signature_health("def f(x: int)")
        assert h["has_return"] is False

    def test_uses_any(self):
        from roam.commands.cmd_py_types import _signature_health

        h = _signature_health("def f(x: Any) -> Any")
        assert h["uses_any"] is True

    def test_old_typing(self):
        from roam.commands.cmd_py_types import _signature_health

        h = _signature_health("def f(x: Optional[int]) -> Dict[str, int]")
        assert h["old_typing"] is True

    def test_self_not_counted(self):
        from roam.commands.cmd_py_types import _signature_health

        h = _signature_health("def method(self, x: int) -> bool")
        assert h["params_typed"] == 1
        assert h["params_untyped"] == 0


# ---------------------------------------------------------------------------
# Phase J/K additions
# ---------------------------------------------------------------------------


class TestAsyncWithMissingRegex:
    def test_aiofiles_open_matches(self):
        from roam.catalog.python_idioms import _ASYNC_OPEN_RE

        assert _ASYNC_OPEN_RE.search("f = aiofiles.open('x')") is not None

    def test_httpx_async_client_matches(self):
        from roam.catalog.python_idioms import _HTTPX_CLIENT_RE

        assert _HTTPX_CLIENT_RE.search("client = httpx.AsyncClient()") is not None

    def test_sync_open_does_not_match_aiofiles_re(self):
        from roam.catalog.python_idioms import _ASYNC_OPEN_RE

        assert _ASYNC_OPEN_RE.search("f = open('x')") is None


class TestTypeEqRegex:
    def test_type_eq_matches(self):
        from roam.catalog.python_idioms import _TYPE_EQ_RE

        assert _TYPE_EQ_RE.search("if type(x) == int:") is not None
        assert _TYPE_EQ_RE.search("type(value) == MyClass") is not None

    def test_isinstance_does_not_match(self):
        from roam.catalog.python_idioms import _TYPE_EQ_RE

        assert _TYPE_EQ_RE.search("if isinstance(x, int):") is None


class TestAsyncNotAwaitedHelpers:
    """The async-not-awaited detector is too DB-dependent to unit-test
    without a fixture project. These tests cover the safe-prefix logic
    and lookbehind regex shape."""

    def test_safe_prefix_skips_await(self):
        # If the line contains ``await fetch()`` the regex's ``(?<![\w.])``
        # still matches ``fetch(`` but the safe-prefix check at runtime
        # examines the 30 chars before. We test the runtime check
        # indirectly via the detector output on the synthetic case
        # in the integration test.
        # Here we just sanity-check the lookbehind: ``foo.bar(`` must
        # NOT match if ``bar`` is in the async name set.
        import re

        pat = re.compile(r"(?<![\w.])(\bbar)\s*\(")
        assert pat.search("foo.bar()") is None
        assert pat.search("bar()") is not None
        assert pat.search("await bar()") is not None  # safe-prefix check happens later
