os.Connection.read(): Support ReadOnlyPrintable

Add flag ReadOnlyPrintable and support it in read(). This omits all
characters not in string.printable from read()'s result.

This comes in handy if systemd spits out bouncing start-job-running
asterisks which find their way back to the reader but close the shell
on him.

Signed-off-by: Jan Lindemann <jan@janware.com>
This commit is contained in:
Jan Lindemann 2023-02-23 14:47:52 +01:00
commit 368d81cb3f

View file

@ -4,6 +4,7 @@ from abc import ABC, abstractmethod
from enum import Enum, IntFlag from enum import Enum, IntFlag
import asyncio import asyncio
import types import types
import string
from jwutils.log import * from jwutils.log import *
from .TestPhases import TestPhases from .TestPhases import TestPhases
@ -21,6 +22,7 @@ class Connection(ABC): # export
FailOnTimeout = 0x01 FailOnTimeout = 0x01
ReadStripNewline = 0x02 ReadStripNewline = 0x02
ReadDecodeToString = 0x04 ReadDecodeToString = 0x04
ReadOnlyPrintable = 0x08
class Info: class Info:
@ -124,10 +126,12 @@ class Connection(ABC): # export
r = r.decode('utf-8', errors='replace') r = r.decode('utf-8', errors='replace')
except Exception as e: except Exception as e:
slog(WARNING, 'Failed to decode string, returning undecoded ({}): "{}"'.format(e, r)) slog(WARNING, 'Failed to decode string, returning undecoded ({}): "{}"'.format(e, r))
return str(r) r = str(r)
if flags & self.Flags.ReadStripNewline: if flags & self.Flags.ReadStripNewline:
if len(r) and r[-1] == '\n': if len(r) and r[-1] == '\n':
r = r[0:-1] r = r[0:-1]
if flags & self.Flags.ReadOnlyPrintable:
r = ''.join(filter(lambda x: x in string.printable, r))
return r return r
async def readline(self, timeout=None): async def readline(self, timeout=None):