FM / test

/opt/imh-python/lib/python3.9/site-packages/cherrypy/test/

/opt/imh-python/lib/python3.9/site-packages/cherrypy/test

Name	Size	Mode	Actions
static/	-	0755	rm
__pycache__/	-	0755	rm
benchmark.py	12563	0644	edit dl rm
checkerdemo.py	1861	0644	edit dl rm
fastcgi.conf	686	0644	edit dl rm
fcgi.conf	486	0644	edit dl rm
helper.py	16369	0644	edit dl rm
logtest.py	8151	0644	edit dl rm
modfastcgi.py	4652	0644	edit dl rm
modfcgid.py	4237	0644	edit dl rm
modpy.py	4978	0644	edit dl rm
modwsgi.py	4834	0644	edit dl rm
sessiondemo.py	5425	0644	edit dl rm
style.css	17	0644	edit dl rm
test.pem	2254	0644	edit dl rm
test_auth_basic.py	4499	0644	edit dl rm
test_auth_digest.py	4454	0644	edit dl rm
test_bus.py	9960	0644	edit dl rm
test_caching.py	14386	0644	edit dl rm
test_config.py	8836	0644	edit dl rm
test_config_server.py	4037	0644	edit dl rm
test_conn.py	30744	0644	edit dl rm
test_core.py	30408	0644	edit dl rm
test_dynamicobjectmapping.py	12331	0644	edit dl rm
test_encoding.py	17535	0644	edit dl rm
test_etags.py	3093	0644	edit dl rm
test_http.py	10939	0644	edit dl rm
test_httputil.py	2412	0644	edit dl rm
test_iterator.py	5724	0644	edit dl rm
test_json.py	2860	0644	edit dl rm
test_logging.py	8315	0644	edit dl rm
test_mime.py	4538	0644	edit dl rm
test_misc_tools.py	7094	0644	edit dl rm
test_native.py	971	0644	edit dl rm
test_objectmapping.py	14504	0644	edit dl rm
test_params.py	1862	0644	edit dl rm
test_plugins.py	334	0644	edit dl rm
test_proxy.py	5630	0644	edit dl rm
test_refleaks.py	1555	0644	edit dl rm
test_request_obj.py	37097	0644	edit dl rm
test_routes.py	2583	0644	edit dl rm
test_session.py	17949	0644	edit dl rm
test_sessionauthenticate.py	2013	0644	edit dl rm
test_states.py	16703	0644	edit dl rm
test_static.py	16702	0644	edit dl rm
test_tools.py	17851	0644	edit dl rm
test_tutorials.py	6964	0644	edit dl rm
test_virtualhost.py	4021	0644	edit dl rm
test_wsgiapps.py	3997	0644	edit dl rm
test_wsgi_ns.py	2812	0644	edit dl rm
test_wsgi_unix_socket.py	2228	0644	edit dl rm
test_wsgi_vhost.py	1034	0644	edit dl rm
test_xmlrpc.py	4584	0644	edit dl rm
webtest.py	262	0644	edit dl rm
_test_decorators.py	951	0644	edit dl rm
_test_states_demo.py	1876	0644	edit dl rm
__init__.py	396	0644	edit dl rm

Edit: /opt/imh-python/lib/python3.9/site-packages/cherrypy/test/test_bus.py (9960B)

"""Publish-subscribe bus tests."""
# pylint: disable=redefined-outer-name

import os
import sys
import threading
import time
import unittest.mock

import pytest

from cherrypy.process import wspbus

CI_ON_MACOS = bool(os.getenv('CI')) and sys.platform == 'darwin'
msg = 'Listener %d on channel %s: %s.'  # pylint: disable=invalid-name

@pytest.fixture
def bus():
    """Return a wspbus instance."""
    return wspbus.Bus()

@pytest.fixture
def log_tracker(bus):
    """Return an instance of bus log tracker."""
    class LogTracker:  # pylint: disable=too-few-public-methods
        """Bus log tracker."""

log_entries = []

def __init__(self, bus):
            def logit(msg, level):  # pylint: disable=unused-argument
                self.log_entries.append(msg)
            bus.subscribe('log', logit)

return LogTracker(bus)

@pytest.fixture
def listener():
    """Return an instance of bus response tracker."""
    class Listner:  # pylint: disable=too-few-public-methods
        """Bus handler return value tracker."""

responses = []

def get_listener(self, channel, index):
            """Return an argument tracking listener."""
            def listener(arg=None):
                self.responses.append(msg % (index, channel, arg))
            return listener

return Listner()

def test_builtin_channels(bus, listener):
    """Test that built-in channels trigger corresponding listeners."""
    expected = []

for channel in bus.listeners:
        for index, priority in enumerate([100, 50, 0, 51]):
            bus.subscribe(
                channel,
                listener.get_listener(channel, index),
                priority,
            )

for channel in bus.listeners:
        bus.publish(channel)
        expected.extend([msg % (i, channel, None) for i in (2, 1, 3, 0)])
        bus.publish(channel, arg=79347)
        expected.extend([msg % (i, channel, 79347) for i in (2, 1, 3, 0)])

assert listener.responses == expected

def test_custom_channels(bus, listener):
    """Test that custom pub-sub channels work as built-in ones."""
    expected = []

custom_listeners = ('hugh', 'louis', 'dewey')
    for channel in custom_listeners:
        for index, priority in enumerate([None, 10, 60, 40]):
            bus.subscribe(
                channel,
                listener.get_listener(channel, index),
                priority,
            )

for channel in custom_listeners:
        bus.publish(channel, 'ah so')
        expected.extend(msg % (i, channel, 'ah so') for i in (1, 3, 0, 2))
        bus.publish(channel)
        expected.extend(msg % (i, channel, None) for i in (1, 3, 0, 2))

assert listener.responses == expected

def test_listener_errors(bus, listener):
    """Test that unhandled exceptions raise channel failures."""
    expected = []
    channels = [c for c in bus.listeners if c != 'log']

for channel in channels:
        bus.subscribe(channel, listener.get_listener(channel, 1))
        # This will break since the lambda takes no args.
        bus.subscribe(channel, lambda: None, priority=20)

for channel in channels:
        with pytest.raises(wspbus.ChannelFailures):
            bus.publish(channel, 123)
        expected.append(msg % (1, channel, 123))

assert listener.responses == expected

def test_start(bus, listener, log_tracker):
    """Test that bus start sequence calls all listeners."""
    num = 3
    for index in range(num):
        bus.subscribe('start', listener.get_listener('start', index))

bus.start()
    try:
        # The start method MUST call all 'start' listeners.
        assert (
            set(listener.responses) ==
            set(msg % (i, 'start', None) for i in range(num)))
        # The start method MUST move the state to STARTED
        # (or EXITING, if errors occur)
        assert bus.state == bus.states.STARTED
        # The start method MUST log its states.
        assert log_tracker.log_entries == ['Bus STARTING', 'Bus STARTED']
    finally:
        # Exit so the atexit handler doesn't complain.
        bus.exit()

def test_stop(bus, listener, log_tracker):
    """Test that bus stop sequence calls all listeners."""
    num = 3

for index in range(num):
        bus.subscribe('stop', listener.get_listener('stop', index))

bus.stop()

# The stop method MUST call all 'stop' listeners.
    assert (set(listener.responses) ==
            set(msg % (i, 'stop', None) for i in range(num)))

# The stop method MUST move the state to STOPPED
    assert bus.state == bus.states.STOPPED

# The stop method MUST log its states.
    assert log_tracker.log_entries == ['Bus STOPPING', 'Bus STOPPED']

def test_graceful(bus, listener, log_tracker):
    """Test that bus graceful state triggers all listeners."""
    num = 3

for index in range(num):
        bus.subscribe('graceful', listener.get_listener('graceful', index))

bus.graceful()

# The graceful method MUST call all 'graceful' listeners.
    assert (
        set(listener.responses) ==
        set(msg % (i, 'graceful', None) for i in range(num)))

# The graceful method MUST log its states.
    assert log_tracker.log_entries == ['Bus graceful']

def test_exit(bus, listener, log_tracker):
    """Test that bus exit sequence is correct."""
    num = 3

for index in range(num):
        bus.subscribe('stop', listener.get_listener('stop', index))
        bus.subscribe('exit', listener.get_listener('exit', index))

bus.exit()

# The exit method MUST call all 'stop' listeners,
    # and then all 'exit' listeners.
    assert (set(listener.responses) ==
            set([msg % (i, 'stop', None) for i in range(num)] +
                [msg % (i, 'exit', None) for i in range(num)]))

# The exit method MUST move the state to EXITING
    assert bus.state == bus.states.EXITING

# The exit method MUST log its states.
    assert (log_tracker.log_entries ==
            ['Bus STOPPING', 'Bus STOPPED', 'Bus EXITING', 'Bus EXITED'])

def test_wait(bus):
    """Test that bus wait awaits for states."""
    def f(method):  # pylint: disable=invalid-name
        time.sleep(0.2)
        getattr(bus, method)()

flow = [
        ('start', [bus.states.STARTED]),
        ('stop', [bus.states.STOPPED]),
        ('start', [bus.states.STARTING, bus.states.STARTED]),
        ('exit', [bus.states.EXITING]),
    ]

for method, states in flow:
        threading.Thread(target=f, args=(method,)).start()
        bus.wait(states)

# The wait method MUST wait for the given state(s).
        assert bus.state in states, 'State %r not in %r' % (bus.state, states)

@pytest.mark.xfail(CI_ON_MACOS, reason='continuous integration on macOS fails')
def test_wait_publishes_periodically(bus):
    """Test that wait publishes each tick."""
    callback = unittest.mock.MagicMock()
    bus.subscribe('main', callback)

def set_start():
        time.sleep(0.05)
        bus.start()
    threading.Thread(target=set_start).start()
    bus.wait(bus.states.STARTED, interval=0.01, channel='main')
    assert callback.call_count > 3

def test_block(bus, log_tracker):
    """Test that bus block waits for exiting."""
    def f():  # pylint: disable=invalid-name
        time.sleep(0.2)
        bus.exit()

def g():  # pylint: disable=invalid-name
        time.sleep(0.4)

threading.Thread(target=f).start()
    threading.Thread(target=g).start()
    threads = [t for t in threading.enumerate() if not t.daemon]
    assert len(threads) == 3

bus.block()

# The block method MUST wait for the EXITING state.
    assert bus.state == bus.states.EXITING

# 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]
    assert len(threads) == 1

# The last message will mention an indeterminable thread name; ignore
    # it
    expected_bus_messages = [
        'Bus STOPPING',
        'Bus STOPPED',
        'Bus EXITING',
        'Bus EXITED',
        'Waiting for child threads to terminate...',
    ]
    bus_msg_num = len(expected_bus_messages)

# If the last message mentions an indeterminable thread name then ignore it
    assert log_tracker.log_entries[:bus_msg_num] == expected_bus_messages
    assert len(log_tracker.log_entries[bus_msg_num:]) <= 1, (
        'No more than one extra log line with the thread name expected'
    )

def test_start_with_callback(bus):
    """Test that callback fires on bus start."""
    try:
        events = []

def f(*args, **kwargs):  # pylint: disable=invalid-name
            events.append(('f', args, kwargs))

def g():  # pylint: disable=invalid-name
            events.append('g')
        bus.subscribe('start', g)
        bus.start_with_callback(f, (1, 3, 5), {'foo': 'bar'})

# Give wait() time to run f()
        time.sleep(0.2)

# The callback method MUST wait for the STARTED state.
        assert bus.state == bus.states.STARTED

# The callback method MUST run after all start methods.
        assert events == ['g', ('f', (1, 3, 5), {'foo': 'bar'})]
    finally:
        bus.exit()

def test_log(bus, log_tracker):
    """Test that bus messages and errors are logged."""
    assert log_tracker.log_entries == []

# Try a normal message.
    expected = []
    for msg_ in ["O mah darlin'"] * 3 + ['Clementiiiiiiiine']:
        bus.log(msg_)
        expected.append(msg_)
        assert log_tracker.log_entries == expected

# Try an error message
    try:
        foo
    except NameError:
        bus.log('You are lost and gone forever', traceback=True)
        lastmsg = log_tracker.log_entries[-1]
        assert 'Traceback' in lastmsg and 'NameError' in lastmsg, (
            'Last log message %r did not contain '
            'the expected traceback.' % lastmsg
        )
    else:
        pytest.fail('NameError was not raised as expected.')

cmd: