6713509976

Committed 31 Oct 2023 11:13PM UTC coverage: 86.773% (+0.7%) from 86.072%

Build # 6713509976

Build Type

push

github-actions

Committed by

web-flow

Commit Message

Merge pull request #699 from TOMToolkit/dev

Multi-Feature Merge. Please Review Carefully.

Run Details

795 of 795 new or added lines in 39 files covered. (100.0%)

8253 of 9511 relevant lines covered (86.77%)

0.87 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

93.75

/tom_targets/sharing.py

import requests

from django.conf import settings
from django.core.exceptions import ImproperlyConfigured

from tom_targets.serializers import TargetSerializer
from tom_dataproducts.sharing import get_destination_target


def share_target_with_tom(share_destination, form_data, target_lists=()):
    """
    Share a target with a remote TOM.
    :param share_destination: The name of the destination TOM as defined in settings.DATA_SHARING
    :param form_data: The form data from the target form
    :param target_lists: The target lists to add the target to in the destination TOM
    :return: The response from the destination TOM
    """
    # Try to get destination tom authentication/URL information
    try:
        destination_tom_base_url = settings.DATA_SHARING[share_destination]['BASE_URL']
        username = settings.DATA_SHARING[share_destination]['USERNAME']
        password = settings.DATA_SHARING[share_destination]['PASSWORD']
    except KeyError as err:
        raise ImproperlyConfigured(f'Check DATA_SHARING configuration for {share_destination}: Key {err} not found.')
    auth = (username, password)
    headers = {'Content-Type': 'application/json', 'Accept': 'application/json'}

    # establish destination TOM URLs
    targets_url = destination_tom_base_url + 'api/targets/'

    # Check if target already exists in destination DB
    destination_target_id, target_search_response = get_destination_target(form_data['target'], targets_url, headers,
                                                                           auth)
    # Handle errors or multiple targets found
    if target_search_response.status_code != 200:
        return target_search_response
    elif isinstance(destination_target_id, list) and len(destination_target_id) > 1:
        return {'message': 'ERROR: Multiple targets with matching name found in destination TOM.'}

    # Build list of targetlists to add target to in destination TOM
    target_dict_list = [{'name': f'Imported From {settings.TOM_NAME}'}]
    for target_list in target_lists:
        target_dict_list.append({'name': target_list.name})

    # Create or update target in destination TOM
    if destination_target_id is None:
        # If target is not in Destination, serialize and create new target.
        serialized_target = TargetSerializer(form_data['target']).data
        # Remove local User Groups
        serialized_target['groups'] = []
        # Add target lists
        serialized_target['target_lists'] = target_dict_list
        target_create_response = requests.post(targets_url, json=serialized_target, headers=headers, auth=auth)
    else:
        # Add target to target lists if it already exists in destination TOM
        update_target_data = {'target_lists': target_dict_list}
        update_target_url = targets_url + f'{destination_target_id}/'
        target_create_response = requests.patch(update_target_url, json=update_target_data, headers=headers, auth=auth)
    return target_create_response

1	import requests	1✔
2
3	from django.conf import settings	1✔
4	from django.core.exceptions import ImproperlyConfigured	1✔
5
6	from tom_targets.serializers import TargetSerializer	1✔
7	from tom_dataproducts.sharing import get_destination_target	1✔
8
9
10	def share_target_with_tom(share_destination, form_data, target_lists=()):	1✔
11	"""
12	Share a target with a remote TOM.
13	:param share_destination: The name of the destination TOM as defined in settings.DATA_SHARING
14	:param form_data: The form data from the target form
15	:param target_lists: The target lists to add the target to in the destination TOM
16	:return: The response from the destination TOM
17	"""
18	# Try to get destination tom authentication/URL information
19	try:	1✔
20	destination_tom_base_url = settings.DATA_SHARING[share_destination]['BASE_URL']	1✔
21	username = settings.DATA_SHARING[share_destination]['USERNAME']	1✔
22	password = settings.DATA_SHARING[share_destination]['PASSWORD']	1✔
23	except KeyError as err:	×
24	raise ImproperlyConfigured(f'Check DATA_SHARING configuration for {share_destination}: Key {err} not found.')	×
25	auth = (username, password)	1✔
26	headers = {'Content-Type': 'application/json', 'Accept': 'application/json'}	1✔
27
28	# establish destination TOM URLs
29	targets_url = destination_tom_base_url + 'api/targets/'	1✔
30
31	# Check if target already exists in destination DB
32	destination_target_id, target_search_response = get_destination_target(form_data['target'], targets_url, headers,	1✔
33	auth)
34	# Handle errors or multiple targets found
35	if target_search_response.status_code != 200:	1✔
36	return target_search_response	1✔
37	elif isinstance(destination_target_id, list) and len(destination_target_id) > 1:	1✔
38	return {'message': 'ERROR: Multiple targets with matching name found in destination TOM.'}	1✔
39
40	# Build list of targetlists to add target to in destination TOM
41	target_dict_list = [{'name': f'Imported From {settings.TOM_NAME}'}]	1✔
42	for target_list in target_lists:	1✔
43	target_dict_list.append({'name': target_list.name})	1✔
44
45	# Create or update target in destination TOM
46	if destination_target_id is None:	1✔
47	# If target is not in Destination, serialize and create new target.
48	serialized_target = TargetSerializer(form_data['target']).data	1✔
49	# Remove local User Groups
50	serialized_target['groups'] = []	1✔
51	# Add target lists
52	serialized_target['target_lists'] = target_dict_list	1✔
53	target_create_response = requests.post(targets_url, json=serialized_target, headers=headers, auth=auth)	1✔
54	else:
55	# Add target to target lists if it already exists in destination TOM
56	update_target_data = {'target_lists': target_dict_list}	1✔
57	update_target_url = targets_url + f'{destination_target_id}/'	1✔
58	target_create_response = requests.patch(update_target_url, json=update_target_data, headers=headers, auth=auth)	1✔
59	return target_create_response	1✔

TOMToolkit / tom_base / 6713509976

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous