Ignore:
Timestamp:
01/11/13 17:15:19 (15 months ago)
Author:
Ales Erjavec <ales.erjavec@…>
Branch:
default
Message:

Added colored/formated output for the stdout/err output, output thread safety.

File:
1 edited

Legend:

Unmodified
Added
Removed
  • Orange/OrangeCanvas/application/tests/test_outputview.py

    r11167 r11257  
     1import multiprocessing.pool 
     2 
    13from datetime import datetime 
     4from threading import current_thread 
    25 
     6from PyQt4.QtCore import Qt, QThread 
    37from ...gui.test import QAppTestCase 
    48 
    5 from ..outputview import OutputText 
     9from ..outputview import OutputView, TextStream 
    610 
    711 
    812class TestOutputView(QAppTestCase): 
    913    def test_outputview(self): 
    10         output = OutputText() 
     14        output = OutputView() 
    1115        output.show() 
    1216 
     
    3943 
    4044        self.app.exec_() 
     45 
     46    def test_formated(self): 
     47        output = OutputView() 
     48        output.show() 
     49 
     50        output.write("A sword day, ") 
     51        with output.formated(color=Qt.red) as f: 
     52            f.write("a red day...\n") 
     53 
     54            with f.formated(color=Qt.green) as f: 
     55                f.write("Actually sir, orcs bleed green.\n") 
     56 
     57        bold = output.formated(weight=100, underline=True) 
     58        bold.write("Shutup") 
     59 
     60        self.app.exec_() 
     61 
     62    def test_threadsafe(self): 
     63        output = OutputView() 
     64        output.resize(500, 300) 
     65        output.show() 
     66 
     67        blue_formater = output.formated(color=Qt.blue) 
     68        red_formater = output.formated(color=Qt.red) 
     69 
     70        correct = [] 
     71 
     72        def check_thread(*args): 
     73            correct.append(QThread.currentThread() == self.app.thread()) 
     74 
     75        blue = TextStream() 
     76        blue.stream.connect(blue_formater.write) 
     77        blue.stream.connect(check_thread) 
     78 
     79        red = TextStream() 
     80        red.stream.connect(red_formater.write) 
     81        red.stream.connect(check_thread) 
     82 
     83        def printer(i): 
     84            if i % 12 == 0: 
     85                fizzbuz = "fizzbuz" 
     86            elif i % 4 == 0: 
     87                fizzbuz = "buz" 
     88            elif i % 3 == 0: 
     89                fizzbuz = "fizz" 
     90            else: 
     91                fizzbuz = str(i) 
     92 
     93            if i % 2: 
     94                writer = blue 
     95            else: 
     96                writer = red 
     97 
     98            writer.write("Greetings from thread {0}. " 
     99                         "This is {1}\n".format(current_thread().name, 
     100                                                fizzbuz)) 
     101 
     102        pool = multiprocessing.pool.ThreadPool(100) 
     103        res = pool.map_async(printer, range(10000)) 
     104 
     105        self.app.exec_() 
     106 
     107        res.wait() 
     108 
     109        self.assertTrue(all(correct)) 
     110        self.assertTrue(len(correct) == 10000) 
Note: See TracChangeset for help on using the changeset viewer.