321 lines
11 KiB
Go
321 lines
11 KiB
Go
// Cthulhu screen reader.
|
|
// Copyright (c) 2026 Storm Dragon
|
|
// Authors: Storm Dragon, Angus Kola
|
|
//
|
|
// This program is free software: you can redistribute it and/or modify it
|
|
// under the terms of the GNU General Public License as published by the Free
|
|
// Software Foundation, either version 3 of the License, or (at your option)
|
|
// any later version.
|
|
//
|
|
// This program is distributed in the hope that it will be useful, but WITHOUT
|
|
// ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
|
|
// FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
|
|
// more details.
|
|
//
|
|
// You should have received a copy of the GNU General Public License along with
|
|
// this program. If not, see <https://www.gnu.org/licenses/gpl-3.0.html>.
|
|
|
|
package app
|
|
|
|
import (
|
|
"context"
|
|
"log/slog"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
|
|
"git.stormux.org/storm/cthulhu/internal/atspi"
|
|
)
|
|
|
|
func TestProcessEventsTerminalScrollbackDeletionBothSessions(t *testing.T) {
|
|
for _, sessionType := range []string{"wayland", "x11"} {
|
|
t.Run(sessionType, func(t *testing.T) {
|
|
t.Setenv("XDG_SESSION_TYPE", sessionType)
|
|
// Both sessions must apply the same event policy, including
|
|
// an X11 launcher which also supplies WAYLAND_DISPLAY.
|
|
t.Setenv("WAYLAND_DISPLAY", "wayland-test")
|
|
speech := &fakeSpeechQueue{spoken: make(chan string, 3)}
|
|
adapter := &fakeAdapter{events: make(chan atspi.Event, 3)}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
completed := make(chan struct{})
|
|
go func() {
|
|
defer close(completed)
|
|
processEvents(ctx, slog.New(slog.DiscardHandler), speech, adapter, 0)
|
|
}()
|
|
t.Cleanup(func() {
|
|
cancel()
|
|
select {
|
|
case <-completed:
|
|
case <-time.After(time.Second):
|
|
t.Error("event loop did not stop")
|
|
}
|
|
})
|
|
|
|
terminal := atspi.NewSnapshot("/accessible/terminal", "", "terminal", "", "Username@Host> ")
|
|
terminal.TextCaretKnown = true
|
|
terminal.BrailleLineStart = 300
|
|
terminal.BrailleCaret = 315
|
|
oldPrompt := "Username@Host> old command\n"
|
|
deletion := atspi.NewEventWithPayload(atspi.TextDeletedEvent, terminal, oldPrompt)
|
|
output := atspi.NewEventWithPayload(atspi.TextInsertedEvent, terminal, "Current command output")
|
|
output.Detail1 = 200
|
|
// A caret event can arrive between the scroll and its insertion.
|
|
caret := atspi.NewEvent(atspi.CaretEvent, terminal)
|
|
adapter.events <- deletion
|
|
adapter.events <- caret
|
|
adapter.events <- output
|
|
|
|
for _, want := range []string{"Current command output"} {
|
|
select {
|
|
case got := <-speech.spoken:
|
|
if got != want {
|
|
t.Fatalf("speech = %q, want %q", got, want)
|
|
}
|
|
case <-time.After(time.Second):
|
|
t.Fatalf("timed out waiting for %q", want)
|
|
}
|
|
}
|
|
for _, deleted := range []string{"x", "command"} {
|
|
edit := atspi.NewEventWithPayload(atspi.TextDeletedEvent, terminal, deleted)
|
|
edit.Detail1 = terminal.BrailleLineStart
|
|
adapter.events <- edit
|
|
select {
|
|
case got := <-speech.spoken:
|
|
if got != deleted {
|
|
t.Fatalf("editing speech = %q, want %q", got, deleted)
|
|
}
|
|
case <-time.After(time.Second):
|
|
t.Fatalf("timed out waiting for deletion %q", deleted)
|
|
}
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestTerminalScrollbackDeletionPreservesEditing(t *testing.T) {
|
|
t.Setenv("XDG_SESSION_TYPE", "wayland")
|
|
terminal := atspi.NewSnapshot("/accessible/terminal", "", "terminal", "", "Prompt")
|
|
terminal.TextCaretKnown = true
|
|
terminal.BrailleLineStart = 100
|
|
terminal.BrailleCaret = 110
|
|
base := atspi.NewEventWithPayload(atspi.TextDeletedEvent, terminal, "Old prompt\n")
|
|
for _, test := range []struct {
|
|
name string
|
|
modify func(*atspi.Event)
|
|
ignore bool
|
|
}{
|
|
{"scroll", func(event *atspi.Event) {}, true},
|
|
{"unicode offsets", func(event *atspi.Event) {
|
|
*event = atspi.NewEventWithPayload(atspi.TextDeletedEvent, terminal, strings.Repeat("猫", 40)+"\n")
|
|
}, true},
|
|
{"backspace", func(event *atspi.Event) { event.Payload = "x"; event.Detail1 = 109 }, false},
|
|
{"delete", func(event *atspi.Event) { event.Payload = "x"; event.Detail1 = 110 }, false},
|
|
{"delete word", func(event *atspi.Event) { event.Payload = "command"; event.Detail1 = 100 }, false},
|
|
{"first character", func(event *atspi.Event) {
|
|
*event = atspi.NewEventWithPayload(atspi.TextDeletedEvent, terminal, "x")
|
|
}, false},
|
|
{"newline", func(event *atspi.Event) { event.Payload = "\n"; event.Detail1 = 100 }, false},
|
|
{"current first line", func(event *atspi.Event) { event.Source.BrailleLineStart = 0 }, false},
|
|
{"range reaches current line", func(event *atspi.Event) { event.Source.BrailleLineStart = 5 }, false},
|
|
{"unknown caret", func(event *atspi.Event) { event.Source.TextCaretKnown = false }, false},
|
|
{"partial line", func(event *atspi.Event) {
|
|
*event = atspi.NewEventWithPayload(atspi.TextDeletedEvent, terminal, "Old prompt")
|
|
}, false},
|
|
{"editor", func(event *atspi.Event) { event.Source.Role = "text" }, false},
|
|
{"insertion", func(event *atspi.Event) { event.Type = atspi.TextInsertedEvent }, false},
|
|
} {
|
|
t.Run(test.name, func(t *testing.T) {
|
|
event := base
|
|
test.modify(&event)
|
|
if got := should_ignore_terminal_scrollback(event); got != test.ignore {
|
|
t.Fatalf("ignore = %v, want %v", got, test.ignore)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestProcessEventsCoalescesTerminalDeletionWithReplacementInsertion(t *testing.T) {
|
|
speech := &fakeSpeechQueue{spoken: make(chan string, 3)}
|
|
adapter := &fakeAdapter{events: make(chan atspi.Event, 3)}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
completed := make(chan struct{})
|
|
go func() {
|
|
processEvents(ctx, slog.New(slog.DiscardHandler), speech, adapter, 0)
|
|
close(completed)
|
|
}()
|
|
|
|
terminal := atspi.NewSnapshot("/accessible/terminal", "", "terminal", "", "")
|
|
output := atspi.NewEventWithPayload(atspi.TextInsertedEvent, terminal, "Command output")
|
|
deletion := atspi.NewEventWithPayload(atspi.TextDeletedEvent, terminal, "old prompt")
|
|
deletion.Detail1 = 12
|
|
prompt := atspi.NewEventWithPayload(atspi.TextInsertedEvent, terminal, "new prompt")
|
|
prompt.Detail1 = 12
|
|
adapter.events <- output
|
|
adapter.events <- deletion
|
|
adapter.events <- prompt
|
|
|
|
for _, expected := range []string{"Command output", "new prompt"} {
|
|
select {
|
|
case text := <-speech.spoken:
|
|
if text != expected {
|
|
t.Fatalf("spoken text = %q, want %q", text, expected)
|
|
}
|
|
case <-time.After(time.Second):
|
|
t.Fatalf("timed out waiting for terminal speech %q", expected)
|
|
}
|
|
}
|
|
select {
|
|
case text := <-speech.spoken:
|
|
t.Fatalf("terminal redraw deletion was presented as %q", text)
|
|
case <-time.After(2 * terminalMutationDebounce):
|
|
}
|
|
|
|
cancel()
|
|
select {
|
|
case <-completed:
|
|
case <-time.After(time.Second):
|
|
t.Fatal("event loop did not stop")
|
|
}
|
|
}
|
|
|
|
func TestProcessEventsPresentsStandaloneTerminalDeletion(t *testing.T) {
|
|
speech := &fakeSpeechQueue{spoken: make(chan string, 1)}
|
|
adapter := &fakeAdapter{events: make(chan atspi.Event, 1)}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
completed := make(chan struct{})
|
|
go func() {
|
|
processEvents(ctx, slog.New(slog.DiscardHandler), speech, adapter, 0)
|
|
close(completed)
|
|
}()
|
|
|
|
terminal := atspi.NewSnapshot("/accessible/terminal", "", "terminal", "", "")
|
|
adapter.events <- atspi.NewEventWithPayload(atspi.TextDeletedEvent, terminal, "x")
|
|
select {
|
|
case text := <-speech.spoken:
|
|
if text != "x" {
|
|
t.Fatalf("spoken deletion = %q, want %q", text, "x")
|
|
}
|
|
case <-time.After(time.Second):
|
|
t.Fatal("standalone terminal deletion was not presented")
|
|
}
|
|
|
|
cancel()
|
|
select {
|
|
case <-completed:
|
|
case <-time.After(time.Second):
|
|
t.Fatal("event loop did not stop")
|
|
}
|
|
}
|
|
|
|
func TestProcessEventsDoesNotCoalesceTerminalChangesAtDifferentOffsets(t *testing.T) {
|
|
speech := &fakeSpeechQueue{spoken: make(chan string, 2)}
|
|
adapter := &fakeAdapter{events: make(chan atspi.Event, 2)}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
completed := make(chan struct{})
|
|
go func() {
|
|
processEvents(ctx, slog.New(slog.DiscardHandler), speech, adapter, 0)
|
|
close(completed)
|
|
}()
|
|
|
|
terminal := atspi.NewSnapshot("/accessible/terminal", "", "terminal", "", "")
|
|
deletion := atspi.NewEventWithPayload(atspi.TextDeletedEvent, terminal, "x")
|
|
deletion.Detail1 = 12
|
|
insertion := atspi.NewEventWithPayload(atspi.TextInsertedEvent, terminal, "y")
|
|
insertion.Detail1 = 13
|
|
adapter.events <- deletion
|
|
adapter.events <- insertion
|
|
|
|
for _, expected := range []string{"x", "y"} {
|
|
select {
|
|
case text := <-speech.spoken:
|
|
if text != expected {
|
|
t.Fatalf("spoken text = %q, want %q", text, expected)
|
|
}
|
|
case <-time.After(time.Second):
|
|
t.Fatalf("timed out waiting for terminal speech %q", expected)
|
|
}
|
|
}
|
|
|
|
cancel()
|
|
select {
|
|
case <-completed:
|
|
case <-time.After(time.Second):
|
|
t.Fatal("event loop did not stop")
|
|
}
|
|
}
|
|
|
|
func TestProcessEventsCoalescesTerminalFocusWithFollowingInsertion(t *testing.T) {
|
|
speech := &fakeSpeechQueue{spoken: make(chan string, 2)}
|
|
adapter := &fakeAdapter{events: make(chan atspi.Event, 2)}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
completed := make(chan struct{})
|
|
go func() {
|
|
processEvents(ctx, slog.New(slog.DiscardHandler), speech, adapter, time.Hour)
|
|
close(completed)
|
|
}()
|
|
|
|
terminal := atspi.NewSnapshot("/accessible/terminal", "Terminal", "terminal", "", "Old prompt")
|
|
focused := atspi.NewEvent(atspi.FocusedStateEvent, terminal)
|
|
focused.Detail1 = 1
|
|
adapter.events <- focused
|
|
adapter.events <- atspi.NewEventWithPayload(atspi.TextInsertedEvent, terminal, "New prompt")
|
|
|
|
select {
|
|
case text := <-speech.spoken:
|
|
if text != "New prompt" {
|
|
t.Fatalf("speech = %q, want only following insertion", text)
|
|
}
|
|
case <-time.After(time.Second):
|
|
t.Fatal("following terminal insertion was not presented")
|
|
}
|
|
select {
|
|
case text := <-speech.spoken:
|
|
t.Fatalf("redundant terminal focus was presented as %q", text)
|
|
default:
|
|
}
|
|
|
|
cancel()
|
|
select {
|
|
case <-completed:
|
|
case <-time.After(time.Second):
|
|
t.Fatal("event loop did not stop")
|
|
}
|
|
}
|
|
|
|
func TestProcessEventsPresentsTerminalFocusWithoutFollowingInsertion(t *testing.T) {
|
|
speech := &fakeSpeechQueue{spoken: make(chan string, 1)}
|
|
adapter := &fakeAdapter{events: make(chan atspi.Event, 1)}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
completed := make(chan struct{})
|
|
go func() {
|
|
processEvents(ctx, slog.New(slog.DiscardHandler), speech, adapter, 0)
|
|
close(completed)
|
|
}()
|
|
|
|
terminal := atspi.NewSnapshot("/accessible/terminal", "Terminal", "terminal", "", "Prompt")
|
|
terminal.ApplicationName = "TestTerminal"
|
|
focused := atspi.NewEvent(atspi.FocusedStateEvent, terminal)
|
|
focused.Detail1 = 1
|
|
adapter.events <- focused
|
|
select {
|
|
case text := <-speech.spoken:
|
|
if text != "Terminal, terminal, Prompt" {
|
|
t.Fatalf("terminal focus speech = %q", text)
|
|
}
|
|
case <-time.After(time.Second):
|
|
t.Fatal("terminal focus was not presented after debounce")
|
|
}
|
|
|
|
cancel()
|
|
select {
|
|
case <-completed:
|
|
case <-time.After(time.Second):
|
|
t.Fatal("event loop did not stop")
|
|
}
|
|
}
|