<s> import os <EOL> import time <EOL> import socket <EOL> import zmq <EOL> from . base import Test , interactive , passive <EOL> TEST_SOCKET = '<STR_LIT>' <EOL> class TestCollectd ( Test ) : <EOL> def setUp ( self ) : <EOL> super ( ) . setUp ( ) <EOL> import zorro . collectd <EOL> def collectd_putval ( self ) : <EOL> self . z . sleep ( <NUM_LIT:0.1> ) <EOL> sock = self . z . collectd . Connection ( unixsock = TEST_SOCKET ) <EOL> self . assertEqual ( ( bytearray ( b'<STR_LIT>' ) , ) , <EOL> sock . putval ( '<STR_LIT>' , [ ( '<STR_LIT>' , '<STR_LIT>' ) ] , time = <NUM_LIT> ) ) <EOL> self . assertEqual ( ( bytearray ( b'<STR_LIT>' ) , ) , <EOL> sock . putnotif ( "<STR_LIT:hello>" , time = <NUM_LIT> ) ) <EOL> self . assertEqual ( ( bytearray ( b'<STR_LIT>' ) , ) , sock . flush ( ) ) <EOL> self . assertEqual ( { '<STR_LIT>' : <NUM_LIT> , '<STR_LIT>' : <NUM_LIT> } , <EOL> sock . getval ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( { '<STR_LIT>' : <NUM_LIT> , '<STR_LIT>' : <NUM_LIT> } , <EOL> dict ( sock . listval ( ) ) ) <EOL> @ interactive ( collectd_putval ) <EOL> def test_hello ( self ) : <EOL> if os . access ( TEST_SOCKET , os . F_OK ) : <EOL> os . unlink ( TEST_SOCKET ) <EOL> sock = socket . socket ( socket . AF_UNIX , socket . SOCK_STREAM ) <EOL> with sock : <EOL> sock . bind ( TEST_SOCKET ) <EOL> sock . listen ( <NUM_LIT> ) <EOL> s , a = sock . accept ( ) <EOL> with s : <EOL> self . assertEqual ( s . recv ( <NUM_LIT> ) , <EOL> b'<STR_LIT>' ) <EOL> s . sendall ( b'<STR_LIT>' ) <EOL> self . assertEqual ( s . recv ( <NUM_LIT> ) , <EOL> b'<STR_LIT>' ) <EOL> s . sendall ( b'<STR_LIT>' ) <EOL> self . assertEqual ( s . recv ( <NUM_LIT> ) , b'<STR_LIT>' ) <EOL> s . sendall ( b'<STR_LIT>' ) <EOL> self . assertEqual ( s . recv ( <NUM_LIT> ) , b'<STR_LIT>' ) <EOL> s . sendall ( b'<STR_LIT>' ) <EOL> self . assertEqual ( s . recv ( <NUM_LIT> ) , b'<STR_LIT>' ) <EOL> s . sendall ( b'<STR_LIT>' ) </s>
