<s> from __future__ import with_statement <EOL> import pytest <EOL> import gnsq <EOL> from integration_server import NsqdIntegrationServer <EOL> @ pytest . mark . slow <EOL> @ pytest . mark . timeout ( <NUM_LIT> ) <EOL> def test_basic ( ) : <EOL> with NsqdIntegrationServer ( ) as server : <EOL> conn = gnsq . Nsqd ( server . address , http_port = server . http_port ) <EOL> assert conn . ping ( ) == b'<STR_LIT:OK>' <EOL> assert '<STR_LIT>' in conn . stats ( ) <EOL> assert '<STR_LIT:version>' in conn . info ( ) <EOL> @ pytest . mark . slow <EOL> @ pytest . mark . timeout ( <NUM_LIT> ) <EOL> def test_topics_channels ( ) : <EOL> with NsqdIntegrationServer ( ) as server : <EOL> conn = gnsq . Nsqd ( server . address , http_port = server . http_port ) <EOL> assert len ( conn . stats ( ) [ '<STR_LIT>' ] ) == <NUM_LIT:0> <EOL> with pytest . raises ( gnsq . errors . NSQHttpError ) : <EOL> conn . delete_topic ( '<STR_LIT>' ) <EOL> conn . create_topic ( '<STR_LIT>' ) <EOL> topics = conn . stats ( ) [ '<STR_LIT>' ] <EOL> assert len ( topics ) == <NUM_LIT:1> <EOL> assert topics [ <NUM_LIT:0> ] [ '<STR_LIT>' ] == '<STR_LIT>' <EOL> conn . delete_topic ( '<STR_LIT>' ) <EOL> assert len ( conn . stats ( ) [ '<STR_LIT>' ] ) == <NUM_LIT:0> <EOL> with pytest . raises ( gnsq . errors . NSQHttpError ) : <EOL> conn . create_channel ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> with pytest . raises ( gnsq . errors . NSQHttpError ) : <EOL> conn . delete_channel ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> conn . create_topic ( '<STR_LIT>' ) <EOL> assert len ( conn . stats ( ) [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) == <NUM_LIT:0> <EOL> conn . create_channel ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> channels = conn . stats ( ) [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] <EOL> assert len ( channels ) == <NUM_LIT:1> <EOL> assert channels [ <NUM_LIT:0> ] [ '<STR_LIT>' ] == '<STR_LIT>' <EOL> conn . delete_channel ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> assert len ( conn . stats ( ) [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) == <NUM_LIT:0> <EOL> @ pytest . mark . slow <EOL> @ pytest . mark . timeout ( <NUM_LIT> ) <EOL> def test_publish ( ) : <EOL> with NsqdIntegrationServer ( ) as server : <EOL> conn = gnsq . Nsqd ( server . address , http_port = server . http_port ) <EOL> conn . publish ( '<STR_LIT>' , b'<STR_LIT>' ) <EOL> assert conn . stats ( ) [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] == <NUM_LIT:1> <EOL> conn . multipublish ( '<STR_LIT>' , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> assert conn . stats ( ) [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] == <NUM_LIT:3> <EOL> conn . multipublish ( '<STR_LIT>' , iter ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) ) <EOL> assert conn . stats ( ) [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] == <NUM_LIT:6> <EOL> conn . empty_topic ( '<STR_LIT>' ) <EOL> assert conn . stats ( ) [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] == <NUM_LIT:0> <EOL> conn . create_topic ( '<STR_LIT>' ) <EOL> conn . create_channel ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> conn . publish ( '<STR_LIT>' , b'<STR_LIT>' ) <EOL> assert conn . stats ( ) [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] == <NUM_LIT:1> <EOL> conn . empty_channel ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> assert conn . stats ( ) [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] == <NUM_LIT:0> </s>
