tests/test_sdl.py

   1 # ==================================================================================
   2 #       Copyright (c) 2019-2020 Nokia
   3 #       Copyright (c) 2018-2020 AT&T Intellectual Property.
   4 #
   5 #   Licensed under the Apache License, Version 2.0 (the "License");
   6 #   you may not use this file except in compliance with the License.
   7 #   You may obtain a copy of the License at
   8 #
   9 #          http://www.apache.org/licenses/LICENSE-2.0
  10 #
  11 #   Unless required by applicable law or agreed to in writing, software
  12 #   distributed under the License is distributed on an "AS IS" BASIS,
  13 #   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14 #   See the License for the specific language governing permissions and
  15 #   limitations under the License.
  16 # ==================================================================================
  17 """
  18 tests data functions
  19 """
  20 import time
  21 from ricxappframe.xapp_sdl import SDLWrapper
  22
  23
  24 NS = "testns"
  25
  26
  27 def test_sdl():
  28     """
  29     test raw sdl functions
  30     """
  31     sdl = SDLWrapper(use_fake_sdl=True)
  32     sdl.set(NS, "as.df1", "data")
  33     sdl.set(NS, "as.df2", "data2")
  34     assert sdl.get(NS, "as.df1") == "data"
  35     assert sdl.get(NS, "as.df2") == "data2"
  36     assert sdl.find_and_get(NS, "as.df1") == {"as.df1": "data"}
  37     assert sdl.find_and_get(NS, "as.df2") == {"as.df2": "data2"}
  38     assert sdl.find_and_get(NS, "as.df") == {"as.df1": "data", "as.df2": "data2"}
  39     assert sdl.find_and_get(NS, "as.d") == {"as.df1": "data", "as.df2": "data2"}
  40     assert sdl.find_and_get(NS, "as.") == {"as.df1": "data", "as.df2": "data2"}
  41     assert sdl.find_and_get(NS, "asd") == {}
  42
  43     # delete 1
  44     sdl.delete(NS, "as.df1")
  45     assert sdl.get(NS, "as.df1") is None
  46     assert sdl.get(NS, "as.df2") == "data2"
  47
  48     # delete 2
  49     sdl.delete(NS, "as.df2")
  50     assert sdl.get(NS, "as.df2") is None
  51
  52     assert sdl.find_and_get(NS, "as.df") == {}
  53     assert sdl.find_and_get(NS, "") == {}
  54
  55
  56 def test_sdl_set_get():
  57     """
  58     test set, get realted sdl methods
  59     """
  60     sdl = SDLWrapper(use_fake_sdl=True)
  61
  62     # set_if
  63     sdl.set(NS, "gs.df1", "old")
  64     assert sdl.get(NS, "gs.df1") == "old"
  65
  66     sdl.set_if(NS, "gs.df1", "young", "new")
  67     assert sdl.get(NS, "gs.df1") == "old"
  68
  69     sdl.set_if(NS, "gs.df1", "old", "new")
  70     assert sdl.get(NS, "gs.df1") == "new"
  71
  72     # set_if_not_exists
  73     sdl.set(NS, "gs.df2", "old")
  74     assert sdl.get(NS, "gs.df2") == "old"
  75
  76     sdl.set_if_not_exists(NS, "gs.df2", "new")
  77     assert sdl.get(NS, "gs.df2") == "old"
  78
  79     sdl.set_if_not_exists(NS, "gs.df3", "new")
  80     assert sdl.get(NS, "gs.df3") == "new"
  81
  82     # find_keys
  83     assert sdl.find_keys(NS, "gs") == ["gs.df1", "gs.df2", "gs.df3"]
  84     assert sdl.find_keys(NS, "gs.df1") == ["gs.df1"]
  85     assert sdl.find_keys(NS, "gs.df2") == ["gs.df2"]
  86     assert sdl.find_keys(NS, "gs.df3") == ["gs.df3"]
  87
  88     # delete_if
  89     sdl.set(NS, "gs.df4", "delete_this")
  90
  91     assert sdl.delete_if(NS, "gs.df4", "delete") is False
  92     assert sdl.delete_if(NS, "gs.df4", "delete_this") is True
  93     assert sdl.get(NS, "gs.df4") is None
  94
  95
  96 def test_sdl_member():
  97     """
  98     test member related sdl methods
  99     """
 100     # add_member, remove_member, get_members
 101     sdl = SDLWrapper(use_fake_sdl=True)
 102
 103     sdl.add_member(NS, "group1", "member1")
 104     assert sdl.is_member(NS, "group1", "member1") is True
 105
 106     sdl.remove_member(NS, "group1", "not_member")
 107     assert sdl.is_member(NS, "group1", "member1") is True
 108
 109     sdl.remove_member(NS, "group1", "member1")
 110     assert sdl.is_member(NS, "group1", "member1") is False
 111
 112     # remove_group, group_size
 113     sdl.add_member(NS, "group2", "member1")
 114     sdl.add_member(NS, "group2", "member2")
 115     assert sdl.group_size(NS, "group2") == 2
 116     sdl.remove_group(NS, "group2")
 117     assert sdl.group_size(NS, "group2") == 0
 118
 119     # get_members
 120     sdl.add_member(NS, "group3", "member1")
 121     sdl.add_member(NS, "group3", "member2")
 122     members = sdl.get_members(NS, "group3")
 123     assert "member1" in members
 124     assert "member2" in members
 125
 126
 127 def test_sdl_set_and_publish_with_handle_events():
 128     """
 129     test set_and_publish* related sdl methods
 130     """
 131     CH = "channel"
 132     EVENT = "event"
 133     CALLED = None
 134
 135     def cb(channel, event):
 136         nonlocal CH
 137         nonlocal EVENT
 138         nonlocal CALLED
 139         # test is cb called
 140         CALLED = True
 141         assert channel == CH
 142         assert event[0] == EVENT
 143
 144     sdl = SDLWrapper(use_fake_sdl=True)
 145     sdl.subscribe_channel(NS, cb, "channel")
 146
 147     # set_and_publish
 148     CALLED = False
 149     sdl.set_and_publish(NS, "channel", "event", "nt.df1", "old")
 150     sdl.handle_events()
 151     assert sdl.get(NS, "nt.df1") == "old"
 152     assert CALLED is True
 153
 154     # set_if_and_publish fail
 155     CALLED = False
 156     sdl.set_if_and_publish(NS, "channel", "event", "nt.df1", "young", "new")
 157     sdl.handle_events()
 158     assert sdl.get(NS, "nt.df1") == "old"
 159     assert CALLED is False
 160     # set_if_and_publish success
 161     sdl.set_if_and_publish(NS, "channel", "event", "nt.df1", "old", "new")
 162     sdl.handle_events()
 163     assert sdl.get(NS, "nt.df1") == "new"
 164     assert CALLED is True
 165
 166     # set_if_not_exists_and_publish fail
 167     CALLED = False
 168     sdl.set_if_not_exists_and_publish(NS, "channel", "event", "nt.df1", "latest")
 169     sdl.handle_events()
 170     assert sdl.get(NS, "nt.df1") == "new"
 171     assert CALLED is False
 172     # set_if_not_exists_and_publish success
 173     sdl.set_if_not_exists_and_publish(
 174         NS, "channel", "event", "nt.df2", "latest")
 175     sdl.handle_events()
 176     assert sdl.get(NS, "nt.df2") == "latest"
 177     assert CALLED is True
 178
 179     sdl.unsubscribe_channel(NS, "channel")
 180
 181
 182 def test_sdl_remove_and_publish_with_start_event_listener():
 183     """
 184     test remove_and_publish* related sdl methods
 185     """
 186     CH = "channel"
 187     EVENT = "event"
 188     CALLED = None
 189
 190     def cb(channel, event):
 191         nonlocal CH
 192         nonlocal EVENT
 193         nonlocal CALLED
 194         CALLED = True
 195         assert channel == CH
 196         assert event[0] == EVENT
 197
 198     sdl = SDLWrapper(use_fake_sdl=True)
 199     sdl.subscribe_channel(NS, cb, "channel")
 200     sdl.start_event_listener()
 201
 202     # remove_and_publish success
 203     CALLED = False
 204     sdl.set(NS, "nt.df1", "old")
 205     sdl.remove_and_publish(NS, "channel", "event", "nt.df1")
 206     time.sleep(0.3)
 207     assert sdl.get(NS, "nt.df1") is None
 208     assert CALLED is True
 209
 210     # remove_if_and_publish
 211     CALLED = False
 212     sdl.set(NS, "nt.df1", "old")
 213     # fail
 214     sdl.remove_if_and_publish(NS, "channel", "event", "nt.df1", "new")
 215     time.sleep(0.3)
 216     assert sdl.get(NS, "nt.df1") == "old"
 217     assert CALLED is False
 218     # success
 219     sdl.remove_if_and_publish(NS, "channel", "event", "nt.df1", "old")
 220     time.sleep(0.3)
 221     assert sdl.get(NS, "nt.df1") is None
 222     assert CALLED is True
 223
 224     # remove_all_and_publish
 225     CALLED = False
 226     sdl.set(NS, "nt.df1", "data1")
 227     sdl.set(NS, "nt.df2", "data2")
 228     sdl.set(NS, "nt.df3", "data3")
 229     sdl.remove_all_and_publish(NS, "channel", "event")
 230     time.sleep(0.3)
 231     assert sdl.get(NS, "nt.df1") is None
 232     assert sdl.get(NS, "nt.df2") is None
 233     assert sdl.get(NS, "nt.df3") is None
 234     assert sdl.find_keys(NS, "*") == []
 235     assert CALLED is True
 236
 237     sdl.unsubscribe_channel(NS, "channel")