File size: 3,293 Bytes
079c32c |
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 |
import pytest
from .test_base import _HTTPErrorGenerator
from ...exception.slave import SlaveErrorCode, \
get_slave_exception_class_by_error_code, get_slave_exception_by_error, SlaveSystemShuttingDown, \
SlaveSlaveConnectionRefused, SlaveSlaveDisconnectionRefused, SlaveSlaveNotConnected, SlaveSlaveAlreadyConnected, \
SlaveTaskRefused, SlaveMasterTokenInvalid, SlaveMasterTokenNotFound, SlaveSelfTokenNotFound, \
SlaveTaskAlreadyExist, SlaveSelfTokenInvalid, SlaveChannelNotFound, SlaveChannelInvalid, SlaveSuccess
@pytest.mark.unittest
class TestInteractionExceptionSlave(_HTTPErrorGenerator):
def test_error_code(self):
assert len(SlaveErrorCode.__members__) == 14
assert SlaveErrorCode.SUCCESS == 0
# noinspection DuplicatedCode
def test_exception_class(self):
assert get_slave_exception_class_by_error_code(SlaveErrorCode.SUCCESS) == SlaveSuccess
assert get_slave_exception_class_by_error_code(SlaveErrorCode.SYSTEM_SHUTTING_DOWN) == SlaveSystemShuttingDown
assert get_slave_exception_class_by_error_code(SlaveErrorCode.CHANNEL_NOT_FOUND) == SlaveChannelNotFound
assert get_slave_exception_class_by_error_code(SlaveErrorCode.CHANNEL_INVALID) == SlaveChannelInvalid
assert get_slave_exception_class_by_error_code(
SlaveErrorCode.MASTER_TOKEN_NOT_FOUND
) == SlaveMasterTokenNotFound
assert get_slave_exception_class_by_error_code(SlaveErrorCode.MASTER_TOKEN_INVALID) == SlaveMasterTokenInvalid
assert get_slave_exception_class_by_error_code(SlaveErrorCode.SELF_TOKEN_NOT_FOUND) == SlaveSelfTokenNotFound
assert get_slave_exception_class_by_error_code(SlaveErrorCode.SELF_TOKEN_INVALID) == SlaveSelfTokenInvalid
assert get_slave_exception_class_by_error_code(
SlaveErrorCode.SLAVE_ALREADY_CONNECTED
) == SlaveSlaveAlreadyConnected
assert get_slave_exception_class_by_error_code(SlaveErrorCode.SLAVE_NOT_CONNECTED) == SlaveSlaveNotConnected
assert get_slave_exception_class_by_error_code(
SlaveErrorCode.SLAVE_CONNECTION_REFUSED
) == SlaveSlaveConnectionRefused
assert get_slave_exception_class_by_error_code(
SlaveErrorCode.SLAVE_DISCONNECTION_REFUSED
) == SlaveSlaveDisconnectionRefused
assert get_slave_exception_class_by_error_code(SlaveErrorCode.TASK_ALREADY_EXIST) == SlaveTaskAlreadyExist
assert get_slave_exception_class_by_error_code(SlaveErrorCode.TASK_REFUSED) == SlaveTaskRefused
def test_get_slave_exception_by_error(self):
err = get_slave_exception_by_error(self._generate_exception(101, 'This is slave shutting down.'))
assert isinstance(err, SlaveSystemShuttingDown)
assert not err.success
assert err.status_code == 400
assert err.code == 101
assert err.message == 'This is slave shutting down.'
assert err.data == {}
err = get_slave_exception_by_error(self._generate_exception(602, 'Task refused.', data={'value': 233}))
assert isinstance(err, SlaveTaskRefused)
assert not err.success
assert err.status_code == 400
assert err.code == 602
assert err.message == 'Task refused.'
assert err.data == {'value': 233}
|