<s> import unittest <EOL> import json <EOL> import msgpackrpc <EOL> from jubatus . graph . client import Graph <EOL> from jubatus . graph . types import * <EOL> from jubatus . common import Datum <EOL> from jubatus_test . test_util import TestUtil <EOL> host = "<STR_LIT:127.0.0.1>" <EOL> port = <NUM_LIT> <EOL> timeout = <NUM_LIT:10> <EOL> class GraphTest ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . config = { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : <NUM_LIT> , <EOL> "<STR_LIT>" : <NUM_LIT:5> <EOL> } <EOL> } <EOL> TestUtil . write_file ( '<STR_LIT>' , json . dumps ( self . config ) ) <EOL> self . srv = TestUtil . fork_process ( '<STR_LIT>' , port , '<STR_LIT>' ) <EOL> try : <EOL> self . cli = Graph ( host , port , "<STR_LIT:name>" ) <EOL> except : <EOL> TestUtil . kill_process ( self . srv ) <EOL> raise <EOL> def tearDown ( self ) : <EOL> if self . cli : <EOL> self . cli . get_client ( ) . close ( ) <EOL> TestUtil . kill_process ( self . srv ) <EOL> def test_node_info ( self ) : <EOL> edge_query = [ [ "<STR_LIT:a>" , "<STR_LIT:b>" ] , [ "<STR_LIT:c>" , "<STR_LIT:d>" ] , [ "<STR_LIT:e>" , "<STR_LIT:f>" ] ] <EOL> node_query = [ [ "<STR_LIT:0>" , "<STR_LIT:1>" ] , [ "<STR_LIT:2>" , "<STR_LIT:3>" ] ] <EOL> p = PresetQuery ( edge_query , node_query ) <EOL> in_edges = [ <NUM_LIT:0> , <NUM_LIT:0> ] <EOL> out_edges = [ <NUM_LIT:0> , <NUM_LIT:0> ] <EOL> Node ( p , in_edges , out_edges ) <EOL> def test_get_client ( self ) : <EOL> self . assertTrue ( isinstance ( self . cli . get_client ( ) , msgpackrpc . client . Client ) ) <EOL> def test_create_node ( self ) : <EOL> nid = self . cli . create_node ( ) <EOL> self . assertEqual ( str ( int ( nid ) ) , nid ) <EOL> def test_remove_node ( self ) : <EOL> nid = self . cli . create_node ( ) <EOL> self . assertEqual ( self . cli . remove_node ( nid ) , True ) <EOL> def test_update_node ( self ) : <EOL> nid = self . cli . create_node ( ) <EOL> prop = { "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" } <EOL> self . assertEqual ( self . cli . update_node ( nid , prop ) , True ) <EOL> def test_create_edge ( self ) : <EOL> src = self . cli . create_node ( ) <EOL> tgt = self . cli . create_node ( ) <EOL> prop = { "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" } <EOL> ei = Edge ( prop , src , tgt ) <EOL> eid = self . cli . create_edge ( tgt , ei ) <EOL> def test_str ( self ) : <EOL> self . assertEqual ( "<STR_LIT>" , <EOL> str ( Node ( { } , [ ] , [ ] ) ) ) <EOL> self . assertEqual ( "<STR_LIT>" , <EOL> str ( PresetQuery ( [ ] , [ ] ) ) ) <EOL> self . assertEqual ( "<STR_LIT>" , <EOL> str ( Edge ( { } , '<STR_LIT:src>' , '<STR_LIT>' ) ) ) <EOL> self . assertEqual ( "<STR_LIT>" , <EOL> str ( ShortestPathQuery ( '<STR_LIT:src>' , '<STR_LIT>' , <NUM_LIT:10> , PresetQuery ( [ ] , [ ] ) ) ) ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> test_suite = unittest . TestLoader ( ) . loadTestsFromTestCase ( GraphTest ) <EOL> unittest . TextTestRunner ( ) . run ( test_suite ) </s>
