File: test_event_handler_console_direct.py

package info (click to toggle)
ros2-colcon-core 0.20.0-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 1,156 kB
  • sloc: python: 10,333; makefile: 7
file content (60 lines) | stat: -rw-r--r-- 2,044 bytes parent folder | download | duplicates (3)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
# Copyright 2016-2018 Dirk Thomas
# Licensed under the Apache License, Version 2.0

from unittest.mock import patch

from colcon_core.event.output import StderrLine
from colcon_core.event.output import StdoutLine
from colcon_core.event_handler.console_direct import ConsoleDirectEventHandler
import pytest


def test_console_direct():
    with patch('sys.stdout') as stdout:
        extension = ConsoleDirectEventHandler()

        event = StdoutLine(b'bytes line')
        extension((event, None))
        assert stdout.buffer.write.call_count == 1
        event = StdoutLine('string line')
        extension((event, None))
        assert stdout.write.call_count == 1

        stdout.buffer.write.reset_mock()
        stdout.write.reset_mock()
        extension(('unknown', None))
        assert stdout.buffer.write.call_count == 0
        assert stdout.write.call_count == 0

    with patch('sys.stderr') as stderr:
        extension = ConsoleDirectEventHandler()

        event = StderrLine(b'bytes line')
        extension((event, None))
        assert stderr.buffer.write.call_count == 1
        event = StderrLine('string line')
        extension((event, None))
        assert stderr.write.call_count == 1

        stderr.buffer.write.reset_mock()
        stderr.write.reset_mock()
        extension(('unknown', None))
        assert stderr.buffer.write.call_count == 0
        assert stderr.write.call_count == 0

    with patch('sys.stdout') as stdout:
        stdout.buffer.write.side_effect = BrokenPipeError()
        stdout.write.side_effect = BrokenPipeError()

        extension = ConsoleDirectEventHandler()

        event = StdoutLine(b'bytes line')
        with pytest.raises(BrokenPipeError):
            extension((event, None))
        assert stdout.buffer.write.call_count == 1
        event = StdoutLine(b'bytes line')
        extension((event, None))
        assert stdout.buffer.write.call_count == 1
        event = StdoutLine('string line')
        extension((event, None))
        assert stdout.write.call_count == 0