<s> import uuid <EOL> import datetime <EOL> import bson <EOL> from st2tests import DbTestCase <EOL> from st2common . util import date as date_utils <EOL> from tests . unit . base import FakeModel , FakeModelDB <EOL> class TestPersistence ( DbTestCase ) : <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> super ( TestPersistence , cls ) . setUpClass ( ) <EOL> cls . access = FakeModel ( ) <EOL> def tearDown ( self ) : <EOL> FakeModelDB . drop_collection ( ) <EOL> super ( TestPersistence , self ) . tearDown ( ) <EOL> def test_crud ( self ) : <EOL> obj1 = FakeModelDB ( name = uuid . uuid4 ( ) . hex , context = { '<STR_LIT:a>' : <NUM_LIT:1> } ) <EOL> obj1 = self . access . add_or_update ( obj1 ) <EOL> obj2 = self . access . get ( name = obj1 . name ) <EOL> self . assertIsNotNone ( obj2 ) <EOL> self . assertEqual ( obj1 . id , obj2 . id ) <EOL> self . assertEqual ( obj1 . name , obj2 . name ) <EOL> self . assertDictEqual ( obj1 . context , obj2 . context ) <EOL> obj1 . name = uuid . uuid4 ( ) . hex <EOL> obj1 = self . access . add_or_update ( obj1 ) <EOL> obj2 = self . access . get ( name = obj1 . name ) <EOL> self . assertIsNotNone ( obj2 ) <EOL> self . assertEqual ( obj1 . id , obj2 . id ) <EOL> self . assertEqual ( obj1 . name , obj2 . name ) <EOL> self . assertDictEqual ( obj1 . context , obj2 . context ) <EOL> self . access . delete ( obj1 ) <EOL> obj2 = self . access . get ( name = obj1 . name ) <EOL> self . assertIsNone ( obj2 ) <EOL> def test_count ( self ) : <EOL> obj1 = FakeModelDB ( name = uuid . uuid4 ( ) . hex , context = { '<STR_LIT:user>' : '<STR_LIT>' } ) <EOL> obj1 = self . access . add_or_update ( obj1 ) <EOL> obj2 = FakeModelDB ( name = uuid . uuid4 ( ) . hex , context = { '<STR_LIT:user>' : '<STR_LIT>' } ) <EOL> obj2 = self . access . add_or_update ( obj2 ) <EOL> self . assertEqual ( self . access . count ( ) , <NUM_LIT:2> ) <EOL> def test_get_all ( self ) : <EOL> obj1 = FakeModelDB ( name = uuid . uuid4 ( ) . hex , context = { '<STR_LIT:user>' : '<STR_LIT>' } ) <EOL> obj1 = self . access . add_or_update ( obj1 ) <EOL> obj2 = FakeModelDB ( name = uuid . uuid4 ( ) . hex , context = { '<STR_LIT:user>' : '<STR_LIT>' } ) <EOL> obj2 = self . access . add_or_update ( obj2 ) <EOL> objs = self . access . get_all ( ) <EOL> self . assertIsNotNone ( objs ) <EOL> self . assertEqual ( len ( objs ) , <NUM_LIT:2> ) <EOL> self . assertListEqual ( list ( objs ) , [ obj1 , obj2 ] ) <EOL> def test_query_by_id ( self ) : <EOL> obj1 = FakeModelDB ( name = uuid . uuid4 ( ) . hex , context = { '<STR_LIT:user>' : '<STR_LIT>' } ) <EOL> obj1 = self . access . add_or_update ( obj1 ) <EOL> obj2 = self . access . get_by_id ( str ( obj1 . id ) ) <EOL> self . assertIsNotNone ( obj2 ) <EOL> self . assertEqual ( obj1 . id , obj2 . id ) <EOL> self . assertEqual ( obj1 . name , obj2 . name ) <EOL> self . assertDictEqual ( obj1 . context , obj2 . context ) <EOL> self . assertRaises ( ValueError , self . access . get_by_id , str ( bson . ObjectId ( ) ) ) <EOL> def test_query_by_name ( self ) : <EOL> obj1 = FakeModelDB ( name = uuid . uuid4 ( ) . hex , context = { '<STR_LIT:user>' : '<STR_LIT>' } ) <EOL> obj1 = self . access . add_or_update ( obj1 ) <EOL> obj2 = self . access . get_by_name ( obj1 . name ) <EOL> self . assertIsNotNone ( obj2 ) <EOL> self . assertEqual ( obj1 . id , obj2 . id ) <EOL> self . assertEqual ( obj1 . name , obj2 . name ) <EOL> self . assertDictEqual ( obj1 . context , obj2 . context ) <EOL> self . assertRaises ( ValueError , self . access . get_by_name , uuid . uuid4 ( ) . hex ) <EOL> def test_query_filter ( self ) : <EOL> obj1 = FakeModelDB ( name = uuid . uuid4 ( ) . hex , context = { '<STR_LIT:user>' : '<STR_LIT>' } ) <EOL> obj1 = self . access . add_or_update ( obj1 ) <EOL> obj2 = FakeModelDB ( name = uuid . uuid4 ( ) . hex , context = { '<STR_LIT:user>' : '<STR_LIT>' } ) <EOL> obj2 = self . access . add_or_update ( obj2 ) <EOL> objs = self . access . query ( context__user = '<STR_LIT>' ) <EOL> self . assertIsNotNone ( objs ) <EOL> self . assertGreater ( len ( objs ) , <NUM_LIT:0> ) <EOL> self . assertEqual ( obj1 . id , objs [ <NUM_LIT:0> ] . id ) <EOL> self . assertEqual ( obj1 . name , objs [ <NUM_LIT:0> ] . name ) <EOL> self . assertDictEqual ( obj1 . context , objs [ <NUM_LIT:0> ] . context ) <EOL> def test_null_filter ( self ) : <EOL> obj1 = FakeModelDB ( name = uuid . uuid4 ( ) . hex ) <EOL> obj1 = self . access . add_or_update ( obj1 ) <EOL> objs = self . access . query ( index = '<STR_LIT:null>' ) <EOL> self . assertEqual ( len ( objs ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( obj1 . id , objs [ <NUM_LIT:0> ] . id ) <EOL> self . assertEqual ( obj1 . name , objs [ <NUM_LIT:0> ] . name ) <EOL> self . assertIsNone ( getattr ( obj1 , '<STR_LIT:index>' , None ) ) <EOL> objs = self . access . query ( index = None ) <EOL> self . assertEqual ( len ( objs ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( obj1 . id , objs [ <NUM_LIT:0> ] . id ) <EOL> self . assertEqual ( obj1 . name , objs [ <NUM_LIT:0> ] . name ) <EOL> self . assertIsNone ( getattr ( obj1 , '<STR_LIT:index>' , None ) ) <EOL> def test_datetime_range ( self ) : <EOL> base = date_utils . add_utc_tz ( datetime . datetime ( <NUM_LIT> , <NUM_LIT:12> , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) ) <EOL> for i in range ( <NUM_LIT> ) : <EOL> timestamp = base + datetime . timedelta ( seconds = i ) <EOL> obj = FakeModelDB ( name = uuid . uuid4 ( ) . hex , timestamp = timestamp ) <EOL> self . access . add_or_update ( obj ) <EOL> dt_range = '<STR_LIT>' <EOL> objs = self . access . query ( timestamp = dt_range ) <EOL> self . assertEqual ( len ( objs ) , <NUM_LIT:10> ) <EOL> self . assertLess ( objs [ <NUM_LIT:0> ] . timestamp , objs [ <NUM_LIT:9> ] . timestamp ) <EOL> dt_range = '<STR_LIT>' <EOL> objs = self . access . query ( timestamp = dt_range ) <EOL> self . assertEqual ( len ( objs ) , <NUM_LIT:10> ) <EOL> self . assertLess ( objs [ <NUM_LIT:9> ] . timestamp , objs [ <NUM_LIT:0> ] . timestamp ) <EOL> def test_pagination ( self ) : <EOL> count = <NUM_LIT:100> <EOL> page_size = <NUM_LIT> <EOL> pages = count / page_size <EOL> users = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> for user in users : <EOL> context = { '<STR_LIT:user>' : user } <EOL> for i in range ( count ) : <EOL> self . access . add_or_update ( FakeModelDB ( name = uuid . uuid4 ( ) . hex , <EOL> context = context , index = i ) ) <EOL> self . assertEqual ( self . access . count ( ) , len ( users ) * count ) <EOL> for user in users : <EOL> for i in range ( pages ) : <EOL> offset = i * page_size <EOL> objs = self . access . query ( context__user = user , order_by = [ '<STR_LIT:index>' ] , <EOL> offset = offset , limit = page_size ) <EOL> self . assertEqual ( len ( objs ) , page_size ) <EOL> for j in range ( page_size ) : <EOL> self . assertEqual ( objs [ j ] . context [ '<STR_LIT:user>' ] , user ) <EOL> self . assertEqual ( objs [ j ] . index , ( i * page_size ) + j ) <EOL> def test_sort_multiple ( self ) : <EOL> count = <NUM_LIT> <EOL> base = date_utils . add_utc_tz ( datetime . datetime ( <NUM_LIT> , <NUM_LIT:12> , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) ) <EOL> for i in range ( count ) : <EOL> category = '<STR_LIT>' if i % <NUM_LIT:2> else '<STR_LIT>' <EOL> timestamp = base + datetime . timedelta ( seconds = i ) <EOL> obj = FakeModelDB ( name = uuid . uuid4 ( ) . hex , timestamp = timestamp , category = category ) <EOL> self . access . add_or_update ( obj ) <EOL> objs = self . access . query ( order_by = [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> self . assertEqual ( len ( objs ) , count ) <EOL> for i in range ( count ) : <EOL> category = '<STR_LIT>' if i < count / <NUM_LIT:2> else '<STR_LIT>' <EOL> self . assertEqual ( objs [ i ] . category , category ) <EOL> self . assertLess ( objs [ <NUM_LIT:0> ] . timestamp , objs [ ( count / <NUM_LIT:2> ) - <NUM_LIT:1> ] . timestamp ) <EOL> self . assertLess ( objs [ count / <NUM_LIT:2> ] . timestamp , objs [ ( count / <NUM_LIT:2> ) - <NUM_LIT:1> ] . timestamp ) <EOL> self . assertLess ( objs [ count / <NUM_LIT:2> ] . timestamp , objs [ count - <NUM_LIT:1> ] . timestamp ) <EOL> def test_escaped_field ( self ) : <EOL> context = { '<STR_LIT>' : '<STR_LIT:abc>' } <EOL> obj1 = FakeModelDB ( name = uuid . uuid4 ( ) . hex , context = context ) <EOL> obj2 = self . access . add_or_update ( obj1 ) <EOL> self . assertIn ( '<STR_LIT>' , context . keys ( ) ) <EOL> self . assertNotIn ( '<STR_LIT>' , context . keys ( ) ) <EOL> self . assertDictEqual ( obj2 . context , context ) <EOL> obj3 = self . access . get ( name = obj2 . name ) <EOL> self . assertIsNotNone ( obj3 ) <EOL> self . assertEqual ( obj3 . id , obj2 . id ) <EOL> self . assertDictEqual ( obj3 . context , context ) </s>
