<s> from oslo_serialization import jsonutils <EOL> from nova import objects <EOL> from nova . scheduler . filters import json_filter <EOL> from nova import test <EOL> from nova . tests . unit . scheduler import fakes <EOL> class TestJsonFilter ( test . NoDBTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestJsonFilter , self ) . setUp ( ) <EOL> self . filt_cls = json_filter . JsonFilter ( ) <EOL> self . json_query = jsonutils . dumps ( <EOL> [ '<STR_LIT>' , [ '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT> ] , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT:200> * <NUM_LIT> ] ] ) <EOL> def test_json_filter_passes ( self ) : <EOL> spec_obj = objects . RequestSpec ( <EOL> flavor = objects . Flavor ( memory_mb = <NUM_LIT> , <EOL> root_gb = <NUM_LIT:200> , <EOL> ephemeral_gb = <NUM_LIT:0> ) , <EOL> scheduler_hints = dict ( query = [ self . json_query ] ) ) <EOL> host = fakes . FakeHostState ( '<STR_LIT>' , '<STR_LIT>' , <EOL> { '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT:200> * <NUM_LIT> } ) <EOL> self . assertTrue ( self . filt_cls . host_passes ( host , spec_obj ) ) <EOL> def test_json_filter_passes_with_no_query ( self ) : <EOL> spec_obj = objects . RequestSpec ( <EOL> flavor = objects . Flavor ( memory_mb = <NUM_LIT> , <EOL> root_gb = <NUM_LIT:200> , <EOL> ephemeral_gb = <NUM_LIT:0> ) , <EOL> scheduler_hints = None ) <EOL> host = fakes . FakeHostState ( '<STR_LIT>' , '<STR_LIT>' , <EOL> { '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : <NUM_LIT:0> } ) <EOL> self . assertTrue ( self . filt_cls . host_passes ( host , spec_obj ) ) <EOL> def test_json_filter_fails_on_memory ( self ) : <EOL> spec_obj = objects . RequestSpec ( <EOL> flavor = objects . Flavor ( memory_mb = <NUM_LIT> , <EOL> root_gb = <NUM_LIT:200> , <EOL> ephemeral_gb = <NUM_LIT:0> ) , <EOL> scheduler_hints = dict ( query = [ self . json_query ] ) ) <EOL> host = fakes . FakeHostState ( '<STR_LIT>' , '<STR_LIT>' , <EOL> { '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT:200> * <NUM_LIT> } ) <EOL> self . assertFalse ( self . filt_cls . host_passes ( host , spec_obj ) ) <EOL> def test_json_filter_fails_on_disk ( self ) : <EOL> spec_obj = objects . RequestSpec ( <EOL> flavor = objects . Flavor ( memory_mb = <NUM_LIT> , <EOL> root_gb = <NUM_LIT:200> , <EOL> ephemeral_gb = <NUM_LIT:0> ) , <EOL> scheduler_hints = dict ( query = [ self . json_query ] ) ) <EOL> host = fakes . FakeHostState ( '<STR_LIT>' , '<STR_LIT>' , <EOL> { '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : ( <NUM_LIT:200> * <NUM_LIT> ) - <NUM_LIT:1> } ) <EOL> self . assertFalse ( self . filt_cls . host_passes ( host , spec_obj ) ) <EOL> def test_json_filter_fails_on_service_disabled ( self ) : <EOL> json_query = jsonutils . dumps ( <EOL> [ '<STR_LIT>' , [ '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT> ] , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT:200> * <NUM_LIT> ] , <EOL> [ '<STR_LIT>' , '<STR_LIT>' ] ] ) <EOL> spec_obj = objects . RequestSpec ( <EOL> flavor = objects . Flavor ( memory_mb = <NUM_LIT> , local_gb = <NUM_LIT:200> ) , <EOL> scheduler_hints = dict ( query = [ json_query ] ) ) <EOL> host = fakes . FakeHostState ( '<STR_LIT>' , '<STR_LIT>' , <EOL> { '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT:200> * <NUM_LIT> } ) <EOL> self . assertFalse ( self . filt_cls . host_passes ( host , spec_obj ) ) <EOL> def test_json_filter_happy_day ( self ) : <EOL> raw = [ '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> [ '<STR_LIT:=>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> [ '<STR_LIT>' , <EOL> [ '<STR_LIT>' , <EOL> [ '<STR_LIT:<>' , '<STR_LIT>' , <NUM_LIT:30> ] , <EOL> [ '<STR_LIT:<>' , '<STR_LIT>' , <NUM_LIT> ] ] , <EOL> [ '<STR_LIT>' , <EOL> [ '<STR_LIT:>>' , '<STR_LIT>' , <NUM_LIT:30> ] , <EOL> [ '<STR_LIT:>>' , '<STR_LIT>' , <NUM_LIT> ] ] ] ] <EOL> spec_obj = objects . RequestSpec ( <EOL> scheduler_hints = dict ( query = [ jsonutils . dumps ( raw ) ] ) ) <EOL> capabilities = { '<STR_LIT>' : '<STR_LIT>' } <EOL> service = { '<STR_LIT>' : False } <EOL> host = fakes . FakeHostState ( '<STR_LIT>' , '<STR_LIT>' , <EOL> { '<STR_LIT>' : <NUM_LIT:10> , <EOL> '<STR_LIT>' : <NUM_LIT:200> , <EOL> '<STR_LIT>' : capabilities , <EOL> '<STR_LIT>' : service } ) <EOL> self . assertTrue ( self . filt_cls . host_passes ( host , spec_obj ) ) <EOL> capabilities = { '<STR_LIT>' : '<STR_LIT>' } <EOL> service = { '<STR_LIT>' : False } <EOL> host = fakes . FakeHostState ( '<STR_LIT>' , '<STR_LIT>' , <EOL> { '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : capabilities , <EOL> '<STR_LIT>' : service } ) <EOL> self . assertTrue ( self . filt_cls . host_passes ( host , spec_obj ) ) <EOL> capabilities = { '<STR_LIT>' : False , '<STR_LIT>' : '<STR_LIT>' } <EOL> service = { '<STR_LIT>' : False } <EOL> host = fakes . FakeHostState ( '<STR_LIT>' , '<STR_LIT>' , <EOL> { '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : capabilities , <EOL> '<STR_LIT>' : service } ) <EOL> self . assertFalse ( self . filt_cls . host_passes ( host , spec_obj ) ) <EOL> capabilities = { '<STR_LIT>' : True , '<STR_LIT>' : '<STR_LIT>' } <EOL> service = { '<STR_LIT>' : False } <EOL> host = fakes . FakeHostState ( '<STR_LIT>' , '<STR_LIT>' , <EOL> { '<STR_LIT>' : <NUM_LIT:30> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : capabilities , <EOL> '<STR_LIT>' : service } ) <EOL> self . assertFalse ( self . filt_cls . host_passes ( host , spec_obj ) ) <EOL> capabilities = { '<STR_LIT>' : True , '<STR_LIT>' : '<STR_LIT>' } <EOL> service = { '<STR_LIT>' : False } <EOL> host = fakes . FakeHostState ( '<STR_LIT>' , '<STR_LIT>' , <EOL> { '<STR_LIT>' : <NUM_LIT:20> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : capabilities , <EOL> '<STR_LIT>' : service } ) <EOL> self . assertFalse ( self . filt_cls . host_passes ( host , spec_obj ) ) <EOL> capabilities = { '<STR_LIT>' : True , '<STR_LIT>' : '<STR_LIT>' } <EOL> service = { '<STR_LIT>' : True } <EOL> host = fakes . FakeHostState ( '<STR_LIT>' , '<STR_LIT>' , <EOL> { '<STR_LIT>' : <NUM_LIT:20> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : capabilities , <EOL> '<STR_LIT>' : service } ) <EOL> self . assertFalse ( self . filt_cls . host_passes ( host , spec_obj ) ) <EOL> def test_json_filter_basic_operators ( self ) : <EOL> host = fakes . FakeHostState ( '<STR_LIT>' , '<STR_LIT>' , { } ) <EOL> ops_to_test = [ <EOL> [ '<STR_LIT:=>' , [ <NUM_LIT:1> , <NUM_LIT:1> ] , True ] , <EOL> [ '<STR_LIT:=>' , [ <NUM_LIT:1> , <NUM_LIT:2> ] , False ] , <EOL> [ '<STR_LIT:<>' , [ <NUM_LIT:1> , <NUM_LIT:2> ] , True ] , <EOL> [ '<STR_LIT:<>' , [ <NUM_LIT:1> , <NUM_LIT:1> ] , False ] , <EOL> [ '<STR_LIT:<>' , [ <NUM_LIT:2> , <NUM_LIT:1> ] , False ] , <EOL> [ '<STR_LIT:>>' , [ <NUM_LIT:2> , <NUM_LIT:1> ] , True ] , <EOL> [ '<STR_LIT:>>' , [ <NUM_LIT:2> , <NUM_LIT:2> ] , False ] , <EOL> [ '<STR_LIT:>>' , [ <NUM_LIT:2> , <NUM_LIT:3> ] , False ] , <EOL> [ '<STR_LIT>' , [ <NUM_LIT:1> , <NUM_LIT:2> ] , True ] , <EOL> [ '<STR_LIT>' , [ <NUM_LIT:1> , <NUM_LIT:1> ] , True ] , <EOL> [ '<STR_LIT>' , [ <NUM_LIT:2> , <NUM_LIT:1> ] , False ] , <EOL> [ '<STR_LIT>' , [ <NUM_LIT:2> , <NUM_LIT:1> ] , True ] , <EOL> [ '<STR_LIT>' , [ <NUM_LIT:2> , <NUM_LIT:2> ] , True ] , <EOL> [ '<STR_LIT>' , [ <NUM_LIT:2> , <NUM_LIT:3> ] , False ] , <EOL> [ '<STR_LIT>' , [ <NUM_LIT:1> , <NUM_LIT:1> ] , True ] , <EOL> [ '<STR_LIT>' , [ <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] , True ] , <EOL> [ '<STR_LIT>' , [ <NUM_LIT:4> , <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] , False ] , <EOL> [ '<STR_LIT>' , [ True ] , False ] , <EOL> [ '<STR_LIT>' , [ False ] , True ] , <EOL> [ '<STR_LIT>' , [ True , False ] , True ] , <EOL> [ '<STR_LIT>' , [ False , False ] , False ] , <EOL> [ '<STR_LIT>' , [ True , True ] , True ] , <EOL> [ '<STR_LIT>' , [ False , False ] , False ] , <EOL> [ '<STR_LIT>' , [ True , False ] , False ] , <EOL> [ '<STR_LIT>' , [ [ '<STR_LIT>' , True , False ] , [ '<STR_LIT:>>' , <NUM_LIT:2> , <NUM_LIT:1> ] ] , True ] ] <EOL> for ( op , args , expected ) in ops_to_test : <EOL> raw = [ op ] + args <EOL> spec_obj = objects . RequestSpec ( <EOL> scheduler_hints = dict ( <EOL> query = [ jsonutils . dumps ( raw ) ] ) ) <EOL> self . assertEqual ( expected , <EOL> self . filt_cls . host_passes ( host , spec_obj ) ) <EOL> raw = [ '<STR_LIT>' , True , False , True , False ] <EOL> spec_obj = objects . RequestSpec ( <EOL> scheduler_hints = dict ( <EOL> query = [ jsonutils . dumps ( raw ) ] ) ) <EOL> self . assertTrue ( self . filt_cls . host_passes ( host , spec_obj ) ) <EOL> raw = [ '<STR_LIT>' , True , True , True ] <EOL> spec_obj = objects . RequestSpec ( <EOL> scheduler_hints = dict ( <EOL> query = [ jsonutils . dumps ( raw ) ] ) ) <EOL> self . assertFalse ( self . filt_cls . host_passes ( host , spec_obj ) ) <EOL> def test_json_filter_unknown_operator_raises ( self ) : <EOL> raw = [ '<STR_LIT>' , <NUM_LIT:1> , <NUM_LIT:2> ] <EOL> spec_obj = objects . RequestSpec ( <EOL> scheduler_hints = dict ( <EOL> query = [ jsonutils . dumps ( raw ) ] ) ) <EOL> host = fakes . FakeHostState ( '<STR_LIT>' , '<STR_LIT>' , <EOL> { } ) <EOL> self . assertRaises ( KeyError , <EOL> self . filt_cls . host_passes , host , spec_obj ) <EOL> def test_json_filter_empty_filters_pass ( self ) : <EOL> host = fakes . FakeHostState ( '<STR_LIT>' , '<STR_LIT>' , <EOL> { } ) <EOL> raw = [ ] <EOL> spec_obj = objects . RequestSpec ( <EOL> scheduler_hints = dict ( <EOL> query = [ jsonutils . dumps ( raw ) ] ) ) <EOL> self . assertTrue ( self . filt_cls . host_passes ( host , spec_obj ) ) <EOL> raw = { } <EOL> spec_obj = objects . RequestSpec ( <EOL> scheduler_hints = dict ( <EOL> query = [ jsonutils . dumps ( raw ) ] ) ) <EOL> self . assertTrue ( self . filt_cls . host_passes ( host , spec_obj ) ) <EOL> def test_json_filter_invalid_num_arguments_fails ( self ) : <EOL> host = fakes . FakeHostState ( '<STR_LIT>' , '<STR_LIT>' , <EOL> { } ) <EOL> raw = [ '<STR_LIT:>>' , [ '<STR_LIT>' , [ '<STR_LIT>' , [ '<STR_LIT>' , [ '<STR_LIT:<>' , [ '<STR_LIT>' , [ '<STR_LIT>' , [ '<STR_LIT>' , ] ] ] ] ] ] ] ] <EOL> spec_obj = objects . RequestSpec ( <EOL> scheduler_hints = dict ( <EOL> query = [ jsonutils . dumps ( raw ) ] ) ) <EOL> self . assertFalse ( self . filt_cls . host_passes ( host , spec_obj ) ) <EOL> raw = [ '<STR_LIT:>>' , <NUM_LIT:1> ] <EOL> spec_obj = objects . RequestSpec ( <EOL> scheduler_hints = dict ( <EOL> query = [ jsonutils . dumps ( raw ) ] ) ) <EOL> self . assertFalse ( self . filt_cls . host_passes ( host , spec_obj ) ) <EOL> def test_json_filter_unknown_variable_ignored ( self ) : <EOL> host = fakes . FakeHostState ( '<STR_LIT>' , '<STR_LIT>' , <EOL> { } ) <EOL> raw = [ '<STR_LIT:=>' , '<STR_LIT>' , <NUM_LIT:1> , <NUM_LIT:1> ] <EOL> spec_obj = objects . RequestSpec ( <EOL> scheduler_hints = dict ( <EOL> query = [ jsonutils . dumps ( raw ) ] ) ) <EOL> self . assertTrue ( self . filt_cls . host_passes ( host , spec_obj ) ) <EOL> raw = [ '<STR_LIT:=>' , '<STR_LIT>' , <NUM_LIT:2> , <NUM_LIT:2> ] <EOL> spec_obj = objects . RequestSpec ( <EOL> scheduler_hints = dict ( <EOL> query = [ jsonutils . dumps ( raw ) ] ) ) <EOL> self . assertTrue ( self . filt_cls . host_passes ( host , spec_obj ) ) </s>
