Commit 1b461899 authored by Sutton's avatar Sutton
Browse files

.

parent 6d1eb8ec
# Copyright © 2020 Amazon Web Services
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# Copyright © 2020 Amazon Web Services
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import json
from typing import List
from osdu_api.clients.base_client import BaseClient
from osdu_api.model.file_dms.register_files import RegisterFiles
from osdu_api.model.file_dms.get_files_delivery import GetFilesDelivery
from osdu_api.model.http_method import HttpMethod
'''
Holds the logic for interfacing with File DMS's api
'''
class FileDMSClient(BaseClient):
def get_file_upload_location(self, bearer_token=None):
return self.make_request(method=HttpMethod.GET, url='{}{}'.format(self.file_dms_url, '/getFileUploadLocation'), bearer_token=bearer_token)
def register_files(self, register_files: RegisterFiles, bearer_token=None):
return self.make_request(method=HttpMethod.POST, url='{}{}'.format(self.file_dms_url, '/registerFiles'),
data=register_files.to_JSON(), bearer_token=bearer_token)
def get_file_delivery_instructions(self, get_files_delivery: GetFilesDelivery, bearer_token=None):
return self.make_request(method=HttpMethod.POST, url='{}{}'.format(self.file_dms_url, '/getFileDeliveryInstructions'),
data=get_files_delivery.to_JSON(), bearer_token=bearer_token)
# Copyright © 2020 Amazon Web Services
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# Copyright © 2020 Amazon Web Services
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from osdu_api.model.base import Base
'''
Represents the attributes needed for OSDU to create a dataset registry record for a file
'''
class File(Base):
def __init__(self, unsigned_url: str, file_name: str, file_description:str):
self.unsignedUrl = unsigned_url
self.fileName = file_name
self.fileDescription = file_description
\ No newline at end of file
# Copyright © 2020 Amazon Web Services
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from typing import List
from osdu_api.model.base import Base
'''
A list of dataset registry record ids used against file dms's get file delivery instructions endpoint
'''
class GetFilesDelivery(Base):
def __init__(self, datasetRegistryRecordIds: List[str]):
self.datasetRegistryRecordIds = datasetRegistryRecordIds
\ No newline at end of file
# Copyright © 2020 Amazon Web Services
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import json
import re
from typing import List
from osdu_api.model.file_dms.file import File
from osdu_api.model.base import Base
'''
A list of files that are used against file dms register files endpoint
'''
class RegisterFiles(Base):
def __init__(self, files: List[File]):
self.files = files
\ No newline at end of file
# Copyright © 2020 Amazon Web Services
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import unittest
import mock
from osdu_api.clients.file_dms.file_dms_client import FileDMSClient
from osdu_api.clients.base_client import BaseClient
from osdu_api.model.http_method import HttpMethod
from osdu_api.model.file_dms.register_files import RegisterFiles
from osdu_api.model.file_dms.file import File
from osdu_api.model.file_dms.get_files_delivery import GetFilesDelivery
class TestFileDMSClient(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 = FileDMSClient()
client.service_principal_token = 'stubbed'
client.file_dms_url = 'stubbed url'
client.headers = {}
# Act
response = client.get_file_upload_location()
# Assert
assert response == make_request_mock.return_value
@mock.patch.object(BaseClient, 'make_request', return_value="response")
@mock.patch.object(BaseClient, '_refresh_service_principal_token', return_value="stubbed")
def test_register_files(self, get_bearer_token_mock, make_request_mock):
# Arrange
client = FileDMSClient()
client.service_principal_token = 'stubbed'
client.file_dms_url = 'stubbed url'
client.headers = {}
cust_file = File('path', 'name', 'desc')
register_files = RegisterFiles([cust_file])
# Act
response = client.register_files(register_files)
# Assert
assert response == make_request_mock.return_value
@mock.patch.object(BaseClient, 'make_request', return_value="response")
@mock.patch.object(BaseClient, '_refresh_service_principal_token', return_value="stubbed")
def test_files_delivery(self, get_bearer_token_mock, make_request_mock):
# Arrange
client = FileDMSClient()
client.service_principal_token = 'stubbed'
client.file_dms_url = 'stubbed url'
client.headers = {}
get_files_delivery = GetFilesDelivery(['id'])
# Act
response = client.get_file_delivery_instructions(get_files_delivery)
# 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