diff --git a/.DS_Store b/.DS_Store new file mode 100644 index 0000000..6186685 Binary files /dev/null and b/.DS_Store differ diff --git a/fabfile/APPspecific.py b/fabfile/APPspecific.py index e1afeef..63fa4e5 100644 --- a/fabfile/APPspecific.py +++ b/fabfile/APPspecific.py @@ -164,9 +164,9 @@ def start_APP_and_check_status(): with cd('{0}'.format(APP_source_dir())): virtualenv('python -m unittest discover test') # run('mkdir -p /tmp/fuse') -# virtualenv('posix_space --cfg test_vo.ini > /tmp/space.log 2>&1') +# virtualenv('posix_space --cfg test_vo_posix.ini > /tmp/space.log 2>&1') # time.sleep(2) -# virtualenv('posix_storage --cfg test_vo.ini > /tmp/storage.log 2>&1') +# virtualenv('posix_storage --cfg test_vo_posix.ini > /tmp/storage.log 2>&1') # time.sleep(2) # virtualenv('python -m pyvospace.client.fuse --host localhost --port 8080 --username test --password test --mountpoint /tmp/fuse/`` > /tmp/fusemnt.log 2>&1') # time.sleep(2) @@ -186,9 +186,9 @@ def sysinitstart_APP_and_check_status(): with cd('{0}'.format(APP_source_dir())): # virtualenv('python -m unittest discover test') # run('mkdir -p /tmp/fuse') - # virtualenv('posix_space --cfg test_vo.ini > /tmp/space.log 2>&1') + # virtualenv('posix_space --cfg test_vo_posix.ini > /tmp/space.log 2>&1') # time.sleep(2) - # virtualenv('posix_storage --cfg test_vo.ini > /tmp/storage.log 2>&1') + # virtualenv('posix_storage --cfg test_vo_posix.ini > /tmp/storage.log 2>&1') # time.sleep(2) # virtualenv('python -m pyvospace.client.fuse --host localhost --port 8080 --username test --password test --mountpoint /tmp/fuse/`` > /tmp/fusemnt.log 2>&1') # time.sleep(2) diff --git a/pyvospace/server/spaces/ngas/README.md b/pyvospace/server/spaces/ngas/README.md new file mode 100644 index 0000000..f25875c --- /dev/null +++ b/pyvospace/server/spaces/ngas/README.md @@ -0,0 +1,62 @@ +# pyvospace-ngas +NGAS plugin for PyVOSpace, implementation in Python3. + +This plugin provides a PyVOSpace implementation that uses an NGAS server as the backing store. The plugin connects to both a PostgreSQL database and one or more federated instances of an NGAS server. + +**Requirements** + +- Python3.6 or greater. +- Docker support for running a PostgreSQL server. +- Access to a running NGAS server or servers. To setup an NGAS server please see the [online documentation](https://ngas.readthedocs.io/en/latest/). +- FUSE filesystem modules (including fusepy). + +**Installation** + +For installation, including setting up the docker database instance, please see the main ICRAR PyVOSpace [documentation page](https://github.com/ICRAR/pyvospace/blob/master/README.md). + +**Basic configuration** + +The file *test_vo_ngas.ini* contains a default configuration, as shown below. + +``` +[Space] +host = localhost +port = 8082 +name = ngas +uri = icrar.org +dsn = postgres://vos_user:vos_user@localhost:5435/vospace +parameters = {} +secret_key = ZlmNyXdQgRhhrC2Wwy-gLZj7Wv6ZtoKH +domain = +use_ssl = 0 + +[Storage] +name = ngas +host = localhost +port = 8083 +parameters = {"staging_dir": "/tmp/ngas/staging/"} +use_ssl = 0 +ngas_servers=[{"hostname" : "localhost", "port" : 7777}, + {"hostname" : "localhost", "port" : 7777}] +``` + +Note the *ngas_servers* line. Multiple NGAS servers may be specified here, and it is assumed that the NGAS servers are synchronised and can work with the same information. At runtime an NGAS server will be chosen at random for both the space and storage servers. + +**Running the Space and Storage servers** + +'ngas_space --cfg test_vo_ngas.ini' +'ngas_storage --cfg test_vo_ngas.ini' + +**Unit tests** + +Unit tests are located in test/test_ngas/\*.py and assume an NGAS server is running on localhost at port 7777. + +**Common errors** + +If you see an error such as, + +``` +raise aiohttp.web.HTTPServerError(reason=f"File {filename_ngas} not deleted properly from NGAS server"), +``` + +it means that the NGAS server is not configured with caching enabled and has refused to delete a file. In this instance check if you want the NGAS server to be able to delete files and then enable caching in the NGAS server configuration. diff --git a/pyvospace/server/spaces/ngas/__init__.py b/pyvospace/server/spaces/ngas/__init__.py new file mode 100644 index 0000000..b8a3701 --- /dev/null +++ b/pyvospace/server/spaces/ngas/__init__.py @@ -0,0 +1 @@ +from pyvospace.server.spaces.ngas.space.ngas_space import * \ No newline at end of file diff --git a/pyvospace/server/spaces/ngas/auth.py b/pyvospace/server/spaces/ngas/auth.py new file mode 100644 index 0000000..b0a3dc8 --- /dev/null +++ b/pyvospace/server/spaces/ngas/auth.py @@ -0,0 +1,313 @@ +# +# ICRAR - International Centre for Radio Astronomy Research +# (c) UWA - The University of Western Australia, 2018 +# Copyright by UWA (in the framework of the ICRAR) +# All rights reserved +# +# This library is free software; you can redistribute it and/or +# modify it under the terms of the GNU Lesser General Public +# License as published by the Free Software Foundation; either +# version 2.1 of the License, or (at your option) any later version. +# +# This library is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# Lesser General Public License for more details. +# +# You should have received a copy of the GNU Lesser General Public +# License along with this library; if not, write to the Free Software +# Foundation, Inc., 59 Temple Place, Suite 330, Boston, +# MA 02111-1307 USA + +import os +import json + +# XML passing +import xml.etree.ElementTree as ElementTree +from datetime import datetime + +from aiohttp import helpers, web +from aiohttp_security.abc import AbstractAuthorizationPolicy +from aiohttp_security import remember, forget +from passlib.hash import pbkdf2_sha256 + +from pyvospace.core.model import PushToSpace, Property, NodeType +from pyvospace.server.spaces.ngas.utils import convert_to_epoch_seconds +import traceback + +PROTECTED_URI = [#'ivo://ivoa.net/vospace/core#title', + 'ivo://ivoa.net/vospace/core#creator', + #'ivo://ivoa.net/vospace/core#subject', + #'ivo://ivoa.net/vospace/core#description', + #'ivo://ivoa.net/vospace/core#publisher', + #'ivo://ivoa.net/vospace/core#contributor', + #'ivo://ivoa.net/vospace/core#date', + 'ivo://ivoa.net/vospace/core#type', + 'ivo://ivoa.net/vospace/core#format', + 'ivo://ivoa.net/vospace/core#identifier', + 'ivo://ivoa.net/vospace/core#source', + 'ivo://ivoa.net/vospace/core#language', + 'ivo://ivoa.net/vospace/core#relation', + 'ivo://ivoa.net/vospace/core#coverage', + 'ivo://ivoa.net/vospace/core#rights', + 'ivo://ivoa.net/vospace/core#availableSpace', + 'ivo://ivoa.net/vospace/core#groupread', + 'ivo://ivoa.net/vospace/core#groupwrite', + 'ivo://ivoa.net/vospace/core#publicread', + 'ivo://ivoa.net/vospace/core#quota', + 'ivo://ivoa.net/vospace/core#length', + 'ivo://ivoa.net/vospace/core#mtime', + 'ivo://ivoa.net/vospace/core#ctime', + 'ivo://ivoa.net/vospace/core#btime'] + + +class DBUserNodeAuthorizationPolicy(AbstractAuthorizationPolicy): + + def __init__(self, space_name, db_pool, ngas_hostname, ngas_port, ngas_session): + super().__init__() + self.space_name = space_name + self.db_pool = db_pool + self.ngas_hostname=ngas_hostname + self.ngas_port=ngas_port + self.ngas_session=ngas_session + + def _any_value_in_lists(self, a, b): + return any(i in a for i in b) + + def _any_property_in_protected(self, a): + return any(i.uri in PROTECTED_URI for i in a) + + async def authorized_userid(self, identity): + async with self.db_pool.acquire() as conn: + results = await conn.fetchrow("select * from users " + "where username=$1 and space_name=$2", + identity, self.space_name) + if not results: + return None + return results['username'] + + def make_dummy_metadata(self, node): + # Special function to make dummy metadata for nodes that are in the + # database but not necessarily in NGAS + if node is not None: + # Set zero size and use now as the date? + st_size = 0 + + # Set creation and modification time to now? + st_mtime = convert_to_epoch_seconds(datetime.today()) + st_ctime = convert_to_epoch_seconds(datetime.today()) + + # Make a zeroed out dictionary for now + struct_statvfs_dict = dict((key, 0) for key in ('f_bavail', 'f_bfree', + 'f_blocks', 'f_bsize', + 'f_favail', 'f_ffree', + 'f_files', 'f_flag', + 'f_frsize', 'f_namemax')) + + prop_length = Property('ivo://ivoa.net/vospace/core#length', st_size) + prop_btime = Property('ivo://ivoa.net/vospace/core#btime', st_mtime) + prop_ctime = Property('ivo://ivoa.net/vospace/core#ctime', st_ctime) + prop_mtime = Property('ivo://ivoa.net/vospace/core#mtime', st_mtime) + prop_statfs = Property('ivo://icrar.org/vospace/core#statfs', json.dumps(struct_statvfs_dict)) + + node.add_property(prop_length) + node.add_property(prop_btime) + node.add_property(prop_ctime) + node.add_property(prop_mtime) + node.add_property(prop_statfs) + + return True + else: + return False + + async def permits(self, identity, permission, context=None): + async with self.db_pool.acquire() as conn: + user = await conn.fetchrow("select * from users " + "where username=$1 and space_name=$2", + identity, self.space_name) + if not user: + raise web.HTTPForbidden(f"{identity} not found.") + + if permission == 'createNode': + parent = context[0] + node = context[1] + modify_properties = self._any_property_in_protected(node.properties.values()) + # User trying to create a protected property + if modify_properties is True: + return False + # allow root node creation + if parent.path == '/' and user['admin']: + return True + else: + # check if the parent container is owned by the user + if parent.owner == identity: + return True + return self._any_value_in_lists(parent.group_write, user['groupwrite']) + + elif permission == 'setNode': + node = context + modify_properties = self._any_property_in_protected(context.properties.values()) + # User trying to update a protected property + if modify_properties is True: + return False + if node.owner == identity: + return True + return self._any_value_in_lists(node.group_write, user['groupwrite']) + + elif permission == 'getNode': + + try: + node = context + + # Check if we are using container types + if node.node_type != NodeType.ContainerNode: + + # Do we work with container types? + ngas_filename=f"{os.path.basename(node.path)}_{node.id}" + + # Url to talk to NGAS + url=f"http://{self.ngas_hostname}:{self.ngas_port}/STATUS" + + # Fetch the STATUS from NGAS + params = {"file_id": ngas_filename} + resp = await self.ngas_session.get(url, params=params) + + # Read all the response content and parse in as XML + lines = await resp.content.read() + + xmltree=ElementTree.fromstring(lines) + + # Create a dictionary of all XML elements in the tree + elements={t.tag : t for t in xmltree.iter()} + + if resp.status!=200 or "FileStatus" not in elements: + return (self.make_dummy_metadata(node)) + + filestatus=elements["FileStatus"] + + # File size + st_size = int(filestatus.get('FileSize')) + # Creation time + st_ctime = filestatus.get("IngestionDate") + # Modification time + st_mtime = filestatus.get('ModificationDate') + if st_mtime == "": + st_mtime = st_ctime + + # Convert these dates to seconds since the UNIX epoch (1,1,1970) + st_ctime = convert_to_epoch_seconds(st_ctime) + st_mtime = convert_to_epoch_seconds(st_mtime) + + # Make a zero'd dictionary for now + struct_statvfs_dict = dict((key, 0) for key in ('f_bavail', 'f_bfree', + 'f_blocks', 'f_bsize', + 'f_favail', 'f_ffree', + 'f_files', 'f_flag', + 'f_frsize', 'f_namemax')) + + prop_length = Property('ivo://ivoa.net/vospace/core#length', st_size) + prop_btime = Property('ivo://ivoa.net/vospace/core#btime', st_mtime) + prop_ctime = Property('ivo://ivoa.net/vospace/core#ctime', st_ctime) + prop_mtime = Property('ivo://ivoa.net/vospace/core#mtime', st_mtime) + prop_statfs = Property('ivo://icrar.org/vospace/core#statfs', json.dumps(struct_statvfs_dict)) + + node.add_property(prop_length) + node.add_property(prop_btime) + node.add_property(prop_ctime) + node.add_property(prop_mtime) + node.add_property(prop_statfs) + + return True + + else: + return self.make_dummy_metadata(node) + + except Exception as e: + traceback.print_exc() + # Revoke permission? + return False + + + elif permission in ('moveNode', 'copyNode'): + src = context[0] + dest = context[1] + if dest.path == '/' and user['admin']: + return True + if src.owner == identity and dest.owner == identity: + return True + if self._any_value_in_lists(src.group_write, user['groupwrite']) and \ + self._any_value_in_lists(dest.group_write, user['groupwrite']): + return True + return False + + elif permission == 'createTransfer': + return True + + elif permission == 'deleteNode': + node = context + if node.owner == identity: + return True + return self._any_value_in_lists(node.group_write, user['groupwrite']) + + elif permission == 'dataTransfer': + job = context + if job.transfer.target.owner == identity: + return True + if isinstance(job.transfer, PushToSpace): + return self._any_value_in_lists(job.transfer.target.group_write, user['groupwrite']) + else: + return self._any_value_in_lists(job.transfer.target.group_read, user['groupread']) or \ + self._any_value_in_lists(job.transfer.target.group_write, user['groupwrite']) + + elif permission in ('runJob', 'abortJob'): + job = context + if job.owner == identity: + return True + + return False + + +class DBUserAuthentication(object): + def __init__(self, space_name, db_pool): + self.db_pool = db_pool + self.space_name = space_name + + async def get_user(self, identity): + async with self.db_pool.acquire() as conn: + return await conn.fetchrow("select * from users where username=$1 and space_name=$2", + identity, self.space_name) + + async def check_credentials(self, username, password): + user = await self.get_user(username) + if not user: + return None + + if pbkdf2_sha256.verify(password, user['password']): + return username + return None + + async def login(self, request): + try: + auth = helpers.BasicAuth.decode(request.headers['Authorization']) + except: + return web.HTTPForbidden() + try: + user = await self.check_credentials(auth.login, auth.password) + if not user: + return web.HTTPForbidden() + + response = web.Response(status=200) + await remember(request, response, user) + return response + except web.HTTPForbidden: + raise + except Exception: + raise web.HTTPInternalServerError() + + async def logout(self, request): + try: + response = web.Response() + await forget(request, response) + return response + except Exception: + raise web.HTTPInternalServerError() diff --git a/pyvospace/server/spaces/ngas/space/__init__.py b/pyvospace/server/spaces/ngas/space/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/pyvospace/server/spaces/ngas/space/__main__.py b/pyvospace/server/spaces/ngas/space/__main__.py new file mode 100644 index 0000000..ed931fd --- /dev/null +++ b/pyvospace/server/spaces/ngas/space/__main__.py @@ -0,0 +1,62 @@ +# +# ICRAR - International Centre for Radio Astronomy Research +# (c) UWA - The University of Western Australia, 2018 +# Copyright by UWA (in the framework of the ICRAR) +# All rights reserved +# +# This library is free software; you can redistribute it and/or +# modify it under the terms of the GNU Lesser General Public +# License as published by the Free Software Foundation; either +# version 2.1 of the License, or (at your option) any later version. +# +# This library is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# Lesser General Public License for more details. +# +# You should have received a copy of the GNU Lesser General Public +# License along with this library; if not, write to the Free Software +# Foundation, Inc., 59 Temple Place, Suite 330, Boston, +# MA 02111-1307 USA + +import os +import ssl +import asyncio +import argparse +import configparser + +from aiohttp import web + +from .ngas_space import NGASSpaceServer + + +def main(args=None): + parser = argparse.ArgumentParser() + parser.add_argument('--cfg', type=str, action='store') + args = parser.parse_args() + + if args.cfg: + cfg_file = args.cfg + else: + app_path = os.path.dirname(os.path.realpath(__file__)) + cfg_file = f"{app_path}/cfg/space.ini" + + config = configparser.ConfigParser() + config.read(cfg_file) + port = config.getint('Space', 'port') + use_ssl = config.getint('Space', 'use_ssl') + + context = None + if bool(use_ssl): + cert_file = config['Space']['cert_file'] + key_file = config['Space']['key_file'] + context = ssl.SSLContext() + context.load_cert_chain(certfile=cert_file, keyfile=key_file) + + loop = asyncio.get_event_loop() + app = loop.run_until_complete(NGASSpaceServer.create(cfg_file)) + web.run_app(app, host='localhost', port=port, ssl_context=context) + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/pyvospace/server/spaces/ngas/space/ngas_space.py b/pyvospace/server/spaces/ngas/space/ngas_space.py new file mode 100644 index 0000000..6abfb8d --- /dev/null +++ b/pyvospace/server/spaces/ngas/space/ngas_space.py @@ -0,0 +1,248 @@ +# +# ICRAR - International Centre for Radio Astronomy Research +# (c) UWA - The University of Western Australia, 2018 +# Copyright by UWA (in the framework of the ICRAR) +# All rights reserved +# +# This library is free software; you can redistribute it and/or +# modify it under the terms of the GNU Lesser General Public +# License as published by the Free Software Foundation; either +# version 2.1 of the License, or (at your option) any later version. +# +# This library is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# Lesser General Public License for more details. +# +# You should have received a copy of the GNU Lesser General Public +# License along with this library; if not, write to the Free Software +# Foundation, Inc., 59 Temple Place, Suite 330, Boston, +# MA 02111-1307 USA + +import json +import numpy as np +import os + +from contextlib import suppress +import aiohttp +from aiohttp_security import setup as setup_security +from aiohttp_security import SessionIdentityPolicy +from aiohttp_session import setup as setup_session +from aiohttp_session.cookie_storage import EncryptedCookieStorage +from typing import List + +from pyvospace.server.space import SpaceServer, AbstractSpace +from pyvospace.core.model import Views, View, Protocols, \ + Node, NodeTextLookup, NodeType, Properties, Property, Protocol,\ + PushToSpace, PullFromSpace, HTTPGet, HTTPSGet, HTTPPut, HTTPSPut, Endpoint, SecurityMethod, UWSJob + +from pyvospace.server.spaces.posix.utils import mkdir +from pyvospace.server.spaces.ngas.auth import DBUserAuthentication, DBUserNodeAuthorizationPolicy +from pyvospace.core.exception import VOSpaceError + +import pdb +import traceback + + +ACCEPTS_VIEWS = { + 'vos:Node': [View('ivo://ivoa.net/vospace/core#anyview')], + 'vos:DataNode': [View('ivo://ivoa.net/vospace/core#anyview')], + 'vos:UnstructuredDataNode': [View('ivo://ivoa.net/vospace/core#anyview')], + 'vos:StructuredDataNode': [View('ivo://ivoa.net/vospace/core#anyview')], + 'vos:ContainerNode': [], + 'vos:LinkNode': [] +} + +PROVIDES_VIEWS = { + 'vos:Node': [View('ivo://ivoa.net/vospace/core#defaultview')], + 'vos:DataNode': [View('ivo://ivoa.net/vospace/core#defaultview')], + 'vos:UnstructuredDataNode': [View('ivo://ivoa.net/vospace/core#defaultview')], + 'vos:StructuredDataNode': [View('ivo://ivoa.net/vospace/core#defaultview')], + 'vos:ContainerNode': [], + 'vos:LinkNode': [] +} + +class NGASSpaceServer(SpaceServer, AbstractSpace): + def __init__(self, cfg_file, *args, **kwargs): + super().__init__(cfg_file, *args, **kwargs) + + self.secret_key = self.config['Space']['secret_key'] + self.domain = self.config['Space']['domain'] + self.name = self.config['Storage']['name'] + self.storage_parameters = json.loads(self.config['Storage']['parameters']) + + # NGAS parameters section - assumes that NGAS servers (if more than one) + # all point to the same storage, i.e they are federated + self.ngas_servers=json.loads(self.config['Storage']['ngas_servers']) + server_index=np.random.choice([n for n in range(0,len(self.ngas_servers))],1)[0] + self.ngas_server=self.ngas_servers[server_index] + + # Extract hostname and port + self.ngas_hostname=self.ngas_server["hostname"] + self.ngas_port=int(self.ngas_server["port"]) + self.ngas_session = aiohttp.ClientSession() + + self.staging_dir = self.storage_parameters['staging_dir'] + if not self.staging_dir: + raise Exception('staging_dir not found.') + + self.authentication = None + + async def setup_space(self): + await super().setup(self) + + # Make the staging directory + await mkdir(self.staging_dir) + + setup_session(self, + EncryptedCookieStorage( + secret_key=self.secret_key.encode(), + cookie_name='PYVOSPACE_COOKIE', + domain=self.domain)) + + self.authentication = DBUserAuthentication(self['space_name'], self['db_pool']) + + setup_security(self, + SessionIdentityPolicy(), + DBUserNodeAuthorizationPolicy(self['space_name'], + self['db_pool'], + self.ngas_hostname, + self.ngas_port, + self.ngas_session)) + + self.router.add_route('POST', '/login', self.authentication.login, name='login') + self.router.add_route('POST', '/logout', self.authentication.logout, name='logout') + + async def shutdown(self): + await super().shutdown() + await self.ngas_session.close() + + @classmethod + async def create(cls, cfg_file, *args, **kwargs): + # Is this method called when a new instance is created? + app = NGASSpaceServer(cfg_file, *args, **kwargs) + await app.setup_space() + return app + + def get_properties(self) -> Properties: + accepts = [Property('ivo://ivoa.net/vospace/core#title', None), + Property('ivo://ivoa.net/vospace/core#creator', None), + Property('ivo://ivoa.net/vospace/core#subject', None), + Property('ivo://ivoa.net/vospace/core#description', None), + Property('ivo://ivoa.net/vospace/core#publisher', None), + Property('ivo://ivoa.net/vospace/core#contributor', None), + Property('ivo://ivoa.net/vospace/core#date', None)] + provides = [] + return Properties(accepts, provides) + + def get_protocols(self) -> Protocols: + security_method = SecurityMethod('ivo://ivoa.net/sso#cookie') + return Protocols(accepts=[], provides=[HTTPGet(security_method=security_method), + HTTPPut(security_method=security_method)]) + + def get_views(self) -> Views: + return Views(accepts=[View('ivo://ivoa.net/vospace/core#anyview')], + provides=[View('ivo://ivoa.net/vospace/core#defaultview')]) + + def get_accept_views(self, node: Node) -> List[View]: + return ACCEPTS_VIEWS[NodeTextLookup[node.node_type]] + + def get_provide_views(self, node: Node) -> List[View]: + return PROVIDES_VIEWS[NodeTextLookup[node.node_type]] + + async def move_storage_node(self, src, dest): + # Files are not moved in NGAS + pass + + async def copy_storage_node(self, src, dest): + # Files are not copied in NGAS + pass + + async def create_storage_node(self, node: Node): + # Files are created when they are uploaded to NGAS + pass + + async def delete_storage_node(self, node : Node): + """Remove a node that in the NGAS database""" + + # Walk over nodes and remove all leaf nodes + for temp_node in node.walk(node): + if temp_node.node_type == NodeType.ContainerNode: + pass + else: + try: + # Make up the NGAS filename + filename_ngas=f'{os.path.basename(temp_node.path)}_{temp_node.id}' + + # Get the details of all files in the NGAS server with this query + params = {"query": "files_like", "like": filename_ngas, "format": "json"} + url = f'http://{self.ngas_hostname}:{self.ngas_port}/QUERY' + resp = await self.ngas_session.get(url, params=params) + lines = await resp.content.read() + + # List of file entries in JSON format + file_entries = json.loads(lines) + + # Loop through each file entry and delete it from the server + for file in file_entries: + + if ("file_id" in file) and (file["file_id"] == filename_ngas): + params = {"file_id": file["file_id"], "disk_id": file["disk_id"], "file_version": file["file_version"]} + url = f'http://{self.ngas_hostname}:{self.ngas_port}/CACHEDEL' + resp = await self.ngas_session.get(url, params=params) + + if (resp.status != 200): + raise aiohttp.web.HTTPServerError(reason=f"File {filename_ngas} not deleted properly from NGAS server") + + except Exception as e: + traceback.print_exc() + + async def get_transfer_protocols(self, job: UWSJob) -> List[Protocol]: + new_protocols = [] + protocols = job.job_info.protocols + security_method = SecurityMethod('ivo://ivoa.net/sso#cookie') + + if isinstance(job.job_info, PushToSpace): + if any(i in [HTTPPut(), HTTPSPut()] for i in protocols) is False: + raise VOSpaceError(400, "Protocol Not Supported.") + + async with self['db_pool'].acquire() as conn: + async with conn.transaction(): + results = await conn.fetch("select * from storage where name=$1", self.name) + + if HTTPPut() in protocols: + for row in results: + if row['https'] is False: + endpoint = Endpoint(f'http://{row["host"]}:{row["port"]}/' + f'vospace/{job.job_info.direction}/{job.job_id}') + new_protocols.append(HTTPPut(endpoint=endpoint, security_method=security_method)) + + if HTTPSPut() in protocols: + for row in results: + if row['https'] is True: + endpoint = Endpoint(f'https://{row["host"]}:{row["port"]}/' + f'vospace/{job.job_info.direction}/{job.job_id}') + new_protocols.append(HTTPPut(endpoint=endpoint, security_method=security_method)) + + elif isinstance(job.job_info, PullFromSpace): + if any(i in [HTTPGet(), HTTPSGet()] for i in protocols) is False: + raise VOSpaceError(400, "Protocol Not Supported.") + + storage = job.job_info.target.storage + if storage is None: + raise VOSpaceError(400, f"{job.job_info.target} not on any storage device.") + + if HTTPGet() in protocols: + endpoint = Endpoint(f'http://{storage.host}:{storage.port}/' + f'vospace/{job.job_info.direction}/{job.job_id}') + new_protocols.append(HTTPGet(endpoint=endpoint, security_method=security_method)) + + if HTTPSGet() in protocols: + endpoint = Endpoint(f'https://{storage.host}:{storage.port}/' + f'vospace/{job.job_info.direction}/{job.job_id}') + new_protocols.append(HTTPSGet(endpoint=endpoint, security_method=security_method)) + + if not new_protocols: + raise VOSpaceError(400, "Protocol Not Supported. No storage found") + + return new_protocols diff --git a/pyvospace/server/spaces/ngas/storage/__init__.py b/pyvospace/server/spaces/ngas/storage/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/pyvospace/server/spaces/ngas/storage/__main__.py b/pyvospace/server/spaces/ngas/storage/__main__.py new file mode 100644 index 0000000..3844665 --- /dev/null +++ b/pyvospace/server/spaces/ngas/storage/__main__.py @@ -0,0 +1,61 @@ +# +# ICRAR - International Centre for Radio Astronomy Research +# (c) UWA - The University of Western Australia, 2018 +# Copyright by UWA (in the framework of the ICRAR) +# All rights reserved +# +# This library is free software; you can redistribute it and/or +# modify it under the terms of the GNU Lesser General Public +# License as published by the Free Software Foundation; either +# version 2.1 of the License, or (at your option) any later version. +# +# This library is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# Lesser General Public License for more details. +# +# You should have received a copy of the GNU Lesser General Public +# License along with this library; if not, write to the Free Software +# Foundation, Inc., 59 Temple Place, Suite 330, Boston, +# MA 02111-1307 USA + +import os +import ssl +import asyncio +import argparse +import configparser + +from aiohttp import web + +from .ngas_storage import NGASStorageServer + +def main(args=None): + parser = argparse.ArgumentParser() + parser.add_argument('--cfg', type=str, action='store') + args = parser.parse_args() + + if args.cfg: + cfg_file = args.cfg + else: + app_path = os.path.dirname(os.path.realpath(__file__)) + cfg_file = f"{app_path}/cfg/storage.ini" + + config = configparser.ConfigParser() + config.read(cfg_file) + port = config.getint('Storage', 'port') + use_ssl = config.getint('Storage', 'use_ssl') + + context = None + if bool(use_ssl): + cert_file = config['Storage']['cert_file'] + key_file = config['Storage']['key_file'] + context = ssl.SSLContext() + context.load_cert_chain(certfile=cert_file, keyfile=key_file) + + loop = asyncio.get_event_loop() + app = loop.run_until_complete(NGASStorageServer.create(cfg_file)) + web.run_app(app, port=port, ssl_context=context) + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/pyvospace/server/spaces/ngas/storage/ngas_storage.py b/pyvospace/server/spaces/ngas/storage/ngas_storage.py new file mode 100644 index 0000000..9f06429 --- /dev/null +++ b/pyvospace/server/spaces/ngas/storage/ngas_storage.py @@ -0,0 +1,393 @@ +# +# ICRAR - International Centre for Radio Astronomy Research +# (c) UWA - The University of Western Australia, 2018 +# Copyright by UWA (in the framework of the ICRAR) +# All rights reserved +# +# This library is free software; you can redistribute it and/or +# modify it under the terms of the GNU Lesser General Public +# License as published by the Free Software Foundation; either +# version 2.1 of the License, or (at your option) any later version. +# +# This library is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# Lesser General Public License for more details. +# +# You should have received a copy of the GNU Lesser General Public +# License along with this library; if not, write to the Free Software +# Foundation, Inc., 59 Temple Place, Suite 330, Boston, +# MA 02111-1307 USA + +import io +import os +import uuid +import asyncio +import aiofiles +import aiohttp +import numpy as np +import pdb +import traceback +import json +import numpy as np + +from aiohttp import web +from aiohttp_security import setup as setup_security +from aiohttp_security import SessionIdentityPolicy +from aiohttp_session import setup as setup_session +from aiohttp_session.cookie_storage import EncryptedCookieStorage +from contextlib import suppress +from concurrent.futures import ProcessPoolExecutor + +from pyvospace.core.model import NodeType, View + +# Not sure if I need these +from pyvospace.server.spaces.posix.utils import mkdir, remove, send_file, rmtree, tar, untar +from pyvospace.server.spaces.ngas.utils import send_stream_to_ngas, send_file_to_ngas, recv_file_from_ngas +from pyvospace.server.storage import HTTPSpaceStorageServer +from pyvospace.server import fuzz, fuzz01 +from pyvospace.server.spaces.ngas.auth import DBUserNodeAuthorizationPolicy +from pyvospace.server.uws import StorageUWSJob + +class NGASStorageServer(HTTPSpaceStorageServer): + def __init__(self, cfg_file, *args, **kwargs): + super().__init__(cfg_file, *args, **kwargs) + + # What is this for? + self.secret_key = self.config['Space']['secret_key'] + self.domain = self.config['Space']['domain'] + + # NGAS parameters section - assumes that NGAS servers (if more than one) + # all point to the same storage, i.e they are federated + # Choose an NGAS server, initially at random + self.ngas_servers=json.loads(self.config['Storage']['ngas_servers']) + server_index=np.random.choice([n for n in range(0,len(self.ngas_servers))],1)[0] + self.ngas_server=self.ngas_servers[server_index] + + # Extract hostname and port + self.ngas_hostname=self.ngas_server["hostname"] + self.ngas_port=int(self.ngas_server["port"]) + self.ngas_session = aiohttp.ClientSession() + + # We need a staging directory + self.staging_dir = self.parameters['staging_dir'] + if not self.staging_dir: + raise Exception('staging_dir not found.') + + # Is a pooled server already available? + self.process_executor = ProcessPoolExecutor(max_workers=32) + self.on_shutdown.append(self.shutdown) + + async def shutdown(self): + loop = asyncio.get_event_loop() + await super().shutdown() + await loop.run_in_executor(None, self.process_executor.shutdown) + # Close the NGAS session + await self.ngas_session.close() + + async def setup(self): + await super().setup() + + # We do need a staging directory + await mkdir(self.staging_dir) + + setup_session(self, + EncryptedCookieStorage( + secret_key=self.secret_key.encode(), + cookie_name='PYVOSPACE_COOKIE', + domain=self.domain)) + + # Does this need to be different? + setup_security(self, + SessionIdentityPolicy(), + DBUserNodeAuthorizationPolicy( self.name, + self.db_pool, + self.ngas_hostname, + self.ngas_port, + self.ngas_session)) + + @classmethod + async def create(cls, cfg_file, *args, **kwargs): + app = NGASStorageServer(cfg_file, *args, **kwargs) + await app.setup() + return app + + async def download(self, job: StorageUWSJob, request: aiohttp.web.Request): + """Download files from the NGAS server""" + path_tree = job.transfer.target.path + + if job.transfer.target.node_type == NodeType.ContainerNode: + + # Download the files from NGAS to a directory, + # tar the directory and send to client + # Should we implement a streaming creation of the tarfile? + + # Make a directory in staging area + path_tree=job.transfer.target.path + stage_dir = os.path.normpath(f'{self.staging_dir}/{uuid.uuid4()}') + stage_path = os.path.normpath(f'{stage_dir}/{path_tree}') + + # make the stage directory + await mkdir(stage_dir) + + # Now loop over each current node in NGAS and fetch files + current_node=job.transfer.target + + # Current node + if job.transfer.view != View('ivo://ivoa.net/vospace/core#tar'): + return web.Response(status=400, text=f'Unsupported Container View. ' + f'View: {job.transfer.view}') + + # Now loop over each node and download to file + for node in current_node.walk(current_node): + node_path=os.path.normpath(f'{stage_dir}/{node.path}') + if node.node_type==NodeType.ContainerNode: + # Make a directory in the node path + await mkdir(node_path) + else: + # Make the local filename + filename_local=node_path + # Make the NGAS filename + filename_ngas=f'{os.path.basename(node_path)}_{node.id}' + + # Fetch a file from NGAS + await recv_file_from_ngas( self.ngas_session, + self.ngas_hostname, + self.ngas_port, + filename_ngas, + filename_local) + + # Tar file + tar_file = os.path.normpath(f'{stage_dir}/{uuid.uuid4()}/{os.path.basename(path_tree)}.tar') + + try: + # Tarring up the sources, I might need to ask the database for all + loop = asyncio.get_event_loop() + await loop.run_in_executor(self.process_executor, tar, + stage_path, tar_file, os.path.basename(path_tree)) + return await send_file(request, os.path.basename(tar_file), tar_file) + finally: + with suppress(Exception): + await asyncio.shield(rmtree(os.path.dirname(tar_file))) + with suppress(Exception): + await asyncio.shield(rmtree(os.path.dirname(stage_dir))) + + else: + + # Get the UUID on the node from the database + id=job.transfer.target.id + + # Filename to be used with the NGAS object store + base_name=os.path.basename(path_tree) + filename_ngas=f'{base_name}_{id}' + + # URL for retrieval from NGAS + url_ngas=f'http://{self.ngas_hostname}:{self.ngas_port}/RETRIEVE' + + # Make up the filename for retrieval from NGAS + params={"file_id" : filename_ngas} + + # Connect to NGAS + resp_ngas = await self.ngas_session.get(url_ngas, params=params) + + # Rudimentry error checking on the NGAS connection + if resp_ngas.status!=200: + raise aiohttp.web.HTTPServerError(reason="Error in connecting to NGAS server") + + # Otherwise create the client + resp_client=web.StreamResponse() + + # Update the headers + resp_client.headers.update(resp_ngas.headers) + + # Change the filename? + resp_client.headers['Content-Disposition']=f'attachment; filename=\"{base_name}\"' + + # Prepare the connection + await resp_client.prepare(request) + + # Read from source and and write destination in buffers + async for chunk in resp_ngas.content.iter_chunked(io.DEFAULT_BUFFER_SIZE): + if chunk: + await resp_client.write(chunk) + + # Finish the stream + await resp_client.write_eof() + return(resp_client) + + + async def upload(self, job: StorageUWSJob, request: aiohttp.web.Request): + # Upload file or container contents from a client to the NGAS server + + # Get the path tree + path_tree = job.transfer.target.path + # Get the base filename + base_name = os.path.basename(path_tree) + + # Check for content length in the headers of the incoming request + # This will inform how we respond to the request + if 'content-length' in request.headers: + content_length=request.headers['content-length'] + else: + content_length=None + + if job.transfer.target.node_type == NodeType.ContainerNode: + + # Incoming client content + reader=request.content + + # Temporary file to stage to + stage_file_name = os.path.join(self.staging_dir, base_name) + + # Temporary UUID for staging the directory + target_id = uuid.uuid4() + + # Path to the top level directory of the node + path_tree = job.transfer.target.path + + try: + # Read from incoming client buffer to temporary file + size = 0 + async with aiofiles.open(stage_file_name, 'wb') as f: + while True: + buffer = await reader.read(io.DEFAULT_BUFFER_SIZE) + if not buffer: + break + else: + await fuzz() + await f.write(buffer) + size += len(buffer) + + if job.transfer.view != View('ivo://ivoa.net/vospace/core#tar'): + return web.Response(status=400, text=f'Unsupported Container View. ' + f'View: {job.transfer.view}') + + # Directory to extract to + extract_dir = os.path.normpath(f'{self.staging_dir}/{target_id}') + + try: + + loop = asyncio.get_event_loop() + + # Untarring process + root_node = await loop.run_in_executor(self.process_executor, + untar, + stage_file_name, + f'{extract_dir}/{path_tree}', + job.transfer.target, + self.storage) + + # Do the walk and upload here + # Walk the tree and upload each file to an NGAS flat object store + # Keep the ID's of old nodes through checking + oldnode=job.transfer.target + + # Copy old id's across and upload to NGAS + + # Flatten the tree + old_node_paths=[node.path for node in oldnode.walk(oldnode)] + old_nodes=[node for node in oldnode.walk(oldnode)] + new_nodes=[node for node in root_node.walk(root_node)] + + # Loop over new nodes + for new_node in new_nodes: + if new_node.path in old_node_paths: + # Get the index of the old node + index_old=old_node_paths.index(new_node.path) + old_node=old_nodes[index_old] + # Copy the ID from the old node to the new node + new_node.id=old_node.id + + if new_node.node_type != NodeType.ContainerNode: + filename_local=os.path.normpath(f'{extract_dir}/{new_node.path}') + filename_ngas=f'{os.path.basename(new_node.path)}_{new_node.id}' + nbytes_transfer = await send_file_to_ngas(self.ngas_session, + self.ngas_hostname, + self.ngas_port, + filename_ngas, + filename_local) + new_node.size=nbytes_transfer + else: + # We have a container type, it doesn't contribute to the upload + # so make it zero size + new_node.size=0 + + + # Now update the database + async with job.transaction() as tr: + #pdb.set_trace() + node = tr.target + node.size = size + node.storage = self.storage + # This saves all nodes under the root node + node.nodes = root_node.nodes + await asyncio.shield(node.save()) + # Let the client know the transaction was successful + return web.Response(status=200) + + except Exception as e: + traceback.print_exception(e, SyntaxError, None) + raise e + + finally: + with suppress(Exception): + await asyncio.shield(rmtree(f'{self.staging_dir}/{target_id}')) + + except Exception as e: + traceback.print_exception(e, SyntaxError, None) + raise e + + else: + + # Get the UUID for the node + id = job.transfer.target.id + ngas_filename=f"{base_name}_{id}" + + if content_length is not None: + # Content length exists, we can forward the stream straight to the NGAS server + nbytes_transfer = await send_stream_to_ngas(request, self.ngas_session, self.ngas_hostname, + self.ngas_port, ngas_filename, self.logger) + else: + # Make up a uuid for the staging of a file + reader=request.content + + # Temporary uuid for the upload of a file + target_id = uuid.uuid4() + base_name = f'{target_id}_{os.path.basename(path_tree)}' + + # Temporary file to stage to + stage_file_name = os.path.join(self.staging_dir, base_name) + + # Need to read in a specific number of bytes? + async with aiofiles.open(stage_file_name, 'wb') as fd: + # We have checked that content-length must exist. + while True: + buffer = await reader.read(io.DEFAULT_BUFFER_SIZE) + if buffer: + await fuzz() + await fd.write(buffer) + else: + break + + # Now the file is on disk, send it + nbytes_transfer = await send_file_to_ngas(self.ngas_session, self.ngas_hostname, self.ngas_port, + ngas_filename, stage_file_name) + + + # Remove the staged file if it exists + with suppress(Exception): + await asyncio.shield(remove(stage_file_name)) + + # Inform the database of new data if size + async with job.transaction() as tr: + if nbytes_transfer is not None: + node = tr.target # get the target node that is associated with the data + node.size = nbytes_transfer # set the size + node.storage = self.storage # set the storage back end so it can be found + await asyncio.shield(fuzz01(2)) + await asyncio.shield(node.save()) # save details to db + + # Let the client know the transaction was successful + return web.Response(status=200) + + diff --git a/pyvospace/server/spaces/ngas/utils.py b/pyvospace/server/spaces/ngas/utils.py new file mode 100644 index 0000000..c5d04eb --- /dev/null +++ b/pyvospace/server/spaces/ngas/utils.py @@ -0,0 +1,212 @@ +# +# ICRAR - International Centre for Radio Astronomy Research +# (c) UWA - The University of Western Australia, 2018 +# Copyright by UWA (in the framework of the ICRAR) +# All rights reserved +# +# This library is free software; you can redistribute it and/or +# modify it under the terms of the GNU Lesser General Public +# License as published by the Free Software Foundation; either +# version 2.1 of the License, or (at your option) any later version. +# +# This library is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# Lesser General Public License for more details. +# +# You should have received a copy of the GNU Lesser General Public +# License along with this library; if not, write to the Free Software +# Foundation, Inc., 59 Temple Place, Suite 330, Boston, +# MA 02111-1307 USA + +import os +import io +from aiofiles.os import stat +import asyncio +import aiohttp +import aiofiles +import datetime +import xml.etree.ElementTree as ElementTree + +from aiohttp import web +from pyvospace.server import fuzz + +class CountedReader: + """A wrapper class to count the number of bytes being sent from a stream""" + def __init__(self, content): + self._content=content + self._size=0 + self._iter=None + + def __aiter__(self): + #self._iter=self._content.__aiter__() + self._iter=self._content.iter_chunked(io.DEFAULT_BUFFER_SIZE) + return self + + async def __anext__(self): + buffer=await self._iter.__anext__() + self._size+=len(buffer) + return buffer + +class ControlledReader: + """A wrapper class to limit the number of bytes returned from a stream + to exactly content_length bytes""" + + def __init__(self, content, content_length): + self._content = content + self._content_length=content_length + self._bytes_read = 0 + self._iter = None + + def __aiter__(self): + return self + + async def __anext__(self): + # What is the minimum number of bytes to read? + bytes_to_read = min(io.DEFAULT_BUFFER_SIZE, self._content_length - self._bytes_read) + if bytes_to_read <= 0: + raise StopAsyncIteration + else: + buffer = await self._content.readexactly(bytes_to_read) + self._bytes_read+=bytes_to_read + return buffer + +def convert_to_epoch_seconds(date): + # Convert a specific date string or date object to a number of seconds since + # the UNIX epoch. + if isinstance(date, str): + dt = datetime.datetime.strptime(date, "%Y-%m-%dT%H:%M:%S.%f") + elif isinstance(date, datetime.date): + dt=date + else: + return None + + # Get the number of seconds since the UNIX epoch. + seconds=str(int((dt - datetime.datetime(1970, 1, 1)).total_seconds())) + return seconds + +async def recv_file_from_ngas(session, hostname, port, filename_ngas, filename_local): + + """Get a single file from NGAS and put it into filename_local""" + + # The URL to contact the NGAS server + url = f'http://{hostname}:{port}/RETRIEVE' + + # Make up the filename for retrieval from NGAS + # How can I get the uuid from the database? + params = {"file_id": filename_ngas} + + # Connect to NGAS + resp_ngas = await session.get(url, params=params) + + # Rudimentry error checking on the NGAS connection + if resp_ngas.status != 200: + raise aiohttp.web.HTTPServerError(reason="Error in connecting to NGAS server") + + # Open the file for writing + async with aiofiles.open(filename_local, 'wb') as fd: + # Connect to the NGAS server and download the file + async for chunk in resp_ngas.content.iter_chunked(io.DEFAULT_BUFFER_SIZE): + if chunk: + await fd.write(chunk) + + +async def send_file_to_ngas(session, hostname, port, filename_ngas, filename_local): + + #pdb.set_trace() + + """Send a single file to an NGAS server""" + try: + + # Create parameters for the upload + params = {"filename": filename_ngas, + "file_id" : filename_ngas, + "mime_type": "application/octet-stream"} + + # The URL to contact the NGAS server + url=f'http://{hostname}:{port}/ARCHIVE' + + # Make sure a the file exists + if filename_local is None or not os.path.isfile(filename_local): + raise FileNotFoundError + + # Get the size of the file for content-length + file_size = (await stat(filename_local)).st_size + + if file_size==0: + raise ValueError(f"file {filename_local} has 0 size") + + async with aiofiles.open(filename_local, 'rb') as fd: + # Connect to the NGAS server and upload the file + resp = await session.post(url, params=params, + data=fd, + headers={"content-length" : str(file_size)}) + + if resp.status!=200: + raise aiohttp.ServerConnectionError("Error received in connecting to NGAS server") + + return(file_size) + + except Exception as e: + # Do we do anything here? + raise e + +async def send_stream_to_ngas(request: aiohttp.web.Request, session, hostname, port, filename_ngas, logger): + + """If an incoming POST request has the content-length, send a stream direct to NGAS""" + try: + + # Create parameters for the upload + params = {"filename": filename_ngas, + "file_id" : filename_ngas, + "mime_type": "application/octet-stream"} + + # The URL to contact the NGAS server + url="http://"+str(hostname)+":"+str(port)+"/ARCHIVE" + + # Test for content-length + if 'content-length' not in request.headers: + raise aiohttp.ServerConnectionError("No content-length in header") + + content_length=int(request.headers['Content-Length']) + + if content_length==0: + raise ValueError + + # Create a ControlledReader from the content + reader=ControlledReader(request.content, content_length) + + # Test for proper implementation + if 'transfer-encoding' in request.headers: + if request.headers['transfer-encoding']=="chunked": + raise aiohttp.ServerConnectionError("Error, content length defined but transfer-encoding is chunked") + + # Connect to the NGAS server and upload + resp = await session.post(url, params=params, + data=reader, + headers={"content-length" : str(content_length)}) + + # Handle the response in a specific way, as requested + if resp.status==200: + # Dig into the XML response for output, we are looking for SUCCESS + feedback = await resp.text() + xmltree = ElementTree.fromstring(feedback) + + # Create a dictionary of all XML elements in the response + elements = {t.tag: t for t in xmltree.iter()} + # Status of the NGAS transaction + status=elements["Status"].get("Status") + + if status=="SUCCESS": + return(content_length) + else: + raise aiohttp.ServerConnectionError("Error received in connecting to NGAS server") + return(None) + else: + raise aiohttp.ServerConnectionError("Error received in connecting to NGAS server") + return(None) + + except Exception as e: + # Do we do anything here? + raise e + diff --git a/setup.py b/setup.py index 870c838..103c8fc 100644 --- a/setup.py +++ b/setup.py @@ -21,5 +21,7 @@ 'requests'], entry_points={'console_scripts': [ 'posix_space = pyvospace.server.spaces.posix.space.__main__:main', - 'posix_storage = pyvospace.server.spaces.posix.storage.__main__:main'] - }) \ No newline at end of file + 'posix_storage = pyvospace.server.spaces.posix.storage.__main__:main', + 'ngas_space = pyvospace.server.spaces.ngas.space.__main__:main', + 'ngas_storage = pyvospace.server.spaces.ngas.storage.__main__:main'] + }) diff --git a/test/.DS_Store b/test/.DS_Store new file mode 100644 index 0000000..51f6654 Binary files /dev/null and b/test/.DS_Store differ diff --git a/test/test_ngas/.DS_Store b/test/test_ngas/.DS_Store new file mode 100644 index 0000000..506682d Binary files /dev/null and b/test/test_ngas/.DS_Store differ diff --git a/test/test_ngas/__init__.py b/test/test_ngas/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/test/test_ngas/test_abort.py b/test/test_ngas/test_abort.py new file mode 100644 index 0000000..cded0b8 --- /dev/null +++ b/test/test_ngas/test_abort.py @@ -0,0 +1,142 @@ +# +# ICRAR - International Centre for Radio Astronomy Research +# (c) UWA - The University of Western Australia, 2018 +# Copyright by UWA (in the framework of the ICRAR) +# All rights reserved +# +# This library is free software; you can redistribute it and/or +# modify it under the terms of the GNU Lesser General Public +# License as published by the Free Software Foundation; either +# version 2.1 of the License, or (at your option) any later version. +# +# This library is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# Lesser General Public License for more details. +# +# You should have received a copy of the GNU Lesser General Public +# License along with this library; if not, write to the Free Software +# Foundation, Inc., 59 Temple Place, Suite 330, Boston, +# MA 02111-1307 USA + +import unittest +import asyncio + +from aiohttp import web +from contextlib import suppress + +from pyvospace.core.model import * +from pyvospace.server import set_fuzz, set_fuzz01, wait_fuzz01 +from pyvospace.server.spaces.ngas.storage.ngas_storage import NGASStorageServer +from test_base import TestBase + + +class TestAbort(TestBase): + + def setUp(self): + super().setUp() + self.loop.run_until_complete(self._setup()) + ngas_server = self.loop.run_until_complete(NGASStorageServer.create(self.config_filename)) + + self.ngas_runner = web.AppRunner(ngas_server) + self.loop.run_until_complete(self.ngas_runner.setup()) + self.ngas_site = web.TCPSite(self.ngas_runner, 'localhost', 8083) + self.loop.run_until_complete(self.ngas_site.start()) + + async def _setup(self): + if not os.path.exists('/tmp/download'): + os.makedirs('/tmp/download') + await self.create_file('/tmp/datafile.dat') + + def tearDown(self): + self.loop.run_until_complete(self.delete('http://localhost:8080/vospace/nodes/syncdatanode')) + self.loop.run_until_complete(self.ngas_runner.shutdown()) + self.loop.run_until_complete(self.ngas_runner.cleanup()) + super().tearDown() + + def test_push_shield_abort(self): + async def run(): + node = Node('/syncdatanode') + push = PushToSpace(node, [HTTPPut()]) + transfer = await self.sync_transfer_node(push, 200) + put_end = transfer.protocols[0].endpoint.url + set_fuzz01(True) + + async def push(url): + with suppress(Exception): + await self.push_to_space(url, '/tmp/datafile.dat', expected_status=200) + + push_task = asyncio.ensure_future(push(put_end)) + await wait_fuzz01() + push_task.cancel() + await push_task + set_fuzz01(False) + + node = Node('/syncdatanode') + pull = PullFromSpace(node, [HTTPGet()]) + transfer = await self.sync_transfer_node(pull) + end_get = transfer.protocols[0].endpoint.url + await self.pull_from_space(end_get, '/tmp/download/', expected_status=(200,)) + + self.loop.run_until_complete(run()) + + def test_push_to_space_sync_push_abort(self): + async def run(): + node = Node('/syncdatanode') + push = PushToSpace(node, [HTTPPut()]) + transfer = await self.sync_transfer_node(push, 200) + put_end = transfer.protocols[0].endpoint.url + job_id = os.path.basename(put_end) + set_fuzz(True) + + async def defer_abort(job_id): + await asyncio.sleep(0.5) + await self.change_job_state(job_id, state='PHASE=ABORT', expected_status=200) + await self.poll_job(job_id, poll_until=('ABORTED', 'ERROR'), expected_status='ABORTED') + + tasks = [ + asyncio.ensure_future(self.push_to_space(put_end, '/tmp/datafile.dat', expected_status=400)), + asyncio.ensure_future(defer_abort(job_id)) + ] + + await asyncio.gather(*tasks) + set_fuzz(False) + await self.poll_job(job_id, poll_until=('ABORTED', 'ERROR'), expected_status='ABORTED') + + self.loop.run_until_complete(run()) + + def test_push_to_space_sync_pull_abort(self): + async def run(): + node = Node('/syncdatanode') + push = PushToSpace(node, [HTTPPut()]) + transfer = await self.sync_transfer_node(push, 200) + put_end = transfer.protocols[0].endpoint.url + await self.push_to_space(put_end, '/tmp/datafile.dat', expected_status=200) + + pull = PullFromSpace(node, [HTTPGet()]) + transfer = await self.sync_transfer_node(pull) + pull_end = transfer.protocols[0].endpoint.url + + job_id = os.path.basename(pull_end) + set_fuzz(True) + + async def defer_abort(job_id): + await asyncio.sleep(0.5) + await self.change_job_state(job_id, state='PHASE=ABORT', expected_status=200) + await self.poll_job(job_id, poll_until=('ABORTED', 'ERROR'), expected_status='ABORTED') + + tasks = [ + asyncio.ensure_future(self.pull_from_space_defer_error(pull_end, '/tmp/download/')), + asyncio.ensure_future(defer_abort(job_id)) + ] + # client recv should fail + with self.assertRaises(IOError): + await asyncio.gather(*tasks) + + set_fuzz(False) + await self.poll_job(job_id, poll_until=('ABORTED', 'ERROR'), expected_status='ABORTED') + + self.loop.run_until_complete(run()) + +if __name__ == '__main__': + unittest.main() \ No newline at end of file diff --git a/test/test_ngas/test_base.py b/test/test_ngas/test_base.py new file mode 100644 index 0000000..7a1ea87 --- /dev/null +++ b/test/test_ngas/test_base.py @@ -0,0 +1,354 @@ +# +# ICRAR - International Centre for Radio Astronomy Research +# (c) UWA - The University of Western Australia, 2018 +# Copyright by UWA (in the framework of the ICRAR) +# All rights reserved +# +# This library is free software; you can redistribute it and/or +# modify it under the terms of the GNU Lesser General Public +# License as published by the Free Software Foundation; either +# version 2.1 of the License, or (at your option) any later version. +# +# This library is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# Lesser General Public License for more details. +# +# You should have received a copy of the GNU Lesser General Public +# License along with this library; if not, write to the Free Software +# Foundation, Inc., 59 Temple Place, Suite 330, Boston, +# MA 02111-1307 USA + +import io +import sys +import asyncio +import json +import aiofiles +import aiofiles.os +import aiohttp +import logging +import unittest +import configparser +import xml.etree.ElementTree as ET +import requests +import socket +import tarfile +import pdb + +from aiohttp import web +from urllib.parse import urlencode +from passlib.hash import pbkdf2_sha256 + +from pyvospace.server.spaces.ngas.space.ngas_space import NGASSpaceServer +from pyvospace.core.model import * + + +class TestBase(unittest.TestCase): + + def setUp(self): + logging.basicConfig(stream=sys.stderr) + logging.getLogger('test').setLevel(logging.DEBUG) + self.log = logging.getLogger('test') + self.session = None + self.loop = asyncio.new_event_loop() + asyncio.set_event_loop(None) + + self.config_filename = 'test_vo_ngas.ini' + config = configparser.ConfigParser() + if not os.path.exists(self.config_filename): + storage_details = json.dumps( + {'root_dir': '/tmp/ngas/storage/', + 'staging_dir': '/tmp/ngas/staging/'}) + + config['Space'] = {'host': 'localhost', + 'port': 8082, + 'name': 'ngas', + 'uri': 'icrar.org', + 'dsn': 'postgres://vos_user:vos_user@localhost:5435/vospace', + 'parameters': '{}', + 'secret_key': 'ZlmNyXdQgRhhrC2Wwy-gLZj7Wv6ZtoKH', + 'domain': '', + 'use_ssl': 0 + } + print("running this") + config['Storage'] = {'name': 'ngas', + 'host': 'localhost', + 'port': 8083, + 'parameters': storage_details, + 'use_ssl': 0, + 'ngas_servers' : [{"hostname" : "localhost", "port" : 7777}, + {"hostname" : "localhost", "port" : 7777}] + } + + with open(self.config_filename, 'w') as conf: + config.write(conf) + + self.app = self.loop.run_until_complete(NGASSpaceServer.create(self.config_filename)) + self.runner = web.AppRunner(self.app) + self.loop.run_until_complete(self.runner.setup()) + if not hasattr(socket, 'SO_REUSEPORT'): + reuse_port = False + else: + reuse_port = True + site = web.TCPSite(self.runner, 'localhost', 8080, + reuse_address=True, reuse_port=reuse_port) + self.loop.run_until_complete(site.start()) + + user = ['test', pbkdf2_sha256.hash('test'), [], [], 'ngas', True] + self.loop.run_until_complete(self.create_user(self.app['db_pool'], *user)) + self.session = self.loop.run_until_complete(self._login('test', 'test')) + + def tearDown(self): + if self.session: + self.loop.run_until_complete(self._logout()) + self.loop.run_until_complete(self.session.close()) + + self.loop.run_until_complete(self.runner.shutdown()) + self.loop.run_until_complete(self.runner.cleanup()) + self.loop.close() + + async def create_user(self, db_pool, username, password, group_read, group_write, space_name, admin): + async with db_pool.acquire() as conn: + await conn.fetchrow("insert into users (username, password, groupread, " + "groupwrite, space_name, admin) values($1, $2, $3, $4, $5, $6) " + "on conflict (username, space_name) do nothing", + username, password, group_read, group_write, space_name, admin) + + + async def create_tar(self, file_name): + try: + await aiofiles.os.stat(file_name) + except FileNotFoundError: + if not os.path.exists('/tmp/tar'): + os.makedirs('/tmp/tar') + if not os.path.exists('/tmp/tar/dir1/dir2'): + os.makedirs('/tmp/tar/dir1/dir2') + await self.create_file('/tmp/tar/test1', blocksize=2) + await self.create_file('/tmp/tar/test2', blocksize=64) + await self.create_file('/tmp/tar/test3', blocksize=1024) + await self.create_file('/tmp/tar/dir1/test1', blocksize=2048) + await self.create_file('/tmp/tar/dir1/dir2/test2', blocksize=128) + with tarfile.open(file_name, "w") as tar: + tar.add('/tmp/tar') + + async def create_file(self, file_name, blocksize=1024): + try: + await aiofiles.os.stat(file_name) + except FileNotFoundError: + async with aiofiles.open(file_name, mode='wb') as f: + await f.truncate(blocksize*io.DEFAULT_BUFFER_SIZE) + + async def file_sender(self, file_name=None): + async with aiofiles.open(file_name, 'rb') as f: + chunk = await f.read(64 * 1024) + while chunk: + yield chunk + chunk = await f.read(64 * 1024) + + async def _login(self, username, password): + session = aiohttp.ClientSession(auth=aiohttp.BasicAuth(username, password)) + async with session.post(f'http://localhost:8080/login') as resp: + response = await resp.text() + self.assertEqual(resp.status, 200, msg=response) + return session + + async def _logout(self): + async with self.session.post(f'http://localhost:8080/logout') as resp: + response = await resp.text() + self.assertEqual(resp.status, 200, msg=response) + + async def post(self, url, **kwargs): + #async with aiohttp.ClientSession() as session: + async with self.session.post(url, **kwargs) as resp: + return resp.status, await resp.text() + + async def delete(self, url): + #async with aiohttp.ClientSession() as session: + async with self.session.delete(url) as resp: + return resp.status, await resp.text() + + async def put(self, url, **kwargs): + #async with aiohttp.ClientSession() as session: + async with self.session.put(url, **kwargs) as resp: + return resp.status, await resp.text() + + async def get(self, url, **kwargs): + #async with aiohttp.ClientSession() as session: + async with self.session.get(url, **kwargs) as resp: + return resp.status, await resp.text() + + async def create_node(self, node, expected_status=201): + status, response = await self.put(f'http://localhost:8080/vospace/nodes/{node.path}', + data=node.tostring()) + self.assertEqual(status, expected_status, msg=response) + return response + + async def get_node(self, path, params, expected_status=200): + status, response = await self.get(f'http://localhost:8080/vospace/nodes/{path}', params=params) + self.assertEqual(expected_status, status, msg=response) + if status == 200: + return Node.fromstring(response) + return None + + async def set_node_properties(self, node, expected_status=200): + status, response = await self.post(f'http://localhost:8080/vospace/nodes/{node.path}', data=node.tostring()) + self.assertEqual(expected_status, status, msg=response) + return response + + async def transfer_node(self, transfer): + status, response = await self.post('http://localhost:8080/vospace/transfers', data=transfer.tostring()) + self.assertEqual(200, status, msg=response) + return UWSJob.fromstring(response) + + async def sync_transfer_node(self, transfer, expected_status=200): + status, response = await self.post('http://localhost:8080/vospace/synctrans', data=transfer.tostring()) + self.assertEqual(expected_status, status, msg=response) + if status == 200: + return Transfer.fromstring(response) + return response + + async def sync_transfer_node_parameterised(self, transfer, expected_status=200): + status, response = await self.post('http://localhost:8080/vospace/synctrans', + params=transfer.tomap()) + self.assertEqual(expected_status, status, msg=response) + if status == 200: + return Transfer.fromstring(response) + return response + + async def sync_pull_from_space_parameterised(self, transfer, output_path, expected_status=(200,)): + loop = asyncio.get_event_loop() + await loop.run_in_executor(None, self.download_file_product, transfer, output_path) + + async def delete_node(self, node): + return await self.delete(f'http://localhost:8080/vospace/nodes/{node.path}') + + def get_job_id(self, response): + root = ET.fromstring(response) + job_id = root.find('{http://www.ivoa.net/xml/UWS/v1.0}jobId') + self.assertIsNotNone(job_id.text) + return job_id.text + + async def get_job_details(self, job_id): + status, response = await self.get(f'http://localhost:8080/vospace/transfers/{job_id}', params=None) + return response + + def extract_transfer_details(self, response): + root = ET.fromstring(response) + results = root.find('{http://www.ivoa.net/xml/UWS/v1.0}results') + for result in results: + if result.attrib['id'] == 'transferDetails': + return f'http://localhost:8080{result.attrib["{http://www.w3.org/1999/xlink}href"]}' + return None + + async def get_transfer_details(self, job_id, expected_status=200): + status, response = await self.get(f'http://localhost:8080/vospace/transfers/{job_id}' + f'/results/transferDetails', params=None) + self.assertEqual(expected_status, status, msg=response) + if status == 200: + return Transfer.fromstring(response) + return None + + async def change_job_state(self, job_id, state='PHASE=RUN', expected_status=200): + status, response = await self.post(f'http://localhost:8080/vospace/transfers/{job_id}/phase', + data=state) + self.assertEqual(status, expected_status, msg=response) + return status, response + + async def poll_job(self, job_id, poll_until=('COMPLETED', 'ERROR'), expected_status='COMPLETED'): + while True: + status, response = await self.get(f'http://localhost:8080/vospace/transfers/{job_id}/phase') + self.assertEqual(200, status, msg=response) + if response in poll_until: + break + await asyncio.sleep(0.1) + self.assertEqual(response, expected_status, msg=response) + + async def get_error_summary(self, job_id, error_contains): + status, response = await self.get(f'http://localhost:8080/vospace/transfers/{job_id}/error', + params=None) + self.assertEqual(200, status, msg=response) + job = UWSJob.fromstring(response) + self.assertIsNotNone(job.error) + self.assertTrue(error_contains in job.error, msg=job.error) + + async def push_to_space(self, url, file_path, expected_status=200): + async with aiohttp.ClientSession(cookie_jar=self.session.cookie_jar) as session: + async with session.put(url, data=self.file_sender(file_name=file_path)) as resp: + response = await resp.text() + self.assertEqual(resp.status, expected_status, msg=response) + + async def push_to_space_with_content_length(self, url, file_path, expected_status=200): + async with aiohttp.ClientSession(cookie_jar=self.session.cookie_jar) as session: + async with session.put(url, + data=self.file_sender(file_name=file_path), + headers={'Content-Length': str(os.path.getsize(file_path))}) as resp: + response = await resp.text() + self.assertEqual(resp.status, expected_status, msg=response) + + async def push_to_space_defer_error(self, url, file_path): + async with aiohttp.ClientSession(cookie_jar=self.session.cookie_jar) as session: + async with session.put(url, data=self.file_sender(file_name=file_path)) as resp: + response = await resp.text() + return resp.status, response + + async def pull_from_space(self, url, output_path, expected_status=(200,)): + async with aiohttp.ClientSession(cookie_jar=self.session.cookie_jar) as session: + async with session.get(url) as resp: + self.assertIn(resp.status, expected_status) + if resp.status == 200: + hdr_length = int(resp.headers[aiohttp.hdrs.CONTENT_LENGTH]) + path = f"{output_path}/{resp.content_disposition.filename}" + downloaded = 0 + try: + async with aiofiles.open(path, mode='wb') as out_file: + while downloaded < hdr_length: + buff = await resp.content.read(65536) + if not buff: + break + await out_file.write(buff) + downloaded += len(buff) + self.assertEqual(hdr_length, downloaded, f"Header: {hdr_length} != Recv: {downloaded}") + except Exception as e: + raise IOError(str(e)) + + async def pull_from_space_defer_error(self, url, output_path): + async with aiohttp.ClientSession(cookie_jar=self.session.cookie_jar) as session: + async with session.get(url) as resp: + if resp.status == 200: + hdr_length = int(resp.headers[aiohttp.hdrs.CONTENT_LENGTH]) + path = f"{output_path}/{resp.content_disposition.filename}" + downloaded = 0 + try: + async with aiofiles.open(path, mode='wb') as out_file: + while downloaded < hdr_length: + buff = await resp.content.read(65536) + if not buff: + break + await out_file.write(buff) + downloaded += len(buff) + self.assertEqual(hdr_length, downloaded, f"Header: {hdr_length} != Recv: {downloaded}") + return resp.status + except Exception as e: + raise IOError(str(e)) + + def download_file_product(self, transfer, output_path): + t = self.session.cookie_jar.filter_cookies('http://localhost:8080/') + cookies = {} + for _, value in t.items(): + cookies[value.key] = value.value + with requests.post(url='http://localhost:8080/vospace/synctrans', + params=urlencode(transfer.tomap()), + cookies=cookies, + stream=True, + verify=False) as r: + r.raise_for_status() + path = f"{output_path}/test" + with open(path, 'wb') as f: + for chunk in r.iter_content(chunk_size=65536): + if chunk: + f.write(chunk) + + file_size = os.path.getsize(path) + content_length = int(r.headers['content-length']) + if file_size != content_length: + raise IOError('size mismatch') \ No newline at end of file diff --git a/test/test_ngas/test_copy_move.py b/test/test_ngas/test_copy_move.py new file mode 100644 index 0000000..5442c01 --- /dev/null +++ b/test/test_ngas/test_copy_move.py @@ -0,0 +1,324 @@ +# +# ICRAR - International Centre for Radio Astronomy Research +# (c) UWA - The University of Western Australia, 2018 +# Copyright by UWA (in the framework of the ICRAR) +# All rights reserved +# +# This library is free software; you can redistribute it and/or +# modify it under the terms of the GNU Lesser General Public +# License as published by the Free Software Foundation; either +# version 2.1 of the License, or (at your option) any later version. +# +# This library is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# Lesser General Public License for more details. +# +# You should have received a copy of the GNU Lesser General Public +# License along with this library; if not, write to the Free Software +# Foundation, Inc., 59 Temple Place, Suite 330, Boston, +# MA 02111-1307 USA + +import unittest + +from pyvospace.core.model import * +from test_base import TestBase + +class TestCopyMove(TestBase): + + def tearDown(self): + self.loop.run_until_complete(self.delete('http://localhost:8080/vospace/nodes/root1')) + self.loop.run_until_complete(self.delete('http://localhost:8080/vospace/nodes/root2')) + self.loop.run_until_complete(self.delete('http://localhost:8080/vospace/nodes/root3')) + self.loop.run_until_complete(self.delete('http://localhost:8080/vospace/nodes/root4')) + self.loop.run_until_complete(self.delete('http://localhost:8080/vospace/nodes/data0')) + self.loop.run_until_complete(self.delete('http://localhost:8080/vospace/nodes/data1')) + self.loop.run_until_complete(self.delete('http://localhost:8080/vospace/nodes/data2')) + self.loop.run_until_complete(self.delete('http://localhost:8080/vospace/nodes/data3')) + self.loop.run_until_complete(self.delete('http://localhost:8080/vospace/nodes/test1')) + self.loop.run_until_complete(self.delete('http://localhost:8080/vospace/nodes/newnode')) + super().tearDown() + + def test_move_node(self): + async def run(): + root1 = ContainerNode('root1') + status, response = await self.put('http://localhost:8080/vospace/nodes/root1', + data=root1.tostring()) + self.assertEqual(201, status, msg=response) + + root2 = ContainerNode('/root2') + status, response = await self.put('http://localhost:8080/vospace/nodes/root2', + data=root2.tostring()) + self.assertEqual(201, status, msg=response) + + properties = [Property('ivo://ivoa.net/vospace/core#title', "Test1", True), + Property('ivo://ivoa.net/vospace/core#description', "Test2", True)] + node1 = ContainerNode('/root1/test2', properties=properties) + status, response = await self.put('http://localhost:8080/vospace/nodes/root1/test2', + data=node1.tostring()) + self.assertEqual(201, status, msg=response) + + node2 = ContainerNode('/root1/test2/test3') + status, response = await self.put('http://localhost:8080/vospace/nodes/root1/test2/test3', + data=node2.tostring()) + self.assertEqual(201, status, msg=response) + + node3 = ContainerNode('/root1/test2/test4') + status, response = await self.put('http://localhost:8080/vospace/nodes/root1/test2/test4', + data=node3.tostring()) + self.assertEqual(201, status, msg=response) + + # Check consistency of tree + params = {'detail': 'max'} + status, response = await self.get('http://localhost:8080/vospace/nodes/root1/test2', + params=params) + self.assertEqual(200, status, msg=response) + + node = Node.fromstring(response) + node.remove_property('ivo://ivoa.net/vospace/core#length') + node.remove_property('ivo://ivoa.net/vospace/core#btime') + node.remove_property('ivo://ivoa.net/vospace/core#ctime') + node.remove_property('ivo://ivoa.net/vospace/core#mtime') + node.remove_property('ivo://icrar.org/vospace/core#statfs') + + orig_node = ContainerNode('/root1/test2', + properties=properties, + nodes=[ContainerNode('/root1/test2/test3'), + ContainerNode('/root1/test2/test4')]) + self.assertEqual(node, orig_node) + + params = {'detail': 'max'} + status, response = await self.get('http://localhost:8080/vospace/nodes/root2', + params=params) + self.assertEqual(200, status, msg=response) + + node = Node.fromstring(response) + node.remove_property('ivo://ivoa.net/vospace/core#length') + node.remove_property('ivo://ivoa.net/vospace/core#btime') + node.remove_property('ivo://ivoa.net/vospace/core#ctime') + node.remove_property('ivo://ivoa.net/vospace/core#mtime') + node.remove_property('ivo://icrar.org/vospace/core#statfs') + orig_node = ContainerNode('/root2') + self.assertEqual(node, orig_node) + + # Move tree from node1 to root2 + mv = Move(ContainerNode('/root1/test2'), ContainerNode('/root2/test2')) + job = await self.transfer_node(mv) + + await self.change_job_state(job.job_id, 'PHASE=RUN') + await self.poll_job(job.job_id, expected_status='COMPLETED') + + # Check tree has been moved from node1 to root2 + params = {'detail': 'max'} + status, response = await self.get('http://localhost:8080/vospace/nodes/root2/test2', + params=params) + self.assertEqual(200, status, msg=response) + + node = Node.fromstring(response) + node.remove_property('ivo://ivoa.net/vospace/core#length') + node.remove_property('ivo://ivoa.net/vospace/core#btime') + node.remove_property('ivo://ivoa.net/vospace/core#ctime') + node.remove_property('ivo://ivoa.net/vospace/core#mtime') + node.remove_property('ivo://icrar.org/vospace/core#statfs') + moved_node = ContainerNode('/root2/test2', + properties=properties, + nodes=[ContainerNode('/root2/test2/test3'), + ContainerNode('/root2/test2/test4')]) + self.assertEqual(node, moved_node) + + params = {'detail': 'max'} + status, response = await self.get('http://localhost:8080/vospace/nodes/root1', + params=params) + self.assertEqual(200, status, msg=response) + + node = Node.fromstring(response) + node.remove_property('ivo://ivoa.net/vospace/core#length') + node.remove_property('ivo://ivoa.net/vospace/core#btime') + node.remove_property('ivo://ivoa.net/vospace/core#ctime') + node.remove_property('ivo://ivoa.net/vospace/core#mtime') + node.remove_property('ivo://icrar.org/vospace/core#statfs') + orig_node = ContainerNode('/root1') + self.assertEqual(node, orig_node) + + self.loop.run_until_complete(run()) + + def test_move_to_existing_child_node(self): + async def run(): + node1 = ContainerNode('/data1') + node3 = ContainerNode('/data3') + node12 = Node('/data1/data2') + node32 = Node('/data3/data2') + + await self.create_node(node1) + await self.create_node(node3) + await self.create_node(node12) + await self.create_node(node32) + + mv = Move(node12, node3) + job = await self.transfer_node(mv) + + await self.change_job_state(job.job_id, 'PHASE=RUN') + await self.poll_job(job.job_id, expected_status='ERROR') + await self.get_error_summary(job.job_id, error_contains='Duplicate') + + self.loop.run_until_complete(run()) + + def test_rename_node(self): + async def run(): + node0 = Node('/data0') + await self.create_node(node0) + + mv = Move(Node('/data0'), Node('/newnode')) + job = await self.transfer_node(mv) + await self.change_job_state(job.job_id, 'PHASE=RUN') + await self.poll_job(job.job_id, expected_status='COMPLETED') + + mv = Move(Node('/newnode'), Node('/newnode/newnode')) + job = await self.transfer_node(mv) + await self.change_job_state(job.job_id, 'PHASE=RUN') + await self.poll_job(job.job_id, expected_status='ERROR') + + node0 = ContainerNode('/data0') + await self.create_node(node0) + + mv = Move(Node('/newnode'), Node('/data0/newnode')) + job = await self.transfer_node(mv) + await self.change_job_state(job.job_id, 'PHASE=RUN') + await self.poll_job(job.job_id, expected_status='COMPLETED') + + self.loop.run_until_complete(run()) + + def test_invalid_copy_move(self): + async def run(): + node0 = Node('/data0') + node1 = ContainerNode('/data1') + node2 = ContainerNode('/data2') + node3 = ContainerNode('/data1/data4') + node4 = Node('/data2/data4') + + # create nodes for invalid tests + await self.create_node(node0) + await self.create_node(node1) + await self.create_node(node2) + await self.create_node(node3) + await self.create_node(node4) + + # Invalid Jobid + await self.change_job_state(1234, 'PHASE=RUN', expected_status=400) + + # Source node doesn't exist + mv = Move(Node('/data11'), node2) + job = await self.transfer_node(mv) + await self.change_job_state(job.job_id, 'PHASE=RUN') + await self.poll_job(job.job_id, expected_status='ERROR') + await self.get_error_summary(job.job_id, "Node Not Found") + + # Destination node doesn't exist + mv = Move(Node('/data0'), ContainerNode('/doesnotexist/data0')) + job = await self.transfer_node(mv) + await self.change_job_state(job.job_id, 'PHASE=RUN') + await self.poll_job(job.job_id, expected_status='ERROR') + await self.get_error_summary(job.job_id, "Node Not Found") + + # move node1 -> node2 + mv = Move(node1, node2) + job = await self.transfer_node(mv) + + # Invalid Phase + await self.change_job_state(job.job_id, 'PHASE=STOP', expected_status=400) + + # delete node before move + await self.delete('http://localhost:8080/vospace/nodes/data1') + + # start move job + await self.change_job_state(job.job_id, 'PHASE=RUN') + await self.poll_job(job.job_id, expected_status='ERROR') + # Check error, node should not exist + await self.get_error_summary(job.job_id, "Node Not Found") + + # Create the deleted node from previous test + await self.create_node(node1) + + # Invalid move to a non-container + mv = Move(node1, node0) + job = await self.transfer_node(mv) + await self.change_job_state(job.job_id, 'PHASE=RUN') + await self.poll_job(job.job_id, expected_status='ERROR') + await self.get_error_summary(job.job_id, error_contains='Duplicate Node') + + # Invalid move if node already exists in destination tree + await self.create_node(node3) + mv = Move(node3, node2) + job = await self.transfer_node(mv) + await self.change_job_state(job.job_id, 'PHASE=RUN') + await self.poll_job(job.job_id, expected_status='ERROR') + await self.get_error_summary(job.job_id, error_contains='Duplicate Node') + + # Move parent to child which should be invalid because node1 is node3s parent + mv = Move(node1, node3) + job = await self.transfer_node(mv) + await self.change_job_state(job.job_id, 'PHASE=RUN') + await self.poll_job(job.job_id, expected_status='ERROR') + await self.get_error_summary(job.job_id, error_contains='Invalid URI.') + + self.loop.run_until_complete(run()) + + def test_copy_node(self): + async def run(): + root1 = ContainerNode('/root3') + await self.create_node(root1) + + root2 = ContainerNode('/root4') + await self.create_node(root2) + + properties = [Property('ivo://ivoa.net/vospace/core#title', "Test1", True), + Property('ivo://ivoa.net/vospace/core#description', "Test2", True)] + node1 = ContainerNode('/root3/test1', properties=properties) + await self.create_node(node1) + + properties1 = [Property('ivo://ivoa.net/vospace/core#title', "Hello", True), + Property('ivo://ivoa.net/vospace/core#description', "There", True)] + node2 = Node('/root3/test1/test2', properties=properties1) + await self.create_node(node2) + + # Copy tree from node1 to root2 + mv = Copy(ContainerNode('/root3/test1'), ContainerNode('/root4/test1')) + job = await self.transfer_node(mv) + await self.change_job_state(job.job_id, 'PHASE=RUN') + await self.poll_job(job.job_id, expected_status='COMPLETED') + + # Just chek there isn't any transfer details for a move or copy + await self.get_transfer_details(job.job_id, expected_status=400) + + # Check tree has been moved from node1 to root2 + params = {'detail': 'max'} + node = await self.get_node('root4/test1', params) + + copy_node = ContainerNode('/root4/test1', + properties=properties, + nodes=[Node('/root4/test1/test2')]) + node.remove_property('ivo://ivoa.net/vospace/core#length') + node.remove_property('ivo://ivoa.net/vospace/core#btime') + node.remove_property('ivo://ivoa.net/vospace/core#ctime') + node.remove_property('ivo://ivoa.net/vospace/core#mtime') + node.remove_property('ivo://icrar.org/vospace/core#statfs') + self.assertEqual(node, copy_node) + + # check original node is still there + params = {'detail': 'max'} + node = await self.get_node('root3/test1', params) + orig_node = ContainerNode('/root3/test1', + properties=properties, + nodes=[Node('/root3/test1/test2')]) + node.remove_property('ivo://ivoa.net/vospace/core#length') + node.remove_property('ivo://ivoa.net/vospace/core#btime') + node.remove_property('ivo://ivoa.net/vospace/core#ctime') + node.remove_property('ivo://ivoa.net/vospace/core#mtime') + node.remove_property('ivo://icrar.org/vospace/core#statfs') + self.assertEqual(node, orig_node) + + self.loop.run_until_complete(run()) + + +if __name__ == '__main__': + unittest.main() \ No newline at end of file diff --git a/test/test_ngas/test_create.py b/test/test_ngas/test_create.py new file mode 100644 index 0000000..b129978 --- /dev/null +++ b/test/test_ngas/test_create.py @@ -0,0 +1,268 @@ +# +# ICRAR - International Centre for Radio Astronomy Research +# (c) UWA - The University of Western Australia, 2018 +# Copyright by UWA (in the framework of the ICRAR) +# All rights reserved +# +# This library is free software; you can redistribute it and/or +# modify it under the terms of the GNU Lesser General Public +# License as published by the Free Software Foundation; either +# version 2.1 of the License, or (at your option) any later version. +# +# This library is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# Lesser General Public License for more details. +# +# You should have received a copy of the GNU Lesser General Public +# License along with this library; if not, write to the Free Software +# Foundation, Inc., 59 Temple Place, Suite 330, Boston, +# MA 02111-1307 USA + +import unittest +import xml.etree.ElementTree as ET + +from xml.etree.ElementTree import tostring + +from test_base import TestBase +from pyvospace.core.model import * +import pdb + + +class TestCreate(TestBase): + + def tearDown(self): + self.loop.run_until_complete(self.delete('http://localhost:8080/vospace/nodes/test1')) + self.loop.run_until_complete(self.delete('http://localhost:8080/vospace/nodes/datanode')) + super().tearDown() + + def test_container_model(self): + root = ContainerNode('/test') + root.insert_node_into_tree(ContainerNode('/test/test1')) + root.insert_node_into_tree(ContainerNode('/test/test1/test3')) + with self.assertRaises(InvalidArgument): + root.insert_node_into_tree(DataNode('/test/test1/test3/')) + with self.assertRaises(InvalidArgument): + root.insert_node_into_tree(Node('/test/test1/test3/test3/test4')) + root.insert_node_into_tree(Node('/test/test1/'), True) + with self.assertRaises(InvalidArgument): + root.insert_node_into_tree(Node('/test/'), True) + self.assertEqual(root, ContainerNode('/test', nodes=[Node('/test/test1/')])) + + # add to a node deeper in the tree + root1 = ContainerNode('/root1/test2/test3') + root1.insert_node_into_tree(ContainerNode('/root1/test2/test3/test4')) + nodes = [n for n in Node.walk(root1)] + self.assertEqual(len(nodes), 2) + + # check root + root1 = ContainerNode('/') + root1.insert_node_into_tree(ContainerNode('///root1')) + nodes = [n for n in Node.walk(root1)] + self.assertEqual(len(nodes), 2) + + + def test_create_delete(self): + async def run(): + node = ContainerNode('test1') + await self.create_node(node) + + node = Node('/test1/zzz') + await self.create_node(node) + + node = ContainerNode('/test1/test2') + await self.create_node(node) + + node = ContainerNode('/test1/test2/anothercontainer') + await self.create_node(node) + + node = DataNode('/test1/test2/anothercontainer/test.tar') + await self.create_node(node) + + for i in range(100): + node = Node(f'/test1/{i}') + await self.create_node(node) + + self.loop.run_until_complete(run()) + + def test_get_protocol(self): + async def run(): + status, response = await self.get('http://localhost:8080/vospace/protocols', params=None) + self.assertEqual(200, status, msg=response) + + self.loop.run_until_complete(run()) + + def test_get_views(self): + async def run(): + status, response = await self.get('http://localhost:8080/vospace/views', params=None) + self.assertEqual(200, status, msg=response) + + self.loop.run_until_complete(run()) + + def test_get_properties(self): + async def run(): + properties = [Property('ivo://ivoa.net/vospace/core#title', "Hello1", False), + Property('ivo://ivoa.net/vospace/core#description', "Hello2", False)] + node1 = ContainerNode('/test1', properties=properties) + await self.create_node(node1) + + status, response = await self.get('http://localhost:8080/vospace/properties', params=None) + self.assertEqual(200, status, msg=response) + + self.loop.run_until_complete(run()) + + def test_set_properties(self): + async def run(): + properties = [Property('ivo://ivoa.net/vospace/core#title', "Hello1", False), + Property('ivo://ivoa.net/vospace/core#description', "Hello2", False)] + node1 = ContainerNode('/test1', properties=properties) + + await self.create_node(node1) + + # Set properties + properties = [Property('ivo://ivoa.net/vospace/core#title', "NewTitle"), + DeleteProperty('ivo://ivoa.net/vospace/core#description')] + node1 = ContainerNode('/test1', properties=properties) + + # Node doesnt exist + await self.set_node_properties(Node('/test2'), expected_status=404) + + await self.set_node_properties(node1) + + params = {'detail': 'max'} + node = await self.get_node('test1', params) + node.remove_property('ivo://ivoa.net/vospace/core#length') + node.remove_property('ivo://ivoa.net/vospace/core#btime') + node.remove_property('ivo://ivoa.net/vospace/core#ctime') + node.remove_property('ivo://ivoa.net/vospace/core#mtime') + node.remove_property('ivo://icrar.org/vospace/core#statfs') + prop = [Property('ivo://ivoa.net/vospace/core#title', "NewTitle", False)] + orig_node = ContainerNode('/test1', properties=prop) + self.assertEqual(node, orig_node) + + self.loop.run_until_complete(run()) + + def test_create_node_fail(self): + async def run(): + # XML Parse Error + status, response = await self.put('http://localhost:8080/vospace/nodes/', + data='') + self.assertEqual(500, status, msg=response) + + # URI name in node does not match URL path + node = Node('data1') + status, response = await self.put('http://localhost:8080/vospace/nodes/data', + data=node.tostring()) + self.assertEqual(400, status, msg=response) + + # Invalid attribute + node = Node('data1') + root = ET.fromstring(node.tostring()) + root.attrib.pop('uri') + xml = tostring(root) + status, response = await self.put('http://localhost:8080/vospace/nodes/data1', + data=xml) + self.assertEqual(400, status, msg=response) + + # Invalid node type + node = Node('data1') + root = ET.fromstring(node.tostring()) + root.attrib['{http://www.w3.org/2001/XMLSchema-instance}type'] = 'vos:unknown' + xml = tostring(root) + status, response = await self.put('http://localhost:8080/vospace/nodes/data1', + data=xml) + self.assertEqual(400, status, msg=response) + + self.loop.run_until_complete(run()) + + def test_create_node(self): + async def run(): + properties = [Property('ivo://ivoa.net/vospace/core#title', "Test", True), + Property('ivo://ivoa.net/vospace/core#description', "Hello", True)] + node = ContainerNode('test1', properties=properties) + await self.create_node(node) + + node1 = ContainerNode('/test1/test2') + await self.create_node(node1) + + node2 = Node('/test1/data') + await self.create_node(node2) + + node3 = DataNode('/datanode') + await self.create_node(node3) + + data_node = await self.get_node('/datanode', params={'detail': 'max'}) + self.assertGreater(len(data_node.provides), 0) + self.assertGreater(len(data_node.accepts), 0) + + # Test failure cases of getNode + params = {'detail': 'max'} + await self.get_node('/', params, expected_status=200) + + # Duplicate Node + await self.create_node(node1, expected_status=409) + + # Link Node + link_node = LinkNode('/test1/test2/test3', 'http://google.com') + await self.create_node(link_node) + + node = await self.get_node('/test1/test2/test3', params={'detail': 'max'}) + self.assertEqual(node.target, link_node.target) + + # Check that Link Node is in Path + node3 = Node('/test1/test2/test3/test4') + await self.create_node(node3, expected_status=400) + + # Node not found + params = {'detail': 'max'} + await self.get_node('mynode', params, expected_status=404) + + # Container Node not found + node_not_found = Node('/test1/test2/test10/test11') + await self.create_node(node_not_found, expected_status=404) + + # Invalid parameters + params = {'detail': 'invalid'} + await self.get_node('test1', params, expected_status=400) + + params = {'limit': -1} + await self.get_node('test1', params, expected_status=400) + + params = {'limit': "string"} + await self.get_node('test1', params, expected_status=400) + + # get only one child of the container node, other should not be there. + # Should be in alphabetical order + params = {'detail': 'max', 'limit': 1} + node = await self.get_node('test1', params) + node.remove_property('ivo://ivoa.net/vospace/core#length') + node.remove_property('ivo://ivoa.net/vospace/core#btime') + node.remove_property('ivo://ivoa.net/vospace/core#ctime') + node.remove_property('ivo://ivoa.net/vospace/core#mtime') + node.remove_property('ivo://icrar.org/vospace/core#statfs') + cmp_node = ContainerNode('/test1', + properties=properties, + nodes=[Node('/test1/data')]) + + self.assertEqual(node, cmp_node) + + params = {'detail': 'min'} + node = await self.get_node('test1', params) + + # detail = min should not have properties + cmp_node = ContainerNode('/test1', + properties=properties, + nodes=[Node('/test1/data'), + ContainerNode('/test1/test2')]) + self.assertNotEqual(node, cmp_node) + + cmp_node = ContainerNode('test1', + nodes=[Node('/test1/data'), + ContainerNode('/test1/test2')]) + + self.assertEqual(node, cmp_node) + + self.loop.run_until_complete(run()) + +if __name__ == '__main__': + unittest.main() \ No newline at end of file diff --git a/test/test_ngas/test_push_pull.py b/test/test_ngas/test_push_pull.py new file mode 100644 index 0000000..f87151d --- /dev/null +++ b/test/test_ngas/test_push_pull.py @@ -0,0 +1,325 @@ +# +# ICRAR - International Centre for Radio Astronomy Research +# (c) UWA - The University of Western Australia, 2018 +# Copyright by UWA (in the framework of the ICRAR) +# All rights reserved +# +# This library is free software; you can redistribute it and/or +# modify it under the terms of the GNU Lesser General Public +# License as published by the Free Software Foundation; either +# version 2.1 of the License, or (at your option) any later version. +# +# This library is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# Lesser General Public License for more details. +# +# You should have received a copy of the GNU Lesser General Public +# License along with this library; if not, write to the Free Software +# Foundation, Inc., 59 Temple Place, Suite 330, Boston, +# MA 02111-1307 USA + +import unittest +import asyncio + +from aiohttp import web + +from pyvospace.core.model import * +from pyvospace.server import set_fuzz, set_busy_fuzz +from pyvospace.server.spaces.ngas.storage.ngas_storage import NGASStorageServer +from test_base import TestBase + + +class TestPushPull(TestBase): + + def setUp(self): + super().setUp() + self.loop.run_until_complete(self._setup()) + ngas_server = self.loop.run_until_complete(NGASStorageServer.create(self.config_filename)) + + self.ngas_runner = web.AppRunner(ngas_server) + self.loop.run_until_complete(self.ngas_runner.setup()) + self.ngas_site = web.TCPSite(self.ngas_runner, 'localhost', 8083) + self.loop.run_until_complete(self.ngas_site.start()) + + async def _setup(self): + if not os.path.exists('/tmp/download'): + os.makedirs('/tmp/download') + await self.create_file('/tmp/datafile.dat') + await self.create_tar('/tmp/mytar.tar.gz') + + def tearDown(self): + self.loop.run_until_complete(self.delete('http://localhost:8080/vospace/nodes/root')) + self.loop.run_until_complete(self.delete('http://localhost:8080/vospace/nodes/datanode')) + self.loop.run_until_complete(self.delete('http://localhost:8080/vospace/nodes/syncdatanode')) + self.loop.run_until_complete(self.delete('http://localhost:8080/vospace/nodes/syncdatanode1.fits')) + self.loop.run_until_complete(self.delete('http://localhost:8080/vospace/nodes/root/mytar.tar.gz')) + self.loop.run_until_complete(self.ngas_runner.shutdown()) + self.loop.run_until_complete(self.ngas_runner.cleanup()) + super().tearDown() + + def test_push_to_container(self): + async def run(): + root_node = ContainerNode('/root') + await self.create_node(root_node) + + node = DataNode('/root/mytar.tar.gz', + properties=[Property('ivo://ivoa.net/vospace/core#title', "mytar.tar.gz", True)]) + await self.create_node(node) + + security_method = SecurityMethod('ivo://ivoa.net/sso#cookie') + + # push tar to node + container_push = PushToSpace(node, [HTTPPut(security_method=security_method)], + view=View('ivo://ivoa.net/vospace/core#tar'), + params=[Parameter("ivo://ivoa.net/vospace/core#length", 1234)]) + + transfer = await self.sync_transfer_node(container_push) + put_end = transfer.protocols[0].endpoint.url + await self.push_to_space(put_end, '/tmp/mytar.tar.gz', expected_status=200) + + # push to container node + container_push = PushToSpace(root_node, [HTTPPut(security_method=security_method)], + view=View('ivo://ivoa.net/vospace/core#tar'), + params=[Parameter("ivo://ivoa.net/vospace/core#length", 1234)]) + + transfer = await self.sync_transfer_node(container_push) + put_end = transfer.protocols[0].endpoint.url + await self.push_to_space(put_end, '/tmp/mytar.tar.gz', expected_status=200) + + pull = PullFromSpace(root_node, [HTTPGet()], view=View('ivo://ivoa.net/vospace/core#tar')) + transfer = await self.sync_transfer_node(pull) + pull_end = transfer.protocols[0].endpoint.url + await self.pull_from_space(pull_end, '/tmp/download/') + + pull = PullFromSpace(node, [HTTPGet()], view=View('ivo://ivoa.net/vospace/core#tar')) + transfer = await self.sync_transfer_node(pull) + pull_end = transfer.protocols[0].endpoint.url + await self.pull_from_space(pull_end, '/tmp/download/') + + self.loop.run_until_complete(run()) + + def test_push_to_space_sync_node_delete(self): + async def run(): + node = Node('/syncdatanode') + push = PushToSpace(node, [HTTPPut()]) + set_fuzz(True) + + async def defer_delete(node): + await asyncio.sleep(0.5) + await self.delete_node(node) + + tasks = [ + asyncio.ensure_future(self.sync_transfer_node(push, 200)), + asyncio.ensure_future(defer_delete(node)) + ] + + await asyncio.gather(*tasks) + set_fuzz(False) + + self.loop.run_until_complete(run()) + + def test_push_to_space_sync(self): + async def run(): + container_node = ContainerNode('/syncdatanode', + properties=[Property('ivo://ivoa.net/vospace/core#title', + "syncdatanode", True)]) + await self.create_node(container_node) + + node = DataNode('/syncdatanode/syncdatanode1.fits', + properties=[Property('ivo://ivoa.net/vospace/core#title', "syncdatanode1.fits", True), + Property('ivo://ivoa.net/vospace/core#contributor', "dave", True)]) + + await self.create_node(node) + + # push to container node + container_push = PushToSpace(container_node, [HTTPPut()], + params=[Parameter("ivo://ivoa.net/vospace/core#length", 1234)]) + + transfer = await self.sync_transfer_node(container_push) + put_end = transfer.protocols[0].endpoint.url + await self.push_to_space(put_end, '/tmp/datafile.dat', expected_status=400) + + # push to leaf node + push = PushToSpace(node, [HTTPPut()], + params=[Parameter("ivo://ivoa.net/vospace/core#length", 1234)]) + + transfer = await self.sync_transfer_node(push) + put_end = transfer.protocols[0].endpoint.url + await self.push_to_space(put_end, '/tmp/datafile.dat', expected_status=200) + + # retrieve leaf data + push = PullFromSpace(node, [HTTPGet()]) + transfer = await self.sync_transfer_node(push) + pull_end = transfer.protocols[0].endpoint.url + await self.pull_from_space(pull_end, '/tmp/download/') + + self.loop.run_until_complete(run()) + + def test_push_to_space_sync_parameterised(self): + async def run(): + node = Node('/syncdatanode1.fits') + push = PushToSpace(node, [HTTPPut()]) + transfer = await self.sync_transfer_node_parameterised(push) + put_end = transfer.protocols[0].endpoint.url + await self.push_to_space(put_end, '/tmp/datafile.dat', expected_status=200) + + pull = PullFromSpace(node, [HTTPGet()]) + await self.sync_pull_from_space_parameterised(pull, '/tmp/download/') + + self.loop.run_until_complete(run()) + + def test_push_to_space_async(self): + async def run(): + node1 = ContainerNode('/datanode') + await self.create_node(node1) + + node1 = ContainerNode('/datanode/datanode1') + await self.create_node(node1) + + node = Node('/datanode/datanode1/datanode2') + push = PushToSpace(node, [HTTPPut()]) + + job = await self.transfer_node(push) + # Job that is not in the correct phase + # This means that the node is not yet associated with the job. + # It gets associated when the job is run. + await self.push_to_space(f'http://localhost:8083/vospace/pushToVoSpace/{job.job_id}', + '/tmp/datafile.dat', expected_status=400) + + # Get transfer details, should be in invalid state because its not Executing + await self.get_transfer_details(job.job_id, expected_status=400) + await self.change_job_state(job.job_id) + await self.poll_job(job.job_id, poll_until=('EXECUTING', 'ERROR'), expected_status='EXECUTING') + + # Get transferDetails + transfer = await self.get_transfer_details(job.job_id, expected_status=200) + end = transfer.protocols[0].endpoint.url + + # badly formed job id + await self.push_to_space('http://localhost:8083/vospace/pushToVoSpace/1234', + '/tmp/datafile.dat', expected_status=400) + + # job that doesn't exist + await self.push_to_space('http://localhost:8083/vospace/pushToVoSpace/1324a40b-4c6a-453b-a756-cd41ca4b7408', + '/tmp/datafile.dat', expected_status=404) + + tasks = [ + asyncio.ensure_future(self.push_to_space_defer_error(end, '/tmp/datafile.dat')), + asyncio.ensure_future(self.push_to_space_defer_error(end, '/tmp/datafile.dat')) + ] + + result = [] + finished, unfinished = await asyncio.wait(tasks) + self.assertEqual(len(finished), 2) + for i in finished: + result.append((await i)[0]) + + self.assertIn(200, result) + self.assertIn(400, result) + + self.loop.run_until_complete(run()) + + def test_push_to_space_async_error(self): + async def run(): + # can not push data to container node + node1 = ContainerNode('/datanode', + properties=[Property('ivo://ivoa.net/vospace/core#title', "datanode", True)]) + await self.create_node(node1) + + leaf = ContainerNode('/datanode/datanode1', + properties=[Property('ivo://ivoa.net/vospace/core#title', "datanode", True)]) + await self.create_node(leaf) + + push = PushToSpace(node1, [HTTPPut()]) + job = await self.transfer_node(push) + await self.change_job_state(job.job_id) + await self.poll_job(job.job_id, poll_until=('EXECUTING', 'ERROR'), expected_status='EXECUTING') + + transfer = await self.get_transfer_details(job.job_id, expected_status=200) + end = transfer.protocols[0].endpoint.url + await self.push_to_space(end, '/tmp/datafile.dat', expected_status=400) + + await self.delete('http://localhost:8080/vospace/nodes/datanode') + + # can not push data to linknode + node1 = LinkNode('/datanode', 'http://google.com') + await self.create_node(node1) + + push = PushToSpace(node1, [HTTPPut()]) + job = await self.transfer_node(push) + await self.change_job_state(job.job_id) + await self.poll_job(job.job_id, poll_until=('EXECUTING', 'ERROR'), expected_status='ERROR') + + await self.delete('http://localhost:8080/vospace/nodes/datanode') + + # delete node before job execute + node1 = Node('/datanode') + await self.create_node(node1) + + push = PushToSpace(node1, [HTTPPut()]) + job = await self.transfer_node(push) + await self.change_job_state(job.job_id) + await self.poll_job(job.job_id, poll_until=('EXECUTING', 'ERROR'), expected_status='EXECUTING') + await self.delete('http://localhost:8080/vospace/nodes/datanode') + + transfer = await self.get_transfer_details(job.job_id, expected_status=200) + end = transfer.protocols[0].endpoint.url + await self.push_to_space(end, '/tmp/datafile.dat', expected_status=404) + + self.loop.run_until_complete(run()) + + def test_push_to_space_concurrent(self): + async def run(): + node1 = ContainerNode('/datanode') + await self.create_node(node1) + + node1 = ContainerNode('/datanode/datanode1') + await self.create_node(node1) + + node1 = Node('/datanode/datanode1/datanode2.dat') + await self.create_node(node1) + + node = Node('/datanode/datanode1/datanode2.dat') + push = PushToSpace(node, [HTTPPut()]) + + job = await self.transfer_node(push) + await self.change_job_state(job.job_id) + await self.poll_job(job.job_id, poll_until=('EXECUTING', 'ERROR'), expected_status='EXECUTING') + + transfer = await self.get_transfer_details(job.job_id, expected_status=200) + end1 = transfer.protocols[0].endpoint.url + + # start job to upload to same node + job = await self.transfer_node(push) + await self.change_job_state(job.job_id) + await self.poll_job(job.job_id, poll_until=('EXECUTING', 'ERROR'), expected_status='EXECUTING') + + transfer = await self.get_transfer_details(job.job_id, expected_status=200) + end2 = transfer.protocols[0].endpoint.url + + set_fuzz(True) + + # concurrent upload + tasks = [ + asyncio.ensure_future(self.push_to_space_defer_error(end1, '/tmp/datafile.dat')), + asyncio.ensure_future(self.push_to_space_defer_error(end2, '/tmp/datafile.dat')) + ] + + set_fuzz(False) + + result = [] + finished, unfinished = await asyncio.wait(tasks) + self.assertEqual(len(finished), 2) + for i in finished: + r = await i + result.append(r[0]) + + self.assertIn(200, result) + self.assertIn(400, result) + + self.loop.run_until_complete(run()) + + +if __name__ == '__main__': + unittest.main() \ No newline at end of file diff --git a/test/test_ngas/test_simple.py b/test/test_ngas/test_simple.py new file mode 100644 index 0000000..c1d9c2d --- /dev/null +++ b/test/test_ngas/test_simple.py @@ -0,0 +1,210 @@ +# +# ICRAR - International Centre for Radio Astronomy Research +# (c) UWA - The University of Western Australia, 2018 +# Copyright by UWA (in the framework of the ICRAR) +# All rights reserved +# +# This library is free software; you can redistribute it and/or +# modify it under the terms of the GNU Lesser General Public +# License as published by the Free Software Foundation; either +# version 2.1 of the License, or (at your option) any later version. +# +# This library is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# Lesser General Public License for more details. +# +# You should have received a copy of the GNU Lesser General Public +# License along with this library; if not, write to the Free Software +# Foundation, Inc., 59 Temple Place, Suite 330, Boston, +# MA 02111-1307 USA + +import unittest +import asyncio +import logging +import hashlib +import io +import filecmp +import pdb + +import aiofiles +from aiohttp import web + +from pyvospace.core.model import * +from pyvospace.server import set_fuzz, set_busy_fuzz +from pyvospace.server.spaces.ngas.storage.ngas_storage import NGASStorageServer +from test_base import TestBase + +class TestPushPull(TestBase): + + def setUp(self): + + logger=logging.getLogger("aiohttp.web") + logger.setLevel(logging.DEBUG) + + super().setUp() + self.loop.run_until_complete(self._setup()) + ngas_server = self.loop.run_until_complete(NGASStorageServer.create(self.config_filename, logger=logger)) + + self.ngas_runner = web.AppRunner(ngas_server) + self.loop.run_until_complete(self.ngas_runner.setup()) + self.ngas_site = web.TCPSite(self.ngas_runner, 'localhost', 8083) + self.loop.run_until_complete(self.ngas_site.start()) + + async def _setup(self): + if not os.path.exists('/tmp/download'): + os.makedirs('/tmp/download') + await self.create_file('/tmp/datafile.dat') + await self.create_tar('/tmp/mytar.tar.gz') + + def tearDown(self): + self.loop.run_until_complete(self.delete('http://localhost:8080/vospace/nodes/root')) + self.loop.run_until_complete(self.delete('http://localhost:8080/vospace/nodes/datanode')) + self.loop.run_until_complete(self.delete('http://localhost:8080/vospace/nodes/syncdatanode')) + self.loop.run_until_complete(self.delete('http://localhost:8080/vospace/nodes/syncdatanode1.fits')) + self.loop.run_until_complete(self.delete('http://localhost:8080/vospace/nodes/root/mytar.tar.gz')) + self.loop.run_until_complete(self.ngas_runner.shutdown()) + self.loop.run_until_complete(self.ngas_runner.cleanup()) + super().tearDown() + + def test_push_pull_simple_chunked(self): + async def run(): + root_node = ContainerNode('/root') + await self.create_node(root_node) + + # Make a file with content in it + test_file="/tmp/datafile.dat" + test_basename=os.path.basename(test_file) + + test_bytes=1234 + with open(test_file,"wb") as fd: + fd.write(os.urandom(test_bytes)) + + node = DataNode('/root/datafile.dat', + properties=[Property('ivo://ivoa.net/vospace/core#title', test_basename, True), + Property('ivo://ivoa.net/vospace/core#contributor', "dave", True)]) + await self.create_node(node) + + # Push to leaf node + push = PushToSpace(node, [HTTPPut()], params=[Parameter("ivo://ivoa.net/vospace/core#length", test_bytes)]) + # + transfer = await self.sync_transfer_node(push) + put_end = transfer.protocols[0].endpoint.url + await self.push_to_space(put_end, test_file, expected_status=200) + + # # Pull from leaf node + pull = PullFromSpace(node, [HTTPGet()]) + transfer = await self.sync_transfer_node(pull) + pull_end = transfer.protocols[0].endpoint.url + await self.pull_from_space(pull_end, '/tmp/download/') + + # Do it again + transfer = await self.sync_transfer_node(push) + put_end = transfer.protocols[0].endpoint.url + await self.push_to_space(put_end, test_file, expected_status=200) + + # # Pull from leaf node + pull = PullFromSpace(node, [HTTPGet()]) + transfer = await self.sync_transfer_node(pull) + pull_end = transfer.protocols[0].endpoint.url + await self.pull_from_space(pull_end, '/tmp/download/') + + # Compare the two files + result=filecmp.cmp(test_file, '/tmp/download/'+test_basename) + + self.assertEqual(result, True, msg="Downloaded file not the same as uploaded file") + + self.loop.run_until_complete(run()) + + + def test_push_pull_simple_with_content_length(self): + async def run(): + root_node = ContainerNode('/root') + await self.create_node(root_node) + + # Make a file with content in it + test_file = "/tmp/datafile.dat" + test_basename = os.path.basename(test_file) + + test_bytes = 1234 + with open(test_file, "wb") as fd: + fd.write(os.urandom(test_bytes)) + + node = DataNode('/root/datafile.dat', + properties=[Property('ivo://ivoa.net/vospace/core#title', test_basename, True), + Property('ivo://ivoa.net/vospace/core#contributor', "dave", True)]) + await self.create_node(node) + + # Push to leaf node + push = PushToSpace(node, [HTTPPut()], + params=[Parameter("ivo://ivoa.net/vospace/core#length", test_bytes)]) + # + transfer = await self.sync_transfer_node(push) + put_end = transfer.protocols[0].endpoint.url + await self.push_to_space_with_content_length(put_end, test_file, expected_status=200) + + # # Pull from leaf node + pull = PullFromSpace(node, [HTTPGet()]) + transfer = await self.sync_transfer_node(pull) + pull_end = transfer.protocols[0].endpoint.url + await self.pull_from_space(pull_end, '/tmp/download/') + + # Compare the two files + result = filecmp.cmp(test_file, '/tmp/download/' + test_basename) + + self.assertEqual(result, True, msg="Downloaded file not the same as uploaded file") + + self.loop.run_until_complete(run()) + + def test_push_to_container(self): + async def run(): + + root_node = ContainerNode('/root') + await self.create_node(root_node) + + node = DataNode('/root/mytar.tar.gz', + properties=[Property('ivo://ivoa.net/vospace/core#title', "mytar.tar.gz", True)]) + await self.create_node(node) + + security_method = SecurityMethod('ivo://ivoa.net/sso#cookie') + + # push tar to node + container_push = PushToSpace(node, [HTTPPut(security_method=security_method)], + view=View('ivo://ivoa.net/vospace/core#tar'), + params=[Parameter("ivo://ivoa.net/vospace/core#length", 1234)]) + + transfer = await self.sync_transfer_node(container_push) + put_end = transfer.protocols[0].endpoint.url + await self.push_to_space(put_end, '/tmp/mytar.tar.gz', expected_status=200) + + # push to container node + container_push = PushToSpace(root_node, [HTTPPut(security_method=security_method)], + view=View('ivo://ivoa.net/vospace/core#tar'), + params=[Parameter("ivo://ivoa.net/vospace/core#length", 1234)]) + + #pdb.set_trace() + + transfer = await self.sync_transfer_node(container_push) + put_end = transfer.protocols[0].endpoint.url + await self.push_to_space(put_end, '/tmp/mytar.tar.gz', expected_status=200) + + # Do it again + transfer = await self.sync_transfer_node(container_push) + put_end = transfer.protocols[0].endpoint.url + await self.push_to_space(put_end, '/tmp/mytar.tar.gz', expected_status=200) + + pull = PullFromSpace(root_node, [HTTPGet()], view=View('ivo://ivoa.net/vospace/core#tar')) + transfer = await self.sync_transfer_node(pull) + pull_end = transfer.protocols[0].endpoint.url + await self.pull_from_space(pull_end, '/tmp/download/') + + pull = PullFromSpace(node, [HTTPGet()], view=View('ivo://ivoa.net/vospace/core#tar')) + transfer = await self.sync_transfer_node(pull) + pull_end = transfer.protocols[0].endpoint.url + await self.pull_from_space(pull_end, '/tmp/download/') + + self.loop.run_until_complete(run()) + + +if __name__ == '__main__': + unittest.main() diff --git a/test/test_ngas/test_vo_ngas.ini b/test/test_ngas/test_vo_ngas.ini new file mode 100644 index 0000000..7e9e1e1 --- /dev/null +++ b/test/test_ngas/test_vo_ngas.ini @@ -0,0 +1,19 @@ +[Space] +host = localhost +port = 8082 +name = ngas +uri = icrar.org +dsn = postgres://vos_user:vos_user@localhost:5435/vospace +parameters = {} +secret_key = ZlmNyXdQgRhhrC2Wwy-gLZj7Wv6ZtoKH +domain = +use_ssl = 0 + +[Storage] +name = ngas +host = localhost +port = 8083 +parameters = {"staging_dir": "/tmp/ngas/staging/"} +use_ssl = 0 +ngas_servers=[{"hostname" : "localhost", "port" : 7777}, + {"hostname" : "localhost", "port" : 7777}] diff --git a/test/test_posix/__init__.py b/test/test_posix/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/test/test_abort.py b/test/test_posix/test_abort.py similarity index 100% rename from test/test_abort.py rename to test/test_posix/test_abort.py diff --git a/test/test_base.py b/test/test_posix/test_base.py similarity index 99% rename from test/test_base.py rename to test/test_posix/test_base.py index f3865d6..b4e46c0 100644 --- a/test/test_base.py +++ b/test/test_posix/test_base.py @@ -52,7 +52,7 @@ def setUp(self): self.loop = asyncio.new_event_loop() asyncio.set_event_loop(None) - self.config_filename = 'test_vo.ini' + self.config_filename = 'test_vo_posix.ini' config = configparser.ConfigParser() if not os.path.exists(self.config_filename): storage_details = json.dumps( diff --git a/test/test_copy_move.py b/test/test_posix/test_copy_move.py similarity index 100% rename from test/test_copy_move.py rename to test/test_posix/test_copy_move.py diff --git a/test/test_create.py b/test/test_posix/test_create.py similarity index 100% rename from test/test_create.py rename to test/test_posix/test_create.py diff --git a/test/test_push_pull.py b/test/test_posix/test_push_pull.py similarity index 100% rename from test/test_push_pull.py rename to test/test_posix/test_push_pull.py diff --git a/test_vo_ngas.ini b/test_vo_ngas.ini new file mode 100644 index 0000000..7e9e1e1 --- /dev/null +++ b/test_vo_ngas.ini @@ -0,0 +1,19 @@ +[Space] +host = localhost +port = 8082 +name = ngas +uri = icrar.org +dsn = postgres://vos_user:vos_user@localhost:5435/vospace +parameters = {} +secret_key = ZlmNyXdQgRhhrC2Wwy-gLZj7Wv6ZtoKH +domain = +use_ssl = 0 + +[Storage] +name = ngas +host = localhost +port = 8083 +parameters = {"staging_dir": "/tmp/ngas/staging/"} +use_ssl = 0 +ngas_servers=[{"hostname" : "localhost", "port" : 7777}, + {"hostname" : "localhost", "port" : 7777}] diff --git a/test_vo_posix.ini b/test_vo_posix.ini new file mode 100644 index 0000000..5abf016 --- /dev/null +++ b/test_vo_posix.ini @@ -0,0 +1,17 @@ +[Space] +host = localhost +port = 8080 +name = posix +uri = icrar.org +dsn = postgres://vos_user:vos_user@localhost:5435/vospace +parameters = {} +secret_key = ZlmNyXdQgRhhrC2Wwy-gLZj7Wv6ZtoKH +domain = +use_ssl = 0 + +[Storage] +name = posix +host = localhost +port = 8081 +parameters = {"root_dir": "/tmp/posix/storage/", "staging_dir": "/tmp/posix/staging/"} +use_ssl = 0 \ No newline at end of file