4606800856

Committed 04 Apr 2023 10:40AM UTC coverage: 57.843% (-0.08%) from 57.923%

Build Type

push

github

Committed by Christian Muehlhaeuser

Commit Message

fix: Cloud Shell supports RGB colors

Run Details

3 of 3 new or added lines in 1 file covered. (100.0%)

531 of 918 relevant lines covered (57.84%)

8.86 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

8.16

/termenv_unix.go

//go:build darwin || dragonfly || freebsd || linux || netbsd || openbsd || solaris
// +build darwin dragonfly freebsd linux netbsd openbsd solaris

package termenv

import (
        "fmt"
        "io"
        "strconv"
        "strings"
        "time"

        "golang.org/x/sys/unix"
)

const (
        // timeout for OSC queries
        OSCTimeout = 5 * time.Second
)

// ColorProfile returns the supported color profile:
// Ascii, ANSI, ANSI256, or TrueColor.
func (o *Output) ColorProfile() Profile {
        if !o.isTTY() {
                return Ascii
        }

        if o.environ.Getenv("GOOGLE_CLOUD_SHELL") == "true" {
                return TrueColor
        }

        term := o.environ.Getenv("TERM")
        colorTerm := o.environ.Getenv("COLORTERM")

        switch strings.ToLower(colorTerm) {
        case "24bit":
                fallthrough
        case "truecolor":
                if strings.HasPrefix(term, "screen") {
                        // tmux supports TrueColor, screen only ANSI256
                        if o.environ.Getenv("TERM_PROGRAM") != "tmux" {
                                return ANSI256
                        }
                }
                return TrueColor
        case "yes":
                fallthrough
        case "true":
                return ANSI256
        }

        switch term {
        case "xterm-kitty":
                return TrueColor
        case "linux":
                return ANSI
        }

        if strings.Contains(term, "256color") {
                return ANSI256
        }
        if strings.Contains(term, "color") {
                return ANSI
        }
        if strings.Contains(term, "ansi") {
                return ANSI
        }

        return Ascii
}

func (o Output) foregroundColor() Color {
        s, err := o.termStatusReport(10)
        if err == nil {
                c, err := xTermColor(s)
                if err == nil {
                        return c
                }
        }

        colorFGBG := o.environ.Getenv("COLORFGBG")
        if strings.Contains(colorFGBG, ";") {
                c := strings.Split(colorFGBG, ";")
                i, err := strconv.Atoi(c[0])
                if err == nil {
                        return ANSIColor(i)
                }
        }

        // default gray
        return ANSIColor(7)
}

func (o Output) backgroundColor() Color {
        s, err := o.termStatusReport(11)
        if err == nil {
                c, err := xTermColor(s)
                if err == nil {
                        return c
                }
        }

        colorFGBG := o.environ.Getenv("COLORFGBG")
        if strings.Contains(colorFGBG, ";") {
                c := strings.Split(colorFGBG, ";")
                i, err := strconv.Atoi(c[len(c)-1])
                if err == nil {
                        return ANSIColor(i)
                }
        }

        // default black
        return ANSIColor(0)
}

func (o *Output) waitForData(timeout time.Duration) error {
        fd := o.TTY().Fd()
        tv := unix.NsecToTimeval(int64(timeout))
        var readfds unix.FdSet
        readfds.Set(int(fd))

        for {
                n, err := unix.Select(int(fd)+1, &readfds, nil, nil, &tv)
                if err == unix.EINTR {
                        continue
                }
                if err != nil {
                        return err
                }
                if n == 0 {
                        return fmt.Errorf("timeout")
                }

                break
        }

        return nil
}

func (o *Output) readNextByte() (byte, error) {
        if !o.unsafe {
                if err := o.waitForData(OSCTimeout); err != nil {
                        return 0, err
                }
        }

        var b [1]byte
        n, err := o.TTY().Read(b[:])
        if err != nil {
                return 0, err
        }

        if n == 0 {
                panic("read returned no data")
        }

        return b[0], nil
}

// readNextResponse reads either an OSC response or a cursor position response:
//   - OSC response: "\x1b]11;rgb:1111/1111/1111\x1b\\"
//   - cursor position response: "\x1b[42;1R"
func (o *Output) readNextResponse() (response string, isOSC bool, err error) {
        start, err := o.readNextByte()
        if err != nil {
                return "", false, err
        }

        // first byte must be ESC
        for start != ESC {
                start, err = o.readNextByte()
                if err != nil {
                        return "", false, err
                }
        }

        response += string(start)

        // next byte is either '[' (cursor position response) or ']' (OSC response)
        tpe, err := o.readNextByte()
        if err != nil {
                return "", false, err
        }

        response += string(tpe)

        var oscResponse bool
        switch tpe {
        case '[':
                oscResponse = false
        case ']':
                oscResponse = true
        default:
                return "", false, ErrStatusReport
        }

        for {
                b, err := o.readNextByte()
                if err != nil {
                        return "", false, err
                }

                response += string(b)

                if oscResponse {
                        // OSC can be terminated by BEL (\a) or ST (ESC)
                        if b == BEL || strings.HasSuffix(response, string(ESC)) {
                                return response, true, nil
                        }
                } else {
                        // cursor position response is terminated by 'R'
                        if b == 'R' {
                                return response, false, nil
                        }
                }

                // both responses have less than 25 bytes, so if we read more, that's an error
                if len(response) > 25 {
                        break
                }
        }

        return "", false, ErrStatusReport
}

func (o Output) termStatusReport(sequence int) (string, error) {
        // screen/tmux can't support OSC, because they can be connected to multiple
        // terminals concurrently.
        term := o.environ.Getenv("TERM")
        if strings.HasPrefix(term, "screen") || strings.HasPrefix(term, "tmux") {
                return "", ErrStatusReport
        }

        tty := o.TTY()
        if tty == nil {
                return "", ErrStatusReport
        }

        if !o.unsafe {
                fd := int(tty.Fd())
                // if in background, we can't control the terminal
                if !isForeground(fd) {
                        return "", ErrStatusReport
                }

                t, err := unix.IoctlGetTermios(fd, tcgetattr)
                if err != nil {
                        return "", fmt.Errorf("%s: %s", ErrStatusReport, err)
                }
                defer unix.IoctlSetTermios(fd, tcsetattr, t) //nolint:errcheck

                noecho := *t
                noecho.Lflag = noecho.Lflag &^ unix.ECHO
                noecho.Lflag = noecho.Lflag &^ unix.ICANON
                if err := unix.IoctlSetTermios(fd, tcsetattr, &noecho); err != nil {
                        return "", fmt.Errorf("%s: %s", ErrStatusReport, err)
                }
        }

        // first, send OSC query, which is ignored by terminal which do not support it
        fmt.Fprintf(tty, OSC+"%d;?"+ST, sequence)

        // then, query cursor position, should be supported by all terminals
        fmt.Fprintf(tty, CSI+"6n")

        // read the next response
        res, isOSC, err := o.readNextResponse()
        if err != nil {
                return "", fmt.Errorf("%s: %s", ErrStatusReport, err)
        }

        // if this is not OSC response, then the terminal does not support it
        if !isOSC {
                return "", ErrStatusReport
        }

        // read the cursor query response next and discard the result
        _, _, err = o.readNextResponse()
        if err != nil {
                return "", err
        }

        // fmt.Println("Rcvd", res[1:])
        return res, nil
}

// EnableVirtualTerminalProcessing enables virtual terminal processing on
// Windows for w and returns a function that restores w to its previous state.
// On non-Windows platforms, or if w does not refer to a terminal, then it
// returns a non-nil no-op function and no error.
func EnableVirtualTerminalProcessing(w io.Writer) (func() error, error) {
        return func() error { return nil }, nil
}

1	//go:build darwin \|\| dragonfly \|\| freebsd \|\| linux \|\| netbsd \|\| openbsd \|\| solaris
2	// +build darwin dragonfly freebsd linux netbsd openbsd solaris
3
4	package termenv
5
6	import (
7	"fmt"
8	"io"
9	"strconv"
10	"strings"
11	"time"
12
13	"golang.org/x/sys/unix"
14	)
15
16	const (
17	// timeout for OSC queries
18	OSCTimeout = 5 * time.Second
19	)
20
21	// ColorProfile returns the supported color profile:
22	// Ascii, ANSI, ANSI256, or TrueColor.
23	func (o *Output) ColorProfile() Profile {	14✔
24	if !o.isTTY() {	27✔
25	return Ascii	13✔
26	}	13✔
27
28	if o.environ.Getenv("GOOGLE_CLOUD_SHELL") == "true" {	1✔
29	return TrueColor	×
30	}	×
31
32	term := o.environ.Getenv("TERM")	1✔
33	colorTerm := o.environ.Getenv("COLORTERM")	1✔
34		1✔
35	switch strings.ToLower(colorTerm) {	1✔
36	case "24bit":	×
37	fallthrough	×
38	case "truecolor":	×
39	if strings.HasPrefix(term, "screen") {	×
40	// tmux supports TrueColor, screen only ANSI256	×
41	if o.environ.Getenv("TERM_PROGRAM") != "tmux" {	×
42	return ANSI256	×
43	}	×
44	}
45	return TrueColor	×
46	case "yes":	×
47	fallthrough	×
48	case "true":	×
49	return ANSI256	×
50	}
51
52	switch term {	1✔
53	case "xterm-kitty":	×
54	return TrueColor	×
55	case "linux":	×
56	return ANSI	×
57	}
58
59	if strings.Contains(term, "256color") {	1✔
60	return ANSI256	×
61	}	×
62	if strings.Contains(term, "color") {	1✔
63	return ANSI	×
64	}	×
65	if strings.Contains(term, "ansi") {	1✔
66	return ANSI	×
67	}	×
68
69	return Ascii	1✔
70	}
71
72	func (o Output) foregroundColor() Color {	×
73	s, err := o.termStatusReport(10)	×
74	if err == nil {	×
75	c, err := xTermColor(s)	×
76	if err == nil {	×
77	return c	×
78	}	×
79	}
80
81	colorFGBG := o.environ.Getenv("COLORFGBG")	×
82	if strings.Contains(colorFGBG, ";") {	×
83	c := strings.Split(colorFGBG, ";")	×
84	i, err := strconv.Atoi(c[0])	×
85	if err == nil {	×
86	return ANSIColor(i)	×
87	}	×
88	}
89
90	// default gray
91	return ANSIColor(7)	×
92	}
93
94	func (o Output) backgroundColor() Color {	×
95	s, err := o.termStatusReport(11)	×
96	if err == nil {	×
97	c, err := xTermColor(s)	×
98	if err == nil {	×
99	return c	×
100	}	×
101	}
102
103	colorFGBG := o.environ.Getenv("COLORFGBG")	×
104	if strings.Contains(colorFGBG, ";") {	×
105	c := strings.Split(colorFGBG, ";")	×
106	i, err := strconv.Atoi(c[len(c)-1])	×
107	if err == nil {	×
108	return ANSIColor(i)	×
109	}	×
110	}
111
112	// default black
113	return ANSIColor(0)	×
114	}
115
116	func (o *Output) waitForData(timeout time.Duration) error {	×
117	fd := o.TTY().Fd()	×
118	tv := unix.NsecToTimeval(int64(timeout))	×
119	var readfds unix.FdSet	×
120	readfds.Set(int(fd))	×
121		×
122	for {	×
123	n, err := unix.Select(int(fd)+1, &readfds, nil, nil, &tv)	×
124	if err == unix.EINTR {	×
125	continue	×
126	}
127	if err != nil {	×
128	return err	×
129	}	×
130	if n == 0 {	×
131	return fmt.Errorf("timeout")	×
132	}	×
133
134	break	×
135	}
136
137	return nil	×
138	}
139
140	func (o *Output) readNextByte() (byte, error) {	×
141	if !o.unsafe {	×
142	if err := o.waitForData(OSCTimeout); err != nil {	×
143	return 0, err	×
144	}	×
145	}
146
147	var b [1]byte	×
148	n, err := o.TTY().Read(b[:])	×
149	if err != nil {	×
150	return 0, err	×
151	}	×
152
153	if n == 0 {	×
154	panic("read returned no data")	×
155	}
156
157	return b[0], nil	×
158	}
159
160	// readNextResponse reads either an OSC response or a cursor position response:
161	// - OSC response: "\x1b]11;rgb:1111/1111/1111\x1b\\"
162	// - cursor position response: "\x1b[42;1R"
163	func (o *Output) readNextResponse() (response string, isOSC bool, err error) {	×
164	start, err := o.readNextByte()	×
165	if err != nil {	×
166	return "", false, err	×
167	}	×
168
169	// first byte must be ESC
170	for start != ESC {	×
171	start, err = o.readNextByte()	×
172	if err != nil {	×
173	return "", false, err	×
174	}	×
175	}
176
177	response += string(start)	×
178		×
179	// next byte is either '[' (cursor position response) or ']' (OSC response)	×
180	tpe, err := o.readNextByte()	×
181	if err != nil {	×
182	return "", false, err	×
183	}	×
184
185	response += string(tpe)	×
186		×
187	var oscResponse bool	×
188	switch tpe {	×
189	case '[':	×
190	oscResponse = false	×
191	case ']':	×
192	oscResponse = true	×
193	default:	×
194	return "", false, ErrStatusReport	×
195	}
196
197	for {	×
198	b, err := o.readNextByte()	×
199	if err != nil {	×
200	return "", false, err	×
201	}	×
202
203	response += string(b)	×
204		×
205	if oscResponse {	×
206	// OSC can be terminated by BEL (\a) or ST (ESC)	×
207	if b == BEL \|\| strings.HasSuffix(response, string(ESC)) {	×
208	return response, true, nil	×
209	}	×
210	} else {	×
211	// cursor position response is terminated by 'R'	×
212	if b == 'R' {	×
213	return response, false, nil	×
214	}	×
215	}
216
217	// both responses have less than 25 bytes, so if we read more, that's an error
218	if len(response) > 25 {	×
219	break	×
220	}
221	}
222
223	return "", false, ErrStatusReport	×
224	}
225
226	func (o Output) termStatusReport(sequence int) (string, error) {	×
227	// screen/tmux can't support OSC, because they can be connected to multiple	×
228	// terminals concurrently.	×
229	term := o.environ.Getenv("TERM")	×
230	if strings.HasPrefix(term, "screen") \|\| strings.HasPrefix(term, "tmux") {	×
231	return "", ErrStatusReport	×
232	}	×
233
234	tty := o.TTY()	×
235	if tty == nil {	×
236	return "", ErrStatusReport	×
237	}	×
238
239	if !o.unsafe {	×
240	fd := int(tty.Fd())	×
241	// if in background, we can't control the terminal	×
242	if !isForeground(fd) {	×
243	return "", ErrStatusReport	×
244	}	×
245
246	t, err := unix.IoctlGetTermios(fd, tcgetattr)	×
247	if err != nil {	×
248	return "", fmt.Errorf("%s: %s", ErrStatusReport, err)	×
249	}	×
250	defer unix.IoctlSetTermios(fd, tcsetattr, t) //nolint:errcheck	×
251		×
252	noecho := *t	×
253	noecho.Lflag = noecho.Lflag &^ unix.ECHO	×
254	noecho.Lflag = noecho.Lflag &^ unix.ICANON	×
255	if err := unix.IoctlSetTermios(fd, tcsetattr, &noecho); err != nil {	×
256	return "", fmt.Errorf("%s: %s", ErrStatusReport, err)	×
257	}	×
258	}
259
260	// first, send OSC query, which is ignored by terminal which do not support it
261	fmt.Fprintf(tty, OSC+"%d;?"+ST, sequence)	×
262		×
263	// then, query cursor position, should be supported by all terminals	×
264	fmt.Fprintf(tty, CSI+"6n")	×
265		×
266	// read the next response	×
267	res, isOSC, err := o.readNextResponse()	×
268	if err != nil {	×
269	return "", fmt.Errorf("%s: %s", ErrStatusReport, err)	×
270	}	×
271
272	// if this is not OSC response, then the terminal does not support it
273	if !isOSC {	×
274	return "", ErrStatusReport	×
275	}	×
276
277	// read the cursor query response next and discard the result
278	_, _, err = o.readNextResponse()	×
279	if err != nil {	×
280	return "", err	×
281	}	×
282
283	// fmt.Println("Rcvd", res[1:])
284	return res, nil	×
285	}
286
287	// EnableVirtualTerminalProcessing enables virtual terminal processing on
288	// Windows for w and returns a function that restores w to its previous state.
289	// On non-Windows platforms, or if w does not refer to a terminal, then it
290	// returns a non-nil no-op function and no error.
291	func EnableVirtualTerminalProcessing(w io.Writer) (func() error, error) {	1✔
292	return func() error { return nil }, nil	2✔
293	}

muesli / termenv / 4606800856

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous