New feature : copy technologies from a clientspace to another

tests/test_copy_technologies.py

@@ -0,0 +1,59 @@
+from toolbox.api.datagalaxy_api import DataGalaxyApiAuthentication, Token
+from toolbox.api.datagalaxy_api_technologies import DataGalaxyApiTechnology
+from toolbox.commands.copy_technologies import copy_technologies
+import pytest as pytest
+
+
+def list_mock_technologies_without_custom():
+    return [
+        {
+            'creationUserId': '00000000-0000-0000-0000-000000000000',
+        }
+    ]
+
+
+def test_copy_technologies_when_nothing_on_source(mocker):
+    """
+    Scenario 1. error
+    :param mocker:
+    :return: raise Exception
+    """
+    client_space_mock = mocker.patch.object(Token, 'get_client_space_id', autospec=True)
+    client_space_mock.return_value = 'cid'
+    api_authenticate_mock = mocker.patch.object(DataGalaxyApiAuthentication, 'authenticate', autospec=True)
+    api_authenticate_mock.return_value = 'token'
+    technologies_list_mock = mocker.patch.object(DataGalaxyApiTechnology, 'list_technologies', autospec=True)
+    technologies_list_mock.return_value = []
+
+    # ASSERT / VERIFY
+    with pytest.raises(Exception, match='no technology found on source clientspace'):
+        copy_technologies(
+            url_source='url_source',
+            url_target='url_target',
+            integration_token_source_value='integration_token_source',
+            integration_token_target_value='integration_token_target'
+        )
+
+
+def test_copy_technologies_when_no_custom_technology_on_source(mocker):
+    """
+    Scenario 1. error
+    :param mocker:
+    :return: raise Exception
+    """
+    client_space_mock = mocker.patch.object(Token, 'get_client_space_id', autospec=True)
+    client_space_mock.return_value = 'cid'
+    api_authenticate_mock = mocker.patch.object(DataGalaxyApiAuthentication, 'authenticate', autospec=True)
+    api_authenticate_mock.return_value = 'token'
+    technologies_list_mock = mocker.patch.object(DataGalaxyApiTechnology, 'list_technologies', autospec=True)
+    technologies_list_mock.return_value = list_mock_technologies_without_custom()
+
+    # ASSERT / VERIFY
+    result = copy_technologies(
+            url_source='url_source',
+            url_target='url_target',
+            integration_token_source_value='integration_token_source',
+            integration_token_target_value='integration_token_target'
+        )
+
+    assert result == 0

toolbox/__main__.py

@@ -3,6 +3,7 @@
 import sys
 
 from toolbox.commands.copy_attributes import copy_attributes_parse, copy_attributes
+from toolbox.commands.copy_technologies import copy_technologies_parse, copy_technologies
 from toolbox.commands.copy_usages import copy_usages, copy_usages_parse
 from toolbox.commands.copy_dataprocessings import copy_dataprocessings, copy_dataprocessings_parse
 from toolbox.commands.delete_attributes import delete_attributes_parse, delete_attributes
@@ -23,6 +24,7 @@ def run(args):
                         action="store_true")
     subparsers = parser.add_subparsers(help='sub-command help', dest='subparsers_name')
     copy_attributes_parse(subparsers)
+    copy_technologies_parse(subparsers)
     delete_attributes_parse(subparsers)
     copy_glossary_parse(subparsers)
     copy_usages_parse(subparsers)
@@ -45,6 +47,17 @@ def run(args):
         logging.info("<<< copy_attributes")
         return 0
 
+    if result.subparsers_name == 'copy-technologies':
+        logging.info(">>> copy_technologies")
+        copy_technologies(
+            result.url_source,
+            result.url_target,
+            result.token_source,
+            result.token_target
+        )
+        logging.info("<<< copy_technologies")
+        return 0
+
     if result.subparsers_name == 'delete-attributes':
         logging.info(">>> delete_attributes")
         delete_attributes(result.url, result.token)

toolbox/api/datagalaxy_api_technologies.py

@@ -0,0 +1,29 @@
+import logging
+import requests as requests
+
+
+class DataGalaxyApiTechnology:
+    def __init__(self, url: str, access_token: str):
+        self.url = url
+        self.access_token = access_token
+
+    def list_technologies(self) -> list:
+        headers = {'Authorization': f"Bearer {self.access_token}"}
+        response = requests.get(f"{self.url}/technologies", headers=headers)
+        code = response.status_code
+        body_json = response.json()
+        if code != 200:
+            raise Exception(body_json['error'])
+
+        logging.info(f'list_technologies - {len(body_json["technologies"])} technologies found')
+        result = body_json['technologies']
+        return result
+
+    def insert_technology(self, technology) -> object:
+        headers = {'Authorization': f"Bearer {self.access_token}"}
+        response = requests.post(f"{self.url}/technologies", json=technology, headers=headers)
+        code = response.status_code
+        body_json = response.json()
+        if code != 201:
+            raise Exception(body_json['error'])
+        return body_json

toolbox/commands/copy_technologies.py

@@ -0,0 +1,77 @@
+import logging
+
+from toolbox.api.datagalaxy_api import get_access_token, Token
+from toolbox.api.datagalaxy_api_technologies import DataGalaxyApiTechnology
+
+
+def copy_technologies(url_source: str,
+                      url_target: str,
+                      integration_token_source_value: str,
+                      integration_token_target_value: str) -> int:
+    integration_token_source = Token(integration_token_source_value)
+    integration_token_target = Token(integration_token_target_value)
+
+    source_client_space_id = integration_token_source.get_client_space_id()
+
+    access_token_source = get_access_token(url_source, integration_token_source)
+    access_token_target = get_access_token(url_target, integration_token_target)
+
+    technologies_api_source = DataGalaxyApiTechnology(url=url_source, access_token=access_token_source)
+    technologies_api_target = DataGalaxyApiTechnology(url=url_target, access_token=access_token_target)
+
+    source_technologies = technologies_api_source.list_technologies()
+    target_technologies = technologies_api_target.list_technologies()
+
+    if len(source_technologies) == 0:
+        raise Exception('no technology found on source clientspace')
+
+    # custom technologies have a "creationUserId" that is not set on the default value
+    source_custom_technologies = list(filter(lambda technology: technology['creationUserId'] != "00000000-0000-0000-0000-000000000000", source_technologies))
+
+    if len(source_custom_technologies) == 0:
+        logging.info(f'copy_technologies - No custom technology found on source client_space: {source_client_space_id}, aborting.')
+        return 0
+
+    logging.info(f'copy_technologies - {len(source_custom_technologies)} custom technologies found on client_space: {source_client_space_id}')
+
+    target_technology_codes = list(map(lambda t: t['technologyCode'], target_technologies))
+
+    count_created_technologies = 0
+    for source_custom in source_custom_technologies:
+        # We check that the custom technology code does not already exists on the target clientspace
+        if source_custom['technologyCode'] in target_technology_codes:
+            logging.info(f'copy_technologies - {source_custom["technologyCode"]} already exists in target')
+            continue
+
+        # That is not the case, so we create the technology in the target clientspace
+        logging.info(f'copy_technologies - {source_custom["technologyCode"]} does not exists in target')
+        technologies_api_target.insert_technology(source_custom)
+        count_created_technologies += 1
+
+    logging.info(f'copy_technologies - {count_created_technologies} technologies were created on target client_space: {source_client_space_id}')
+    return count_created_technologies
+
+
+def copy_technologies_parse(subparsers):
+    # create the parser for the "copy-technologies" command
+    copy_technologies_parse = subparsers.add_parser('copy-technologies', help='copy-technologies help')
+    copy_technologies_parse.add_argument(
+        '--url-source',
+        type=str,
+        help='url source',
+        required=True)
+    copy_technologies_parse.add_argument(
+        '--url-target',
+        type=str,
+        help='url target',
+        required=True)
+    copy_technologies_parse.add_argument(
+        '--token-source',
+        type=str,
+        help='integration source token',
+        required=True)
+    copy_technologies_parse.add_argument(
+        '--token-target',
+        type=str,
+        help='integration target token',
+        required=True)