# Path: gnsq/backofftimer.py
# class BackoffTimer(object):
#     def __init__(self, ratio=1, max_interval=None, min_interval=None):
#         self.c = 0
#         self.ratio = ratio
# 
#         self.max_interval = max_interval
#         self.min_interval = min_interval
# 
#     def is_reset(self):
#         return self.c == 0
# 
#     def reset(self):
#         self.c = 0
#         return self
# 
#     def success(self):
#         self.c = max(self.c - 1, 0)
#         return self
# 
#     def failure(self):
#         self.c += 1
#         return self
# 
#     def get_interval(self):
#         k = pow(2, self.c) - 1
#         interval = random.random() * k * self.ratio
# 
#         if self.max_interval is not None:
#             interval = min(interval, self.max_interval)
# 
#         if self.min_interval is not None:
#             interval = max(interval, self.min_interval)
# 
#         return interval
# 
# Path: gnsq/protocol.py
# MAGIC_V2 = b'  V2'
# NEWLINE = b'\n'
# SPACE = b' '
# EMPTY = b''
# HEARTBEAT = b'_heartbeat_'
# OK = b'OK'
# IDENTIFY = b'IDENTIFY'
# AUTH = b'AUTH'
# SUB = b'SUB'
# PUB = b'PUB'
# MPUB = b'MPUB'
# DPUB = b'DPUB'
# RDY = b'RDY'
# FIN = b'FIN'
# REQ = b'REQ'
# TOUCH = b'TOUCH'
# CLS = b'CLS'
# NOP = b'NOP'
# FRAME_TYPE_RESPONSE = 0
# FRAME_TYPE_ERROR = 1
# FRAME_TYPE_MESSAGE = 2
# VALID_NAME_RE = re.compile(r'^[\.a-zA-Z0-9_-]+(#ephemeral)?$')
# def _valid_name(name):
# def valid_topic_name(topic):
# def valid_channel_name(channel):
# def assert_valid_topic_name(topic):
# def assert_valid_channel_name(channel):
# def unpack_size(data):
# def unpack_response(data):
# def unpack_message(data):
# def _packsize(data):
# def _packbody(body):
# def _encode_param(data):
# def _command(cmd, body, *params):
# def identify(data):
# def auth(secret):
# def subscribe(topic_name, channel_name):
# def publish(topic_name, data):
# def multipublish_body(messages):
# def multipublish(topic_name, messages):
# def deferpublish(topic_name, data, delay_ms):
# def ready(count):
# def finish(message_id):
# def requeue(message_id, timeout=0):
# def touch(message_id):
# def close():
# def nop():


import pytest
from six.moves import range
from gnsq import BackoffTimer
from gnsq import protocol as nsq
def test_channel_names(name, good):
    assert nsq.valid_channel_name(name) == good


def test_assert_topic():
    assert nsq.assert_valid_topic_name('topic') is None

    with pytest.raises(ValueError):
        nsq.assert_valid_topic_name('invalid name with space')


def test_assert_channel():
    assert nsq.assert_valid_channel_name('channel') is None

    with pytest.raises(ValueError):
        nsq.assert_valid_channel_name('invalid name with space')


def test_invalid_commands():
    with pytest.raises(TypeError):
        nsq.requeue('1234', None)

    with pytest.raises(TypeError):
        nsq.ready(None)

    with pytest.raises(ValueError):
        nsq.ready(-1)


def test_backoff_timer():

# Path: gnsq/backofftimer.py
# class BackoffTimer(object):
#     def __init__(self, ratio=1, max_interval=None, min_interval=None):
#         self.c = 0
#         self.ratio = ratio
# 
#         self.max_interval = max_interval
#         self.min_interval = min_interval
# 
#     def is_reset(self):
#         return self.c == 0
# 
#     def reset(self):
#         self.c = 0
#         return self
# 
#     def success(self):
#         self.c = max(self.c - 1, 0)
#         return self
# 
#     def failure(self):
#         self.c += 1
#         return self
# 
#     def get_interval(self):
#         k = pow(2, self.c) - 1
#         interval = random.random() * k * self.ratio
# 
#         if self.max_interval is not None:
#             interval = min(interval, self.max_interval)
# 
#         if self.min_interval is not None:
#             interval = max(interval, self.min_interval)
# 
#         return interval
# 
# Path: gnsq/protocol.py
# MAGIC_V2 = b'  V2'
# NEWLINE = b'\n'
# SPACE = b' '
# EMPTY = b''
# HEARTBEAT = b'_heartbeat_'
# OK = b'OK'
# IDENTIFY = b'IDENTIFY'
# AUTH = b'AUTH'
# SUB = b'SUB'
# PUB = b'PUB'
# MPUB = b'MPUB'
# DPUB = b'DPUB'
# RDY = b'RDY'
# FIN = b'FIN'
# REQ = b'REQ'
# TOUCH = b'TOUCH'
# CLS = b'CLS'
# NOP = b'NOP'
# FRAME_TYPE_RESPONSE = 0
# FRAME_TYPE_ERROR = 1
# FRAME_TYPE_MESSAGE = 2
# VALID_NAME_RE = re.compile(r'^[\.a-zA-Z0-9_-]+(#ephemeral)?$')
# def _valid_name(name):
# def valid_topic_name(topic):
# def valid_channel_name(channel):
# def assert_valid_topic_name(topic):
# def assert_valid_channel_name(channel):
# def unpack_size(data):
# def unpack_response(data):
# def unpack_message(data):
# def _packsize(data):
# def _packbody(body):
# def _encode_param(data):
# def _command(cmd, body, *params):
# def identify(data):
# def auth(secret):
# def subscribe(topic_name, channel_name):
# def publish(topic_name, data):
# def multipublish_body(messages):
# def multipublish(topic_name, messages):
# def deferpublish(topic_name, data, delay_ms):
# def ready(count):
# def finish(message_id):
# def requeue(message_id, timeout=0):
# def touch(message_id):
# def close():
# def nop():


# Path: tests/test_basic.py
import pytest
from six.moves import range
from gnsq import BackoffTimer
from gnsq import protocol as nsq

def test_channel_names(name, good):
    assert nsq.valid_channel_name(name) == good


def test_assert_topic():
    assert nsq.assert_valid_topic_name('topic') is None

    with pytest.raises(ValueError):
        nsq.assert_valid_topic_name('invalid name with space')


def test_assert_channel():
    assert nsq.assert_valid_channel_name('channel') is None

    with pytest.raises(ValueError):
        nsq.assert_valid_channel_name('invalid name with space')


def test_invalid_commands():
    with pytest.raises(TypeError):
        nsq.requeue('1234', None)

    with pytest.raises(TypeError):
        nsq.ready(None)

    with pytest.raises(ValueError):
        nsq.ready(-1)


def test_backoff_timer():

    timer = BackoffTimer(max_interval=1000)