Commit 2fb049a9 authored by Sutton's avatar Sutton
Browse files

Adding legal client capabilities

parent e454e27f
import sys, os
import importlib
from configparser import ConfigParser
from configparser import SafeConfigParser
import requests
from osdu_api.model.http_method import HttpMethod
......@@ -34,7 +34,7 @@ class BaseClient:
entitlements_module_name=entitlements_client
'''
def _parse_config(self):
config_parser = ConfigParser()
config_parser = SafeConfigParser(os.environ)
config_file_name = 'osdu_api.ini'
found_names = config_parser.read(config_file_name)
if config_file_name not in found_names:
......@@ -45,6 +45,7 @@ class BaseClient:
self.search_url = config_parser.get('environment', 'search_url')
self.data_workflow_url = config_parser.get('environment', 'data_workflow_url')
self.file_dms_url = config_parser.get('environment', 'file_dms_url')
self.legal_url = config_parser.get('environment', 'legal_url')
self.dataset_registry_url = config_parser.get('environment', 'dataset_registry_url')
self.use_service_principal = config_parser.get('environment', 'use_service_principal')
......
import json
from typing import List
from osdu_api.clients.base_client import BaseClient
from osdu_api.model.legal.legal_tag import LegalTag
from osdu_api.model.legal.update_legal_tag import UpdateLegalTag
from osdu_api.model.legal.legal_tag_names import LegalTagNames
from osdu_api.model.http_method import HttpMethod
'''
Holds the logic for interfacing with Legal's api
'''
class LegalClient(BaseClient):
def get_legal_tag(self, legal_tag_name: str, bearer_token=None):
return self.make_request(method=HttpMethod.GET, url='{}{}/{}'.format(self.legal_url, '/legaltags', legal_tag_name), bearer_token=bearer_token)
def create_legal_tag(self, legal_tag: LegalTag, bearer_token=None):
return self.make_request(method=HttpMethod.POST, url='{}{}'.format(self.legal_url, '/legaltags'),
data=legal_tag.to_JSON(), bearer_token=bearer_token)
def delete_legal_tag(self, legal_tag_name: str, bearer_token=None):
return self.make_request(method=HttpMethod.DELETE, url='{}{}/{}'.format(self.legal_url, '/legaltags', legal_tag_name), bearer_token=bearer_token)
def list_legal_tags(self, bearer_token=None):
return self.make_request(method=HttpMethod.GET, url='{}{}'.format(self.legal_url, '/legaltags'), bearer_token=bearer_token)
def update_legal_tag(self, update_legal_tag: UpdateLegalTag, bearer_token=None):
return self.make_request(method=HttpMethod.PUT, url='{}{}'.format(self.legal_url, '/legaltags'),
data=update_legal_tag.to_JSON(), bearer_token=bearer_token)
def get_legal_tags(self, legal_tag_names: LegalTagNames, bearer_token=None):
return self.make_request(method=HttpMethod.POST, url='{}{}'.format(self.legal_url, '/legaltags:batchRetrieve'),
data=legal_tag_names.to_JSON(), bearer_token=bearer_token)
def validate_legal_tags(self, legal_tag_names: LegalTagNames, bearer_token=None):
return self.make_request(method=HttpMethod.POST, url='{}{}'.format(self.legal_url, '/legaltags:validate'),
data=legal_tag_names.to_JSON(), bearer_token=bearer_token)
def get_legal_tag_properties(self, bearer_token=None):
return self.make_request(method=HttpMethod.GET, url='{}{}'.format(self.legal_url, '/legaltags:properties'), bearer_token=bearer_token)
import json
from osdu_api.data_workflow.data_workflow_client import DataWorkflowClient
from osdu_api.dataset_registry.dataset_registry_client import DatasetRegistryClient
from osdu_api.file_dms.file_dms_client import FileDMSClient
from osdu_api.clients.data_workflow.data_workflow_client import DataWorkflowClient
from osdu_api.clients.dataset_registry.dataset_registry_client import DatasetRegistryClient
from osdu_api.clients.file_dms.file_dms_client import FileDMSClient
from osdu_api.model.file_dms.file import File
from osdu_api.model.file_dms.register_files import RegisterFiles
from osdu_api.model.storage.record import Record
......
from osdu_api.clients.legal.legal_client import LegalClient
from osdu_api.model.legal.legal_tag import LegalTag
from osdu_api.model.legal.legal_tag_properties import LegalTagProperties
legal_client = LegalClient()
legal_tag_properties = LegalTagProperties(['US'], 'A1234', 2222222222222, 'default', 'Public Domain Data', 'Public', 'No Personal Data', 'EAR99')
legal_tag = LegalTag('public-usa-dataset-3', legal_tag_properties, 'a default legal tag')
response = legal_client.create_legal_tag(legal_tag)
print(response.status_code)
response = legal_client.get_legal_tag('opendes-public-usa-dataset-3')
print(response.content)
[environment]
data_partition_id=opendes
storage_url=$STORAGE_BASE_URL/api/storage/v2
search_url=$SEARCH_BASE_URL/api/search/v2
data_workflow_url=$DATA_WORKFLOW_URL/api/data-workflow/v1
file_dms_url=$FILE_DMS_URL/api/filedms/v2
dataset_registry_url=$DATASET_REGISTRY_URL/api/dataset-registry/v1
storage_url=%(STORAGE_BASE_URL)s/api/storage/v2
search_url=%(SEARCH_BASE_URL)s/api/search/v2
legal_url=%(LEGAL_BASE_URL)s/api/legal/v1
data_workflow_url=%(DATA_WORKFLOW_BASE_URL)s/api/data-workflow/v1
file_dms_url=%(FILE_DMS_BASE_URL)s/api/filedms/v2
dataset_registry_url=%(DATASET_REGISTRY_BASE_URL)s/api/dataset-registry/v1
use_service_principal=True
[provider]
name=aws
service_principal_module_name=service_principal_util
token_url=$TOKEN_URL
aws_oauth_custom_scope=$CUSTOM_SCOPE
client_id_ssm_path=/osdu/$ENVIRONMENT/client-credentials-client-id
client_secret_name=/osdu/$ENVIRONMENT/client_credenials_secret
token_url=%(AUTH_TOKEN_URL)s
aws_oauth_custom_scope=%(CUSTOM_SCOPE)s
client_id_ssm_path=/osdu/%(ENVIRONMENT)s/client-credentials-client-id
client_secret_name=/osdu/%(ENVIRONMENT)s/client_credenials_secret
client_secret_dict_key=client_credentials_client_secret
region_name=us-east-1
\ No newline at end of file
region_name=%(AWS_REGION)s
\ No newline at end of file
......@@ -3,4 +3,5 @@ from enum import Enum
class HttpMethod(Enum):
GET = 0
POST = 1
PUT = 2
\ No newline at end of file
PUT = 2
DELETE = 3
\ No newline at end of file
from osdu_api.model.base import Base
from osdu_api.model.legal.legal_tag_properties import LegalTagProperties
class LegalTag(Base):
def __init__(self, name: str, properties: LegalTagProperties, description: str):
self.name = name
self.properties = properties
self.description = description
\ No newline at end of file
from typing import List
from osdu_api.model.base import Base
class LegalTagNames(Base):
def __init__(self, names: List[str]):
self.names = names
\ No newline at end of file
from typing import List
from osdu_api.model.base import Base
class LegalTagProperties(Base):
def __init__(self, country_of_origin: List[str], contract_id: str, expiration_date: int, originator: str,
data_type: str, security_classification: str, personal_data: str, export_classification: str):
self.countryOfOrigin = country_of_origin
self.contractId = contract_id
self.expirationDate = expiration_date
self.originator = originator
self.dataType = data_type
self.securityClassification = security_classification
self.personalData = personal_data
self.exportClassification = export_classification
\ No newline at end of file
from osdu_api.model.base import Base
class UpdateLegalTag(Base):
def __init__(self, name: str, contract_id: str, expiration_date: int, description: str):
self.name = name
self.contractId = contract_id
self.expirationDate = expiration_date
self.description = description
\ No newline at end of file
import os
import base64
import boto3
import requests
......@@ -47,7 +47,7 @@ def _get_secret(session, region_name, secret_name, secret_dict_key):
def get_service_principal_token():
config_parser = ConfigParser()
config_parser = ConfigParser(os.environ)
config_file_name = 'osdu_api.ini'
found_names = config_parser.read(config_file_name)
if config_file_name not in found_names:
......
[environment]
data_partition_id=opendes
storage_url=$STORAGE_BASE_URL/api/storage/v2
search_url=$SEARCH_BASE_URL/api/search/v2
data_workflow_url=$DATA_WORKFLOW_URL/api/data-workflow/v1
file_dms_url=$FILE_DMS_URL/api/filedms/v2
dataset_registry_url=$DATASET_REGISTRY_URL/api/dataset-registry/v1
storage_url=%(STORAGE_BASE_URL)s/api/storage/v2
search_url=%(SEARCH_BASE_URL)s/api/search/v2
legal_url=%(LEGAL_BASE_URL)s/api/legal/v1
data_workflow_url=%(DATA_WORKFLOW_BASE_URL)s/api/data-workflow/v1
file_dms_url=%(FILE_DMS_BASE_URL)s/api/filedms/v2
dataset_registry_url=%(DATASET_REGISTRY_BASE_URL)s/api/dataset-registry/v1
use_service_principal=True
[provider]
name=aws
service_principal_module_name=service_principal_util
token_url=$TOKEN_URL
aws_oauth_custom_scope=$CUSTOM_SCOPE
client_id_ssm_path=/osdu/$ENVIRONMENT/client-credentials-client-id
client_secret_name=/osdu/$ENVIRONMENT/client_credenials_secret
token_url=%(AUTH_TOKEN_URL)s
aws_oauth_custom_scope=%(CUSTOM_SCOPE)s
client_id_ssm_path=/osdu/%(ENVIRONMENT)s/client-credentials-client-id
client_secret_name=/osdu/%(ENVIRONMENT)s/client_credenials_secret
client_secret_dict_key=client_credentials_client_secret
region_name=us-east-1
\ No newline at end of file
region_name=%(AWS_REGION)s
\ No newline at end of file
import unittest
import mock
from osdu_api.clients.legal.legal_client import LegalClient
from osdu_api.clients.base_client import BaseClient
class TestLegalClient(unittest.TestCase):
@mock.patch.object(BaseClient, 'make_request', return_value="response")
@mock.patch.object(BaseClient, '_refresh_service_principal_token', return_value="stubbed")
def test_get_file_upload(self, get_bearer_token_mock, make_request_mock):
# Arrange
client = LegalClient()
client.service_principal_token = 'stubbed'
client.legal_url = 'stubbed url'
client.headers = {}
# Act
response = client.get_legal_tag('name')
# Assert
assert response == make_request_mock.return_value
\ No newline at end of file
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment