<s> from __future__ import unicode_literals <EOL> import datetime <EOL> from pytz import utc <EOL> from tornado import gen <EOL> from tornado . ioloop import IOLoop <EOL> from tornado . testing import gen_test <EOL> from nose . plugins . attrib import attr <EOL> from goblin . tests . base import BaseGoblinTestCase <EOL> from goblin . models import ( Vertex , V , Edge , IN , OUT , BOTH , GREATER_THAN , <EOL> LESS_THAN ) <EOL> from goblin import properties <EOL> class Person ( Vertex ) : <EOL> name = properties . String ( ) <EOL> age = properties . Integer ( ) <EOL> class Course ( Vertex ) : <EOL> name = properties . String ( ) <EOL> credits = properties . Double ( ) <EOL> class ResearchGroup ( Vertex ) : <EOL> name = properties . String ( ) <EOL> class EnrolledIn ( Edge ) : <EOL> date_enrolled = properties . DateTime ( ) <EOL> enthusiasm = properties . Integer ( default = <NUM_LIT:5> ) <EOL> class TaughtBy ( Edge ) : <EOL> overall_mood = properties . String ( default = '<STR_LIT>' ) <EOL> class BelongsTo ( Edge ) : <EOL> member_since = properties . DateTime ( ) <EOL> class SupervisedBy ( Edge ) : <EOL> pass <EOL> class BaseTraversalTestCase ( BaseGoblinTestCase ) : <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> super ( BaseTraversalTestCase , cls ) . setUpClass ( ) <EOL> loop = IOLoop . current ( ) <EOL> @ gen . coroutine <EOL> def build_graph ( ) : <EOL> """<STR_LIT>""" <EOL> cls . jon = yield Person . create ( name = '<STR_LIT>' , age = <NUM_LIT> ) <EOL> cls . eric = yield Person . create ( name = '<STR_LIT>' , age = <NUM_LIT> ) <EOL> cls . blake = yield Person . create ( name = '<STR_LIT>' , age = <NUM_LIT> ) <EOL> cls . physics = yield Course . create ( name = '<STR_LIT>' , credits = <NUM_LIT:1.0> ) <EOL> cls . beekeeping = yield Course . create ( <EOL> name = '<STR_LIT>' , credits = <NUM_LIT> ) <EOL> cls . theoretics = yield Course . create ( <EOL> name = '<STR_LIT>' , credits = - <NUM_LIT> ) <EOL> cls . dist_dev = yield ResearchGroup . create ( <EOL> name = '<STR_LIT>' ) <EOL> cls . eric_in_physics = yield EnrolledIn . create ( <EOL> cls . eric , cls . physics , <EOL> date_enrolled = datetime . datetime . now ( tz = utc ) , enthusiasm = <NUM_LIT:10> ) <EOL> cls . jon_in_beekeeping = yield EnrolledIn . create ( <EOL> cls . jon , cls . beekeeping , <EOL> date_enrolled = datetime . datetime . now ( tz = utc ) , enthusiasm = <NUM_LIT:1> ) <EOL> cls . jon_in_dist_dev = yield BelongsTo . create ( <EOL> cls . jon , cls . dist_dev , <EOL> member_since = datetime . datetime . now ( tz = utc ) ) <EOL> cls . blake_in_theoretics = yield EnrolledIn . create ( <EOL> cls . blake , cls . theoretics , <EOL> date_enrolled = datetime . datetime . now ( tz = utc ) , enthusiasm = <NUM_LIT:8> ) <EOL> cls . blake_beekeeping = yield TaughtBy . create ( <EOL> cls . beekeeping , cls . blake , overall_mood = '<STR_LIT>' ) <EOL> cls . jon_physics = yield TaughtBy . create ( <EOL> cls . physics , cls . jon , overall_mood = '<STR_LIT>' ) <EOL> cls . eric_theoretics = yield TaughtBy . create ( <EOL> cls . theoretics , cls . eric , overall_mood = '<STR_LIT>' ) <EOL> cls . jon_eric = yield SupervisedBy . create ( cls . eric , cls . jon ) <EOL> loop . run_sync ( build_graph ) <EOL> @ classmethod <EOL> def tearDownClass ( cls ) : <EOL> loop = IOLoop . current ( ) <EOL> @ gen . coroutine <EOL> def destroy_graph ( ) : <EOL> yield cls . eric_theoretics . delete ( ) <EOL> yield cls . jon_physics . delete ( ) <EOL> yield cls . blake_beekeeping . delete ( ) <EOL> yield cls . blake_in_theoretics . delete ( ) <EOL> yield cls . jon_in_beekeeping . delete ( ) <EOL> yield cls . eric_in_physics . delete ( ) <EOL> yield cls . jon_eric . delete ( ) <EOL> yield cls . jon_in_dist_dev . delete ( ) <EOL> yield cls . dist_dev . delete ( ) <EOL> yield cls . theoretics . delete ( ) <EOL> yield cls . beekeeping . delete ( ) <EOL> yield cls . physics . delete ( ) <EOL> yield cls . blake . delete ( ) <EOL> yield cls . eric . delete ( ) <EOL> yield cls . jon . delete ( ) <EOL> loop . run_sync ( destroy_graph ) <EOL> super ( BaseTraversalTestCase , cls ) . tearDownClass ( ) <EOL> @ attr ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> class TestVertexTraversals ( BaseTraversalTestCase ) : <EOL> @ gen_test <EOL> def test_inV_works ( self ) : <EOL> """<STR_LIT>""" <EOL> stream = yield self . jon . inV ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertEqual ( len ( results ) , <NUM_LIT:2> ) <EOL> self . assertIn ( self . physics , results ) <EOL> self . assertIn ( self . eric , results ) <EOL> stream = yield self . physics . inV ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertEqual ( len ( results ) , <NUM_LIT:1> ) <EOL> self . assertIn ( self . eric , results ) <EOL> stream = yield self . eric . inV ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertEqual ( len ( results ) , <NUM_LIT:1> ) <EOL> self . assertIn ( self . theoretics , results ) <EOL> stream = yield self . theoretics . inV ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertEqual ( len ( results ) , <NUM_LIT:1> ) <EOL> self . assertIn ( self . blake , results ) <EOL> stream = yield self . beekeeping . inV ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertEqual ( len ( results ) , <NUM_LIT:1> ) <EOL> self . assertIn ( self . jon , results ) <EOL> stream = yield self . blake . inV ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertEqual ( len ( results ) , <NUM_LIT:1> ) <EOL> self . assertIn ( self . beekeeping , results ) <EOL> @ gen_test <EOL> def test_inE_traversals ( self ) : <EOL> """<STR_LIT>""" <EOL> stream = yield self . jon . inE ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertEqual ( len ( results ) , <NUM_LIT:2> ) <EOL> self . assertIn ( self . jon_physics , results ) <EOL> self . assertIn ( self . jon_eric , results ) <EOL> @ gen_test <EOL> def test_outV_traversals ( self ) : <EOL> """<STR_LIT>""" <EOL> stream = yield self . eric . outV ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertEqual ( len ( results ) , <NUM_LIT:2> ) <EOL> self . assertIn ( self . physics , results ) <EOL> self . assertIn ( self . jon , results ) <EOL> @ gen_test <EOL> def test_outE_traverals ( self ) : <EOL> """<STR_LIT>""" <EOL> stream = yield self . blake . outE ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertEqual ( len ( results ) , <NUM_LIT:1> ) <EOL> self . assertIn ( self . blake_in_theoretics , results ) <EOL> @ gen_test <EOL> def test_bothE_traversals ( self ) : <EOL> """<STR_LIT>""" <EOL> stream = yield self . jon . bothE ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertEqual ( len ( results ) , <NUM_LIT:4> ) <EOL> self . assertIn ( self . jon_physics , results ) <EOL> self . assertIn ( self . jon_in_beekeeping , results ) <EOL> self . assertIn ( self . jon_in_dist_dev , results ) <EOL> self . assertIn ( self . jon_eric , results ) <EOL> @ gen_test <EOL> def test_bothV_traversals ( self ) : <EOL> """<STR_LIT>""" <EOL> stream = yield self . blake . bothV ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertEqual ( len ( results ) , <NUM_LIT:2> ) <EOL> self . assertIn ( self . beekeeping , results ) <EOL> @ attr ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> class TestVertexCentricQueries ( BaseTraversalTestCase ) : <EOL> @ gen_test <EOL> def test_out_step ( self ) : <EOL> stream = yield V ( self . jon ) . out_step ( ) . get ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertIn ( self . beekeeping , results ) <EOL> @ gen_test <EOL> def test_in ( self ) : <EOL> stream = yield V ( self . jon ) . in_step ( ) . get ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertIn ( self . physics , results ) <EOL> self . assertIn ( self . eric , results ) <EOL> @ gen_test <EOL> def test_both ( self ) : <EOL> stream = yield V ( self . jon ) . both ( ) . get ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertIn ( self . physics , results ) <EOL> self . assertIn ( self . beekeeping , results ) <EOL> @ gen_test <EOL> def test_out_labels ( self ) : <EOL> stream = yield V ( self . jon ) . out_step ( EnrolledIn ) . get ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertEqual ( len ( results ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( results [ <NUM_LIT:0> ] . name , "<STR_LIT>" ) <EOL> stream = yield V ( self . jon ) . out_step ( BelongsTo ) . get ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertEqual ( len ( results ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( results [ <NUM_LIT:0> ] . name , "<STR_LIT>" ) <EOL> stream = yield V ( self . jon ) . out_step ( BelongsTo , EnrolledIn ) . get ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertEqual ( len ( results ) , <NUM_LIT:2> ) <EOL> self . assertIn ( self . dist_dev , results ) <EOL> self . assertIn ( self . beekeeping , results ) <EOL> @ gen_test <EOL> def test_in_labels ( self ) : <EOL> stream = yield V ( self . jon ) . in_step ( TaughtBy ) . get ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertEqual ( len ( results ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( results [ <NUM_LIT:0> ] . name , "<STR_LIT>" ) <EOL> stream = yield V ( self . jon ) . in_step ( SupervisedBy ) . get ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertEqual ( len ( results ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( results [ <NUM_LIT:0> ] . name , "<STR_LIT>" ) <EOL> stream = yield V ( self . jon ) . in_step ( SupervisedBy , TaughtBy ) . get ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertEqual ( len ( results ) , <NUM_LIT:2> ) <EOL> self . assertIn ( self . physics , results ) <EOL> self . assertIn ( self . eric , results ) <EOL> @ gen_test <EOL> def test_both_labels ( self ) : <EOL> stream = yield V ( self . jon ) . both ( EnrolledIn ) . get ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertEqual ( len ( results ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( results [ <NUM_LIT:0> ] . name , "<STR_LIT>" ) <EOL> stream = yield V ( self . jon ) . both ( BelongsTo ) . get ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertEqual ( len ( results ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( results [ <NUM_LIT:0> ] . name , "<STR_LIT>" ) <EOL> stream = yield V ( self . jon ) . both ( BelongsTo , EnrolledIn ) . get ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertEqual ( len ( results ) , <NUM_LIT:2> ) <EOL> self . assertIn ( self . dist_dev , results ) <EOL> self . assertIn ( self . beekeeping , results ) <EOL> stream = yield V ( self . jon ) . both ( TaughtBy ) . get ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertEqual ( len ( results ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( results [ <NUM_LIT:0> ] . name , "<STR_LIT>" ) <EOL> stream = yield V ( self . jon ) . both ( SupervisedBy ) . get ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertEqual ( len ( results ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( results [ <NUM_LIT:0> ] . name , "<STR_LIT>" ) <EOL> stream = yield V ( self . jon ) . both ( SupervisedBy , TaughtBy ) . get ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertEqual ( len ( results ) , <NUM_LIT:2> ) <EOL> self . assertIn ( self . physics , results ) <EOL> self . assertIn ( self . eric , results ) <EOL> stream = yield V ( self . jon ) . both ( <EOL> SupervisedBy , TaughtBy , BelongsTo , EnrolledIn ) . get ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertEqual ( len ( results ) , <NUM_LIT:4> ) <EOL> self . assertIn ( self . physics , results ) <EOL> self . assertIn ( self . eric , results ) <EOL> self . assertIn ( self . dist_dev , results ) <EOL> self . assertIn ( self . beekeeping , results ) <EOL> @ gen_test <EOL> def test_out_v ( self ) : <EOL> stream = yield V ( self . blake ) . in_e ( ) . out_v ( ) . get ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertEqual ( len ( results ) , <NUM_LIT:1> ) <EOL> self . assertIn ( self . beekeeping , results ) <EOL> @ gen_test <EOL> def test_out_v ( self ) : <EOL> stream = yield V ( self . blake ) . out_e ( ) . in_v ( ) . get ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertEqual ( len ( results ) , <NUM_LIT:1> ) <EOL> self . assertIn ( self . theoretics , results ) <EOL> @ gen_test <EOL> def test_both_v ( self ) : <EOL> stream = yield V ( self . blake ) . out_e ( ) . both_v ( ) . get ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertEqual ( len ( results ) , <NUM_LIT:2> ) <EOL> self . assertIn ( self . theoretics , results ) <EOL> self . assertIn ( self . blake , results ) <EOL> @ gen_test <EOL> def test_other_v ( self ) : <EOL> stream = yield V ( self . jon ) . both_e ( ) . other_v ( ) . get ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertEqual ( len ( results ) , <NUM_LIT:4> ) <EOL> self . assertIn ( self . beekeeping , results ) <EOL> self . assertIn ( self . dist_dev , results ) <EOL> self . assertIn ( self . eric , results ) <EOL> self . assertIn ( self . physics , results ) <EOL> @ gen_test <EOL> def test_both_v_multiple ( self ) : <EOL> stream = yield V ( self . jon ) . both_e ( ) . both_v ( ) . get ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertEqual ( len ( results ) , <NUM_LIT:8> ) <EOL> self . assertIn ( self . beekeeping , results ) <EOL> self . assertIn ( self . dist_dev , results ) <EOL> self . assertIn ( self . eric , results ) <EOL> self . assertIn ( self . physics , results ) <EOL> self . assertIn ( self . jon , results ) <EOL> @ gen_test <EOL> def test_out_e ( self ) : <EOL> stream = yield V ( self . jon ) . out_e ( ) . get ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertTrue ( len ( results ) , <NUM_LIT:2> ) <EOL> self . assertIn ( self . jon_in_dist_dev , results ) <EOL> self . assertIn ( self . jon_in_beekeeping , results ) <EOL> stream = yield V ( self . jon ) . out_e ( BelongsTo ) . get ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertTrue ( len ( results ) , <NUM_LIT:1> ) <EOL> self . assertIn ( self . jon_in_dist_dev , results ) <EOL> stream = yield V ( self . jon ) . out_e ( EnrolledIn ) . get ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertTrue ( len ( results ) , <NUM_LIT:1> ) <EOL> self . assertIn ( self . jon_in_beekeeping , results ) <EOL> stream = yield V ( self . jon ) . out_e ( EnrolledIn , BelongsTo ) . get ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertTrue ( len ( results ) , <NUM_LIT:2> ) <EOL> self . assertIn ( self . jon_in_dist_dev , results ) <EOL> self . assertIn ( self . jon_in_beekeeping , results ) <EOL> @ gen_test <EOL> def test_in_e ( self ) : <EOL> stream = yield V ( self . jon ) . in_e ( ) . get ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertTrue ( len ( results ) , <NUM_LIT:2> ) <EOL> self . assertIn ( self . jon_eric , results ) <EOL> self . assertIn ( self . jon_physics , results ) <EOL> stream = yield V ( self . jon ) . in_e ( SupervisedBy ) . get ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertTrue ( len ( results ) , <NUM_LIT:1> ) <EOL> self . assertIn ( self . jon_eric , results ) <EOL> stream = yield V ( self . jon ) . in_e ( TaughtBy ) . get ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertTrue ( len ( results ) , <NUM_LIT:1> ) <EOL> self . assertIn ( self . jon_physics , results ) <EOL> stream = yield V ( self . jon ) . in_e ( SupervisedBy , TaughtBy ) . get ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertTrue ( len ( results ) , <NUM_LIT:2> ) <EOL> self . assertIn ( self . jon_eric , results ) <EOL> self . assertIn ( self . jon_physics , results ) <EOL> @ gen_test <EOL> def test_both_e ( self ) : <EOL> stream = yield V ( self . jon ) . both_e ( ) . get ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertTrue ( len ( results ) , <NUM_LIT:4> ) <EOL> self . assertIn ( self . jon_eric , results ) <EOL> self . assertIn ( self . jon_physics , results ) <EOL> self . assertIn ( self . jon_in_dist_dev , results ) <EOL> self . assertIn ( self . jon_in_beekeeping , results ) <EOL> stream = yield V ( self . jon ) . both_e ( SupervisedBy ) . get ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertTrue ( len ( results ) , <NUM_LIT:1> ) <EOL> self . assertIn ( self . jon_eric , results ) <EOL> stream = yield V ( self . jon ) . both_e ( EnrolledIn ) . get ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertTrue ( len ( results ) , <NUM_LIT:1> ) <EOL> self . assertIn ( self . jon_in_beekeeping , results ) <EOL> stream = yield V ( self . jon ) . both_e ( SupervisedBy , BelongsTo ) . get ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertTrue ( len ( results ) , <NUM_LIT:2> ) <EOL> self . assertIn ( self . jon_eric , results ) <EOL> self . assertIn ( self . jon_in_dist_dev , results ) <EOL> stream = yield V ( self . jon ) . both_e ( <EOL> SupervisedBy , BelongsTo , EnrolledIn , TaughtBy ) . get ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertTrue ( len ( results ) , <NUM_LIT:4> ) <EOL> self . assertIn ( self . jon_eric , results ) <EOL> self . assertIn ( self . jon_physics , results ) <EOL> self . assertIn ( self . jon_in_dist_dev , results ) <EOL> self . assertIn ( self . jon_in_beekeeping , results ) <EOL> @ gen_test <EOL> def test_has ( self ) : <EOL> stream = yield V ( self . jon ) . out_step ( ) . has ( <EOL> ResearchGroup . get_property_by_name ( "<STR_LIT:name>" ) , <EOL> "<STR_LIT>" ) . get ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertEqual ( len ( results ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( results [ <NUM_LIT:0> ] , self . dist_dev ) <EOL> @ gen_test <EOL> def test_has_lt_gt ( self ) : <EOL> stream = yield V ( self . jon ) . out_step ( ) . has ( <EOL> Course . get_property_by_name ( "<STR_LIT>" ) , <NUM_LIT> , GREATER_THAN ) . get ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertEqual ( len ( results ) , <NUM_LIT:1> ) <EOL> stream = yield V ( self . jon ) . out_step ( ) . has ( <EOL> Course . get_property_by_name ( "<STR_LIT>" ) , <NUM_LIT> , LESS_THAN ) . get ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertEqual ( len ( results ) , <NUM_LIT:0> ) <EOL> @ gen_test <EOL> def test_has_label ( self ) : <EOL> stream = yield V ( self . jon ) . out_step ( ) . has_label ( <EOL> ResearchGroup . get_label ( ) ) . get ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertEqual ( len ( results ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( results [ <NUM_LIT:0> ] , self . dist_dev ) <EOL> stream = yield V ( self . jon ) . out_step ( ) . has_label ( <EOL> ResearchGroup . get_label ( ) , Course . get_label ( ) ) . get ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertEqual ( len ( results ) , <NUM_LIT:2> ) <EOL> self . assertIn ( self . dist_dev , results ) <EOL> self . assertIn ( self . beekeeping , results ) <EOL> @ gen_test <EOL> def test_has_id ( self ) : <EOL> stream = yield V ( self . jon ) . out_step ( ) . has_id ( <EOL> self . dist_dev . id ) . get ( ) <EOL> results = yield stream . read ( ) <EOL> self . assertEqual ( len ( results ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( results [ <NUM_LIT:0> ] , self . dist_dev ) <EOL> stream = yield V ( self . jon ) . out_step ( ) . has_id ( <EOL> self . dist_dev . id , self . beekeeping . id ) . get ( ) <EOL> results = yield stream . read ( ) </s>
