ronantakizawa/codereview-bench huggingface dataset:
Repository: Textualize/textual
File: src/textual/_node_list.py (Python)

Code before:
from __future__ import annotations

from typing import Iterator, overload, TYPE_CHECKING
from weakref import ref

import rich.repr

if TYPE_CHECKING:
    from .dom import DOMNode


@rich.repr.auto
class NodeList:
    """
    A container for widgets that forms one level of hierarchy.

    Although named a list, widgets may appear only once, making them more like a set.

    """

    def __init__(self, nodes: list[DOMNode] = None) -> None:
        self._nodes: list[DOMNode] = [] if nodes is None else nodes

    def __bool__(self) -> bool:
        return bool(self._nodes)

    def __length_hint__(self) -> int:
        return len(self._nodes)

    def __rich_repr__(self) -> rich.repr.Result:
        yield self._nodes

    def __len__(self) -> int:
        return len(self._nodes)

    def __contains__(self, widget: DOMNode) -> bool:
        return widget in self._nodes

    def _append(self, widget: DOMNode) -> None:
        if widget not in self._nodes:
            self._nodes.append(widget)

    def _clear(self) -> None:
        del self._nodes[:]

    def __iter__(self) -> Iterator[DOMNode]:

Review comment (suggestion):
I was thinking that being able to optionally provide an initial list of nodes to this constructor could be useful?
I can revert this change if you think it's better to not do this of course :slightly_smiling_face: 

(I checked the [NodeList](https://devdocs.io/dom/nodelist) Web API, but there is no constructor on this one anyway so we cannot do any features comparison here :sweat_smile: )

Code after:
from __future__ import annotations

from typing import Iterator, overload, TYPE_CHECKING
from weakref import ref

import rich.repr

if TYPE_CHECKING:
    from .dom import DOMNode


@rich.repr.auto
class NodeList:
    """
    A container for widgets that forms one level of hierarchy.

    Although named a list, widgets may appear only once, making them more like a set.

    """

    def __init__(self) -> None:
        self._nodes: list[DOMNode] = []

    def __bool__(self) -> bool:
        return bool(self._nodes)

    def __length_hint__(self) -> int:
        return len(self._nodes)

    def __rich_repr__(self) -> rich.repr.Result:
        yield self._nodes

    def __len__(self) -> int:
        return len(self._nodes)

    def __contains__(self, widget: DOMNode) -> bool:
        return widget in self._nodes

    def _append(self, widget: DOMNode) -> None:
        if widget not in self._nodes:
            self._nodes.append(widget)

    def _clear(self) -> None:
        del self._nodes[:]

    def __iter__(self) -> Iterator[DOMNode]:
Repository: Textualize/textual
File: tests/test_reactive.py (Python)

Code before:
    class SomeWidget(Widget):
        test_1: var[int] = var(0)
        test_2: var[int] = var(0, init=False)

        def watch_test_1(self) -> None:
            logs.append("test_1")

        def watch_test_2(self) -> None:
            logs.append("test_2")

    class InitOverrideApp(App[None]):
        def compose(self) -> ComposeResult:
            yield SomeWidget()

        def on_mount(self) -> None:
            def watch_test_2_extra() -> None:
                logs.append("test_2_extra")

            self.watch(self.query_one(SomeWidget), "test_2", watch_test_2_extra)

    app = InitOverrideApp()
    async with app.run_test():
        assert logs == ["test_1", "test_2_extra"]
        app.query_one(SomeWidget).test_2 = 73
        assert logs.count("test_2_extra") == 2
        assert "test_2" in logs


Review comment (suggestion):
Let's make sure that the watcher fires exactly once instead of at least once.
```suggestion
        assert logs.count("test_2") == 1
```

Code after:
        test_1: var[int] = var(0)
        test_2: var[int] = var(0, init=False)

        def watch_test_1(self) -> None:
            logs.append("test_1")

        def watch_test_2(self) -> None:
            logs.append("test_2")

    class InitOverrideApp(App[None]):
        def compose(self) -> ComposeResult:
            yield SomeWidget()

        def on_mount(self) -> None:
            def watch_test_2_extra() -> None:
                logs.append("test_2_extra")

            self.watch(self.query_one(SomeWidget), "test_2", watch_test_2_extra)

    app = InitOverrideApp()
    async with app.run_test():
        assert logs == ["test_1", "test_2_extra"]
        app.query_one(SomeWidget).test_2 = 73
        assert logs.count("test_2_extra") == 2
        assert logs.count("test_2") == 1


async def test_external_watch_init_does_not_propagate_to_externals() -> None:
    """Regression test for https://github.com/Textualize/textual/issues/3878.

    Make sure that when setting an extra watcher programmatically and `init` is set,
    we init only the new watcher and not the other ones (even if they were
    added dynamically with `watch`), but at the same time make sure all watchers
    work in regular circumstances.
    """

    logs: list[str] = []

    class SomeWidget(Widget):
        test_var: var[int] = var(0)

    class MyApp(App[None]):
        def compose(self) -> ComposeResult:
            yield SomeWidget()

        def add_first_watcher(self) -> None:
            def first_callback() -> None:
                logs.append("first")

            self.watch(self.query_one(SomeWidget), "test_var", first_callback)
Repository: Textualize/textual
File: tests/test_reactive.py (Python)

Code before:


async def test_no_duplicate_external_watchers() -> None:
    """Make sure we skip duplicated watchers."""

    counter = 0

    class Holder(Widget):
        attr = var(None)

    class MyApp(App[None]):
        def __init__(self) -> None:
            super().__init__()
            self.holder = Holder()

        def on_mount(self) -> None:
            self.watch(self.holder, "attr", self.callback)
            self.watch(self.holder, "attr", self.callback)

        def callback(self) -> None:
            nonlocal counter
            counter += 1

    app = MyApp()
    async with app.run_test():
        before = counter
        app.holder.attr = 73
        after = counter
    assert after == before + 1


async def test_external_watch_init_does_not_propagate() -> None:
    """Regression test for https://github.com/Textualize/textual/issues/3878.

    Make sure that when setting an extra watcher programmatically and `init` is set,
    we init only the new watcher and not the other ones, but at the same
    time make sure both watchers work in regular circumstances.
    """

    logs: list[str] = []

    class SomeWidget(Widget):
        test_1: var[int] = var(0)
        test_2: var[int] = var(0, init=False)

        def watch_test_1(self) -> None:
            logs.append("test_1")

        def watch_test_2(self) -> None:
            logs.append("test_2")


Review comment (refactor):
Any reason why you do the before and after stuff here instead of `assert counter == 1` before the assignment and then `assert counter == 2` after the assignment? That would seem simpler and a more accurate test.

As it stands, this test wouldn't catch duplicate watchers firing on `init`.

Code after:


async def test_no_duplicate_external_watchers() -> None:
    """Make sure we skip duplicated watchers."""

    counter = 0

    class Holder(Widget):
        attr = var(None)

    class MyApp(App[None]):
        def __init__(self) -> None:
            super().__init__()
            self.holder = Holder()

        def on_mount(self) -> None:
            self.watch(self.holder, "attr", self.callback)
            self.watch(self.holder, "attr", self.callback)

        def callback(self) -> None:
            nonlocal counter
            counter += 1

    app = MyApp()
    async with app.run_test():
        assert counter == 1
        app.holder.attr = 73
        assert counter == 2


async def test_external_watch_init_does_not_propagate() -> None:
    """Regression test for https://github.com/Textualize/textual/issues/3878.

    Make sure that when setting an extra watcher programmatically and `init` is set,
    we init only the new watcher and not the other ones, but at the same
    time make sure both watchers work in regular circumstances.
    """

    logs: list[str] = []

    class SomeWidget(Widget):
        test_1: var[int] = var(0)
        test_2: var[int] = var(0, init=False)

        def watch_test_1(self) -> None:
            logs.append("test_1")

        def watch_test_2(self) -> None:
            logs.append("test_2")

    class InitOverrideApp(App[None]):
Repository: Textualize/textual
File: tests/test_pilot.py (Python)

Code before:

    class TestApp(App[None]):
        def on_key(self, event: events.Key) -> None:
            keys_pressed.append(event.character)

    async with TestApp().run_test() as pilot:
        for char in KEY_CHARACTERS_TO_TEST:
            await pilot.press(char)
            assert keys_pressed[-1] == char


async def test_pilot_exception_catching_app_compose():
    """Ensuring that test frameworks are aware of exceptions
    inside compose methods when running via Pilot run_test()."""

    class FailingApp(App):
        def compose(self) -> ComposeResult:
            1 / 0
            yield Label("Beep")

    with pytest.raises(ZeroDivisionError):
        async with FailingApp().run_test():
            pass


async def test_pilot_exception_cathing_widget_compose():
    class SomeScreen(Screen[None]):
        def compose(self) -> ComposeResult:
            1 / 0
            yield Label("Beep")

    class FailingApp(App[None]):
        def on_mount(self) -> None:
            self.push_screen(SomeScreen())

    with pytest.raises(ZeroDivisionError):
        async with FailingApp().run_test():
            pass


async def test_pilot_exception_catching_action():
    """Ensure that exceptions inside action handlers are presented
    to the test framework when running via Pilot run_test()."""

    class FailingApp(App):
        BINDINGS = [Binding("b", "beep", "beep")]

        def action_beep(self) -> None:
            1 / 0

    with pytest.raises(ZeroDivisionError):

Review comment (suggestion):
```suggestion
async def test_pilot_exception_catching_widget_compose():
```

Code after:
    keys_pressed = []

    class TestApp(App[None]):
        def on_key(self, event: events.Key) -> None:
            keys_pressed.append(event.character)

    async with TestApp().run_test() as pilot:
        for char in KEY_CHARACTERS_TO_TEST:
            await pilot.press(char)
            assert keys_pressed[-1] == char


async def test_pilot_exception_catching_app_compose():
    """Ensuring that test frameworks are aware of exceptions
    inside compose methods when running via Pilot run_test()."""

    class FailingApp(App):
        def compose(self) -> ComposeResult:
            1 / 0
            yield Label("Beep")

    with pytest.raises(ZeroDivisionError):
        async with FailingApp().run_test():
            pass


async def test_pilot_exception_catching_widget_compose():
    class SomeScreen(Screen[None]):
        def compose(self) -> ComposeResult:
            1 / 0
            yield Label("Beep")

    class FailingApp(App[None]):
        def on_mount(self) -> None:
            self.push_screen(SomeScreen())

    with pytest.raises(ZeroDivisionError):
        async with FailingApp().run_test():
            pass


async def test_pilot_exception_catching_action():
    """Ensure that exceptions inside action handlers are presented
    to the test framework when running via Pilot run_test()."""

    class FailingApp(App):
        BINDINGS = [Binding("b", "beep", "beep")]

        def action_beep(self) -> None:
            1 / 0
Repository: Textualize/textual
File: src/textual/renderables/bar.py (Python)

Code before:

        # Fire actions when certain ranges are clicked (e.g. for tabs)
        for range_name, (start, end) in self.clickable_ranges.items():
            output_bar.apply_meta(
                {"@click": f"range_clicked('{range_name}')"}, start, end
            )

        yield output_bar


def _apply_gradient(text: Text, gradient: Gradient, width: int) -> None:
    """Apply a gradient to a Rich Text instance.

    Args:
        text: A Text object.
        gradient: A Textual gradient.
        width: Width of gradient.
    """
    from_color = Style.from_color
    get_rich_color = gradient.get_rich_color

    max_width = width - 1
    if not max_width:
        text.stylize(from_color(gradient.get_color(0).rich_color))
        return
    text_length = len(text)
    for offset in range(text_length):
        bar_offset = text_length - offset
        text.stylize(
            from_color(get_rich_color(bar_offset / max_width)),
            offset,
            offset + 1,
        )


Review comment (suggestion):
Also just to check - if width == 0, then max_width will be -1 and we'll go into the loop which I don't think is intentional? (Should this be `max_width <= 0`?)

Code after:

        # Fire actions when certain ranges are clicked (e.g. for tabs)
        for range_name, (start, end) in self.clickable_ranges.items():
            output_bar.apply_meta(
                {"@click": f"range_clicked('{range_name}')"}, start, end
            )

        yield output_bar


def _apply_gradient(text: Text, gradient: Gradient, width: int) -> None:
    """Apply a gradient to a Rich Text instance.

    Args:
        text: A Text object.
        gradient: A Textual gradient.
        width: Width of gradient.
    """
    if not width:
        return
    assert width > 0
    from_color = Style.from_color
    get_rich_color = gradient.get_rich_color

    max_width = width - 1
    if not max_width:
        text.stylize(from_color(gradient.get_color(0).rich_color))
        return
    text_length = len(text)
    for offset in range(text_length):
        bar_offset = text_length - offset
        text.stylize(
            from_color(get_rich_color(bar_offset / max_width)),
            offset,
            offset + 1,
        )
Repository: Textualize/textual
File: src/textual/events.py (Python)

Code before:
    """

    __slots__ = ["time", "count", "callback"]

    def __init__(
        self,
        timer: "TimerClass",
        time: float,
        count: int = 0,
        callback: TimerCallback | None = None,
    ) -> None:
        super().__init__()
        self.timer = timer
        self.time = time
        self.count = count
        self.callback = callback

    def __rich_repr__(self) -> rich.repr.Result:
        yield self.timer.name
        yield "count", self.count


class Enter(Event, bubble=True, verbose=True):
    """Sent when the mouse is moved over a widget.

    Note that this widget bubbles, so a widget may receive this event when the mouse
    moves over a child widget. Check the `node` attribute for the widget directly under
    tne mouse.

    - [X] Bubbles
    - [X] Verbose
    """

    __slots__ = ["node"]

    def __init__(self, node: DOMNode) -> None:
        self.node = node
        """The node directly under the mouse."""
        super().__init__()


class Leave(Event, bubble=True, verbose=True):
    """Sent when the mouse is moved away from a widget, or if a widget is
    programmatically disabled while hovered.

    Note that this widget bubbles, so a widget may receive Leave events for any child widgets.
    Check the `node` parameter for the original widget that was previously under the mouse.


    - [X] Bubbles
    - [X] Verbose

Review comment (suggestion):
```suggestion
    Note that this event bubbles, so a widget may receive this event when the mouse
```

Code after:
    """

    __slots__ = ["time", "count", "callback"]

    def __init__(
        self,
        timer: "TimerClass",
        time: float,
        count: int = 0,
        callback: TimerCallback | None = None,
    ) -> None:
        super().__init__()
        self.timer = timer
        self.time = time
        self.count = count
        self.callback = callback

    def __rich_repr__(self) -> rich.repr.Result:
        yield self.timer.name
        yield "count", self.count


class Enter(Event, bubble=True, verbose=True):
    """Sent when the mouse is moved over a widget.

    Note that this event bubbles, so a widget may receive this event when the mouse
    moves over a child widget. Check the `node` attribute for the widget directly under
    the mouse.

    - [X] Bubbles
    - [X] Verbose
    """

    __slots__ = ["node"]

    def __init__(self, node: DOMNode) -> None:
        self.node = node
        """The node directly under the mouse."""
        super().__init__()


class Leave(Event, bubble=True, verbose=True):
    """Sent when the mouse is moved away from a widget, or if a widget is
    programmatically disabled while hovered.

    Note that this widget bubbles, so a widget may receive Leave events for any child widgets.
    Check the `node` parameter for the original widget that was previously under the mouse.


    - [X] Bubbles
    - [X] Verbose
Repository: Textualize/textual
File: src/textual/app.py (Python)

Code before:
        else:
            raise SuspendNotSupported(
                "App.suspend is not supported in this environment."
            )

    def action_suspend_process(self) -> None:
        """Suspend the process into the background.

        Note:
            On Unix and Unix-like systems a `SIGTSTP` is sent to the
            application's process. Currently on Windows and when running
            under Textual Web this is a non-operation.
        """
        # Check if we're in an environment that permits this kind of
        # suspend.
        if not WINDOWS and self._driver is not None and self._driver.can_suspend:
            # First, ensure that the suspend signal gets published while
            # we're still in application mode.
            self._suspend_signal()
            # With that out of the way, send the SIGTSTP signal.
            os.kill(os.getpid(), signal.SIGTSTP)
            # NOTE: There is no call to publish the resume signal here, this
            # will be handled by the driver posting a SignalResume event
            # (see the event handler on App._resume_signal) above.

    def open_url(self, url: str) -> None:
        """Open a URL in the default web browser.

        Args:
            url: The URL to open.
        """
        if self._driver is not None:
            self._driver.open_url(url)


Review comment (suggestion):
I think we should add a `blank` boolean. If True it would open a tab, otherwise it would open in the same tab. Default should be True.

Locally, this attribute would be ignored.

Code after:
        else:
            raise SuspendNotSupported(
                "App.suspend is not supported in this environment."
            )

    def action_suspend_process(self) -> None:
        """Suspend the process into the background.

        Note:
            On Unix and Unix-like systems a `SIGTSTP` is sent to the
            application's process. Currently on Windows and when running
            under Textual Web this is a non-operation.
        """
        # Check if we're in an environment that permits this kind of
        # suspend.
        if not WINDOWS and self._driver is not None and self._driver.can_suspend:
            # First, ensure that the suspend signal gets published while
            # we're still in application mode.
            self._suspend_signal()
            # With that out of the way, send the SIGTSTP signal.
            os.kill(os.getpid(), signal.SIGTSTP)
            # NOTE: There is no call to publish the resume signal here, this
            # will be handled by the driver posting a SignalResume event
            # (see the event handler on App._resume_signal) above.

    def open_url(self, url: str, *, new_tab: bool = True) -> None:
        """Open a URL in the default web browser.

        Args:
            url: The URL to open.
            new_tab: Whether to open the URL in a new tab.
        """
        if self._driver is not None:
            self._driver.open_url(url, new_tab)
Repository: Textualize/textual
File: src/textual/_parser.py (Python)

Code before:
    """Reads a single character, but does not advance the parser position."""

    timeout: float | None = None
    """Optional timeout in seconds."""


T = TypeVar("T")
TokenCallback = Callable[[T], None]


class Parser(Generic[T]):
    """Base class for a simple parser."""

    read1 = Read1
    peek1 = Peek1

    def __init__(self) -> None:
        self._eof = False
        self._tokens: Deque[T] = deque()
        self._gen = self.parse(self._tokens.append)
        self._awaiting: Read1 | Peek1 = next(self._gen)
        self._timeout_time: float | None = None

    @property
    def is_eof(self) -> bool:
        """Is the parser at the end of the fine (i.e. exhausted)?"""
        return self._eof

    def tick(self) -> Iterable[T]:
        """Call at regular intervals to check for timeouts."""
        if self._timeout_time is not None and get_time() >= self._timeout_time:
            self._timeout_time = None
            self._awaiting = self._gen.throw(ParseTimeout())
            while self._tokens:
                yield self._tokens.popleft()

    def feed(self, data: str) -> Iterable[T]:
        """Feed data to be parsed.

        Args:
            data: Data to parser.

        Raises:
            ParseError: If the data could not be parsed.

        Yields:
            T: A generic data type.
        """
        if self._eof:
            raise ParseError("end of file reached") from None


Review comment (suggestion):
```suggestion
        """Is the parser at the end of the file (i.e. exhausted)?"""
```

Code after:
    """Reads a single character, but does not advance the parser position."""

    timeout: float | None = None
    """Optional timeout in seconds."""


T = TypeVar("T")
TokenCallback = Callable[[T], None]


class Parser(Generic[T]):
    """Base class for a simple parser."""

    read1 = Read1
    peek1 = Peek1

    def __init__(self) -> None:
        self._eof = False
        self._tokens: Deque[T] = deque()
        self._gen = self.parse(self._tokens.append)
        self._awaiting: Read1 | Peek1 = next(self._gen)
        self._timeout_time: float | None = None

    @property
    def is_eof(self) -> bool:
        """Is the parser at the end of the file (i.e. exhausted)?"""
        return self._eof

    def tick(self) -> Iterable[T]:
        """Call at regular intervals to check for timeouts."""
        if self._timeout_time is not None and get_time() >= self._timeout_time:
            self._timeout_time = None
            self._awaiting = self._gen.throw(ParseTimeout())
            while self._tokens:
                yield self._tokens.popleft()

    def feed(self, data: str) -> Iterable[T]:
        """Feed data to be parsed.

        Args:
            data: Data to parser.

        Raises:
            ParseError: If the data could not be parsed.

        Yields:
            T: A generic data type.
        """
        if self._eof:
            raise ParseError("end of file reached") from None