update tests

h5rdmtoolbox/database/zenodo/config.py

@@ -2,7 +2,6 @@
 import configparser
 import os
 import pathlib
-import requests
 from typing import Union
 
 from h5rdmtoolbox.utils import create_tbx_logger
@@ -27,7 +26,6 @@ def get_api_token(sandbox: bool,
     if env_token is not None:
         env_token = env_token.strip()
         logger.debug('Took token from environment variable ZENODO_API_TOKEN.')
-        verify_token(sandbox, env_token)
         return env_token
     zenodo_ini_filename = _parse_ini_file(zenodo_ini_filename)
     config = configparser.ConfigParser()
@@ -40,24 +38,9 @@ def get_api_token(sandbox: bool,
         raise ValueError(f'No API token found in {zenodo_ini_filename}. Please verify the correctness of the file '
                          f'{zenodo_ini_filename}. The api_token entry must be in the section [zenodo] or '
                          f'[zenodo:sandbox].')
-    verify_token(sandbox, api_token)
     return api_token
 
 
-def verify_token(sandbox: bool, api_token: str):
-    # validate the token
-    if sandbox:
-        url = 'https://sandbox.zenodo.org/api/deposit/depositions'
-    else:
-        url = 'https://zenodo.org/api/deposit/depositions'
-    r = requests.get(url,
-                     params={'access_token': api_token})
-    try:
-        r.raise_for_status()
-    except requests.exceptions.HTTPError:
-        raise ValueError(f'Zenodo api token is invalid: {api_token}')
-
-
 def set_api_token(sandbox: bool,
                   api_token: str,
                   zenodo_ini_filename: Union[str, pathlib.Path] = None):

h5rdmtoolbox/database/zenodo/deposit.py

@@ -1,4 +1,5 @@
 import abc
+import appdirs
 import pathlib
 import requests
 from typing import List, Union
@@ -51,14 +52,26 @@ def publish(self):
     def add_file(self, filename):
         """Add a file to the deposit."""
 
+    @abc.abstractmethod
+    def download_files(self):
+        """Download all (!) files from Zenodo."""
+
+    @abc.abstractmethod
+    def download_file(self, name):
+        """Download a specific file from Zenodo."""
+
+    @abc.abstractmethod
+    def get_filenames(self):
+        """Get a list of all filenames."""
+
 
 class ZenodoRecordInterface(AbstractZenodoRecord):
 
     def __init__(self,
                  metadata: Metadata,
                  deposit_id: Union[int, None] = None,
                  file_or_filenames: Union[List[Union[str, pathlib.Path]], None] = None):
-        self.deposit_id = deposit_id
+        self.deposit_id = int(deposit_id)
         self.metadata = metadata
         self.filenames = []
 
@@ -74,6 +87,8 @@ def __init__(self,
 
     def _get(self, raise_for_status: bool = False):
         """Get the deposit from Zenodo."""
+        if self.deposit_id is None:
+            raise ValueError('The deposit_id must be set.')
         base_url = self.base_url + f'/{self.deposit_id}'
         r = requests.get(
             base_url,
@@ -97,14 +112,22 @@ def create(self):
         data = {
             'metadata': self.metadata.model_dump()
         }
+        if self.deposit_id is not None:
+            url = self.base_url + f'/{self.deposit_id}/actions/edit'
+        else:
+            url = self.base_url
         r = requests.post(
-            self.base_url,
+            url,
             json=data,
             params={"access_token": self.api_token},
             headers={"Content-Type": "application/json"}
         )
+        logger.debug(f'creating deposit on Zenodo: request response: {r.json()}')
         r.raise_for_status()
-        self.deposit_id = r.json()['id']
+        if self.deposit_id is None:
+            self.deposit_id = r.json()['id']
+        else:
+            assert self.deposit_id == r.json()['id']
         self._push_files()
         return self.deposit_id
 
@@ -150,6 +173,52 @@ def add_file(self, filename):
             raise FileNotFoundError(f'{filename} does not exist.')
         self.filenames.append(filename)
 
+    def get_filenames(self):
+        """Get a list of all filenames."""
+        r = self._get()
+        return [f['filename'] for f in r.json()['files']]
+
+    def download_files(self, target_folder: Union[str, pathlib.Path] = None):
+        """Download all (!) files from Zenodo."""
+        r = self._get()
+        download_files = []
+        for f in r.json()['files']:
+            if target_folder is None:
+                target_folder = pathlib.Path(appdirs.user_data_dir('h5rdmtoolbox')) / 'zenodo_downloads' / str(
+                    self.deposit_id)
+                target_folder.mkdir(exist_ok=True, parents=True)
+            else:
+                target_folder = pathlib.Path(target_folder)
+            fname = f["filename"]
+            target_filename = target_folder / fname
+            bucket_dict = requests.get(f['links']['self'],
+                                       params={'access_token': self.api_token}).json()
+            logger.debug(f'downloading file "{fname}" to "{target_filename}"')
+            download_files.append(target_filename)
+            with open(target_filename, 'wb') as file:
+                file.write(requests.get(bucket_dict['links']['self']).content)
+        return download_files
+
+    def download_file(self, name, target_folder: Union[str, pathlib.Path] = None):
+        """Download a single file from Zenodo."""
+        if target_folder is None:
+            target_folder = pathlib.Path(appdirs.user_data_dir('h5rdmtoolbox')) / 'zenodo_downloads' / str(
+                self.deposit_id)
+            target_folder.mkdir(exist_ok=True, parents=True)
+        else:
+            target_folder = pathlib.Path(target_folder)
+        r = self._get()
+        for f in r.json()['files']:
+            if f['filename'] == name:
+                fname = f["filename"]
+                target_filename = target_folder / fname
+                bucket_dict = requests.get(f['links']['self'],
+                                           params={'access_token': self.api_token}).json()
+                logger.debug(f'downloading file "{fname}" to "{target_filename}"')
+                with open(target_filename, 'wb') as file:
+                    file.write(requests.get(bucket_dict['links']['self']).content)
+                return target_filename
+
 
 class ZenodoRecord(ZenodoRecordInterface):
     base_url = 'https://zenodo.org/api/deposit/depositions'

h5rdmtoolbox/database/zenodo/metadata.py

@@ -22,9 +22,7 @@
 
 def verify_version(version: str) -> str:
     """Verify that version is a valid as defined in https://semver.org/"""
-    re_pattern = '^(?P<major>0|[1-9]\d*)\.(?P<minor>0|[1-9]\d*)\.(?P<patch>0|[1-9]\d*)(?:-(?P<prerelease>(?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*)(?:\.(?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*))*))?(?:\+(?P<buildmetadata>[0-9a-zA-Z-]+(?:\.[0-9a-zA-Z-]+)*))?$'
-    # print(f'check {version}')
-    assert re.match(pattern=re_pattern, string=version) is not None
+    re_pattern = r'^(?P<major>0|[1-9]\d*)\.(?P<minor>0|[1-9]\d*)\.(?P<patch>0|[1-9]\d*)(?:-(?P<prerelease>(?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*)(?:\.(?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*))*))?(?:\+(?P<buildmetadata>[0-9a-zA-Z-]+(?:\.[0-9a-zA-Z-]+)*))?$'
     return re.match(pattern=re_pattern, string=version) is not None
 
 
@@ -131,7 +129,7 @@ class Metadata(BaseModel):
     image_type: Optional[ImageType] = None  # if upload_type == image
     keywords: List[str]
     notes: Optional[str] = None
-    contributors: Optional[List[Contributor]] = None
+    contributors: Optional[List[Contributor]] = []
     access_right: Optional[Literal["open", "closed", "restricted", "embargoed"]] = "open"
     publication_date: Optional[Union[str, datetime]] = Field(default_factory=datetime.today)
     license: Optional[str] = "cc-by-4.0"
@@ -189,18 +187,15 @@ def _parse_date(value):
 
 
 if __name__ == '__main__':
-    # testing
-    # assert verify_version("0.1.0-rc.1+build.1")
-
     meta = Metadata(version="0.1.0-rc.1+build.1",
                     title='h5rdmtoolbox',
                     description='A toolbox for managing HDF5-based research data management',
                     creators=[Creator(name="Probst, Matthias",
                                       affiliation="KIT - ITS",
-                                      orcid="0000-0003-4423-4370")],
+                                      orcid="0000-0001-8729-0482")],
                     contributors=[Contributor(name="Probst, Matthias",
                                               affiliation="KIT - ITS",
-                                              orcid="0000-0003-4423-4370",
+                                              orcid="0000-0001-8729-0482",
                                               type="ContactPerson")],
                     publication_type='other',
                     access_right='open',

tests/conventions/test_conventions.py

@@ -1,3 +1,4 @@
+import appdirs
 import pathlib
 import pint
 import requests
@@ -15,6 +16,8 @@
 
 __this_dir__ = pathlib.Path(__file__).parent
 
+ZENODO_TUTORIAL_CONVENTION_DEPOSIT_ID = 8363
+
 
 class TestConventions(unittest.TestCase):
 
@@ -31,6 +34,44 @@ def setUp(self) -> None:
         from h5rdmtoolbox.conventions import logger
         logger.setLevel('DEBUG')
 
+    def test_upload_convention(self):
+        cv_yaml_filename = tutorial.get_standard_attribute_yaml_filename()
+        print(cv_yaml_filename)
+        # upload to zenodo sandbox
+        from h5rdmtoolbox.database.zenodo.deposit import ZenodoSandboxRecord
+        from h5rdmtoolbox.database.zenodo.metadata import Metadata, Creator
+        from datetime import datetime
+        meta = Metadata(
+            version="1.0.0",
+            title='H5TBX Tutorial Convention Definition',
+            description='The convention file used in tests and documentation as part of '
+                        f'the h5rdmtoolbox={h5tbx.__version__}.',
+            creators=[Creator(name="Probst, Matthias",
+                              affiliation="Karlsruhe Institute of Technology, Institute for Thermal Turbomachinery",
+                              orcid="0000-0001-8729-0482")],
+            upload_type='other',
+            access_right='open',
+            keywords=['h5rdmtoolbox', 'tutorial', 'convention'],
+            publication_date=datetime.now(),
+        )
+        zsr = ZenodoSandboxRecord(deposit_id=ZENODO_TUTORIAL_CONVENTION_DEPOSIT_ID,
+                                  metadata=meta,
+                                  file_or_filenames=cv_yaml_filename)
+        if not zsr.exists():
+            zsr.deposit_id = None
+            zsr.create()
+        else:
+            zsr.update()
+
+        # download file from zenodo deposit:
+        self.assertEqual(1, len(zsr.get_filenames()))
+        zsr.download_files()
+        zsr.download_file('tutorial_convention.yaml')
+        download_dir = pathlib.Path(appdirs.user_data_dir('h5rdmtoolbox')) / 'zenodo_downloads'
+        self.assertTrue(
+            (download_dir / f'{ZENODO_TUTORIAL_CONVENTION_DEPOSIT_ID}' / 'tutorial_convention.yaml').exists()
+        )
+
     def test_delete(self):
         cv = h5tbx.conventions.Convention.from_yaml(__this_dir__ / 'simple_cv.yaml')
         self.assertTrue(cv.name in sys.modules)

tests/database/test_zenodo.py

@@ -48,10 +48,10 @@ def test_create_new_deposit(self):
             description='A toolbox for managing HDF5-based research data management',
             creators=[Creator(name="Probst, Matthias",
                               affiliation="KIT - ITS",
-                              orcid="0000-0003-4423-4370")],
+                              orcid="0000-0001-8729-0482")],
             contributors=[Contributor(name="Probst, Matthias",
                                       affiliation="KIT - ITS",
-                                      orcid="0000-0003-4423-4370",
+                                      orcid="0000-0001-8729-0482",
                                       type="ContactPerson")],
             upload_type='image',
             image_type='photo',