def test():
    ''' pcap_bro.py: Unit test'''

    # This worker test requires a local server running
    workbench = zerorpc.Client(timeout=300, heartbeat=60)
    workbench.connect("tcp://127.0.0.1:4242")

    # Generate the input data for this worker
    data_path = os.path.join(os.path.dirname(os.path.realpath(__file__)), '../data/pcap/http.pcap')
    md5 = workbench.store_sample(open(data_path, 'rb').read(), 'http.pcap', 'pcap')
    input_data = workbench.get_sample(md5)

    # Execute the worker (unit test)
    worker = PcapBro()
    output = worker.execute(input_data)
    print '\n<<< Unit Test >>>'
    pprint.pprint(output)

    # Execute the worker (server test)
    output = workbench.work_request('pcap_bro', md5)
    print '\n<<< Server Test >>>'
    pprint.pprint(output)

    # Open a bunch of pcaps
    data_dir = os.path.join(os.path.dirname(os.path.realpath(__file__)), '../data/pcap')
    file_list = [os.path.join(data_dir, child) for child in os.listdir(data_dir)]
    pcap_md5s = []
    for filename in file_list:

        # Skip OS generated files
        if '.DS_Store' in filename: 
            continue

        with open(filename, 'rb') as pcap_file:
            base_name = os.path.basename(filename)
            pcap_md5s.append(workbench.store_sample(pcap_file.read(), base_name, 'pcap'))

    # Now store the sample set
    set_md5 = workbench.store_sample_set(pcap_md5s)
    print set_md5

    # Execute the worker (unit test)
    output = worker.execute({'sample_set': {'md5_list': pcap_md5s}})
    print '\n<<< Unit Test >>>'
    pprint.pprint(output)

    # Execute the worker (server test)
    output = workbench.work_request('pcap_bro', set_md5)
    print '\n<<< Server Test >>>'
    pprint.pprint(output)
def test ( ) : workbench = zerorpc . Client ( timeout = 300 , heartbeat = 60 ) workbench . connect ( "tcp://127.0.0.1:4242" ) data_path = os . path . join ( os . path . dirname ( os . path . realpath ( __file__ ) ) , '../data/pcap/http.pcap' ) md5 = workbench . store_sample ( open ( data_path , 'rb' ) . read ( ) , 'http.pcap' , 'pcap' ) input_data = workbench . get_sample ( md5 ) worker = PcapBro ( ) output = worker . execute ( input_data ) print '\n<<< Unit Test >>>' pprint . pprint ( output ) output = workbench . work_request ( 'pcap_bro' , md5 ) print '\n<<< Server Test >>>' pprint . pprint ( output ) data_dir = os . path . join ( os . path . dirname ( os . path . realpath ( __file__ ) ) , '../data/pcap' ) file_list = [ os . path . join ( data_dir , child ) for child in os . listdir ( data_dir ) ] pcap_md5s = [ ] for filename in file_list : if '.DS_Store' in filename : continue with open ( filename , 'rb' ) as pcap_file : base_name = os . path . basename ( filename ) pcap_md5s . append ( workbench . store_sample ( pcap_file . read ( ) , base_name , 'pcap' ) ) set_md5 = workbench . store_sample_set ( pcap_md5s ) print set_md5 output = worker . execute ( { 'sample_set' : { 'md5_list' : pcap_md5s } } ) print '\n<<< Unit Test >>>' pprint . pprint ( output ) output = workbench . work_request ( 'pcap_bro' , set_md5 ) print '\n<<< Server Test >>>' pprint . pprint ( output )
pcap_bro.py: Unit test