<s> import socket <EOL> import time <EOL> from collections import defaultdict <EOL> from circus . fixed_threading import Thread <EOL> from zmq . eventloop import ioloop <EOL> from circus . stats import collector as collector_module <EOL> from circus . stats . collector import SocketStatsCollector , WatcherStatsCollector <EOL> from circus . tests . support import TestCase , EasyTestSuite <EOL> class TestCollector ( TestCase ) : <EOL> def setUp ( self ) : <EOL> self . socks = [ ] <EOL> self . clients = [ ] <EOL> self . fds = [ ] <EOL> self . pids = { } <EOL> def tearDown ( self ) : <EOL> for sock , _ , _ in self . socks : <EOL> sock . close ( ) <EOL> for sock in self . clients : <EOL> sock . close ( ) <EOL> def _get_streamer ( self ) : <EOL> for i in range ( <NUM_LIT:10> ) : <EOL> sock = socket . socket ( socket . AF_INET , socket . SOCK_STREAM ) <EOL> sock . bind ( ( '<STR_LIT:localhost>' , <NUM_LIT:0> ) ) <EOL> sock . listen ( <NUM_LIT:1> ) <EOL> self . socks . append ( ( sock , '<STR_LIT>' , sock . fileno ( ) ) ) <EOL> self . fds . append ( sock . fileno ( ) ) <EOL> client = socket . socket ( socket . AF_INET , socket . SOCK_STREAM ) <EOL> client . connect ( sock . getsockname ( ) ) <EOL> self . clients . append ( client ) <EOL> class FakeStreamer ( object ) : <EOL> stats = [ ] <EOL> def __init__ ( this ) : <EOL> this . sockets = self . socks <EOL> @ property <EOL> def circus_pids ( this ) : <EOL> return self . circus_pids <EOL> def get_pids ( this , name ) : <EOL> return self . pids [ name ] <EOL> @ property <EOL> def publisher ( this ) : <EOL> return this <EOL> def publish ( this , name , stat ) : <EOL> this . stats . append ( stat ) <EOL> self . streamer = FakeStreamer ( ) <EOL> return self . streamer <EOL> def _get_collector ( self , collector_class ) : <EOL> self . _get_streamer ( ) <EOL> class Collector ( Thread ) : <EOL> def __init__ ( this , streamer ) : <EOL> Thread . __init__ ( this ) <EOL> this . streamer = streamer <EOL> this . loop = ioloop . IOLoop ( ) <EOL> this . daemon = True <EOL> def run ( self ) : <EOL> collector = collector_class ( <EOL> self . streamer , '<STR_LIT>' , callback_time = <NUM_LIT:0.1> , <EOL> io_loop = self . loop ) <EOL> collector . start ( ) <EOL> self . loop . start ( ) <EOL> def stop ( self ) : <EOL> self . loop . add_callback ( self . loop . stop ) <EOL> self . loop . add_callback ( self . loop . close ) <EOL> return Collector ( self . streamer ) <EOL> def test_watcherstats ( self ) : <EOL> calls = defaultdict ( int ) <EOL> info = [ ] <EOL> for i in range ( <NUM_LIT:2> ) : <EOL> info . append ( { <EOL> '<STR_LIT>' : <NUM_LIT> + i , <EOL> '<STR_LIT>' : [ ] , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:0.0> + i / <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:0.0> , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT:username>' : '<STR_LIT>' } ) <EOL> def _get_info ( pid ) : <EOL> try : <EOL> data = info [ calls [ pid ] ] . copy ( ) <EOL> except IndexError : <EOL> raise collector_module . util . NoSuchProcess ( pid ) <EOL> data [ '<STR_LIT>' ] = pid <EOL> calls [ pid ] += <NUM_LIT:1> <EOL> return data <EOL> old_info = collector_module . util . get_info <EOL> try : <EOL> collector_module . util . get_info = _get_info <EOL> self . pids [ '<STR_LIT>' ] = [ <NUM_LIT> , <NUM_LIT> ] <EOL> collector = WatcherStatsCollector ( self . _get_streamer ( ) , '<STR_LIT>' ) <EOL> stats = list ( collector . collect_stats ( ) ) <EOL> self . assertEqual ( len ( stats ) , <NUM_LIT:3> ) <EOL> stats = list ( collector . collect_stats ( ) ) <EOL> self . assertEqual ( len ( stats ) , <NUM_LIT:3> ) <EOL> stats = list ( collector . collect_stats ( ) ) <EOL> self . assertEqual ( len ( stats ) , <NUM_LIT:1> ) <EOL> self . circus_pids = { <NUM_LIT> : '<STR_LIT>' } <EOL> self . pids [ '<STR_LIT>' ] = [ <NUM_LIT> ] <EOL> collector = WatcherStatsCollector ( self . _get_streamer ( ) , '<STR_LIT>' ) <EOL> stats = list ( collector . collect_stats ( ) ) <EOL> self . assertEqual ( stats [ <NUM_LIT:0> ] [ '<STR_LIT:name>' ] , '<STR_LIT>' ) <EOL> finally : <EOL> collector_module . util . get_info = old_info <EOL> def test_collector_aggregation ( self ) : <EOL> collector = WatcherStatsCollector ( self . _get_streamer ( ) , '<STR_LIT>' ) <EOL> aggregate = { } <EOL> for i in range ( <NUM_LIT:0> , <NUM_LIT:10> ) : <EOL> pid = <NUM_LIT:1000> + i <EOL> aggregate [ pid ] = { <EOL> '<STR_LIT>' : <NUM_LIT> , '<STR_LIT>' : [ ] , <EOL> '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:0.0> + i / <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:0.0> + i // <NUM_LIT:10> , <EOL> '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:username>' : '<STR_LIT>' , '<STR_LIT>' : pid , '<STR_LIT:name>' : '<STR_LIT>' } <EOL> res = collector . _aggregate ( aggregate ) <EOL> self . assertEqual ( res [ '<STR_LIT>' ] , <NUM_LIT:0> ) <EOL> self . assertEqual ( len ( res [ '<STR_LIT>' ] ) , <NUM_LIT:10> ) <EOL> self . assertEqual ( res [ '<STR_LIT>' ] , <NUM_LIT> ) <EOL> def test_collector_aggregation_when_unknown_values ( self ) : <EOL> collector = WatcherStatsCollector ( self . _get_streamer ( ) , '<STR_LIT>' ) <EOL> aggregate = { } <EOL> for i in range ( <NUM_LIT:0> , <NUM_LIT:10> ) : <EOL> pid = <NUM_LIT:1000> + i <EOL> aggregate [ pid ] = { <EOL> '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : [ ] , '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : <NUM_LIT:0> , '<STR_LIT>' : pid , <EOL> '<STR_LIT:username>' : '<STR_LIT>' , '<STR_LIT>' : pid , '<STR_LIT:name>' : '<STR_LIT>' } <EOL> res = collector . _aggregate ( aggregate ) <EOL> self . assertEqual ( res [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( len ( res [ '<STR_LIT>' ] ) , <NUM_LIT:10> ) <EOL> self . assertEqual ( res [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> def test_socketstats ( self ) : <EOL> collector = self . _get_collector ( SocketStatsCollector ) <EOL> collector . start ( ) <EOL> time . sleep ( <NUM_LIT:1.> ) <EOL> for i in range ( <NUM_LIT:10> ) : <EOL> for client in self . clients : <EOL> client . send ( b'<STR_LIT>' ) <EOL> collector . stop ( ) <EOL> for s , _ , _ in self . socks : <EOL> s . close ( ) <EOL> self . assertTrue ( len ( self . streamer . stats ) > <NUM_LIT:2> ) <EOL> stat = self . streamer . stats [ <NUM_LIT:0> ] <EOL> self . assertTrue ( stat [ '<STR_LIT>' ] in self . fds ) <EOL> self . assertTrue ( stat [ '<STR_LIT>' ] > <NUM_LIT:1> ) <EOL> test_suite = EasyTestSuite ( __name__ ) </s>
