Method expect

locust/test/subprocess_utils.py:121–141

(self, to_expect, *, stream="stderr")

    # Check output logs from last found (stateful)
    def expect(self, to_expect, *, stream="stderr"):
        __tracebackhide__ = True

        if stream == "stdout":
            buffer = self.stdout_output
            cursor = self._stdout_cursor
        else:
            buffer = self.stderr_output
            cursor = self._stderr_cursor

        start_time = time.time()
        while time.time() - start_time < self.expect_timeout:
            new_lines = buffer[cursor:]
            for idx, line in enumerate(new_lines):
                if to_expect in line:
                    cursor += idx + 1
                    return
            time.sleep(0.05)

        output = "\n".join(buffer[-5:])
        self.on_fail(f"Timed out waiting for '{to_expect}' after {self.expect_timeout} seconds. Got\n{output}")

    # Check all output logs (stateless)
    def expect_any(self, to_expect, *, stream="stderr"):