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
|
import pytest
import subprocess
def test_garbage(poolcounter, clients):
"""Bad commands result in error"""
client = clients.get(1)
client.send('GARBAGE')
assert client.receive() == 'ERROR BAD_COMMAND'
def test_locking_just_lockname(poolcounter, clients, lock_type):
"""Locking with just lock name results in error"""
client = clients.get(1)
client.send('%s l' % lock_type)
assert client.receive() == 'ERROR BAD_SYNTAX'
def test_locking_no_workers(poolcounter, clients, lock_type):
"""Locking with no workers results in error"""
client = clients.get(1)
client.send('%s l 0 1 1' % lock_type)
assert client.receive() == 'ERROR BAD_SYNTAX'
def test_locking_no_queue(poolcounter, clients, lock_type):
"""Locking with no queue results in error"""
client = clients.get(1)
client.send('%s l 1 0 1' % lock_type)
assert client.receive() == 'ERROR BAD_SYNTAX'
def test_locking_non_integers(poolcounter, clients, lock_type):
"""Locking with no queue results in error"""
client = clients.get(1)
client.send('%s l GARBAGE 0 1' % lock_type)
assert client.receive() == 'ERROR BAD_SYNTAX'
@pytest.mark.xfail # flaky, sometimes timeouts in final recv
def test_locking_while_waiting(poolcounter, clients, lock_type):
"""Locking while waiting is an error"""
client1, client2 = clients.get(2)
client2.send('%s l 1 10 10' % lock_type)
assert client2.receive() == 'LOCKED'
client1.send('%s l 1 10 10' % lock_type)
client1.send('%s l 1 10 10' % lock_type)
assert client1.receive() == 'ERROR WAIT_FOR_RESPONSE'
def test_invalid_listen(poolcounter_path):
with pytest.raises(subprocess.CalledProcessError) as e:
subprocess.check_call([poolcounter_path, '-l', 'bad'])
assert e.value.returncode == 1
|
