Source Code for Module backend.satellite_tools.req_channels

  1  # 
  2  # Copyright (c) 2008--2017 Red Hat, Inc. 
  3  # 
  4  # This software is licensed to you under the GNU General Public License, 
  5  # version 2 (GPLv2). There is NO WARRANTY for this software, express or 
  6  # implied, including the implied warranties of MERCHANTABILITY or FITNESS 
  7  # FOR A PARTICULAR PURPOSE. You should have received a copy of GPLv2 
  8  # along with this software; if not, see 
  9  # http://www.gnu.org/licenses/old-licenses/gpl-2.0.txt. 
 10  # 
 11  # Red Hat trademarks are not licensed under GPLv2. No permission is 
 12  # granted to use or replicate Red Hat trademarks that are incorporated 
 13  # in this software or its documentation. 
 14  # 
 15  # 
 16   
 17  from spacewalk.common import usix 
 18   
 19   
 20 -class RequestedChannels: 
 21   
 22      """Bookkeeping of the state of various channels 
 23      Argument to constructor is the list of requested channels 
 24      """ 
 25      # Simplify the getters/setters/resetters 
 26      __lists = [ 
 27          '_requested_imported',  # requested and previously imported 
 28          '_requested_new',  # requested and NOT previously imported 
 29          '_requested_channels',  # Union of the above two 
 30          '_not_requested',  # not requested but available channels 
 31          '_end_of_service',  # requested, once available, but no longer supported 
 32          '_typos',  # requested, but probably a typo 
 33      ] 
 34   
 35 -    def __init__(self, requested=()): 
 36          # All the requested channels 
 37          self._requested = {} 
 38          # All available channels 
 39          self._available = {} 
 40          # All imported channels 
 41          self._imported = {} 
 42   
 43          # These will be computed 
 44          # We could have initialized them in a loop, but pychecker would 
 45          # complain that the data member does not exist 
 46          self._requested_imported = [] 
 47          self._requested_new = [] 
 48          self._requested_channels = [] 
 49          self._not_requested = [] 
 50          self._end_of_service = [] 
 51          self._typos = [] 
 52   
 53          for l in self.__lists: 
 54              assert hasattr(self, l), "Class does not initialize %s" % l 
 55              assert isinstance(getattr(self, l), usix.ListType) 
 56   
 57          # Initialize the requested channels 
 58          self.request(requested) 
 59   
 60 -    def request(self, requested): 
 61          assert isinstance(requested, usix.ListType) 
 62   
 63          self._requested.clear() 
 64          for c in requested: 
 65              self._requested[c] = None 
 66          return self 
 67   
 68 -    def _add(self, name, channel): 
 69          if name not in ['_available', '_imported']: 
 70              raise AttributeError('add' + name) 
 71          getattr(self, name)[channel] = None 
 72          return self 
 73   
 74 -    def _set(self, name, channel_list): 
 75          if name not in ['_available', '_imported']: 
 76              raise AttributeError('set' + name) 
 77          assert isinstance(channel_list, usix.ListType) 
 78          h = getattr(self, name) 
 79          h.clear() 
 80          for c in channel_list: 
 81              h[c] = None 
 82          return self 
 83   
 84 -    def _reset(self, name): 
 85          del getattr(self, name)[:] 
 86          return self 
 87   
 88 -    def _get(self, name): 
 89          return getattr(self, name) 
 90   
 91 -    def reset(self): 
 92          self._available.clear() 
 93          self._imported.clear() 
 94          self._reset_computed() 
 95   
 96 -    def _reset_computed(self): 
 97          for name in self.__lists: 
 98              del getattr(self, name)[:] 
 99          return self 
100   
101 -    def _print_values(self): 
102          for name in self.__lists: 
103              print("Contents of %s: %s" % (name, getattr(self, name))) 
104          return self 
105   
106 -    def compute(self): 
107          self._reset_computed() 
108          available = self._available.copy() 
109          imported = self._imported.copy() 
110          for c in self._requested: 
111              if c in self._available: 
112                  del available[c] 
113                  # Channel exists 
114                  if c in self._imported: 
115                      del imported[c] 
116                      self._requested_imported.append(c) 
117                      continue 
118                  self._requested_new.append(c) 
119                  continue 
120              # Requested channel not available 
121              if c in self._imported: 
122                  del imported[c] 
123                  self._end_of_service.append(c) 
124                  continue 
125              # Typo 
126              self._typos.append(c) 
127   
128          for c in available.keys(): 
129              if c in imported: 
130                  # Available, already imported 
131                  del imported[c] 
132              # Available, not imported 
133              self._not_requested.append(c) 
134   
135          # The rest are channels that were once imported, but now are 
136          # unavailable 
137          self._end_of_service.extend(list(imported.keys())) 
138   
139          self._requested_channels.extend(self._requested_new) 
140          self._requested_channels.extend(self._requested_imported) 
141   
142          # Sort the lists 
143          for name in self.__lists: 
144              getattr(self, name).sort() 
145          return self 
146   
147 -    def __getattr__(self, name): 
148          if name.startswith('add'): 
149              return Method(name[3:], self._add) 
150          if name.startswith('get'): 
151              return Method(name[3:], self._get) 
152          if name.startswith('set'): 
153              return Method(name[3:], self._set) 
154          if name.startswith('reset'): 
155              return Method(name[5:], self._reset) 
156          raise AttributeError(name) 
157   
158   
159 -class Method: 
160      # pylint: disable=R0903 
161   
162 -    def __init__(self, name, func): 
163          self._func = func 
164          self._name = name 
165   
166 -    def __call__(self, *args, **kwargs): 
167          return self._func(self._name, *args, **kwargs) 
168   
169  # Test functions 
170   
171   
172 -def _verify_expectations(c, expectations): 
173      for k, expected in expectations.items(): 
174          method_name = 'get' + k 
175          val = getattr(c, method_name)() 
176          if val == expected: 
177              print("ok: %s = %s" % (method_name, expected)) 
178          else: 
179              print("FAILED: %s: expected %s, got %s" % (method_name, expected, 
180                                                         val)) 
181   
182   
183 -def test1(requested, available, imported, expectations): 
184      c = RequestedChannels(requested) 
185      # Available channels 
186      for av in available: 
187          c.add_available(av) 
188      # Already impoted 
189      for i in imported: 
190          c.add_imported(i) 
191   
192      c.compute() 
193      _verify_expectations(c, expectations) 
194   
195   
196 -def test2(requested, available, imported, expectations): 
197      c = RequestedChannels(requested) 
198      # Available channels 
199      c.set_available(available) 
200      # Already impoted 
201      c.set_imported(imported) 
202   
203      c.compute() 
204      _verify_expectations(c, expectations) 
205   
206   
207 -def test(): 
208      requested = ['a', 'b', 'c', 'd'] 
209      available = ['a', 'd', 'e', 'f'] 
210      imported = ['d', 'e', 'h'] 
211      expectations = { 
212          '_requested_imported': ['d'], 
213          '_requested_new': ['a'], 
214          '_not_requested': ['e', 'f'], 
215          '_end_of_service': ['h'], 
216          '_typos': ['b', 'c'], 
217          '_requested_channels': ['a', 'd'], 
218      } 
219      print("Running test1") 
220      test1(requested, available, imported, expectations) 
221      print("Running test2") 
222      test2(requested, available, imported, expectations) 
223   
224  if __name__ == '__main__': 
225      test() 
226