MCPcopy Create free account
hub / github.com/ceph/ceph / LogStream

Class LogStream

qa/tasks/cephfs_test_runner.py:47–60  ·  view source on GitHub ↗

Source from the content-addressed store, hash-verified

45
46
47class LogStream(object):
48 def __init__(self):
49 self.buffer = ""
50
51 def write(self, data):
52 self.buffer += data
53 if "\n" in self.buffer:
54 lines = self.buffer.split("\n")
55 for line in lines[:-1]:
56 log.info(line)
57 self.buffer = lines[-1]
58
59 def flush(self):
60 pass
61
62
63class InteractiveFailureResult(unittest.TextTestResult):

Callers 1

taskFunction · 0.70

Calls

no outgoing calls

Tested by 1

taskFunction · 0.56