adapter/acumos/tests/test_fed.py

   1 # ============LICENSE_START====================================================
   2 # org.onap.dcae
   3 # =============================================================================
   4 # Copyright (c) 2019-2020 AT&T Intellectual Property. All rights reserved.
   5 # =============================================================================
   6 # Copyright (c) 2021 highstreet technologies GmbH. All rights reserved.
   7 # =============================================================================
   8 # Licensed under the Apache License, Version 2.0 (the "License");
   9 # you may not use this file except in compliance with the License.
  10 # You may obtain a copy of the License at
  11 #
  12 #      http://www.apache.org/licenses/LICENSE-2.0
  13 #
  14 # Unless required by applicable law or agreed to in writing, software
  15 # distributed under the License is distributed on an "AS IS" BASIS,
  16 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  17 # See the License for the specific language governing permissions and
  18 # limitations under the License.
  19 # ============LICENSE_END======================================================
  20
  21 import json
  22 import requests
  23
  24 from testing_helpers import get_fixture_path as get_test_file
  25
  26 from aoconversion import docker_gen as aoc_docker_gen
  27 from aoconversion import scanner as aoc_scanner
  28
  29 #
  30 # General mocking
  31 #
  32
  33
  34 class _MockModule:
  35     """
  36     Class to mock packages.
  37     Just a quick way to create an object with specified attributes
  38     """
  39     def __init__(self, **kwargs):
  40         for k, v in kwargs.items():
  41             setattr(self, k, v)
  42
  43
  44 #
  45 # Mocking for package "docker"
  46 #
  47
  48
  49 class _MockAPIClient:
  50     """
  51     Class to mock docker.APIClient class.
  52     """
  53     def __init__(self, base_url, version=None, user_agent='xxx'):
  54         pass
  55
  56     def build(self, path, rm, tag):
  57         return [b'some message', b'another message']
  58
  59     def push(self, repository, auth_config, stream):
  60         return [b'some message', b'another message']
  61
  62     def images(self, x):
  63         return True
  64
  65
  66 def _mock_kwargs_from_env(**kwargs):
  67     """
  68     Method to mock docker.utils.kwargs_from_env method.
  69     """
  70     return {'base_url': None}
  71
  72
  73 _mockdocker = _MockModule(
  74     APIClient=_MockAPIClient,
  75     utils=_MockModule(kwargs_from_env=_mock_kwargs_from_env))
  76
  77
  78 #
  79 # Mocking for requests.get
  80 #
  81
  82
  83 class _r:
  84     """
  85     Fake responses for mocking requests.get
  86     """
  87     def __init__(self, json=None, file=None, data=None):
  88         self.jx = json
  89         self.fx = file
  90         self.dx = data
  91         self.status_code = 200
  92
  93     @property
  94     def text(self):
  95         return self._raw().decode()
  96
  97     def raise_for_status(self):
  98         pass
  99
 100     def _raw(self):
 101         if self.dx is None:
 102             if self.fx is not None:
 103                 with open(self.fx, 'rb') as f:
 104                     self.dx = f.read()
 105             elif self.jx is not None:
 106                 self.dx = json.dumps(self.jx, sort_keys=True).encode()
 107             else:
 108                 self.dx = b''
 109         return self.dx
 110
 111     def iter_content(self, bsize=-1):
 112         buf = self._raw()
 113         pos = 0
 114         lim = len(buf)
 115         if bsize <= 0:
 116             bsize = lim
 117         while pos + bsize < lim:
 118             yield buf[pos:pos + bsize]
 119             pos = pos + bsize
 120         yield buf[pos:]
 121
 122     def json(self):
 123         if self.jx is None:
 124             self.jx = json.loads(self._raw().decode())
 125         return self.jx
 126
 127
 128 def _mockwww(responses):
 129     def _op(path, json=None, auth=None, cert=None, verify=None, stream=False):
 130         return responses[path]
 131     return _op
 132
 133
 134 _mockpostdata = {
 135     'https://onboarding/dataformats': _r({'dataFormatUrl': 'https://onboarding/dataformats/somedfid'}),
 136     'https://onboarding/components': _r({'componentUrl': 'https://onboarding/components/somedxid'}),
 137 }
 138
 139 _mockpatchdata = {
 140     'https://onboarding/dataformats/somedfid': _r({}),
 141     'https://onboarding/components/somedxid': _r({}),
 142 }
 143
 144 _mockwebdata = {
 145     'https://acumos/catalogs': _r({'content': [{'catalogId': 'c1'}]}),
 146     'https://acumos/solutions?catalogId=c1': _r({'content': [{'solutionId': 's1', 'name': 'example-model', 'ratingAverageTenths': 17}]}),
 147     'https://acumos/solutions/s1/revisions': _r({'content': [{'revisionId': 'r1'}]}),
 148     'https://acumos/solutions/s1/revisions/r1': _r({'content': {
 149         'version': 'v1',
 150         'modified': '2019-01-01T00:00:00Z',
 151         'artifacts': [
 152             {'artifactId': 'a1', 'name': 'xxx.other'},
 153             {'artifactId': 'a2', 'name': 'xxx.proto'},
 154             {'artifactId': 'a3', 'name': 'xxx.zip'},
 155             {'artifactId': 'a4', 'name': 'xxx.json'},
 156         ]}
 157     }),
 158     'https://acumos/artifacts/a2/content': _r(file=get_test_file('models/example-model/model.proto')),
 159     'https://acumos/artifacts/a3/content': _r(data=b'dummy zip archive data'),
 160     'https://acumos/artifacts/a4/content': _r(file=get_test_file('models/example-model/metadata.json')),
 161 }
 162
 163
 164 #
 165 # End mocking tools
 166 #
 167
 168
 169 def test_aoconversion(mock_schemas, tmpdir, monkeypatch):
 170     config = aoc_scanner.Config(dcaeurl='http://dcaeurl', dcaeuser='dcaeuser', onboardingurl='https://onboarding', onboardinguser='obuser', onboardingpass='obpass', acumosurl='https://acumos', certfile=None, dockerregistry='dockerregistry', dockeruser='registryuser', dockerpass='registrypassword', http_proxy='', https_proxy='', no_proxy='')
 171     monkeypatch.setattr(aoc_docker_gen, 'APIClient', _mockdocker.APIClient)
 172     monkeypatch.setattr(requests, 'get', _mockwww(_mockwebdata))
 173     monkeypatch.setattr(requests, 'post', _mockwww(_mockpostdata))
 174     monkeypatch.setattr(requests, 'patch', _mockwww(_mockpatchdata))
 175     aoc_scanner.scan(config)
 176     aoc_scanner.scan(config)
 177
 178
 179 def test__derefconfig():
 180     config_path = get_test_file('config.yaml')
 181     assert aoc_scanner._derefconfig('@' + config_path) == 'dcaeurl: https://git.onap.org/dcaegen2/platform/plain/mod'