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
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
|
# -*- mode:python; coding:utf-8 -*-
import os
import time
import multiprocessing as mp
import socket
from unittest import TestCase
from pyDoubles.framework import empty_spy, empty_stub, \
when, assert_that_was_called
import go2
def foo_filter(value):
return value == 'foo'
# providers -> queue.put -> QueueReactor -> PathStorage.add -> menu.add_entry
# PathStorage.flush -> menu.add_entry
# key -> Reactor -> -> menu.input_handler
def pipe():
a, b = socket.socketpair()
read_end = a.makefile()
write_end = b.makefile('w', 0)
return read_end, write_end
class Test_Search(TestCase):
def setUp(self):
go2.config = go2.get_config()
def create_scenario(self, menu_size=26):
self.queue = mp.Queue()
self.read_end, self.write_end = pipe()
self.out = empty_spy()
self.reactor = go2.QueueReactor()
self.menu = go2.Menu(self.reactor, self.out, max_size=menu_size)
self.input_handler = go2.UserChoiceHandler(self.menu)
self.path_buffer = go2.PathBuffer(self.menu)
self.reactor.queue_add_watch(
self.queue,
go2.QueueExtractor(self.path_buffer.add))
def register_input(self):
self.reactor.io_add_watch(self.read_end, self.input_handler)
def put(self, path):
self.queue.put(go2.PathItem(path))
def test_choice_first_entry_with_previous_ignored(self):
# given
self.create_scenario()
self.register_input()
self.path_buffer.add_filter(foo_filter)
# when
self.put('foo')
self.put('bar')
self.write_end.write('a')
self.reactor.run()
# then
self.assertEquals(self.path_buffer.groups[0], ['bar'])
self.assertEquals(self.menu.entries, ['bar'])
assert_that_was_called(self.out.write).with_args('a: bar')
self.assertEquals(self.menu.target, 'bar')
def test_full_menu(self):
# given
self.create_scenario(menu_size=1)
self.register_input()
# when
self.put('foo')
self.put('bar')
self.reactor.process_pending()
# then
assert_that_was_called(self.out.write).with_args('a: foo')
def atest_list_mode(self):
def raw_formatter(index, path, suffix):
return path + '\n'
paths = ['foo', 'bar']
# given
pool = empty_stub()
when(pool.has_tasks).then_return(False)
self.create_scenario()
self.reactor.timeout_add(250, EndChecker(
pool, self.path_buffer, self.menu))
self.menu.formatter = raw_formatter
# when
for i in paths:
self.put(i)
self.reactor.run()
# then
for i in paths:
assert_that_was_called(self.out.write).with_args(i + '\n')
def test_unicode(self):
os.path.exists('ñandáéíóú')
# given: un pool con tareas
# when: el pool queda ocioso
# then: el menu espera una elección
# given: menu con una sola entrada
# when: todas las tareas han terminado
# then: elegir la única entrada sin preguntar
# given: un reactor
# when: el reactor acaba
# then: enviar las rutas alternativas al menú
# given: un reactor y un menú
# when: el menú se llena
# then: el menu espera una elección
# condiciones de fin
# - ESC o CTRL-C - CancelException
# - Ninguna coincidencia
# - Solo una coincidencia
# - Elegir entre varias coincidencias
# - Menu lleno, elegir
|
