<s> """<STR_LIT>""" <EOL> import sys <EOL> import uuid <EOL> import time <EOL> from unittest import TestCase <EOL> from ovs . dal . exceptions import * <EOL> from ovs . extensions . generic import fakesleep <EOL> from ovs . extensions . storage . persistent . dummystore import DummyPersistentStore <EOL> from ovs . extensions . storage . volatile . dummystore import DummyVolatileStore <EOL> from ovs . extensions . storage . persistentfactory import PersistentFactory <EOL> from ovs . extensions . storage . volatilefactory import VolatileFactory <EOL> from ovs . dal . hybrids . t_testdisk import TestDisk <EOL> from ovs . dal . hybrids . t_testemachine import TestEMachine <EOL> from ovs . dal . hybrids . t_testmachine import TestMachine <EOL> from ovs . dal . hybrids . t_teststoragedriver import TestStorageDriver <EOL> from ovs . dal . hybrids . t_teststoragerouter import TestStorageRouter <EOL> from ovs . dal . hybrids . t_testvpool import TestVPool <EOL> from ovs . dal . datalist import DataList <EOL> from ovs . dal . helpers import Descriptor , Toolbox <EOL> from ovs . extensions . generic . volatilemutex import VolatileMutex , NoLockAvailableException <EOL> class Basic ( TestCase ) : <EOL> """<STR_LIT>""" <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> """<STR_LIT>""" <EOL> PersistentFactory . store = DummyPersistentStore ( ) <EOL> PersistentFactory . store . clean ( ) <EOL> PersistentFactory . store . clean ( ) <EOL> VolatileFactory . store = DummyVolatileStore ( ) <EOL> VolatileFactory . store . clean ( ) <EOL> VolatileFactory . store . clean ( ) <EOL> fakesleep . monkey_patch ( ) <EOL> @ classmethod <EOL> def setUp ( cls ) : <EOL> """<STR_LIT>""" <EOL> PersistentFactory . store = DummyPersistentStore ( ) <EOL> PersistentFactory . store . clean ( ) <EOL> VolatileFactory . store = DummyVolatileStore ( ) <EOL> VolatileFactory . store . clean ( ) <EOL> DataList . test_hooks = { } <EOL> @ classmethod <EOL> def tearDownClass ( cls ) : <EOL> """<STR_LIT>""" <EOL> fakesleep . monkey_restore ( ) <EOL> def test_invalidobject ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertRaises ( ObjectNotFoundException , TestDisk , uuid . uuid4 ( ) , None ) <EOL> def test_newobject_delete ( self ) : <EOL> """<STR_LIT>""" <EOL> disk = TestDisk ( ) <EOL> disk . name = '<STR_LIT>' <EOL> disk . save ( ) <EOL> guid = disk . guid <EOL> self . assertIsNotNone ( guid , '<STR_LIT>' ) <EOL> disk . delete ( ) <EOL> self . assertRaises ( Exception , TestDisk , guid , None ) <EOL> def test_discard ( self ) : <EOL> """<STR_LIT>""" <EOL> disk = TestDisk ( ) <EOL> disk . name = '<STR_LIT>' <EOL> disk . save ( ) <EOL> disk . name = '<STR_LIT>' <EOL> disk . discard ( ) <EOL> self . assertEqual ( disk . name , '<STR_LIT>' , '<STR_LIT>' ) <EOL> def test_updateproperty ( self ) : <EOL> """<STR_LIT>""" <EOL> disk = TestDisk ( ) <EOL> disk . name = '<STR_LIT:test>' <EOL> disk . description = '<STR_LIT>' <EOL> self . assertIs ( disk . name , '<STR_LIT:test>' , '<STR_LIT>' ) <EOL> self . assertIs ( disk . description , '<STR_LIT>' , '<STR_LIT>' ) <EOL> def test_preinit ( self ) : <EOL> """<STR_LIT>""" <EOL> disk = TestDisk ( data = { '<STR_LIT:name>' : '<STR_LIT>' } ) <EOL> disk . save ( ) <EOL> self . assertEqual ( disk . name , '<STR_LIT>' , '<STR_LIT>' ) <EOL> disk = TestDisk ( data = { '<STR_LIT:name>' : '<STR_LIT>' , '<STR_LIT:foo>' : '<STR_LIT:bar>' } ) <EOL> disk . save ( ) <EOL> self . assertEqual ( disk . name , '<STR_LIT>' , '<STR_LIT>' ) <EOL> def test_datapersistent ( self ) : <EOL> """<STR_LIT>""" <EOL> disk = TestDisk ( ) <EOL> guid = disk . guid <EOL> disk . name = '<STR_LIT:test>' <EOL> disk . save ( ) <EOL> disk2 = TestDisk ( guid ) <EOL> self . assertEqual ( disk . name , disk2 . name , '<STR_LIT>' ) <EOL> def test_readonlyproperty ( self ) : <EOL> """<STR_LIT>""" <EOL> disk = TestDisk ( ) <EOL> self . assertIsNotNone ( disk . used_size , '<STR_LIT>' ) <EOL> def test_datastorewins ( self ) : <EOL> """<STR_LIT>""" <EOL> disk = TestDisk ( ) <EOL> disk . name = '<STR_LIT>' <EOL> disk . save ( ) <EOL> disk2 = TestDisk ( disk . guid , datastore_wins = True ) <EOL> disk . name = '<STR_LIT>' <EOL> disk . save ( ) <EOL> disk2 . name = '<STR_LIT>' <EOL> disk2 . save ( ) <EOL> self . assertEqual ( disk2 . name , '<STR_LIT>' , '<STR_LIT>' ) <EOL> def test_datastoreloses ( self ) : <EOL> """<STR_LIT>""" <EOL> disk = TestDisk ( ) <EOL> disk . name = '<STR_LIT>' <EOL> disk . save ( ) <EOL> disk2 = TestDisk ( disk . guid , datastore_wins = False ) <EOL> disk . name = '<STR_LIT>' <EOL> disk . save ( ) <EOL> disk2 . name = '<STR_LIT>' <EOL> disk2 . save ( ) <EOL> self . assertEqual ( disk2 . name , '<STR_LIT>' , '<STR_LIT>' ) <EOL> def test_silentdatarefresh ( self ) : <EOL> """<STR_LIT>""" <EOL> disk = TestDisk ( ) <EOL> disk . name = '<STR_LIT>' <EOL> disk . save ( ) <EOL> disk2 = TestDisk ( disk . guid , datastore_wins = False ) <EOL> disk . name = '<STR_LIT>' <EOL> disk . save ( ) <EOL> disk2 . name = '<STR_LIT>' <EOL> disk2 . save ( ) <EOL> disk . save ( ) <EOL> self . assertEqual ( disk2 . name , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( disk . name , '<STR_LIT>' , '<STR_LIT>' ) <EOL> def test_datastoreraises ( self ) : <EOL> """<STR_LIT>""" <EOL> disk = TestDisk ( ) <EOL> disk . name = '<STR_LIT>' <EOL> disk . save ( ) <EOL> disk2 = TestDisk ( disk . guid , datastore_wins = None ) <EOL> disk . name = '<STR_LIT>' <EOL> disk . save ( ) <EOL> disk2 . name = '<STR_LIT>' <EOL> self . assertRaises ( ConcurrencyException , disk2 . save ) <EOL> def test_volatileproperty ( self ) : <EOL> """<STR_LIT>""" <EOL> disk = TestDisk ( ) <EOL> disk . size = <NUM_LIT> <EOL> value = disk . used_size <EOL> time . sleep ( <NUM_LIT:2> ) <EOL> self . assertEqual ( disk . used_size , value , '<STR_LIT>' ) <EOL> time . sleep ( <NUM_LIT:2> ) <EOL> self . assertEqual ( disk . used_size , value , '<STR_LIT>' ) <EOL> time . sleep ( <NUM_LIT:2> ) <EOL> self . assertNotEqual ( disk . used_size , value , '<STR_LIT>' ) <EOL> def test_persistency ( self ) : <EOL> """<STR_LIT>""" <EOL> disk = TestDisk ( ) <EOL> disk . name = '<STR_LIT:test>' <EOL> disk . save ( ) <EOL> disk2 = TestDisk ( disk . guid ) <EOL> self . assertFalse ( disk2 . _metadata [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> disk3 = TestDisk ( disk . guid ) <EOL> self . assertTrue ( disk3 . _metadata [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> DummyVolatileStore ( ) . delete ( disk . _key ) <EOL> disk4 = TestDisk ( disk . guid ) <EOL> self . assertFalse ( disk4 . _metadata [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> def test_queries ( self ) : <EOL> """<STR_LIT>""" <EOL> machine = TestMachine ( ) <EOL> machine . name = '<STR_LIT>' <EOL> machine . save ( ) <EOL> for i in xrange ( <NUM_LIT:0> , <NUM_LIT:20> ) : <EOL> disk = TestDisk ( ) <EOL> disk . name = '<STR_LIT>' . format ( i ) <EOL> disk . size = i <EOL> if i < <NUM_LIT:10> : <EOL> disk . machine = machine <EOL> else : <EOL> disk . storage = machine <EOL> disk . save ( ) <EOL> self . assertEqual ( len ( machine . disks ) , <NUM_LIT:10> , '<STR_LIT>' ) <EOL> list_1 = DataList ( TestDisk , { '<STR_LIT:type>' : DataList . where_operator . AND , <EOL> '<STR_LIT>' : [ ( '<STR_LIT:size>' , DataList . operator . EQUALS , <NUM_LIT:1> ) ] } ) <EOL> self . assertEqual ( len ( list_1 ) , <NUM_LIT:1> , '<STR_LIT>' ) <EOL> list_2 = DataList ( TestDisk , { '<STR_LIT:type>' : DataList . where_operator . AND , <EOL> '<STR_LIT>' : [ ( '<STR_LIT:size>' , DataList . operator . EQUALS , <NUM_LIT:1> ) ] } ) <EOL> found_object = list_2 [ <NUM_LIT:0> ] <EOL> self . assertEqual ( found_object . name , '<STR_LIT>' , '<STR_LIT>' ) <EOL> list_3 = DataList ( TestDisk , { '<STR_LIT:type>' : DataList . where_operator . AND , <EOL> '<STR_LIT>' : [ ( '<STR_LIT:size>' , DataList . operator . GT , <NUM_LIT:3> ) , <EOL> ( '<STR_LIT:size>' , DataList . operator . LT , <NUM_LIT:6> ) ] } ) <EOL> self . assertEqual ( len ( list_3 ) , <NUM_LIT:2> , '<STR_LIT>' ) <EOL> list_4 = DataList ( TestDisk , { '<STR_LIT:type>' : DataList . where_operator . OR , <EOL> '<STR_LIT>' : [ ( '<STR_LIT:size>' , DataList . operator . LT , <NUM_LIT:3> ) , <EOL> ( '<STR_LIT:size>' , DataList . operator . GT , <NUM_LIT:6> ) ] } ) <EOL> self . assertGreaterEqual ( len ( list_4 ) , <NUM_LIT:16> , '<STR_LIT>' ) <EOL> list_5 = DataList ( TestDisk , { '<STR_LIT:type>' : DataList . where_operator . AND , <EOL> '<STR_LIT>' : [ ( '<STR_LIT>' , DataList . operator . EQUALS , machine . guid ) , <EOL> { '<STR_LIT:type>' : DataList . where_operator . OR , <EOL> '<STR_LIT>' : [ ( '<STR_LIT:size>' , DataList . operator . LT , <NUM_LIT:3> ) , <EOL> ( '<STR_LIT:size>' , DataList . operator . GT , <NUM_LIT:6> ) ] } ] } ) <EOL> self . assertEqual ( len ( list_5 ) , <NUM_LIT:6> , '<STR_LIT>' ) <EOL> list_6 = DataList ( TestDisk , { '<STR_LIT:type>' : DataList . where_operator . AND , <EOL> '<STR_LIT>' : [ ( '<STR_LIT:size>' , DataList . operator . LT , <NUM_LIT:3> ) , <EOL> ( '<STR_LIT:size>' , DataList . operator . GT , <NUM_LIT:6> ) ] } ) <EOL> self . assertEqual ( len ( list_6 ) , <NUM_LIT:0> , '<STR_LIT>' ) <EOL> list_7 = DataList ( TestDisk , { '<STR_LIT:type>' : DataList . where_operator . OR , <EOL> '<STR_LIT>' : [ ( '<STR_LIT>' , DataList . operator . EQUALS , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , DataList . operator . EQUALS , - <NUM_LIT:1> ) , <EOL> { '<STR_LIT:type>' : DataList . where_operator . AND , <EOL> '<STR_LIT>' : [ ( '<STR_LIT:size>' , DataList . operator . GT , <NUM_LIT:3> ) , <EOL> ( '<STR_LIT:size>' , DataList . operator . LT , <NUM_LIT:6> ) ] } ] } ) <EOL> self . assertEqual ( len ( list_7 ) , <NUM_LIT:2> , '<STR_LIT>' ) <EOL> list_8 = DataList ( TestDisk , { '<STR_LIT:type>' : DataList . where_operator . AND , <EOL> '<STR_LIT>' : [ ( '<STR_LIT>' , DataList . operator . EQUALS , '<STR_LIT>' ) , <EOL> ( '<STR_LIT:name>' , DataList . operator . EQUALS , '<STR_LIT>' ) ] } ) <EOL> self . assertEqual ( len ( list_8 ) , <NUM_LIT:1> , '<STR_LIT>' ) <EOL> list_9 = DataList ( TestDisk , { '<STR_LIT:type>' : DataList . where_operator . AND , <EOL> '<STR_LIT>' : [ ( '<STR_LIT:size>' , DataList . operator . GT , <NUM_LIT:3> ) , <EOL> { '<STR_LIT:type>' : DataList . where_operator . AND , <EOL> '<STR_LIT>' : [ ( '<STR_LIT:size>' , DataList . operator . LT , <NUM_LIT:6> ) ] } ] } ) <EOL> self . assertEqual ( len ( list_9 ) , <NUM_LIT:2> , '<STR_LIT>' ) <EOL> list_10 = DataList ( TestDisk , { '<STR_LIT:type>' : DataList . where_operator . OR , <EOL> '<STR_LIT>' : [ ( '<STR_LIT:size>' , DataList . operator . LT , <NUM_LIT:3> ) , <EOL> { '<STR_LIT:type>' : DataList . where_operator . OR , <EOL> '<STR_LIT>' : [ ( '<STR_LIT:size>' , DataList . operator . GT , <NUM_LIT:6> ) ] } ] } ) <EOL> self . assertGreaterEqual ( len ( list_10 ) , <NUM_LIT:16> , '<STR_LIT>' ) <EOL> list_11 = DataList ( TestDisk , { '<STR_LIT:type>' : DataList . where_operator . AND , <EOL> '<STR_LIT>' : [ ( '<STR_LIT>' , DataList . operator . EQUALS , '<STR_LIT>' ) ] } ) <EOL> self . assertEqual ( len ( list_11 ) , <NUM_LIT:10> , '<STR_LIT>' ) <EOL> list_12 = DataList ( TestDisk , { '<STR_LIT:type>' : DataList . where_operator . AND , <EOL> '<STR_LIT>' : [ ( '<STR_LIT:name>' , DataList . operator . EQUALS , '<STR_LIT>' ) ] } ) <EOL> self . assertEqual ( len ( list_12 ) , <NUM_LIT:1> , '<STR_LIT>' ) <EOL> list_13 = DataList ( TestDisk , { '<STR_LIT:type>' : DataList . where_operator . AND , <EOL> '<STR_LIT>' : [ ( '<STR_LIT:name>' , DataList . operator . EQUALS , '<STR_LIT>' , False ) ] } ) <EOL> self . assertEqual ( len ( list_13 ) , <NUM_LIT:1> , '<STR_LIT>' ) <EOL> list_14 = DataList ( TestDisk , { '<STR_LIT:type>' : DataList . where_operator . AND , <EOL> '<STR_LIT>' : [ ( '<STR_LIT:name>' , DataList . operator . EQUALS , '<STR_LIT>' ) ] } ) <EOL> self . assertEqual ( len ( list_14 ) , <NUM_LIT:0> , '<STR_LIT>' ) <EOL> def test_invalidpropertyassignment ( self ) : <EOL> """<STR_LIT>""" <EOL> disk = TestDisk ( ) <EOL> disk . size = <NUM_LIT:100> <EOL> with self . assertRaises ( TypeError ) : <EOL> disk . machine = TestDisk ( ) <EOL> def test_recursive ( self ) : <EOL> """<STR_LIT>""" <EOL> machine = TestMachine ( ) <EOL> machine . name = '<STR_LIT>' <EOL> machine . save ( ) <EOL> disks = [ ] <EOL> for i in xrange ( <NUM_LIT:0> , <NUM_LIT:10> ) : <EOL> disk = TestDisk ( ) <EOL> disk . name = '<STR_LIT>' . format ( i ) <EOL> if i % <NUM_LIT:2> : <EOL> disk . machine = machine <EOL> else : <EOL> disk . machine = machine <EOL> self . assertEqual ( disk . machine . name , '<STR_LIT>' , '<STR_LIT>' ) <EOL> disk . machine = None <EOL> self . assertIsNone ( disk . machine , '<STR_LIT>' ) <EOL> disks . append ( disk ) <EOL> disk . save ( ) <EOL> counter = <NUM_LIT:1> <EOL> for disk in machine . disks : <EOL> disk . size = counter <EOL> counter += <NUM_LIT:1> <EOL> machine . save ( recursive = True ) <EOL> disk = TestDisk ( machine . disks [ <NUM_LIT:0> ] . guid ) <EOL> self . assertEqual ( disk . size , <NUM_LIT:1> , '<STR_LIT>' ) <EOL> disk . machine . name = '<STR_LIT>' <EOL> disk . save ( recursive = True ) <EOL> machine2 = TestMachine ( machine . guid ) <EOL> self . assertEqual ( machine2 . disks [ <NUM_LIT:1> ] . size , <NUM_LIT:2> , '<STR_LIT>' ) <EOL> self . assertEqual ( machine2 . name , '<STR_LIT>' , '<STR_LIT>' ) <EOL> def test_descriptors ( self ) : <EOL> """<STR_LIT>""" <EOL> with self . assertRaises ( RuntimeError ) : <EOL> _ = Descriptor ( ) . descriptor <EOL> with self . assertRaises ( RuntimeError ) : <EOL> _ = Descriptor ( ) . get_object ( ) <EOL> def test_relationcache ( self ) : <EOL> """<STR_LIT>""" <EOL> machine = TestMachine ( ) <EOL> machine . name = '<STR_LIT>' <EOL> machine . save ( ) <EOL> disk1 = TestDisk ( ) <EOL> disk1 . name = '<STR_LIT>' <EOL> disk1 . save ( ) <EOL> disk2 = TestDisk ( ) <EOL> disk2 . name = '<STR_LIT>' <EOL> disk2 . save ( ) <EOL> disk3 = TestDisk ( ) <EOL> disk3 . name = '<STR_LIT>' <EOL> disk3 . save ( ) <EOL> self . assertEqual ( len ( machine . disks ) , <NUM_LIT:0> , '<STR_LIT>' ) <EOL> disk1 . machine = machine <EOL> disk1 . save ( ) <EOL> self . assertEqual ( len ( machine . disks ) , <NUM_LIT:1> , '<STR_LIT>' ) <EOL> disk2 . machine = machine <EOL> disk2 . save ( ) <EOL> self . assertEqual ( len ( machine . disks ) , <NUM_LIT:2> , '<STR_LIT>' ) <EOL> disk3 . machine = machine <EOL> disk3 . save ( ) <EOL> self . assertEqual ( len ( machine . disks ) , <NUM_LIT:3> , '<STR_LIT>' ) <EOL> machine . disks [ <NUM_LIT:0> ] . name = '<STR_LIT>' <EOL> machine . disks [ <NUM_LIT:1> ] . name = '<STR_LIT>' <EOL> machine . disks [ <NUM_LIT:2> ] . name = '<STR_LIT>' <EOL> disk1 . machine = None <EOL> disk1 . save ( ) <EOL> disk2 . machine = None <EOL> disk2 . save ( ) <EOL> self . assertEqual ( len ( machine . disks ) , <NUM_LIT:1> , '<STR_LIT>' ) <EOL> def test_datalistactions ( self ) : <EOL> """<STR_LIT>""" <EOL> machine = TestMachine ( ) <EOL> machine . name = '<STR_LIT>' <EOL> machine . save ( ) <EOL> sizes = [ <NUM_LIT:7> , <NUM_LIT:2> , <NUM_LIT:0> , <NUM_LIT:4> , <NUM_LIT:6> , <NUM_LIT:1> , <NUM_LIT:5> , <NUM_LIT:0> , <NUM_LIT:3> , <NUM_LIT:8> ] <EOL> guids = [ ] <EOL> disks = [ ] <EOL> disk = None <EOL> for i in xrange ( <NUM_LIT:0> , <NUM_LIT:10> ) : <EOL> disk = TestDisk ( ) <EOL> disk . name = '<STR_LIT>' . format ( i ) <EOL> disk . size = sizes [ i ] <EOL> disk . machine = machine <EOL> disk . save ( ) <EOL> disks . append ( disk ) <EOL> guids . append ( disk . guid ) <EOL> self . assertEqual ( machine . disks . count ( disk ) , <NUM_LIT:1> , '<STR_LIT>' ) <EOL> self . assertGreaterEqual ( machine . disks . index ( disk ) , <NUM_LIT:0> , '<STR_LIT>' ) <EOL> machine . disks . sort ( ) <EOL> guids . sort ( ) <EOL> self . assertEqual ( machine . disks [ <NUM_LIT:0> ] . guid , guids [ <NUM_LIT:0> ] , '<STR_LIT>' ) <EOL> machine . disks . reverse ( ) <EOL> self . assertEqual ( machine . disks [ - <NUM_LIT:1> ] . guid , guids [ <NUM_LIT:0> ] , '<STR_LIT>' ) <EOL> machine . disks . sort ( ) <EOL> self . assertEqual ( machine . disks [ <NUM_LIT:0> ] . guid , guids [ <NUM_LIT:0> ] , '<STR_LIT>' ) <EOL> disks = DataList ( TestDisk , { '<STR_LIT:type>' : DataList . where_operator . AND , '<STR_LIT>' : [ ] } ) <EOL> disks . sort ( key = lambda a : a . size ) <EOL> self . assertEqual ( disks [ <NUM_LIT:0> ] . size , <NUM_LIT:0> , '<STR_LIT>' ) <EOL> self . assertEqual ( disks [ <NUM_LIT:4> ] . size , <NUM_LIT:3> , '<STR_LIT>' ) <EOL> disks . sort ( key = lambda a : a . name ) <EOL> filtered = disks [ <NUM_LIT:1> : <NUM_LIT:4> ] <EOL> self . assertEqual ( filtered [ <NUM_LIT:0> ] . name , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( filtered [ <NUM_LIT:2> ] . name , '<STR_LIT>' , '<STR_LIT>' ) <EOL> fields = [ ( '<STR_LIT:name>' , True ) , ( '<STR_LIT:size>' , False ) ] <EOL> for field in fields : <EOL> disks . sort ( key = lambda a : Toolbox . extract_key ( a , field [ <NUM_LIT:0> ] ) , reverse = field [ <NUM_LIT:1> ] ) <EOL> self . assertEqual ( disks [ <NUM_LIT:0> ] . size , <NUM_LIT:0> , '<STR_LIT>' ) <EOL> self . assertEqual ( disks [ <NUM_LIT:1> ] . size , <NUM_LIT:0> , '<STR_LIT>' ) <EOL> self . assertEqual ( disks [ <NUM_LIT:0> ] . name , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( disks [ <NUM_LIT:1> ] . name , '<STR_LIT>' , '<STR_LIT>' ) <EOL> fields = [ ( '<STR_LIT:name>' , False ) , ( '<STR_LIT>' , False ) ] <EOL> for field in fields : <EOL> disks . sort ( key = lambda a : Toolbox . extract_key ( a , field [ <NUM_LIT:0> ] ) , reverse = field [ <NUM_LIT:1> ] ) <EOL> self . assertEqual ( disks [ <NUM_LIT:0> ] . predictable , <NUM_LIT:0> , '<STR_LIT>' ) <EOL> self . assertEqual ( disks [ <NUM_LIT:1> ] . predictable , <NUM_LIT:0> , '<STR_LIT>' ) <EOL> self . assertEqual ( disks [ <NUM_LIT:2> ] . predictable , <NUM_LIT:1> , '<STR_LIT>' ) <EOL> self . assertEqual ( disks [ <NUM_LIT:0> ] . name , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( disks [ <NUM_LIT:1> ] . name , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( disks [ <NUM_LIT:2> ] . name , '<STR_LIT>' , '<STR_LIT>' ) <EOL> def test_listcache ( self ) : <EOL> """<STR_LIT>""" <EOL> keys = [ '<STR_LIT>' , None ] <EOL> for key in keys : <EOL> disk0 = TestDisk ( ) <EOL> disk0 . name = '<STR_LIT>' <EOL> disk0 . save ( ) <EOL> list_cache = DataList ( TestDisk , key = key , <EOL> query = { '<STR_LIT:type>' : DataList . where_operator . AND , <EOL> '<STR_LIT>' : [ ( '<STR_LIT>' , DataList . operator . EQUALS , '<STR_LIT>' ) ] } ) <EOL> list_cache . _execute_query ( ) <EOL> self . assertFalse ( list_cache . from_cache , '<STR_LIT>' . format ( key ) ) <EOL> self . assertEqual ( len ( list_cache ) , <NUM_LIT:0> , '<STR_LIT>' . format ( key ) ) <EOL> machine = TestMachine ( ) <EOL> machine . name = '<STR_LIT>' <EOL> machine . save ( ) <EOL> disk1 = TestDisk ( ) <EOL> disk1 . name = '<STR_LIT>' <EOL> disk1 . machine = machine <EOL> disk1 . save ( ) <EOL> list_cache = DataList ( TestDisk , key = key , <EOL> query = { '<STR_LIT:type>' : DataList . where_operator . AND , <EOL> '<STR_LIT>' : [ ( '<STR_LIT>' , DataList . operator . EQUALS , '<STR_LIT>' ) ] } ) <EOL> list_cache . _execute_query ( ) <EOL> self . assertFalse ( list_cache . from_cache , '<STR_LIT>' . format ( key ) ) <EOL> self . assertEqual ( len ( list_cache ) , <NUM_LIT:1> , '<STR_LIT>' . format ( key ) ) <EOL> list_cache = DataList ( TestDisk , key = key , <EOL> query = { '<STR_LIT:type>' : DataList . where_operator . AND , <EOL> '<STR_LIT>' : [ ( '<STR_LIT>' , DataList . operator . EQUALS , '<STR_LIT>' ) ] } ) <EOL> list_cache . _execute_query ( ) <EOL> self . assertTrue ( list_cache . from_cache , '<STR_LIT>' . format ( key ) ) <EOL> disk2 = TestDisk ( ) <EOL> disk2 . machine = machine <EOL> disk2 . name = '<STR_LIT>' <EOL> disk2 . save ( ) <EOL> list_cache = DataList ( TestDisk , key = key , <EOL> query = { '<STR_LIT:type>' : DataList . where_operator . AND , <EOL> '<STR_LIT>' : [ ( '<STR_LIT>' , DataList . operator . EQUALS , '<STR_LIT>' ) ] } ) <EOL> list_cache . _execute_query ( ) <EOL> self . assertFalse ( list_cache . from_cache , '<STR_LIT>' . format ( key ) ) <EOL> self . assertEqual ( len ( list_cache ) , <NUM_LIT:2> , '<STR_LIT>' . format ( key ) ) <EOL> machine . name = '<STR_LIT:x>' <EOL> machine . save ( ) <EOL> list_cache = DataList ( TestDisk , key = key , <EOL> query = { '<STR_LIT:type>' : DataList . where_operator . AND , <EOL> '<STR_LIT>' : [ ( '<STR_LIT>' , DataList . operator . EQUALS , '<STR_LIT>' ) ] } ) <EOL> list_cache . _execute_query ( ) <EOL> self . assertFalse ( list_cache . from_cache , '<STR_LIT>' . format ( key ) ) <EOL> self . assertEqual ( len ( list_cache ) , <NUM_LIT:0> , '<STR_LIT>' . format ( key ) ) <EOL> def test_emptyquery ( self ) : <EOL> """<STR_LIT>""" <EOL> amount = DataList ( TestDisk , { '<STR_LIT:type>' : DataList . where_operator . AND , <EOL> '<STR_LIT>' : [ ( '<STR_LIT>' , DataList . operator . EQUALS , '<STR_LIT>' ) ] } ) <EOL> self . assertEqual ( len ( amount ) , <NUM_LIT:0> , '<STR_LIT>' ) <EOL> def test_nofilterquery ( self ) : <EOL> """<STR_LIT>""" <EOL> disk1 = TestDisk ( ) <EOL> disk1 . name = '<STR_LIT>' <EOL> disk1 . save ( ) <EOL> disk2 = TestDisk ( ) <EOL> disk2 . name = '<STR_LIT>' <EOL> disk2 . save ( ) <EOL> amount = DataList ( TestDisk , key = '<STR_LIT>' , <EOL> query = { '<STR_LIT:type>' : DataList . where_operator . AND , <EOL> '<STR_LIT>' : [ ] } ) <EOL> self . assertEqual ( len ( amount ) , <NUM_LIT:2> , '<STR_LIT>' . format ( amount ) ) <EOL> disk3 = TestDisk ( ) <EOL> disk3 . name = '<STR_LIT>' <EOL> disk3 . save ( ) <EOL> amount = DataList ( TestDisk , key = '<STR_LIT>' , <EOL> query = { '<STR_LIT:type>' : DataList . where_operator . AND , <EOL> '<STR_LIT>' : [ ] } ) <EOL> self . assertEqual ( len ( amount ) , <NUM_LIT:3> , '<STR_LIT>' . format ( amount ) ) <EOL> def test_invalidqueries ( self ) : <EOL> """<STR_LIT>""" <EOL> machine = TestMachine ( ) <EOL> machine . name = '<STR_LIT>' <EOL> machine . save ( ) <EOL> disk = TestDisk ( ) <EOL> disk . name = '<STR_LIT>' <EOL> disk . machine = machine <EOL> disk . save ( ) <EOL> setattr ( DataList . where_operator , '<STR_LIT>' , '<STR_LIT>' ) <EOL> with self . assertRaises ( NotImplementedError ) : <EOL> dlist = DataList ( TestDisk , { '<STR_LIT:type>' : DataList . where_operator . SOMETHING , <EOL> '<STR_LIT>' : [ ( '<STR_LIT>' , DataList . operator . EQUALS , '<STR_LIT>' ) ] } ) <EOL> dlist . _execute_query ( ) <EOL> with self . assertRaises ( NotImplementedError ) : <EOL> dlist = DataList ( TestDisk , { '<STR_LIT:type>' : DataList . where_operator . AND , <EOL> '<STR_LIT>' : [ { '<STR_LIT:type>' : DataList . where_operator . SOMETHING , <EOL> '<STR_LIT>' : [ ( '<STR_LIT>' , DataList . operator . EQUALS , '<STR_LIT>' ) ] } ] } ) <EOL> dlist . _execute_query ( ) <EOL> setattr ( DataList . operator , '<STR_LIT>' , '<STR_LIT>' ) <EOL> with self . assertRaises ( NotImplementedError ) : <EOL> dlist = DataList ( TestDisk , { '<STR_LIT:type>' : DataList . where_operator . AND , <EOL> '<STR_LIT>' : [ ( '<STR_LIT>' , DataList . operator . SOMETHING , '<STR_LIT>' ) ] } ) <EOL> dlist . _execute_query ( ) <EOL> def test_clearedcache ( self ) : <EOL> """<STR_LIT>""" <EOL> disk = TestDisk ( ) <EOL> disk . name = '<STR_LIT>' <EOL> disk . save ( ) <EOL> VolatileFactory . store . delete ( disk . _key ) <EOL> disk2 = TestDisk ( disk . guid ) <EOL> self . assertEqual ( disk2 . name , '<STR_LIT>' , '<STR_LIT>' ) <EOL> def test_serialization ( self ) : <EOL> """<STR_LIT>""" <EOL> machine = TestMachine ( ) <EOL> machine . name = '<STR_LIT>' <EOL> machine . save ( ) <EOL> disk = TestDisk ( ) <EOL> disk . name = '<STR_LIT>' <EOL> disk . machine = machine <EOL> disk . save ( ) <EOL> dictionary = disk . serialize ( ) <EOL> self . assertIn ( '<STR_LIT:name>' , dictionary , '<STR_LIT>' ) <EOL> self . assertEqual ( dictionary [ '<STR_LIT:name>' ] , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertIn ( '<STR_LIT>' , dictionary , '<STR_LIT>' ) <EOL> self . assertEqual ( dictionary [ '<STR_LIT>' ] , machine . guid , <EOL> '<STR_LIT>' ) <EOL> dictionary = disk . serialize ( depth = <NUM_LIT:1> ) <EOL> self . assertIn ( '<STR_LIT>' , dictionary , '<STR_LIT>' ) <EOL> self . assertEqual ( dictionary [ '<STR_LIT>' ] [ '<STR_LIT:name>' ] , '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> def test_volatiemutex ( self ) : <EOL> """<STR_LIT>""" <EOL> mutex = VolatileMutex ( '<STR_LIT:test>' ) <EOL> mutex . acquire ( ) <EOL> mutex . acquire ( ) <EOL> mutex . release ( ) <EOL> mutex . release ( ) <EOL> mutex . _volatile . add ( mutex . key ( ) , <NUM_LIT:1> , <NUM_LIT:10> ) <EOL> with self . assertRaises ( NoLockAvailableException ) : <EOL> mutex . acquire ( wait = <NUM_LIT:1> ) <EOL> mutex . _volatile . delete ( mutex . key ( ) ) <EOL> mutex . acquire ( ) <EOL> time . sleep ( <NUM_LIT:0.5> ) <EOL> mutex . release ( ) <EOL> def test_typesafety ( self ) : <EOL> """<STR_LIT>""" <EOL> disk = TestDisk ( ) <EOL> disk . name = '<STR_LIT:test>' <EOL> disk . name = u'<STR_LIT:test>' <EOL> disk . name = None <EOL> disk . size = <NUM_LIT:100> <EOL> disk . size = <NUM_LIT> <EOL> disk . order = <NUM_LIT:100> <EOL> with self . assertRaises ( TypeError ) : <EOL> disk . order = <NUM_LIT> <EOL> with self . assertRaises ( TypeError ) : <EOL> disk . __dict__ [ '<STR_LIT>' ] = None <EOL> disk . wrong_type_data = '<STR_LIT:string>' <EOL> _ = disk . wrong_type <EOL> with self . assertRaises ( TypeError ) : <EOL> disk . type = '<STR_LIT>' <EOL> disk . type = '<STR_LIT>' <EOL> def test_ownrelations ( self ) : <EOL> """<STR_LIT>""" <EOL> pdisk = TestDisk ( ) <EOL> pdisk . name = '<STR_LIT>' <EOL> pdisk . save ( ) <EOL> cdisk1 = TestDisk ( ) <EOL> cdisk1 . name = '<STR_LIT>' <EOL> cdisk1 . size = <NUM_LIT:100> <EOL> cdisk1 . parent = pdisk <EOL> cdisk1 . save ( ) <EOL> cdisk2 = TestDisk ( ) <EOL> cdisk2 . name = '<STR_LIT>' <EOL> cdisk2 . size = <NUM_LIT:100> <EOL> cdisk2 . parent = pdisk <EOL> cdisk2 . save ( ) <EOL> self . assertEqual ( len ( pdisk . children ) , <NUM_LIT:2> , '<STR_LIT>' . format ( len ( pdisk . children ) ) ) <EOL> self . assertEqual ( cdisk1 . parent . name , '<STR_LIT>' , '<STR_LIT>' ) <EOL> data = DataList ( TestDisk , { '<STR_LIT:type>' : DataList . where_operator . AND , <EOL> '<STR_LIT>' : [ ( '<STR_LIT>' , DataList . operator . EQUALS , '<STR_LIT>' ) ] } ) <EOL> self . assertEqual ( len ( data ) , <NUM_LIT:2> , '<STR_LIT>' . format ( len ( data ) ) ) <EOL> cdisk2 . parent = None <EOL> cdisk2 . save ( ) <EOL> data = DataList ( TestDisk , { '<STR_LIT:type>' : DataList . where_operator . AND , <EOL> '<STR_LIT>' : [ ( '<STR_LIT>' , DataList . operator . EQUALS , '<STR_LIT>' ) ] } ) <EOL> self . assertEqual ( len ( data ) , <NUM_LIT:1> , '<STR_LIT>' . format ( len ( data ) ) ) <EOL> def test_copy ( self ) : <EOL> """<STR_LIT>""" <EOL> machine = TestMachine ( ) <EOL> machine . name = '<STR_LIT>' <EOL> machine . save ( ) <EOL> disk1 = TestDisk ( ) <EOL> disk1 . name = '<STR_LIT>' <EOL> disk1 . size = <NUM_LIT:100> <EOL> disk1 . order = <NUM_LIT:1> <EOL> disk1 . type = '<STR_LIT>' <EOL> disk1 . machine = machine <EOL> disk1 . save ( ) <EOL> disk2 = TestDisk ( ) <EOL> disk2 . copy ( disk1 ) <EOL> self . assertEqual ( disk2 . name , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( disk2 . size , <NUM_LIT:100> , '<STR_LIT>' ) <EOL> self . assertEqual ( disk2 . order , <NUM_LIT:1> , '<STR_LIT>' ) <EOL> self . assertEqual ( disk2 . type , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( disk2 . machine , None , '<STR_LIT>' ) <EOL> disk3 = TestDisk ( ) <EOL> disk3 . copy ( disk1 , include_relations = True ) <EOL> self . assertEqual ( disk3 . machine . name , '<STR_LIT>' , '<STR_LIT>' ) <EOL> disk4 = TestDisk ( ) <EOL> disk4 . copy ( disk1 , include = [ '<STR_LIT:name>' ] ) <EOL> self . assertEqual ( disk4 . name , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( disk4 . size , <NUM_LIT:0> , '<STR_LIT>' ) <EOL> self . assertEqual ( disk4 . machine , None , '<STR_LIT>' ) <EOL> disk5 = TestDisk ( ) <EOL> disk5 . copy ( disk1 , exclude = [ '<STR_LIT:name>' ] ) <EOL> self . assertEqual ( disk5 . name , None , '<STR_LIT>' ) <EOL> self . assertEqual ( disk5 . size , <NUM_LIT:100> , '<STR_LIT>' ) <EOL> self . assertEqual ( disk5 . machine , None , '<STR_LIT>' ) <EOL> def test_querydynamic ( self ) : <EOL> """<STR_LIT>""" <EOL> def _get_disks ( ) : <EOL> return DataList ( TestDisk , { '<STR_LIT:type>' : DataList . where_operator . AND , <EOL> '<STR_LIT>' : [ ( '<STR_LIT>' , DataList . operator . NOT_EQUALS , - <NUM_LIT:1> ) ] } ) <EOL> disk1 = TestDisk ( ) <EOL> disk1 . name = '<STR_LIT>' <EOL> disk1 . size = <NUM_LIT:100> <EOL> disk1 . save ( ) <EOL> disk2 = TestDisk ( ) <EOL> disk2 . name = '<STR_LIT>' <EOL> disk2 . size = <NUM_LIT:100> <EOL> disk2 . save ( ) <EOL> query_result = _get_disks ( ) <EOL> self . assertEqual ( len ( query_result ) , <NUM_LIT:2> , '<STR_LIT>' . format ( len ( query_result ) ) ) <EOL> self . assertFalse ( query_result . from_cache , '<STR_LIT>' ) <EOL> query_result = _get_disks ( ) <EOL> query_result . _execute_query ( ) <EOL> self . assertFalse ( query_result . from_cache , '<STR_LIT>' ) <EOL> def test_delete_abandoning ( self ) : <EOL> """<STR_LIT>""" <EOL> machine = TestMachine ( ) <EOL> machine . name = '<STR_LIT>' <EOL> machine . save ( ) <EOL> disk_1 = TestDisk ( ) <EOL> disk_1 . name = '<STR_LIT>' <EOL> disk_1 . machine = machine <EOL> disk_1 . save ( ) <EOL> disk_2 = TestDisk ( ) <EOL> disk_2 . name = '<STR_LIT>' <EOL> disk_2 . machine = machine <EOL> disk_2 . save ( ) <EOL> self . assertRaises ( LinkedObjectException , machine . delete ) <EOL> disk_3 = TestDisk ( disk_1 . guid ) <EOL> self . assertIsNotNone ( disk_3 . machine , '<STR_LIT>' ) <EOL> _ = machine . disks <EOL> disk_2 . delete ( ) <EOL> machine . delete ( abandon = [ '<STR_LIT>' ] ) <EOL> disk_4 = TestDisk ( disk_1 . guid ) <EOL> self . assertIsNone ( disk_4 . machine , '<STR_LIT>' ) <EOL> def test_save_deleted ( self ) : <EOL> """<STR_LIT>""" <EOL> disk = TestDisk ( ) <EOL> disk . name = '<STR_LIT>' <EOL> disk . save ( ) <EOL> disk . delete ( ) <EOL> self . assertRaises ( ObjectNotFoundException , disk . save , '<STR_LIT>' ) <EOL> def test_itemchange_during_list_build ( self ) : <EOL> """<STR_LIT>""" <EOL> def _inject_new ( datalist_object ) : <EOL> """<STR_LIT>""" <EOL> _ = datalist_object <EOL> disk_x = TestDisk ( ) <EOL> disk_x . name = '<STR_LIT:test>' <EOL> disk_x . save ( ) <EOL> def _inject_delete ( datalist_object ) : <EOL> """<STR_LIT>""" <EOL> _ = datalist_object <EOL> disk_1 . delete ( ) <EOL> def _inject_update ( datalist_object ) : <EOL> """<STR_LIT>""" <EOL> _ = datalist_object <EOL> disk_2 . name = '<STR_LIT:x>' <EOL> disk_2 . save ( ) <EOL> disk_z = None <EOL> disk_1 = TestDisk ( ) <EOL> disk_1 . name = '<STR_LIT:test>' <EOL> disk_1 . save ( ) <EOL> disk_2 = TestDisk ( ) <EOL> disk_2 . name = '<STR_LIT:test>' <EOL> disk_2 . save ( ) <EOL> DataList . test_hooks [ '<STR_LIT>' ] = _inject_new <EOL> disks = DataList ( TestDisk , { '<STR_LIT:type>' : DataList . where_operator . AND , <EOL> '<STR_LIT>' : [ ( '<STR_LIT:name>' , DataList . operator . EQUALS , '<STR_LIT:test>' ) ] } ) <EOL> self . assertEqual ( len ( disks ) , <NUM_LIT:2> , '<STR_LIT>' . format ( len ( disks ) ) ) <EOL> del DataList . test_hooks [ '<STR_LIT>' ] <EOL> disks = DataList ( TestDisk , { '<STR_LIT:type>' : DataList . where_operator . AND , <EOL> '<STR_LIT>' : [ ( '<STR_LIT:name>' , DataList . operator . EQUALS , '<STR_LIT:test>' ) ] } ) <EOL> self . assertEqual ( len ( disks ) , <NUM_LIT:3> , '<STR_LIT>' . format ( len ( disks ) ) ) <EOL> VolatileFactory . store . delete ( '<STR_LIT>' ) <EOL> DataList . test_hooks [ '<STR_LIT>' ] = _inject_update <EOL> disks = DataList ( TestDisk , { '<STR_LIT:type>' : DataList . where_operator . AND , <EOL> '<STR_LIT>' : [ ( '<STR_LIT:name>' , DataList . operator . EQUALS , '<STR_LIT:test>' ) ] } ) <EOL> self . assertEqual ( len ( disks ) , <NUM_LIT:3> , '<STR_LIT>' . format ( len ( disks ) ) ) <EOL> del DataList . test_hooks [ '<STR_LIT>' ] <EOL> disks = DataList ( TestDisk , { '<STR_LIT:type>' : DataList . where_operator . AND , <EOL> '<STR_LIT>' : [ ( '<STR_LIT:name>' , DataList . operator . EQUALS , '<STR_LIT:test>' ) ] } ) <EOL> self . assertEqual ( len ( disks ) , <NUM_LIT:2> , '<STR_LIT>' . format ( len ( disks ) ) ) <EOL> VolatileFactory . store . delete ( '<STR_LIT>' ) <EOL> DataList . test_hooks [ '<STR_LIT>' ] = _inject_delete <EOL> disks = DataList ( TestDisk , { '<STR_LIT:type>' : DataList . where_operator . AND , <EOL> '<STR_LIT>' : [ ( '<STR_LIT:name>' , DataList . operator . EQUALS , '<STR_LIT:test>' ) ] } ) <EOL> self . assertEqual ( len ( disks ) , <NUM_LIT:2> , '<STR_LIT>' . format ( len ( disks ) ) ) <EOL> del DataList . test_hooks [ '<STR_LIT>' ] <EOL> disks = DataList ( TestDisk , { '<STR_LIT:type>' : DataList . where_operator . AND , <EOL> '<STR_LIT>' : [ ( '<STR_LIT:name>' , DataList . operator . EQUALS , '<STR_LIT:test>' ) ] } ) <EOL> self . assertEqual ( len ( disks ) , <NUM_LIT:1> , '<STR_LIT>' . format ( len ( disks ) ) ) <EOL> _ = disk_z <EOL> def test_guid_query ( self ) : <EOL> """<STR_LIT>""" <EOL> machine = TestMachine ( ) <EOL> machine . name = '<STR_LIT>' <EOL> machine . save ( ) <EOL> disk = TestDisk ( ) <EOL> disk . name = '<STR_LIT:test>' <EOL> disk . machine = machine <EOL> disk . save ( ) <EOL> disks = DataList ( TestDisk , { '<STR_LIT:type>' : DataList . where_operator . AND , <EOL> '<STR_LIT>' : [ ( '<STR_LIT>' , DataList . operator . EQUALS , machine . guid ) ] } ) <EOL> self . assertEqual ( len ( disks ) , <NUM_LIT:1> , '<STR_LIT>' . format ( len ( disks ) ) ) <EOL> def test_1_to_1 ( self ) : <EOL> """<STR_LIT>""" <EOL> machine = TestMachine ( ) <EOL> machine . name = '<STR_LIT>' <EOL> machine . save ( ) <EOL> self . assertIsNone ( machine . one , '<STR_LIT>' ) <EOL> self . assertIsNone ( machine . one_guid , '<STR_LIT>' ) <EOL> disk = TestDisk ( ) <EOL> disk . name = '<STR_LIT:test>' <EOL> disk . one = machine <EOL> disk . save ( ) <EOL> self . assertIsNotNone ( machine . one , '<STR_LIT>' ) <EOL> self . assertEqual ( machine . one . name , '<STR_LIT:test>' , '<STR_LIT>' ) <EOL> self . assertEqual ( disk . one . name , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( machine . one_guid , disk . guid , '<STR_LIT>' ) <EOL> with self . assertRaises ( RuntimeError ) : <EOL> machine . one = disk <EOL> def test_auto_inheritance ( self ) : <EOL> """<STR_LIT>""" <EOL> machine = TestMachine ( ) <EOL> self . assertEqual ( Descriptor ( machine . __class__ ) , Descriptor ( TestEMachine ) , '<STR_LIT>' ) <EOL> def test_relation_inheritance ( self ) : <EOL> """<STR_LIT>""" <EOL> machine = TestMachine ( ) <EOL> machine . name = '<STR_LIT>' <EOL> machine . save ( ) <EOL> disk = TestDisk ( ) <EOL> disk . name = '<STR_LIT>' <EOL> disk . machine = machine <EOL> disk . save ( ) <EOL> machine . the_disk = disk <EOL> machine . save ( ) <EOL> disk2 = TestDisk ( disk . guid ) <EOL> self . assertEqual ( Descriptor ( disk2 . machine . __class__ ) , Descriptor ( TestEMachine ) , '<STR_LIT>' ) <EOL> def test_extended_property ( self ) : <EOL> """<STR_LIT>""" <EOL> machine = TestEMachine ( ) <EOL> machine . name = '<STR_LIT>' <EOL> machine . extended = '<STR_LIT>' <EOL> machine . save ( ) <EOL> machine2 = TestEMachine ( machine . guid ) <EOL> self . assertEqual ( machine2 . name , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( machine2 . extended , '<STR_LIT>' , '<STR_LIT>' ) <EOL> def test_extended_filter ( self ) : <EOL> """<STR_LIT>""" <EOL> machine1 = TestMachine ( ) <EOL> machine1 . name = '<STR_LIT>' <EOL> machine1 . save ( ) <EOL> machine2 = TestEMachine ( ) <EOL> machine2 . name = '<STR_LIT>' <EOL> machine2 . save ( ) <EOL> data = DataList ( TestMachine , { '<STR_LIT:type>' : DataList . where_operator . AND , <EOL> '<STR_LIT>' : [ ] } ) <EOL> self . assertEqual ( len ( data ) , <NUM_LIT:2> , '<STR_LIT>' . format ( len ( data ) ) ) <EOL> data = DataList ( TestEMachine , { '<STR_LIT:type>' : DataList . where_operator . AND , <EOL> '<STR_LIT>' : [ ] } ) <EOL> self . assertEqual ( len ( data ) , <NUM_LIT:2> , '<STR_LIT>' . format ( len ( data ) ) ) <EOL> def test_mandatory_fields ( self ) : <EOL> """<STR_LIT>""" <EOL> machine = TestMachine ( ) <EOL> machine . extended = '<STR_LIT>' <EOL> machine . name = '<STR_LIT>' <EOL> machine . save ( ) <EOL> disk = TestDisk ( ) <EOL> [ _ for _ in disk . _relations if _ . name == '<STR_LIT>' ] [ <NUM_LIT:0> ] . mandatory = True <EOL> disk . name = None <EOL> with self . assertRaises ( MissingMandatoryFieldsException ) as exception : <EOL> disk . save ( ) <EOL> self . assertIn ( '<STR_LIT:name>' , exception . exception . message , '<STR_LIT>' . format ( exception . exception . message ) ) <EOL> self . assertIn ( '<STR_LIT>' , exception . exception . message , '<STR_LIT>' . format ( exception . exception . message ) ) <EOL> disk . name = '<STR_LIT>' <EOL> disk . machine = machine <EOL> disk . save ( ) <EOL> disk . description = '<STR_LIT:test>' <EOL> disk . storage = machine <EOL> disk . save ( ) <EOL> [ _ for _ in disk . _relations if _ . name == '<STR_LIT>' ] [ <NUM_LIT:0> ] . mandatory = False <EOL> def test_versioning ( self ) : <EOL> """<STR_LIT>""" <EOL> machine = TestMachine ( ) <EOL> machine . name = '<STR_LIT>' <EOL> machine . save ( ) <EOL> self . assertEqual ( machine . _data [ '<STR_LIT>' ] , <NUM_LIT:1> , '<STR_LIT>' . format ( machine . _data [ '<STR_LIT>' ] ) ) <EOL> machine . save ( ) <EOL> self . assertEqual ( machine . _data [ '<STR_LIT>' ] , <NUM_LIT:2> , '<STR_LIT>' . format ( machine . _data [ '<STR_LIT>' ] ) ) <EOL> machine_x = TestMachine ( machine . guid ) <EOL> machine_x . name = '<STR_LIT>' <EOL> machine_x . save ( ) <EOL> self . assertTrue ( machine . updated_on_datastore ( ) , '<STR_LIT>' ) <EOL> machine . name = '<STR_LIT>' <EOL> machine . save ( ) <EOL> self . assertEqual ( machine . _data [ '<STR_LIT>' ] , <NUM_LIT:4> , '<STR_LIT>' . format ( machine . _data [ '<STR_LIT>' ] ) ) <EOL> self . assertFalse ( machine . updated_on_datastore ( ) , '<STR_LIT>' ) <EOL> def test_outdated_listobjects ( self ) : <EOL> """<STR_LIT>""" <EOL> machine = TestMachine ( ) <EOL> machine . name = '<STR_LIT>' <EOL> machine . save ( ) <EOL> disk1 = TestDisk ( ) <EOL> disk1 . name = '<STR_LIT>' <EOL> disk1 . machine = machine <EOL> disk1 . save ( ) <EOL> disk2 = TestDisk ( ) <EOL> disk2 . name = '<STR_LIT>' <EOL> disk2 . machine = machine <EOL> disk2 . save ( ) <EOL> self . assertListEqual ( [ '<STR_LIT>' , '<STR_LIT>' ] , sorted ( [ disk . name for disk in machine . disks ] ) , '<STR_LIT>' ) <EOL> disk2 . name = '<STR_LIT>' <EOL> self . assertListEqual ( [ '<STR_LIT>' , '<STR_LIT>' ] , sorted ( [ disk . name for disk in machine . disks ] ) , '<STR_LIT>' ) <EOL> disk2 . save ( ) <EOL> self . assertListEqual ( [ '<STR_LIT>' , '<STR_LIT>' ] , sorted ( [ disk . name for disk in machine . disks ] ) , '<STR_LIT>' ) <EOL> def test_invalidonetoone ( self ) : <EOL> """<STR_LIT>""" <EOL> machine = TestMachine ( ) <EOL> machine . name = '<STR_LIT>' <EOL> machine . save ( ) <EOL> self . assertIsNone ( machine . one , '<STR_LIT>' ) <EOL> disk1 = TestDisk ( ) <EOL> disk1 . name = '<STR_LIT>' <EOL> disk1 . one = machine <EOL> disk1 . save ( ) <EOL> self . assertEqual ( machine . one , disk1 , '<STR_LIT>' ) <EOL> disk2 = TestDisk ( ) <EOL> disk2 . name = '<STR_LIT>' <EOL> disk2 . one = machine <EOL> disk2 . save ( ) <EOL> with self . assertRaises ( InvalidRelationException ) : <EOL> _ = machine . one <EOL> def test_object_cleanup ( self ) : <EOL> """<STR_LIT>""" <EOL> machine = TestMachine ( ) <EOL> machine . name = '<STR_LIT>' <EOL> machine . save ( ) <EOL> disk1 = TestDisk ( ) <EOL> disk1 . name = '<STR_LIT>' <EOL> disk1 . machine = machine <EOL> disk1 . save ( ) <EOL> disk2 = TestDisk ( ) <EOL> disk2 . name = '<STR_LIT>' <EOL> disk2 . machine = machine <EOL> disk2 . save ( ) <EOL> for disk in machine . disks : <EOL> disk . delete ( ) <EOL> machine . delete ( ) <EOL> def test_relation_set_build ( self ) : <EOL> """<STR_LIT>""" <EOL> machine = TestMachine ( ) <EOL> machine . name = '<STR_LIT>' <EOL> machine . save ( ) <EOL> disk1 = TestDisk ( ) <EOL> disk1 . name = '<STR_LIT>' <EOL> disk1 . machine = machine <EOL> disk1 . save ( ) <EOL> disk2 = TestDisk ( ) <EOL> disk2 . name = '<STR_LIT>' <EOL> disk2 . machine = machine <EOL> disk2 . save ( ) <EOL> datalist = DataList . get_relation_set ( TestDisk , '<STR_LIT>' , TestEMachine , '<STR_LIT>' , machine . guid ) <EOL> datalist . _execute_query ( ) <EOL> self . assertIsNone ( datalist . from_cache , '<STR_LIT>' ) <EOL> self . assertEqual ( len ( datalist ) , <NUM_LIT:2> , '<STR_LIT>' ) <EOL> for key in PersistentFactory . store . prefix ( '<STR_LIT>' . format ( machine . guid ) ) : <EOL> PersistentFactory . store . delete ( key ) <EOL> datalist = DataList . get_relation_set ( TestDisk , '<STR_LIT>' , TestEMachine , '<STR_LIT>' , machine . guid ) <EOL> datalist . _execute_query ( ) <EOL> self . assertIsNone ( datalist . from_cache , '<STR_LIT>' ) <EOL> self . assertEqual ( len ( datalist ) , <NUM_LIT:0> , '<STR_LIT>' ) <EOL> def test_instance_checks ( self ) : <EOL> """<STR_LIT>""" <EOL> machine = TestMachine ( ) <EOL> machine . name = '<STR_LIT>' <EOL> machine . save ( ) <EOL> disk = TestDisk ( ) <EOL> disk . name = '<STR_LIT>' <EOL> disk . machine = machine <EOL> disk . save ( ) <EOL> self . assertTrue ( Descriptor . isinstance ( disk , TestDisk ) , '<STR_LIT>' ) <EOL> self . assertFalse ( Descriptor . isinstance ( disk , TestMachine ) , '<STR_LIT>' ) <EOL> self . assertTrue ( Descriptor . isinstance ( disk . machine , TestEMachine ) , '<STR_LIT>' ) <EOL> def test_cache_and_save_racecondition ( self ) : <EOL> """<STR_LIT>""" <EOL> guid = None <EOL> def _update ( ) : <EOL> local_machine = TestMachine ( guid ) <EOL> local_machine . name = '<STR_LIT>' <EOL> local_machine . save ( ) <EOL> machine = TestMachine ( ) <EOL> self . assertIsNone ( machine . _metadata [ '<STR_LIT>' ] , "<STR_LIT>" ) <EOL> machine . name = '<STR_LIT>' <EOL> machine . save ( ) <EOL> guid = machine . guid <EOL> machine = TestMachine ( machine . guid , _hook = _update ) <EOL> self . assertEqual ( machine . name , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertFalse ( machine . _metadata [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> machine = TestMachine ( machine . guid ) <EOL> self . assertFalse ( machine . _metadata [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( machine . name , '<STR_LIT>' , '<STR_LIT>' ) <EOL> machine = TestMachine ( machine . guid ) <EOL> self . assertTrue ( machine . _metadata [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> def test_delete_during_object_load ( self ) : <EOL> """<STR_LIT>""" <EOL> guid = None <EOL> def _delete ( ) : <EOL> local_machine = TestMachine ( guid ) <EOL> local_machine . delete ( ) <EOL> machine = TestMachine ( ) <EOL> machine . name = '<STR_LIT>' <EOL> machine . save ( ) <EOL> guid = machine . guid <EOL> with self . assertRaises ( ObjectNotFoundException ) : <EOL> _ = TestMachine ( guid , _hook = _delete ) <EOL> def test_object_save_reverseindex_build ( self ) : <EOL> """<STR_LIT>""" <EOL> machine = TestMachine ( ) <EOL> machine . name = '<STR_LIT>' <EOL> machine . save ( ) <EOL> key = '<STR_LIT>' . format ( machine . guid ) <EOL> self . assertListEqual ( list ( PersistentFactory . store . prefix ( key ) ) , [ ] , '<STR_LIT>' ) <EOL> disk1 = TestDisk ( ) <EOL> disk1 . name = '<STR_LIT>' <EOL> disk1 . machine = machine <EOL> disk1 . save ( ) <EOL> disk2 = TestDisk ( ) <EOL> disk2 . name = '<STR_LIT>' <EOL> disk2 . machine = machine <EOL> disk2 . save ( ) <EOL> amount = len ( machine . disks ) <EOL> self . assertEqual ( amount , <NUM_LIT:2> , '<STR_LIT>' . format ( amount ) ) <EOL> def test_save_nonexisting_relation ( self ) : <EOL> """<STR_LIT>""" <EOL> machine = TestMachine ( ) <EOL> machine . name = '<STR_LIT>' <EOL> disk1 = TestDisk ( ) <EOL> disk1 . name = '<STR_LIT>' <EOL> disk1 . machine = machine <EOL> with self . assertRaises ( ObjectNotFoundException ) : <EOL> disk1 . save ( ) <EOL> machine . save ( ) <EOL> disk1 . save ( ) <EOL> def test_invalidate_dynamics ( self ) : <EOL> """<STR_LIT>""" <EOL> disk = TestDisk ( ) <EOL> disk . __dict__ [ '<STR_LIT>' ] = <NUM_LIT:0> <EOL> disk . dynamic_value = <NUM_LIT:0> <EOL> disk . name = '<STR_LIT:test>' <EOL> disk . save ( ) <EOL> value = disk . updatable <EOL> self . assertEqual ( value , <NUM_LIT:0> , '<STR_LIT>' ) <EOL> disk . dynamic_value = <NUM_LIT:5> <EOL> value = disk . updatable <EOL> self . assertEqual ( value , <NUM_LIT:0> , '<STR_LIT>' . format ( value ) ) <EOL> time . sleep ( <NUM_LIT:5> ) <EOL> value = disk . updatable <EOL> self . assertEqual ( value , <NUM_LIT:5> , '<STR_LIT>' . format ( value ) ) <EOL> disk . dynamic_value = <NUM_LIT:10> <EOL> value = disk . updatable <EOL> self . assertEqual ( value , <NUM_LIT:5> , '<STR_LIT>' . format ( value ) ) <EOL> disk . invalidate_dynamics ( [ '<STR_LIT>' ] ) <EOL> value = disk . updatable <EOL> self . assertEqual ( value , <NUM_LIT:10> , '<STR_LIT>' . format ( value ) ) <EOL> def test_enumerator ( self ) : <EOL> """<STR_LIT>""" <EOL> from ovs . dal . dataobject import DataObject <EOL> list_items = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> enum = DataObject . enumerator ( '<STR_LIT>' , list_items ) <EOL> self . assertEqual ( enum . __name__ , '<STR_LIT>' , '<STR_LIT>' ) <EOL> for item in list_items : <EOL> self . assertIn ( item , enum , '<STR_LIT>' . format ( item ) ) <EOL> self . assertTrue ( hasattr ( enum , item ) , '<STR_LIT>' . format ( item ) ) <EOL> self . assertNotIn ( '<STR_LIT>' , enum , '<STR_LIT>' ) <EOL> extract = [ item for item in enum ] <EOL> self . assertListEqual ( sorted ( list_items ) , sorted ( extract ) , '<STR_LIT>' ) <EOL> self . assertEqual ( enum . ONE , '<STR_LIT>' , '<STR_LIT>' ) <EOL> dict_items = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } <EOL> enum = DataObject . enumerator ( '<STR_LIT>' , dict_items ) <EOL> self . assertEqual ( enum . __name__ , '<STR_LIT>' , '<STR_LIT>' ) <EOL> for key , value in dict_items . iteritems ( ) : <EOL> self . assertIn ( key , enum , '<STR_LIT>' . format ( key ) ) <EOL> self . assertEqual ( getattr ( enum , key ) , value , "<STR_LIT>" . format ( key , value , getattr ( enum , key ) ) ) <EOL> def test_pop_and_remove_from_datalist ( self ) : <EOL> """<STR_LIT>""" <EOL> disk1 = TestDisk ( ) <EOL> disk2 = TestDisk ( ) <EOL> disk1 . name = '<STR_LIT>' <EOL> disk2 . name = '<STR_LIT>' <EOL> disk1 . save ( ) <EOL> disk2 . save ( ) <EOL> datalist1 = DataList ( TestDisk , { '<STR_LIT:type>' : DataList . where_operator . AND , <EOL> '<STR_LIT>' : [ ] } ) <EOL> self . assertEqual ( len ( datalist1 ) , <NUM_LIT:2> , '<STR_LIT>' ) <EOL> datalist1 . remove ( disk1 ) <EOL> self . assertEqual ( len ( datalist1 ) , <NUM_LIT:1> , '<STR_LIT>' ) <EOL> with self . assertRaises ( ValueError ) : <EOL> datalist1 . remove ( disk1 ) <EOL> datalist1 . remove ( disk2 . guid ) <EOL> self . assertEqual ( len ( datalist1 ) , <NUM_LIT:0> , '<STR_LIT>' ) <EOL> machine1 = TestMachine ( ) <EOL> machine1 . name = '<STR_LIT>' <EOL> machine1 . save ( ) <EOL> with self . assertRaises ( TypeError ) : <EOL> datalist1 . remove ( machine1 ) <EOL> datalist2 = DataList ( TestDisk , { '<STR_LIT:type>' : DataList . where_operator . AND , <EOL> '<STR_LIT>' : [ ] } ) <EOL> self . assertEqual ( len ( datalist2 ) , <NUM_LIT:2> , '<STR_LIT>' ) <EOL> with self . assertRaises ( IndexError ) : <EOL> datalist2 . pop ( <NUM_LIT:3> ) <EOL> datalist2 . pop ( - <NUM_LIT:3> ) <EOL> with self . assertRaises ( ValueError ) : <EOL> datalist2 . pop ( '<STR_LIT:test>' ) <EOL> datalist2 . pop ( <NUM_LIT:0> ) <EOL> datalist2 . pop ( - <NUM_LIT:1> ) <EOL> self . assertEqual ( len ( datalist2 ) , <NUM_LIT:0> , '<STR_LIT>' ) <EOL> with self . assertRaises ( IndexError ) : <EOL> datalist2 . pop ( <NUM_LIT:0> ) <EOL> def test_error_during_save ( self ) : <EOL> """<STR_LIT>""" <EOL> def _raise_error ( ) : <EOL> raise RuntimeError ( ) <EOL> machine = TestMachine ( ) <EOL> machine . name = '<STR_LIT>' <EOL> machine . save ( ) <EOL> disk1 = TestDisk ( ) <EOL> disk1 . name = '<STR_LIT>' <EOL> disk1 . machine = machine <EOL> disk1 . save ( ) <EOL> disk2 = TestDisk ( ) <EOL> disk2 . name = '<STR_LIT>' <EOL> disk2 . machine = machine <EOL> try : <EOL> disk2 . save ( _hook = _raise_error ) <EOL> except RuntimeError : <EOL> pass <EOL> self . assertEqual ( len ( machine . disks ) , <NUM_LIT:1> , '<STR_LIT>' ) <EOL> for disk in machine . disks : <EOL> disk . delete ( ) <EOL> machine . delete ( ) <EOL> def test_shuffle_object_list ( self ) : <EOL> """<STR_LIT>""" <EOL> for i in range ( <NUM_LIT:10> ) : <EOL> disk = TestDisk ( ) <EOL> disk . name = '<STR_LIT>' . format ( i ) <EOL> disk . save ( ) <EOL> datalist = DataList ( TestDisk , { '<STR_LIT:type>' : DataList . where_operator . AND , <EOL> '<STR_LIT>' : [ ] } ) <EOL> starting_order = [ disk . name for disk in datalist ] <EOL> datalist . shuffle ( ) <EOL> new_order = [ disk . name for disk in datalist ] <EOL> self . assertNotEqual ( starting_order , new_order , '<STR_LIT>' ) <EOL> self . assertEqual ( set ( starting_order ) , set ( new_order ) , '<STR_LIT>' ) <EOL> def test_volatile_objects ( self ) : <EOL> """<STR_LIT>""" <EOL> disk = TestDisk ( data = { '<STR_LIT:name>' : '<STR_LIT:test>' } , volatile = True ) <EOL> self . assertEqual ( disk . name , '<STR_LIT:test>' , '<STR_LIT>' ) <EOL> self . assertRaises ( VolatileObjectException , disk . save , '<STR_LIT>' ) <EOL> self . assertRaises ( VolatileObjectException , disk . delete , '<STR_LIT>' ) <EOL> def test_datalist_add ( self ) : <EOL> """<STR_LIT>""" <EOL> disk1 = TestDisk ( ) <EOL> disk1 . name = '<STR_LIT>' <EOL> disk1 . save ( ) <EOL> disk2 = TestDisk ( ) <EOL> disk2 . name = '<STR_LIT>' <EOL> disk2 . save ( ) <EOL> datalist1 = DataList ( TestDisk , { '<STR_LIT:type>' : DataList . where_operator . AND , <EOL> '<STR_LIT>' : [ ( '<STR_LIT:name>' , DataList . operator . EQUALS , '<STR_LIT>' ) ] } ) <EOL> datalist2 = DataList ( TestDisk , { '<STR_LIT:type>' : DataList . where_operator . AND , <EOL> '<STR_LIT>' : [ ( '<STR_LIT:name>' , DataList . operator . EQUALS , '<STR_LIT>' ) ] } ) <EOL> datalist3 = datalist1 + datalist2 <EOL> self . assertEqual ( len ( datalist1 ) , <NUM_LIT:1> , '<STR_LIT>' ) <EOL> self . assertEqual ( len ( datalist2 ) , <NUM_LIT:1> , '<STR_LIT>' ) <EOL> self . assertEqual ( len ( datalist3 ) , <NUM_LIT:2> , '<STR_LIT>' ) <EOL> def test_delete_retries ( self ) : <EOL> """<STR_LIT>""" <EOL> def _change_list ( ) : <EOL> if Basic . _executed is False : <EOL> PersistentFactory . store . set ( '<STR_LIT>' , { } ) <EOL> Basic . _executed = True <EOL> PersistentFactory . store . set ( '<STR_LIT>' , { '<STR_LIT:foo>' : '<STR_LIT:bar>' } ) <EOL> Basic . _executed = False <EOL> disk = TestDisk ( ) <EOL> disk . name = '<STR_LIT>' <EOL> disk . save ( ) <EOL> disk . delete ( _hook = _change_list ) <EOL> def test_object_comparison_on_different_level ( self ) : <EOL> """<STR_LIT>""" <EOL> sr = TestStorageRouter ( ) <EOL> sr . name = '<STR_LIT>' <EOL> sr . save ( ) <EOL> vpool = TestVPool ( ) <EOL> vpool . name = '<STR_LIT>' <EOL> vpool . save ( ) <EOL> sd = TestStorageDriver ( ) <EOL> sd . name = '<STR_LIT>' <EOL> sd . vpool = vpool <EOL> sd . storagerouter = sr <EOL> sd . save ( ) <EOL> error_message = '<STR_LIT>' <EOL> assert ( sr == sr ) is True , error_message <EOL> assert ( sr != sr ) is False , error_message <EOL> assert ( sd . storagerouter == sr ) is True , error_message <EOL> assert ( sd . storagerouter != sr ) is False , error_message <EOL> assert ( vpool . storagedrivers [ <NUM_LIT:0> ] . storagerouter == sr ) is True , error_message <EOL> assert ( vpool . storagedrivers [ <NUM_LIT:0> ] . storagerouter != sr ) is False , error_message <EOL> def test_racecondition_datalist_multiget ( self ) : <EOL> """<STR_LIT>""" <EOL> disk1 = TestDisk ( ) <EOL> disk1 . name = '<STR_LIT>' <EOL> disk1 . save ( ) <EOL> disk2 = TestDisk ( ) <EOL> disk2 . name = '<STR_LIT>' <EOL> disk2 . save ( ) <EOL> datalist = DataList ( TestDisk , { '<STR_LIT:type>' : DataList . where_operator . AND , <EOL> '<STR_LIT>' : [ ] } ) <EOL> self . assertEqual ( len ( datalist ) , <NUM_LIT:2> , '<STR_LIT>' ) <EOL> PersistentFactory . store . delete ( '<STR_LIT>' . format ( disk1 . guid ) ) <EOL> datalist = DataList ( TestDisk , { '<STR_LIT:type>' : DataList . where_operator . AND , <EOL> '<STR_LIT>' : [ ] } ) <EOL> self . assertEqual ( len ( datalist ) , <NUM_LIT:1> , '<STR_LIT>' ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> import unittest <EOL> suite = unittest . TestLoader ( ) . loadTestsFromTestCase ( Basic ) <EOL> result = not unittest . TextTestRunner ( verbosity = <NUM_LIT:2> ) . run ( suite ) . wasSuccessful ( ) <EOL> sys . exit ( result ) </s>
