cmux/tests_v2/test_notifications.py

#!/usr/bin/env python3
"""
Automated tests for notification focus/suppression behavior.

Usage:
    python3 test_notifications.py

Requirements:
    - cmux must be running with the socket controller enabled
"""

import os
import sys
import time
from typing import Optional

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from cmux import cmux, cmuxError


class TestResult:
    def __init__(self, name: str):
        self.name = name
        self.passed = False
        self.message = ""

    def success(self, msg: str = ""):
        self.passed = True
        self.message = msg

    def failure(self, msg: str):
        self.passed = False
        self.message = msg


def wait_for_notifications(client: cmux, expected: int, timeout: float = 2.0) -> list[dict]:
    start = time.time()
    while time.time() - start < timeout:
        items = client.list_notifications()
        if len(items) == expected:
            return items
        time.sleep(0.05)
    return client.list_notifications()

def wait_for_flash_count(client: cmux, surface: str, minimum: int = 1, timeout: float = 2.0) -> int:
    """Poll flash_count until it reaches `minimum` or timeout. Returns final count."""
    start = time.time()
    last = 0
    while time.time() - start < timeout:
        try:
            last = client.flash_count(surface)
        except Exception:
            last = 0
        if last >= minimum:
            return last
        time.sleep(0.05)
    return last


def ensure_two_surfaces(client: cmux) -> list[tuple[int, str, bool]]:
    surfaces = client.list_surfaces()
    if len(surfaces) < 2:
        client.new_split("right")
        time.sleep(0.1)
        surfaces = client.list_surfaces()
    return surfaces


def focused_surface_index(client: cmux) -> int:
    surfaces = client.list_surfaces()
    focused = next((s for s in surfaces if s[2]), None)
    if focused is None:
        raise RuntimeError("No focused surface")
    return focused[0]


def send_osc(client: cmux, sequence: str, surface: Optional[int] = None) -> None:
    """Send an OSC sequence by printing it in the shell."""
    command = f"printf '{sequence}'\\n"
    if surface is None:
        client.send(command)
    else:
        client.send_surface(surface, command)


def test_clear_prior_notifications(client: cmux) -> TestResult:
    result = TestResult("Clear Prior Panel Notifications")
    try:
        client.clear_notifications()
        client.set_app_focus(False)
        client.notify("first")
        time.sleep(0.1)
        client.notify("second")
        items = wait_for_notifications(client, 1)
        if len(items) != 1:
            result.failure(f"Expected 1 notification, got {len(items)}")
        elif items[0]["title"] != "second":
            result.failure(f"Expected latest title 'second', got '{items[0]['title']}'")
        else:
            result.success("Prior panel notifications cleared")
    except Exception as e:
        result.failure(f"Exception: {e}")
    return result


def test_suppress_when_focused(client: cmux) -> TestResult:
    result = TestResult("Suppress When App+Panel Focused")
    try:
        client.clear_notifications()
        client.set_app_focus(True)
        client.notify("focused")
        items = wait_for_notifications(client, 0)
        if len(items) == 0:
            result.success("Suppressed notification when focused")
        else:
            result.failure(f"Expected 0 notifications, got {len(items)}")
    except Exception as e:
        result.failure(f"Exception: {e}")
    return result


def test_not_suppressed_when_inactive(client: cmux) -> TestResult:
    result = TestResult("Allow When App Inactive")
    try:
        client.clear_notifications()
        client.set_app_focus(False)
        client.notify("inactive")
        items = wait_for_notifications(client, 1)
        if len(items) != 1:
            result.failure(f"Expected 1 notification, got {len(items)}")
        elif items[0]["is_read"]:
            result.failure("Expected notification to be unread")
        else:
            result.success("Notification stored when app inactive")
    except Exception as e:
        result.failure(f"Exception: {e}")
    return result


def test_kitty_notification_simple(client: cmux) -> TestResult:
    result = TestResult("Kitty OSC 99 Simple")
    try:
        client.clear_notifications()
        client.set_app_focus(False)
        # Avoid Ghostty's 1s desktop notification rate limit. This test can run
        # immediately after app launch in CI/VM environments.
        time.sleep(1.1)
        surface = focused_surface_index(client)
        send_osc(client, "\\x1b]99;;Kitty Simple\\x1b\\\\", surface)
        items = wait_for_notifications(client, 1)
        if len(items) != 1:
            result.failure(f"Expected 1 notification, got {len(items)}")
        elif items[0]["title"] != "Kitty Simple":
            result.failure(f"Expected title 'Kitty Simple', got '{items[0]['title']}'")
        else:
            result.success("OSC 99 simple notification received")
    except Exception as e:
        result.failure(f"Exception: {e}")
    return result


def test_kitty_notification_chunked(client: cmux) -> TestResult:
    result = TestResult("Kitty OSC 99 Chunked Title/Body")
    try:
        client.clear_notifications()
        client.set_app_focus(False)
        # Avoid Ghostty's 1s desktop notification rate limit.
        time.sleep(1.1)
        surface = focused_surface_index(client)
        send_osc(client, "\\x1b]99;i=kitty:d=0:p=title;Kitty Title\\x1b\\\\", surface)
        time.sleep(0.1)
        items = client.list_notifications()
        if items:
            result.failure("Expected no notification before final chunk")
            return result
        send_osc(client, "\\x1b]99;i=kitty:p=body;Kitty Body\\x1b\\\\", surface)
        items = wait_for_notifications(client, 1)
        if len(items) != 1:
            result.failure(f"Expected 1 notification, got {len(items)}")
        elif items[0]["title"] != "Kitty Title" or items[0]["body"] != "Kitty Body":
            result.failure(
                f"Expected title/body 'Kitty Title'/'Kitty Body', got "
                f"'{items[0]['title']}'/'{items[0]['body']}'"
            )
        else:
            result.success("OSC 99 chunked notification received")
    except Exception as e:
        result.failure(f"Exception: {e}")
    return result


def test_rxvt_notification_osc777(client: cmux) -> TestResult:
    result = TestResult("RXVT OSC 777 Notification")
    try:
        client.clear_notifications()
        client.set_app_focus(False)
        # Avoid Ghostty's 1s desktop notification rate limit.
        time.sleep(1.1)
        surface = focused_surface_index(client)
        command = "printf '\\x1b]777;notify;OSC777 Title;OSC777 Body\\x07'"
        client.send_surface(surface, command + "\\n")
        items = wait_for_notifications(client, 1)
        if len(items) != 1:
            result.failure(f"Expected 1 notification, got {len(items)}")
        elif items[0]["title"] != "OSC777 Title" or items[0]["body"] != "OSC777 Body":
            result.failure(
                f"Expected title/body 'OSC777 Title'/'OSC777 Body', got "
                f"'{items[0]['title']}'/'{items[0]['body']}'"
            )
        else:
            result.success("OSC 777 notification received")
    except Exception as e:
        result.failure(f"Exception: {e}")
    return result


def test_mark_read_on_focus_change(client: cmux) -> TestResult:
    result = TestResult("Mark Read On Panel Focus")
    try:
        client.clear_notifications()
        client.reset_flash_counts()
        surfaces = ensure_two_surfaces(client)
        focused = next((s for s in surfaces if s[2]), None)
        other = next((s for s in surfaces if not s[2]), None)
        if focused is None or other is None:
            result.failure("Unable to identify focused and unfocused surfaces")
            return result

        client.set_app_focus(False)
        client.notify_surface(other[0], "focusread")
        time.sleep(0.1)

        client.set_app_focus(True)
        client.focus_surface(other[0])
        time.sleep(0.1)

        items = client.list_notifications()
        target = next((n for n in items if n["surface_id"] == other[1]), None)
        if target is None:
            result.failure("Expected notification for target surface")
        elif not target["is_read"]:
            result.failure("Expected notification to be marked read on focus")
        else:
            count = wait_for_flash_count(client, other[1], minimum=1, timeout=2.0)
            if count < 1:
                result.failure("Expected flash on panel focus dismissal")
            else:
                result.success("Notification marked read on focus")
    except Exception as e:
        result.failure(f"Exception: {e}")
    return result


def test_mark_read_on_app_active(client: cmux) -> TestResult:
    result = TestResult("Mark Read On App Active")
    try:
        client.clear_notifications()
        client.set_app_focus(False)
        client.notify("activate")
        time.sleep(0.1)

        items = client.list_notifications()
        if not items or items[0]["is_read"]:
            result.failure("Expected unread notification before activation")
            return result

        client.simulate_app_active()
        time.sleep(0.1)

        items = client.list_notifications()
        if not items:
            result.failure("Expected notification to remain after activation")
        elif not items[0]["is_read"]:
            result.failure("Expected notification to be marked read on app active")
        else:
            result.success("Notification marked read on app active")
    except Exception as e:
        result.failure(f"Exception: {e}")
    return result


def test_mark_read_on_tab_switch(client: cmux) -> TestResult:
    result = TestResult("Mark Read On Tab Switch")
    try:
        client.clear_notifications()
        client.set_app_focus(False)
        tab1 = client.current_workspace()
        client.notify("tabswitch")
        time.sleep(0.1)

        tab2 = client.new_workspace()
        time.sleep(0.1)

        client.set_app_focus(True)
        client.select_workspace(tab1)
        time.sleep(0.1)

        items = client.list_notifications()
        target = next((n for n in items if n["workspace_id"] == tab1), None)
        if target is None:
            result.failure("Expected notification for original tab")
        elif not target["is_read"]:
            result.failure("Expected notification to be marked read on tab switch")
        else:
            result.success("Notification marked read on tab switch")
    except Exception as e:
        result.failure(f"Exception: {e}")
    return result


def test_flash_on_tab_switch(client: cmux) -> TestResult:
    result = TestResult("Flash On Tab Switch")
    try:
        client.clear_notifications()
        client.reset_flash_counts()

        tab1 = client.current_workspace()
        surfaces = client.list_surfaces()
        focused = next((s for s in surfaces if s[2]), None)
        if focused is None:
            result.failure("Unable to identify focused surface")
            return result

        client.set_app_focus(False)
        client.notify("tabswitchflash")
        time.sleep(0.1)

        client.new_workspace()
        time.sleep(0.1)

        client.set_app_focus(True)
        client.select_workspace(tab1)
        time.sleep(0.2)

        count = wait_for_flash_count(client, focused[1], minimum=1, timeout=2.0)
        if count < 1:
            result.failure(f"Expected flash count >= 1, got {count}")
        else:
            result.success("Flash triggered on tab switch dismissal")
    except Exception as e:
        result.failure(f"Exception: {e}")
    return result


def test_focus_on_notification_click(client: cmux) -> TestResult:
    result = TestResult("Focus On Notification Click")
    try:
        client.clear_notifications()
        client.reset_flash_counts()

        surfaces = ensure_two_surfaces(client)
        focused = next((s for s in surfaces if s[2]), None)
        other = next((s for s in surfaces if not s[2]), None)
        if focused is None or other is None:
            result.failure("Unable to identify focused and unfocused surfaces")
            return result

        client.set_app_focus(False)
        client.notify_surface(other[0], "notifyfocus")
        time.sleep(0.1)

        client.set_app_focus(True)
        workspace_id = client.current_workspace()
        client.focus_notification(workspace_id, other[0])
        time.sleep(0.2)

        surfaces = client.list_surfaces()
        target = next((s for s in surfaces if s[1] == other[1]), None)
        if target is None or not target[2]:
            result.failure("Expected notification surface to be focused")
            return result

        count = wait_for_flash_count(client, other[1], minimum=1, timeout=2.0)
        if count < 1:
            result.failure(f"Expected flash count >= 1, got {count}")
        else:
            result.success("Notification click focuses and flashes panel")
    except Exception as e:
        result.failure(f"Exception: {e}")
    return result


def test_restore_focus_on_tab_switch(client: cmux) -> TestResult:
    result = TestResult("Restore Focus On Tab Switch")
    try:
        client.clear_notifications()
        client.set_app_focus(True)

        surfaces = ensure_two_surfaces(client)
        focused = next((s for s in surfaces if s[2]), None)
        other = next((s for s in surfaces if not s[2]), None)
        if focused is None or other is None:
            result.failure("Unable to identify focused and unfocused surfaces")
            return result

        client.focus_surface(other[0])
        time.sleep(0.1)

        tab1 = client.current_workspace()
        client.new_workspace()
        time.sleep(0.1)

        client.select_workspace(tab1)
        time.sleep(0.2)

        surfaces = client.list_surfaces()
        target = next((s for s in surfaces if s[1] == other[1]), None)
        if target is None:
            result.failure("Unable to find previously focused surface")
        elif not target[2]:
            result.failure("Expected previously focused surface to be focused after tab switch")
        else:
            result.success("Restored last focused surface after tab switch")
    except Exception as e:
        result.failure(f"Exception: {e}")
    return result


def test_clear_on_tab_close(client: cmux) -> TestResult:
    result = TestResult("Clear On Tab Close")
    try:
        client.clear_notifications()
        client.set_app_focus(False)
        tab1 = client.current_workspace()
        client.notify("closetab")
        time.sleep(0.1)

        items = wait_for_notifications(client, 1)
        if len(items) != 1:
            result.failure(f"Expected 1 notification, got {len(items)}")
            return result

        client.new_workspace()
        time.sleep(0.1)
        client.close_workspace(tab1)
        time.sleep(0.2)

        items = client.list_notifications()
        if items:
            result.failure(f"Expected 0 notifications after tab close, got {len(items)}")
        else:
            result.success("Notifications cleared when tab closed")
    except Exception as e:
        result.failure(f"Exception: {e}")
    return result


def run_tests() -> int:
    results = []
    with cmux() as client:
        results.append(test_clear_prior_notifications(client))
        results.append(test_suppress_when_focused(client))
        results.append(test_not_suppressed_when_inactive(client))
        results.append(test_kitty_notification_simple(client))
        results.append(test_kitty_notification_chunked(client))
        results.append(test_rxvt_notification_osc777(client))
        results.append(test_mark_read_on_focus_change(client))
        results.append(test_mark_read_on_app_active(client))
        results.append(test_mark_read_on_tab_switch(client))
        results.append(test_flash_on_tab_switch(client))
        results.append(test_focus_on_notification_click(client))
        results.append(test_restore_focus_on_tab_switch(client))
        results.append(test_clear_on_tab_close(client))
        client.set_app_focus(None)
        client.clear_notifications()

    print("\nNotification Tests:")
    for r in results:
        status = "PASS" if r.passed else "FAIL"
        msg = f" - {r.message}" if r.message else ""
        print(f"{status}: {r.name}{msg}")

    passed = sum(1 for r in results if r.passed)
    total = len(results)
    if passed == total:
        print("\n🎉 All notification tests passed!")
        return 0
    print(f"\n⚠️  {total - passed} test(s) failed")
    return 1


if __name__ == "__main__":
    sys.exit(run_tests())