Merge pull request #61 from vrk-kpa/LIKA-393_getrest-support

Lika 393 getrest support

ckanext/xroad_integration/harvesters/xroad_harvester.py

@@ -25,7 +25,8 @@
 
 from werkzeug.datastructures import FileStorage as FlaskFileStorage
 
-from .xroad_types import MemberList, Subsystem
+from .xroad_types import MemberList, Subsystem, RestServices
+from ckanext.xroad_integration.xroad_utils import xroad_catalog_query_json, ContentFetchError
 
 try:
     from ckan.common import asbool  # CKAN 2.9
@@ -217,6 +218,25 @@ def fetch_stage(self, harvest_object):
                         else:
                             log.warn(f'Empty OpenApi service description returned for {generate_service_name(service)}')
 
+                    if service.service_type.lower() == 'rest':
+                        try:
+                            path = '/'.join(['getRest',
+                                             dataset['xRoadInstance'],
+                                             dataset['xRoadMemberClass'],
+                                             dataset['xRoadMemberCode'],
+                                             subsystem.subsystem_code,
+                                             service.service_code])
+                            rest_services_data = xroad_catalog_query_json(path)
+                            service.rest_services = RestServices.from_dict(rest_services_data)
+                        except ContentFetchError:
+                            error = f'Could not retrieve REST services {harvest_object.id}'
+                            log.info(error, harvest_object, 'Fetch')
+                            self._save_object_error(error, harvest_object, 'Fetch')
+                        except ValueError:
+                            error = f'Error parsing REST services {harvest_object.id}'
+                            log.info(error, harvest_object, 'Fetch')
+                            self._save_object_error(error, harvest_object, 'Fetch')
+
                 dataset['subsystem_pickled'] = subsystem.serialize()
                 dataset['subsystem_dict'] = json.loads(subsystem.serialize_json())
                 harvest_object.content = json.dumps(dataset)
@@ -341,7 +361,7 @@ def import_stage(self, harvest_object):
                 'name': name,
                 'xroad_servicecode': service.service_code,
                 'xroad_serviceversion': service.service_version,
-                'xroad_service_type': service.serviceType,
+                'xroad_service_type': service.service_type,
                 'harvested_from_xroad': True,
                 'access_restriction_level': 'public'
             }
@@ -373,6 +393,13 @@ def import_stage(self, harvest_object):
                 resource_data['upload'] = FlaskFileStorage(open(file_name, 'rb'), target_name, content_length=content_length)
                 resource_data['format'] = resource_format
                 resource_data['valid_content'] = 'yes' if valid_wsdl else 'no'
+            elif service.service_type.lower() == 'rest':
+                file_name = None
+                if service.rest_services is not None:
+                    endpoints = [endpoint.as_dict()
+                                 for rest_service in service.rest_services.services
+                                 for endpoint in rest_service.endpoints]
+                    resource_data['rest_endpoints'] = {'endpoints': endpoints}
             elif unknown_service_link_url is None:
                 log.warn('Unknown type service %s.%s harvested, but '
                          'ckanext.xroad_integration.unknown_service_link_url is not set!',
@@ -758,10 +785,6 @@ def _is_valid_wsdl(self, text_content):
         return True
 
 
-class ContentFetchError(Exception):
-    pass
-
-
 def generate_service_name(service) -> Optional[str]:
     if service.service_code is None:
         return None

ckanext/xroad_integration/harvesters/xroad_types.py

@@ -2,7 +2,8 @@
 from typing import Optional, List
 from datetime import datetime
 
-from .xroad_types_utils import Base, optional, date_value, class_value, xroad_list_value, xroad_service_version_value
+from .xroad_types_utils import (Base, optional, date_value, class_value, xroad_list_value,
+                                xroad_service_version_value, class_list_value)
 
 
 @dataclass
@@ -13,6 +14,38 @@ class Error(Base):
     detail: str
 
 
+@dataclass
+class RestServiceEndpoint(Base):
+    method: str
+    path: str
+
+
+@dataclass
+class RestService(Base):
+    field_map = {'endpointList': 'endpoints',
+                 'xroadInstance': 'instance',
+                 'memberClass': 'member_class',
+                 'memberCode': 'member_code',
+                 'subsystemCode': 'subsystem_code',
+                 'serviceCode': 'service_code',
+                 'serviceVersion': 'service_version'}
+    value_map = {'endpoints': class_list_value(RestServiceEndpoint)}
+    endpoints: List[RestServiceEndpoint]
+    instance: str
+    member_class: str
+    member_code: str
+    subsystem_code: str
+    service_code: str
+    service_version: str
+
+
+@dataclass
+class RestServices(Base):
+    field_map = {'listOfServices': 'services'}
+    value_map = {'services': class_list_value(RestService)}
+    services: List[RestService]
+
+
 @dataclass
 class ServiceDescription(Base):
     field_map = {'externalId': 'external_id'}
@@ -33,7 +66,8 @@ class ServiceDescription(Base):
 @dataclass
 class Service(Base):
     field_map = {'serviceCode': 'service_code',
-                 'serviceVersion': 'service_version'}
+                 'serviceVersion': 'service_version',
+                 'serviceType': 'service_type'}
     value_map = {
             'service_version': xroad_service_version_value,
             'wsdl': optional(class_value(ServiceDescription)),
@@ -48,10 +82,11 @@ class Service(Base):
     changed: datetime
     fetched: datetime
     service_version: Optional[str] = field(default=None)
-    serviceType: Optional[str] = field(default=None)
+    service_type: Optional[str] = field(default=None)
     wsdl: Optional[ServiceDescription] = field(default=None)
     openapi: Optional[ServiceDescription] = field(default=None)
     removed: Optional[datetime] = field(default=None)
+    rest_services: Optional[RestServices] = field(default=None)
 
 
 @dataclass

ckanext/xroad_integration/harvesters/xroad_types_utils.py

@@ -119,3 +119,9 @@ def xroad_service_version_value(v) -> Optional[str]:
 
 def class_value(cls):
     return cls.from_dict
+
+
+def class_list_value(cls):
+    def parse(items) -> List[cls]:
+        return [cls.from_dict(item) for item in items]
+    return parse

ckanext/xroad_integration/logic/action.py

@@ -6,62 +6,31 @@
 from dateutil import relativedelta
 from sqlalchemy import and_, not_
 
-import requests
 import datetime
 import six
 
 from ckan import model
 from requests.exceptions import ConnectionError
-from requests.adapters import HTTPAdapter
-from requests.packages.urllib3.util.retry import Retry
 from ckan.plugins import toolkit
 from pprint import pformat
-from typing import Dict, Any, List, Union
 
 from ckanext.xroad_integration.model import (XRoadError, XRoadStat, XRoadServiceList, XRoadServiceListMember,
                                              XRoadServiceListSubsystem, XRoadServiceListService,
                                              XRoadServiceListSecurityServer, XRoadBatchResult, XRoadDistinctServiceStat,
                                              XRoadHeartbeat)
+from ckanext.xroad_integration.xroad_utils import xroad_catalog_query_json, ContentFetchError, http
 
 
-# Type for json
-Json = Union[Dict[str, "Json"], List["Json"], str, int, float, bool, None]
-
 # PUBLIC_ORGANIZATION_CLASSES = ['GOV', 'MUN', 'ORG']
 # COMPANY_CLASSES = ['COM']
 
-DEFAULT_TIMEOUT = 3  # seconds
 DEFAULT_DAYS_TO_FETCH = 1
 DEFAULT_LIST_ERRORS_HISTORY_IN_DAYS = 90
 DEFAULT_LIST_ERRORS_PAGE_LIMIT = 20
 
-
-# Add default timeout
-class TimeoutHTTPAdapter(HTTPAdapter):
-    def __init__(self, *args, **kwargs):
-        self.timeout = DEFAULT_TIMEOUT
-        if "timeout" in kwargs:
-            self.timeout = kwargs["timeout"]
-            del kwargs["timeout"]
-        super(TimeoutHTTPAdapter, self).__init__(*args, **kwargs)
-
-
-retry_strategy = Retry(
-    total=3,
-    backoff_factor=1
-)
-
-adapter = TimeoutHTTPAdapter(max_retries=retry_strategy)
-http = requests.Session()
-http.mount("http://", adapter)
-
 log = logging.getLogger(__name__)
 
 
-class ContentFetchError(Exception):
-    pass
-
-
 def update_xroad_organizations(context, data_dict):
     toolkit.check_access('update_xroad_organizations', context)
     harvest_source_list = toolkit.get_action('harvest_source_list')
@@ -496,64 +465,6 @@ def _fetch_error_page(params, queryparams, pagination) -> (int, int):
     return error_data.get('numberOfPages', 0), error_count
 
 
-def xroad_catalog_query(service, params: List = None,
-                        queryparams: Dict[str, Any] = None, content_type='application/json', accept='application/json',
-                        pagination: Dict[str, str] = None):
-    if params is None:
-        params = []
-    if queryparams is None:
-        queryparams = {}
-
-    xroad_catalog_address = toolkit.config.get('ckanext.xroad_integration.xroad_catalog_address', '')  # type: str
-    xroad_catalog_certificate = toolkit.config.get('ckanext.xroad_integration.xroad_catalog_certificate')
-    xroad_client_id = toolkit.config.get('ckanext.xroad_integration.xroad_client_id')
-    xroad_client_certificate = toolkit.config.get('ckanext.xroad_integration.xroad_client_certificate')
-
-    if not xroad_catalog_address.startswith('http'):
-        log.warn("Invalid X-Road catalog url %s" % xroad_catalog_address)
-        raise ContentFetchError("Invalid X-Road catalog url %s" % xroad_catalog_address)
-
-    url = '{address}/{service}'.format(address=xroad_catalog_address, service=service)
-
-    if pagination:
-        queryparams['page'] = pagination['page']
-        queryparams['limit'] = pagination['limit']
-
-    for param in params:
-        url += '/' + param
-
-    headers = {'Accept': accept,
-               'Content-Type': content_type,
-               'X-Road-Client': xroad_client_id}
-
-    certificate_args = {}
-    if xroad_catalog_certificate and os.path.isfile(xroad_catalog_certificate):
-        certificate_args['verify'] = xroad_catalog_certificate
-    else:
-        certificate_args['verify'] = False
-
-    if xroad_client_certificate and os.path.isfile(xroad_client_certificate):
-        certificate_args['cert'] = xroad_client_certificate
-
-    return http.get(url, params=queryparams, headers=headers, **certificate_args)
-
-
-def xroad_catalog_query_json(service, params: List = None, queryparams: Dict[str, Any] = None,
-                             pagination: Dict[str, str] = None) -> Json:
-    if params is None:
-        params = []
-    if queryparams is None:
-        queryparams = {}
-    response = xroad_catalog_query(service, params=params, queryparams=queryparams, pagination=pagination)
-    if response.status_code == 204:
-        log.warning("Received empty response for service %s", service)
-        return
-    try:
-        return response.json()
-    except requests.exceptions.JSONDecodeError as e:
-        raise ContentFetchError(f'Expected JSON: {e}')
-
-
 def fetch_xroad_service_list(context, data_dict):
     toolkit.check_access('fetch_xroad_service_list', context)
 

ckanext/xroad_integration/tests/fixtures.py

@@ -51,7 +51,11 @@
     'get_list_errors_data': {
         'host': '127.0.0.1',
         'port': 9199,
-        'content': 'xroad-catalog-mock-responses/test_list_errors.json'}
+        'content': 'xroad-catalog-mock-responses/test_list_errors.json'},
+    'getRest': {
+        'host': '127.0.0.1',
+        'port': 9200,
+        'content': 'xroad-catalog-mock-responses/test_getrest.json'},
     }
 
 

ckanext/xroad_integration/tests/test_plugin.py

@@ -480,3 +480,17 @@ def test_xroad_get_organizations_empty_data(xroad_rest_adapter_mocks, xroad_rest
     result = call_action('update_xroad_organizations', context=context)
     assert result['success'] is True
     assert result['message'] == 'Updated 0 organizations'
+
+
+@pytest.mark.usefixtures('with_plugins', 'clean_db', 'clean_index', 'xroad_database_setup')
+@pytest.mark.ckan_config('ckan.plugins', 'apicatalog scheming_datasets scheming_organizations fluent harvest '
+                                         'xroad_harvester xroad_integration')
+@pytest.mark.ckan_config('ckanext.xroad_integration.xroad_catalog_address', xroad_rest_service_url('getRest'))
+def test_getrest(xroad_rest_adapter_mocks, xroad_rest_mocks):
+    harvester = XRoadHarvesterPlugin()
+    run_harvest(url=xroad_rest_adapter_url('base'), harvester=harvester, config=json.dumps({"force_all": True}))
+
+    subsystem = call_action('package_show', id='TEST.ORG.000003-3.LargeSubsystem')
+    rest_service = next(s for s in subsystem.get('resources', []) if s['xroad_servicecode'] == 'restService')
+    assert {'method': 'POST', 'path': '/PostSomething/v1'} in rest_service['rest_endpoints']['endpoints']
+    assert {'method': 'GET', 'path': '/ComeGetSome/v1'} in rest_service['rest_endpoints']['endpoints']

ckanext/xroad_integration/tests/xroad_mock/xroad_rest_mock.py

@@ -32,6 +32,19 @@ def getOrganization(business_code='000000-0'):
     @app.route('/listErrors/<instance>/<code>/<member>')
     def list_errors(instance='TEST', code='000000-0', member='some_member'):
         return mock_data
+
+    @app.route('/getRest/<instance>/<member_class>/<member_code>/<subsystem_code>/<service_code>')
+    @app.route('/getRest/<instance>/<member_class>/<member_code>/<subsystem_code>/<service_code>/<service_version>')
+    def getRest(instance, member_class, member_code, subsystem_code, service_code, service_version=None):
+        result = mock_data.copy()
+        for service in result.get('listOfServices', []):
+            service['xroadInstance'] = instance
+            service['memberClass'] = member_class
+            service['memberCode'] = member_code
+            service['subsystemCode'] = subsystem_code
+            service['serviceCode'] = service_code
+        return result
+
     return app