Skip to content

Commit

Permalink
Merge branch 'maintenance/assert-in-test'
Browse files Browse the repository at this point in the history
  • Loading branch information
@webknjaz
webknjaz committed Feb 11, 2021
2 parents f73e0a5 + 5908162 commit f30ed35
Show file tree
Hide file tree
Showing 7 changed files with 111 additions and 128 deletions.
10 changes: 0 additions & 10 deletions magicbus/test/__init__.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -12,16 +12,6 @@
from magicbus.plugins import SimplePlugin


def assertEqual(x, y, msg=None):
if not x == y:
raise AssertionError(msg or '%r != %r' % (x, y))


def assertNotEqual(x, y, msg=None):
if x == y:
raise AssertionError(msg or '%r == %r' % (x, y))


class Process(object):

def __init__(self, args):
Expand Down
126 changes: 62 additions & 64 deletions magicbus/test/test_bus.py
View file
Original file line number Diff line number Diff line change
@@ -1,73 +1,71 @@
import unittest

from magicbus.base import Graph


class GraphTests(unittest.TestCase):
# EXIT_ERROR
# |
# V
# START -----------> EXIT ------> EXITED
# / | A / A
# V | \ / |
# START_ERROR RUN | IDLE STOP_ERROR
# | \ | A
# | V V /
# +----------------> STOP
TRANSITIONS = {
'START': ('RUN', 'STOP', 'EXIT'),
'RUN': 'STOP',
'START_ERROR': 'STOP',
'STOP': 'IDLE',
'IDLE': ('START', 'EXIT'),
'STOP_ERROR': 'EXIT',
'EXIT': 'EXITED',
'EXIT_ERROR': 'EXITED',
}


NEXT = {
('EXIT', 'EXITED'): 'EXITED',
('EXIT_ERROR', 'EXITED'): 'EXITED',
('IDLE', 'EXITED'): 'EXIT',
('IDLE', 'EXIT'): 'EXIT',
('IDLE', 'RUN'): 'START',
('IDLE', 'START'): 'START',
('IDLE', 'STOP'): 'START',
('RUN', 'EXITED'): 'STOP',
('RUN', 'EXIT'): 'STOP',
('RUN', 'IDLE'): 'STOP',
('RUN', 'START'): 'STOP',
('RUN', 'STOP'): 'STOP',
('START', 'EXITED'): 'EXIT',
('START', 'EXIT'): 'EXIT',
('START', 'IDLE'): 'STOP',
('START', 'RUN'): 'RUN',
('START', 'STOP'): 'STOP',
('START_ERROR', 'EXITED'): 'STOP',
('START_ERROR', 'EXIT'): 'STOP',
('START_ERROR', 'IDLE'): 'STOP',
('START_ERROR', 'RUN'): 'STOP',
('START_ERROR', 'START'): 'STOP',
('START_ERROR', 'STOP'): 'STOP',
('STOP', 'EXITED'): 'IDLE',
('STOP', 'EXIT'): 'IDLE',
('STOP', 'IDLE'): 'IDLE',
('STOP', 'RUN'): 'IDLE',
('STOP', 'START'): 'IDLE',
('STOP_ERROR', 'EXITED'): 'EXIT',
('STOP_ERROR', 'EXIT'): 'EXIT',
}

maxDiff = None

# EXIT_ERROR
# |
# V
# START -----------> EXIT ------> EXITED
# / | A / A
# V | \ / |
# START_ERROR RUN | IDLE STOP_ERROR
# | \ | A
# | V V /
# +----------------> STOP
transitions = {
'START': ('RUN', 'STOP', 'EXIT'),
'RUN': 'STOP',
'START_ERROR': 'STOP',
'STOP': 'IDLE',
'IDLE': ('START', 'EXIT'),
'STOP_ERROR': 'EXIT',
'EXIT': 'EXITED',
'EXIT_ERROR': 'EXITED',
}
STATES = set(['START', 'RUN', 'STOP', 'IDLE', 'EXIT', 'EXITED',
'START_ERROR', 'STOP_ERROR', 'EXIT_ERROR'])

next = {
('EXIT', 'EXITED'): 'EXITED',
('EXIT_ERROR', 'EXITED'): 'EXITED',
('IDLE', 'EXITED'): 'EXIT',
('IDLE', 'EXIT'): 'EXIT',
('IDLE', 'RUN'): 'START',
('IDLE', 'START'): 'START',
('IDLE', 'STOP'): 'START',
('RUN', 'EXITED'): 'STOP',
('RUN', 'EXIT'): 'STOP',
('RUN', 'IDLE'): 'STOP',
('RUN', 'START'): 'STOP',
('RUN', 'STOP'): 'STOP',
('START', 'EXITED'): 'EXIT',
('START', 'EXIT'): 'EXIT',
('START', 'IDLE'): 'STOP',
('START', 'RUN'): 'RUN',
('START', 'STOP'): 'STOP',
('START_ERROR', 'EXITED'): 'STOP',
('START_ERROR', 'EXIT'): 'STOP',
('START_ERROR', 'IDLE'): 'STOP',
('START_ERROR', 'RUN'): 'STOP',
('START_ERROR', 'START'): 'STOP',
('START_ERROR', 'STOP'): 'STOP',
('STOP', 'EXITED'): 'IDLE',
('STOP', 'EXIT'): 'IDLE',
('STOP', 'IDLE'): 'IDLE',
('STOP', 'RUN'): 'IDLE',
('STOP', 'START'): 'IDLE',
('STOP_ERROR', 'EXITED'): 'EXIT',
('STOP_ERROR', 'EXIT'): 'EXIT',
}

states = set(['START', 'RUN', 'STOP', 'IDLE', 'EXIT', 'EXITED',
'START_ERROR', 'STOP_ERROR', 'EXIT_ERROR'])
def test_shortest_path():
g = Graph.from_edges(TRANSITIONS)
assert g == NEXT

def test_shortest_path(self):
g = Graph.from_edges(self.transitions)
self.assertEqual(g, self.next)

def test_states(self):
g = Graph(self.next)
self.assertEqual(g.states, self.states)
def test_states():
g = Graph(NEXT)
assert g.states == STATES
7 changes: 3 additions & 4 deletions magicbus/test/test_opsys.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@

from magicbus import bus
from magicbus.plugins import opsys
from magicbus.test import assertEqual, Process, WebAdapter, WebService
from magicbus.test import Process, WebAdapter, WebService
from magicbus.test import WebHandler

# from magicbus.plugins import loggers
Expand Down Expand Up @@ -46,10 +46,9 @@ def test_daemonize(self):
try:
# Just get the pid of the daemonization process.
resp = service.do_GET('/pid')
assertEqual(resp.status, 200)
assert resp.status == 200
page_pid = int(resp.read())
assertEqual(ntob(str(page_pid)),
open(pidfile.pidfile, 'rb').read())
assert ntob(str(page_pid)) == open(pidfile.pidfile, 'rb').read()
finally:
# Shut down the spawned process
service.do_GET('/exit')
Expand Down
67 changes: 32 additions & 35 deletions magicbus/test/test_processbus.py
View file
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
import threading
import time
import unittest

import pytest

from magicbus.base import Bus, ChannelFailures
from magicbus.process import ProcessBus
Expand All @@ -9,7 +10,7 @@
msg = 'Listener %d on channel %s: %s.'


class PublishSubscribeTests(unittest.TestCase):
class TestPublishSubscribe(object):

def get_listener(self, channel, index):
def listener(*args):
Expand All @@ -35,7 +36,7 @@ def test_builtin_channels(self):
b.publish(channel)
expected.extend([msg % (i, channel, ()) for i in (2, 1, 3, 0)])

self.assertEqual(self.responses, expected)
assert self.responses == expected
finally:
# Exit so the atexit handler doesn't complain.
b.transition('EXITED')
Expand All @@ -58,12 +59,13 @@ def test_custom_channels(self):
b.publish(channel)
expected.extend([msg % (i, channel, ()) for i in (1, 3, 0, 2)])

self.assertEqual(self.responses, expected)
assert self.responses == expected

def test_listener_errors(self):
b = Bus()

self.responses, expected = [], []
# FIXME: should this be bigger than 0 non-log channels?
channels = [c for c in b.listeners if c != 'log']

for channel in channels:
Expand All @@ -72,15 +74,14 @@ def test_listener_errors(self):
b.subscribe(channel, lambda: None, priority=20)

for channel in channels:
self.assertRaises(ChannelFailures, b.publish, channel, 123)
with pytest.raises(ChannelFailures): # FIXME: add `match=`
b.publish(channel, 123)
expected.append(msg % (1, channel, (123,)))

self.assertEqual(self.responses, expected)

assert self.responses == expected

class BusMethodTests(unittest.TestCase):

maxDiff = None
class TestBusMethod(object):

def log(self, bus, level=10):
self._log_entries = []
Expand All @@ -92,7 +93,7 @@ def logit(msg_, level):
bus.subscribe('log', logit)

def assertLog(self, entries):
self.assertEqual(self._log_entries, entries)
assert self._log_entries == entries

def get_listener(self, channel, index):
def listener(arg=None):
Expand All @@ -111,13 +112,13 @@ def test_idle_to_run(self):
b.transition('RUN')
try:
# The start method MUST call all 'start' listeners.
self.assertEqual(
set(self.responses),
assert (
set(self.responses) ==
set([msg % (i, 'START', None) for i in range(num)])
)
# The transition method MUST move the state to RUN
# (or START_ERROR, if errors occur)
self.assertEqual(b.state, 'RUN')
assert b.state == 'RUN'

# The start method MUST log its states.
self.assertLog([
Expand All @@ -144,12 +145,12 @@ def test_run_to_idle(self):
b.transition('IDLE')

# The idle transition MUST call all 'stop' listeners.
self.assertEqual(
set(self.responses),
assert (
set(self.responses) ==
set(msg % (i, 'STOP', None) for i in range(num))
)
# The idle method MUST move the state to IDLE
self.assertEqual(b.state, 'IDLE')
assert b.state == 'IDLE'
# The idle method MUST log its states.
self.assertLog([
'Bus state: STOP',
Expand All @@ -173,13 +174,13 @@ def test_idle_to_exit(self):

# The bus MUST call all 'EXIT' listeners,
# and then all 'EXITED' listeners.
self.assertEqual(
set(self.responses),
assert (
set(self.responses) ==
set([msg % (i, 'EXIT', None) for i in range(num)] +
[msg % (i, 'EXITED', None) for i in range(num)])
)
# The bus MUST move the state to EXITED
self.assertEqual(b.state, 'EXITED')
assert b.state == 'EXITED'

# The bus MUST log its states.
self.assertLog([
Expand Down Expand Up @@ -209,7 +210,7 @@ def f(desired_state):

# The wait method MUST wait for the given state(s).
if b.state not in states_to_wait_for:
self.fail('State %r not in %r' % (b.state, states_to_wait_for))
pytest.fail('State %r not in %r' % (b.state, states_to_wait_for))

def test_block(self):
b = ProcessBus()
Expand All @@ -231,23 +232,23 @@ def main_listener():
f_thread.start()
threading.Thread(target=g, name='g').start()
threads = [t for t in threading.enumerate() if not t.daemon]
self.assertEqual(len(threads), 3)
assert len(threads) == 3

b.block()
f_thread.join()

# The block method MUST wait for the EXITED state.
self.assertEqual(b.state, 'EXITED')
assert b.state == 'EXITED'
# The block method MUST wait for ALL non-main, non-daemon threads to
# finish.
threads = [t for t in threading.enumerate() if not t.daemon]
self.assertEqual(len(threads), 1)
assert len(threads) == 1
# The last message will mention an indeterminable thread name; ignore
# it
self.assertEqual(
assert (
[entry for entry in self._log_entries
if not entry.startswith('Publishing')
and not entry.startswith('Waiting')],
and not entry.startswith('Waiting')] ==
[
'Bus state: ENTER',
'Bus state: IDLE',
Expand All @@ -258,9 +259,9 @@ def main_listener():

# While the bus was blocked, it should have published periodically
# to the "main" channel.
self.assertGreater(len(main_calls), 0)
assert len(main_calls) > 0

@unittest.skip("Fails intermittently; https://tinyurl.com/ybwwu4gz")
@pytest.mark.skip("Fails intermittently; https://tinyurl.com/ybwwu4gz")
def test_start_with_callback(self):
b = ProcessBus()
self.log(b)
Expand All @@ -279,9 +280,9 @@ def g():
time.sleep(0.2)

# The callback method MUST wait for the STARTED state.
self.assertEqual(b.state, 'RUN')
assert b.state == 'RUN'
# The callback method MUST run after all start methods.
self.assertEqual(events, ['g', ('f', (1, 3, 5), {'foo': 'bar'})])
assert events == ['g', ('f', (1, 3, 5), {'foo': 'bar'})]
finally:
b.transition('EXITED')

Expand All @@ -304,11 +305,7 @@ def test_log(self):
b.log('You are lost and gone forever', traceback=True)
lastmsg = self._log_entries[-1]
if 'Traceback' not in lastmsg or 'NameError' not in lastmsg:
self.fail('Last log message %r did not contain '
pytest.fail('Last log message %r did not contain '
'the expected traceback.' % lastmsg)
else:
self.fail('NameError was not raised as expected.')


if __name__ == '__main__':
unittest.main()
pytest.fail('NameError was not raised as expected.')
6 changes: 3 additions & 3 deletions magicbus/test/test_servers.py
View file
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
from magicbus.process import ProcessBus
from magicbus.plugins import servers
from magicbus.test import assertEqual, WebService, WebHandler
from magicbus.test import WebService, WebHandler

# from magicbus.plugins import loggers
# loggers.StdoutLogger(bus).subscribe()
Expand Down Expand Up @@ -32,6 +32,6 @@ def test_keyboard_interrupt(self):
# Raise a keyboard interrupt in the HTTP server's main thread.
bus.transition('RUN')
resp = service.do_GET('/ctrlc')
assertEqual(resp.status, 200)
assert resp.status == 200
bus.block()
assertEqual(bus.state, 'EXITED')
assert bus.state == 'EXITED'

0 comments on commit f30ed35

Please sign in to comment.