Compare commits

...
5 Commits
Author SHA1 Message Date
orrisroot 6ab34a4ca6 build: release 1.3.21
- Bump the version to 1.3.21 and close the Unreleased section of the
  changelog as 1.3.21.
2026-11-13 19:03:59 +09:00
orrisroot 96bfceba7a build: require Python 3.11 or later
pydantic-settings and validators, at the versions required, support
Python 3.11 and later only.

- Raise requires-python, drop the 3.10 classifier and move black's
  target to py311.
2026-11-13 19:01:56 +09:00
orrisroot 2c0f4a75c5 build: update dependencies
- Raise the floor for pydantic to 2.14.3, and for black, flake8, isort
  and pyright to their current releases.
2026-11-13 19:01:56 +09:00
orrisroot 3081d7fbb3 fix: report a locked folder as FolderLockedException
Fetching a locked folder before its password is sent returns 200 with
only its listing fields, without metadata, sub_folders and path, so
FoldersApi.retrieve failed with a pydantic ValidationError. The DOI
route fetches the folder before it can see the lock, so a
password-protected DOI folder could not be opened even with its
password.

- Add FolderLockedException, a subclass of UnauthorizedException, and
  raise it from retrieve on that short record and from the lock checks.
- On a DOI folder, send the password and fetch it again when the first
  fetch finds it locked.
2026-11-13 19:01:56 +09:00
orrisroot a2bc9bb642 feat: build access level labels from their bits and project names
Labels came from a fixed table matched by subset, so Storage (no bits)
matched any level the table lacked, and a level was named after the first
entry it contained: the server's Brain/MINDS 2.0 folders (0x10000) showed
as Storage and 0x10104 as CBS or PW Open.

Labels are now composed from the bits. Projects (bit 16 and up) take the
names the server reports to their members through users/current/, and
otherwise Project<n>; restrict slot 1 is CBS and any other Restrict<n>.
A level with no meaning is shown as Unknown (0x...). chacl reads keys
built the same way, refuses public and storage before any request, and
lists the user's projects with their keys when given an invalid level.
2026-11-13 16:53:25 +09:00
13 changed files with 478 additions and 60 deletions
+14
View File
@@ -2,6 +2,20 @@
All notable changes to this project will be documented in this file.
## [1.3.21] - 2026-11-13
### Changed
- Built access level labels from their bits instead of a fixed table of combinations, so a level the server adds later is labelled too. A project level (bit 16 and up) is named after the project as the server reports it to its members, e.g. `Brain/MINDS 2.0 Open`, and otherwise `Project<n> Open`. A restrict network other than CBS is `Restrict<n> Open`; `0x0200`, formerly `5Kikan Open`, is now `Restrict2 Open`.
- `chacl` takes keys built the same way: `cbs_or_pw_open`, `project0_open` and so on. `5kikan_open` is no longer accepted. Its help lists them, an invalid level lists the projects the user belongs to with their keys, and `public` and `storage`, which the server refuses, are refused before any request.
- `ls` asks the server for the user's projects once per run when logged in.
- Required Python 3.11 or later, as `pydantic-settings` and `validators` do.
- Raised the minimum version of `pydantic` to 2.14.3, and of the development tools `black`, `flake8`, `isort` and `pyright` to their current releases.
- Removed `FolderAccessLevel.ACCESS_LEVELS` and `FolderAccessLevelItem`. `FolderSimple.access_level_label()` labels a level with the project names given to it.
### Fixed
- Reported an access level that has no meaning as `Unknown (0x…)` instead of `Storage`. Storage is the absence of every bit, so it matched any level. A level made of known bits is no longer labelled by the first table entry it contained either: `0x10104` was `CBS or PW Open`. This changes the `access_level` value in `ls` output, including JSON.
- Reported a locked folder that was fetched without its password as `FolderLockedException`, a subclass of `UnauthorizedException`, instead of a pydantic `ValidationError`. The server answers with a short record that has no sub-folders or path. A password-protected DOI folder fails this way before its password is sent, so it can now be opened again.
## [1.3.20] - 2026-11-04
### Changed
+3 -1
View File
@@ -199,12 +199,14 @@ mdrs rm -r neurodata:/NIU/Repository/TEST/dataset
### chacl
Change the folder access level
Change the folder access level. The level is `private`, `pw_open`, `cbs_open`, `project<n>_open`, or several open levels joined by `_or_`. `<n>` is the project's bit position; an invalid level lists the projects you belong to with their keys.
```shell
mdrs chacl private neurodata:/NIU/Repository/Private
mdrs chacl cbs_open -r neurodata:/NIU/Repository/CBS_Open
mdrs chacl pw_open -r -p SHARING_PASSWORD neurodata:/NIU/Repository/PW_Open
mdrs chacl cbs_or_pw_open -p SHARING_PASSWORD neurodata:/NIU/Repository/CBS_or_PW_Open
mdrs chacl project0_open neurodata:/NIU/Repository/Project_Open
```
### metadata
+7 -2
View File
@@ -6,7 +6,7 @@ from pydantic.dataclasses import dataclass
from mdrsclient.api.base import BaseApi
from mdrsclient.api.utils import retry_on_expired_token, token_check
from mdrsclient.exceptions import UnauthorizedException
from mdrsclient.exceptions import FolderLockedException, UnauthorizedException
from mdrsclient.models import Folder, FolderSimple
@@ -38,7 +38,12 @@ class FoldersApi(BaseApi):
token_check(self.connection)
response = self.connection.get(url)
self._raise_response_error(response)
ret = TypeAdapter(Folder).validate_python(response.json())
data = response.json()
# A locked folder not yet opened with its password comes back with 200 but only
# its listing fields, without `metadata`, `sub_folders` and `path`.
if data.get("lock") and "sub_folders" not in data:
raise FolderLockedException(f"Folder `{data.get('name', id)}` is locked.")
ret = TypeAdapter(Folder).validate_python(data)
return ret
@retry_on_expired_token
+23
View File
@@ -1,3 +1,4 @@
from dataclasses import field
from typing import Final
import requests
@@ -20,6 +21,13 @@ class UsersCurrentResponseLaboratory:
role: int
@dataclass(frozen=True)
class UsersCurrentResponseProject:
id: int
name: str | None
bit_position: int
@dataclass(frozen=True)
class UsersApiCurrentResponse:
id: int
@@ -35,6 +43,7 @@ class UsersApiCurrentResponse:
is_reviewer: bool
last_login: str # ISO8601
date_joined: str # ISO8601
projects: list[UsersCurrentResponseProject] = field(default_factory=list)
class UsersApi(BaseApi):
@@ -50,6 +59,20 @@ class UsersApi(BaseApi):
user = User(id=obj.id, username=obj.username, laboratory_ids=laboratory_ids, is_reviewer=obj.is_reviewer)
return user
def project_names(self) -> dict[int, str]:
"""Names of the user's projects by bit_position, the bit their access level sets."""
url = self.ENTRYPOINT + "current/"
response = self.connection.get(url)
self._raise_response_error(response)
obj = TypeAdapter(UsersApiCurrentResponse).validate_python(response.json())
names: dict[int, list[str]] = {}
for project in obj.projects:
if project.name:
# bit_position is not unique on the server, and a folder at a shared bit
# belongs to every project holding it, so all of them are named.
names.setdefault(project.bit_position, []).append(project.name)
return {bit: ", ".join(sorted(x)) for bit, x in names.items()}
def token(self, username: str, password: str) -> Token:
# print(self.__class__.__name__ + "::" + sys._getframe().f_code.co_name)
url = self.ENTRYPOINT + "token/"
+39 -7
View File
@@ -1,18 +1,28 @@
from argparse import Namespace
from typing import Any
from typing import Any, Final
import requests
from mdrsclient.commands.base import BaseCommand
from mdrsclient.exceptions import IllegalArgumentException
from mdrsclient.exceptions import IllegalArgumentException, MDRSException
from mdrsclient.models import FolderAccessLevel
class ChaclCommand(BaseCommand):
ACCESS_LEVEL_HELP: Final[str] = (
"access level: private, pw_open, cbs_open, project<n>_open, "
"or several open levels joined by `_or_` (e.g. cbs_or_pw_open). "
"<n> is the project's bit position; an invalid level lists the projects you belong to"
)
@classmethod
def register(cls, parsers: Any) -> None:
chacl_parser = parsers.add_parser("chacl", help="change the folder access level")
chacl_parser.add_argument("access_level", help="access level (private, cbs_open, pw_open)")
chacl_parser.add_argument("access_level", help=cls.ACCESS_LEVEL_HELP)
chacl_parser.add_argument("-r", "--recursive", help="change access levels recursively", action="store_true")
chacl_parser.add_argument("-p", "--password", help="password to set when access level is `pw_open`")
chacl_parser.add_argument(
"-p", "--password", help="password to set when the access level includes `pw` (e.g. pw_open)"
)
chacl_parser.add_argument("remote_path", help="remote folder path (remote:/lab/path/)")
chacl_parser.set_defaults(func=cls.func)
@@ -21,9 +31,10 @@ class ChaclCommand(BaseCommand):
remote_path = str(args.remote_path)
access_level = FolderAccessLevel.key2id(str(args.access_level))
if access_level is None:
raise IllegalArgumentException(
"Invalid `access_level` parameter. must be `private`, `cbs_open` or `pw_open`."
)
raise IllegalArgumentException(cls._invalid_level_message(remote_path))
if access_level in (FolderAccessLevel.STORAGE, FolderAccessLevel.PUBLIC):
# The server refuses both; saying so here spares a round trip and a bare 400.
raise IllegalArgumentException(f"`{args.access_level}` cannot be set as an access level.")
password = str(args.password) if args.password else None
is_recursive = bool(args.recursive)
cls.chacl(remote_path, access_level, is_recursive, password)
@@ -35,3 +46,24 @@ class ChaclCommand(BaseCommand):
client = MdrsClient.from_remote(remote)
client.chacl(remote_path, access_level, is_recursive, password)
@classmethod
def _invalid_level_message(cls, remote_path: str) -> str:
message = (
"Invalid `access_level` parameter. must be `private`, `pw_open`, `cbs_open`, `project<n>_open` "
"or open levels joined by `_or_` (e.g. `cbs_or_pw_open`)."
)
remote = remote_path.split(":", 1)[0] if ":" in remote_path else ""
from mdrsclient.client import MdrsClient
# The project numbers are known only to the server, and only for the user's own
# projects. Not reaching it leaves the message as it is: the error being reported
# is the level, not the connection.
try:
projects = MdrsClient.from_remote(remote).project_names()
except (MDRSException, requests.RequestException):
return message
if not projects:
return message
lines = [f" project{bit}_open: {name}" for bit, name in sorted(projects.items())]
return "\n".join([message, "Projects you belong to:", *lines])
+9 -5
View File
@@ -1,5 +1,6 @@
import json
from argparse import Namespace
from dataclasses import field
from typing import Any, Final
from pydantic.dataclasses import dataclass
@@ -26,6 +27,7 @@ class LsCommandContext:
is_json: bool
is_quiet: bool
is_recursive: bool
projects: dict[int, str] = field(default_factory=dict)
class LsCommand(BaseCommand):
@@ -85,6 +87,7 @@ class LsCommand(BaseCommand):
is_json,
is_quiet,
is_recursive,
client.project_names(),
)
if context.is_json:
cls._ls_json(context, folder, files)
@@ -110,7 +113,7 @@ class LsCommand(BaseCommand):
length = {key: len(label) if not context.is_quiet else 0 for key, label in cls.LABELS.items()}
for sub_folder in folder.sub_folders:
sub_laboratory_name = cls._laboratory_name(context, sub_folder.laboratory_id)
length["acl"] = max(length["acl"], len(sub_folder.access_level_name))
length["acl"] = max(length["acl"], len(sub_folder.access_level_label(context.projects)))
length["laboratory"] = max(length["laboratory"], len(sub_laboratory_name))
length["size"] = max(length["size"], len(str(sub_folder.size)))
length["date"] = max(length["date"], len(sub_folder.updated_at_name))
@@ -119,7 +122,7 @@ class LsCommand(BaseCommand):
length["size"] = max(length["size"], len(str(file.size)))
length["date"] = max(length["date"], len(file.updated_at_name))
length["name"] = max(length["name"], len(file.name))
length["acl"] = max(length["acl"], len(folder.access_level_name))
length["acl"] = max(length["acl"], len(folder.access_level_label(context.projects)))
length["laboratory"] = max(length["laboratory"], len(context.laboratory.name))
return length
@@ -159,14 +162,15 @@ class LsCommand(BaseCommand):
for sub_folder in sorted(folder.sub_folders, key=lambda x: x.name):
sub_laboratory_name = cls._laboratory_name(context, sub_folder.laboratory_id)
sub_folder_type = "[d]" if sub_folder.lock is False else "[l]"
sub_folder_acl = sub_folder.access_level_label(context.projects)
print(
f"{sub_folder_type:{length['type']}}\t{sub_folder.access_level_name:{length['acl']}}\t"
f"{sub_folder_type:{length['type']}}\t{sub_folder_acl:{length['acl']}}\t"
f"{sub_laboratory_name:{length['laboratory']}}\t{sub_folder.size:{length['size']}}\t"
f"{sub_folder.updated_at_name:{length['date']}}\t{sub_folder.name:{length['name']}}"
)
for file in sorted(files, key=lambda x: x.name):
print(
f"{'[f]':{length['type']}}\t{folder.access_level_name:{length['acl']}}\t"
f"{'[f]':{length['type']}}\t{folder.access_level_label(context.projects):{length['acl']}}\t"
f"{context.laboratory.name:{length['laboratory']}}\t{file.size:{length['size']}}\t"
f"{file.updated_at_name:{length['date']}}\t{file.name:{length['name']}}"
)
@@ -183,7 +187,7 @@ class LsCommand(BaseCommand):
"pid": folder.pid,
"name": folder.name,
"size": folder.size,
"access_level": folder.access_level_name,
"access_level": folder.access_level_label(context.projects),
"lock": folder.lock,
"laboratory": cls._laboratory_name(context, folder.laboratory_id),
"description": folder.description,
+6
View File
@@ -28,6 +28,12 @@ class UnauthorizedException(MDRSException):
pass
class FolderLockedException(UnauthorizedException):
"""Thrown when a folder cannot be opened without its password"""
pass
class TokenExpiredException(UnauthorizedException):
"""Thrown when the access token was still valid when sent but had expired on arrival"""
+90 -25
View File
@@ -1,4 +1,5 @@
from typing import Any, Final, NamedTuple
import re
from typing import Any, Final, Mapping
from unicodedata import normalize
from pydantic.dataclasses import dataclass
@@ -6,39 +7,99 @@ from pydantic.dataclasses import dataclass
from mdrsclient.models.utils import iso8601_to_user_friendly
class FolderAccessLevelItem(NamedTuple):
mask: int
key: str
label: str
class FolderAccessLevel:
# Bit Mask
# - bit 0: Is Private
# - bit 1: Is Public
# - bit 2: With Password
# - bit 3-7: (Reserved)
# - bit 8-15: Restricted Open
ACCESS_LEVELS: Final[list[FolderAccessLevelItem]] = [
FolderAccessLevelItem(0x0204, "5kikan_or_pw_open", "5Kikan or PW Open"),
FolderAccessLevelItem(0x0104, "cbs_or_pw_open", "CBS or PW Open"),
FolderAccessLevelItem(0x0200, "5kikan_open", "5Kikan Open"),
FolderAccessLevelItem(0x0100, "cbs_open", "CBS Open"),
FolderAccessLevelItem(0x0004, "pw_open", "PW Open"),
FolderAccessLevelItem(0x0002, "public", "Public"),
FolderAccessLevelItem(0x0001, "private", "Private"),
FolderAccessLevelItem(0x0000, "storage", "Storage"),
]
# - bit 8-15: Restricted Open, one bit per restrict network slot (slot 1 is bit 8)
# - bit 16-: Project Open, one bit per project bit_position (bit_position 0 is bit 16)
STORAGE: Final[int] = 0x0000
PRIVATE: Final[int] = 0x0001
PUBLIC: Final[int] = 0x0002
PASSWORD: Final[int] = 0x0004
RESERVED: Final[int] = 0x00F8
RESTRICT_SHIFT: Final[int] = 8
RESTRICT_SLOTS: Final[int] = 8
PROJECT_SHIFT: Final[int] = 16
# The server keeps no names for its restrict networks, so the one in use is named here.
# Any other slot is named after its number.
RESTRICT_NAMES: Final[dict[int, str]] = {1: "CBS"}
LEVEL_NAMES: Final[dict[int, str]] = {STORAGE: "Storage", PRIVATE: "Private", PUBLIC: "Public"}
@staticmethod
def restrict_name(slot: int) -> str:
return FolderAccessLevel.RESTRICT_NAMES.get(slot, f"Restrict{slot}")
@staticmethod
def project_name(bit_position: int, projects: Mapping[int, str] | None = None) -> str:
name = (projects or {}).get(bit_position)
return name if name else f"Project{bit_position}"
@staticmethod
def id2label(id: int, projects: Mapping[int, str] | None = None) -> str | None:
"""
Label an access level from its bits, e.g. 0x0104 is "CBS or PW Open".
`projects` names the projects by bit_position, as the server tells their members;
any other project gets a name made from its bit_position. None for a level that
has no meaning: Private or Public with any other bit, or a reserved bit.
"""
if id in FolderAccessLevel.LEVEL_NAMES:
return FolderAccessLevel.LEVEL_NAMES[id]
if id < 0 or id & (FolderAccessLevel.PRIVATE | FolderAccessLevel.PUBLIC | FolderAccessLevel.RESERVED):
return None
names = [
FolderAccessLevel.restrict_name(slot)
for slot in range(1, FolderAccessLevel.RESTRICT_SLOTS + 1)
if id & (1 << (FolderAccessLevel.RESTRICT_SHIFT + slot - 1))
]
project_bits = id >> FolderAccessLevel.PROJECT_SHIFT
names += [
FolderAccessLevel.project_name(bit, projects)
for bit in range(project_bits.bit_length())
if project_bits & (1 << bit)
]
if id & FolderAccessLevel.PASSWORD:
names.append("PW")
return " or ".join(names) + " Open"
@staticmethod
def key2id(key: str) -> int | None:
acl = next((x for x in FolderAccessLevel.ACCESS_LEVELS if x.key == key), None)
return acl.mask if acl is not None else None
"""
The access level a key names: `private`, `public`, `storage`, or the parts of an
open level joined by `_or_` and ending in `_open`, e.g. `cbs_or_pw_open`. A part
is `pw`, `cbs`, `restrict<slot>` or `project<bit_position>`.
"""
fixed = {label.lower(): id for id, label in FolderAccessLevel.LEVEL_NAMES.items()}
if key in fixed:
return fixed[key]
if not key.endswith("_open"):
return None
id = 0
for part in key.removesuffix("_open").split("_or_"):
bit = FolderAccessLevel._part2bit(part)
if bit is None:
return None
id |= bit
return id
@staticmethod
def id2label(id: int) -> str | None:
acl = next((x for x in FolderAccessLevel.ACCESS_LEVELS if (x.mask & id) == x.mask), None)
return acl.label if acl is not None else None
def _part2bit(part: str) -> int | None:
if part == "pw":
return FolderAccessLevel.PASSWORD
slots = {name.lower(): slot for slot, name in FolderAccessLevel.RESTRICT_NAMES.items()}
match = re.fullmatch(r"restrict([1-9][0-9]*)", part)
slot = slots.get(part, int(match.group(1)) if match else None)
if slot is not None:
if slot > FolderAccessLevel.RESTRICT_SLOTS:
return None
return 1 << (FolderAccessLevel.RESTRICT_SHIFT + slot - 1)
match = re.fullmatch(r"project(0|[1-9][0-9]*)", part)
if match:
return 1 << (FolderAccessLevel.PROJECT_SHIFT + int(match.group(1)))
return None
@dataclass(frozen=True)
@@ -55,10 +116,14 @@ class FolderSimple:
updated_at: str
restrict_opened_at: str | None
def access_level_label(self, projects: Mapping[int, str] | None = None) -> str:
"""The access level as shown to a user, naming the projects given by bit_position."""
label = FolderAccessLevel.id2label(self.access_level, projects)
return label if label is not None else f"Unknown (0x{self.access_level:x})"
@property
def access_level_name(self) -> str:
label = FolderAccessLevel.id2label(self.access_level)
return label if label is not None else ""
return self.access_level_label()
@property
def lock_name(self) -> str:
+25 -5
View File
@@ -3,13 +3,14 @@ import re
from unicodedata import normalize
from mdrsclient.api import DoiApi, FilesApi, FoldersApi, LaboratoriesApi, UsersApi
from mdrsclient.api.utils import token_check
from mdrsclient.cache import CacheInterface
from mdrsclient.config import ConfigFile, ConfigInterface
from mdrsclient.connection import MDRSConnection
from mdrsclient.exceptions import (
FolderLockedException,
IllegalArgumentException,
MissingConfigurationException,
UnauthorizedException,
UnexpectedException,
)
from mdrsclient.models import File, Folder, Laboratory, Token, User
@@ -66,6 +67,17 @@ class MdrsService:
def logout(self) -> None:
self.connection.logout()
def project_names(self) -> dict[int, str]:
"""
Names of the logged-in user's projects by bit_position, for labelling access levels.
Empty when nobody is logged in: the server names a project only to its members.
"""
token_check(self.connection)
if self.connection.token is None:
return {}
return UsersApi(self.connection).project_names()
def whoami(self) -> User:
user_api = UsersApi(self.connection)
return user_api.current()
@@ -91,7 +103,7 @@ class MdrsService:
raise UnexpectedException(f"Folder `{path}` not found.")
if folders[0].lock:
if password is None:
raise UnauthorizedException(f"Folder `{path}` is locked.")
raise FolderLockedException(f"Folder `{path}` is locked.")
folder_api.auth(folders[0].id, password)
return folder_api.retrieve(folders[0].id)
@@ -195,12 +207,20 @@ class MdrsService:
)
folder_api = FoldersApi(self.connection)
folder = folder_api.retrieve(doi_resp.folder_id)
# The folder is found by its id, so whether it is locked is known only from what
# retrieve returns: a full record, or the short one of a locked folder.
folder: Folder | None
try:
folder = folder_api.retrieve(doi_resp.folder_id)
except FolderLockedException:
folder = None
if folder.lock:
if folder is None or folder.lock:
if password is None:
raise UnauthorizedException(f"Folder for DOI `{doi_clean}` is locked.")
raise FolderLockedException(f"Folder for DOI `{doi_clean}` is locked.")
folder_api.auth(doi_resp.folder_id, password)
if folder is None:
folder = folder_api.retrieve(doi_resp.folder_id)
lab_api = LaboratoriesApi(self.connection)
labs = lab_api.list()
+8 -9
View File
@@ -7,20 +7,19 @@ name = "mdrsclient"
[project]
name = "mdrs-client-python"
version = "1.3.20"
version = "1.3.21"
description = "The mdrs-client-python is python library and a command-line client for up- and downloading files to and from MDRS based repository."
authors = [
{ name = "Yoshihiro OKUMURA", email = "yoshihiro.okumura@riken.jp" }
]
license = { text = "MIT" }
readme = "README.md"
requires-python = ">=3.10"
requires-python = ">=3.11"
classifiers = [
"Development Status :: 4 - Beta",
"Environment :: Console",
"Intended Audience :: Developers",
"Intended Audience :: Science/Research",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Programming Language :: Python :: 3.13",
@@ -33,7 +32,7 @@ dependencies = [
"requests>=2.34.2",
"requests-toolbelt>=1.0.0",
"python-dotenv>=1.2.4",
"pydantic>=2.14.2",
"pydantic>=2.14.3",
"pydantic-settings>=2.16.2",
"PyJWT>=2.15.1",
"validators>=0.36.0",
@@ -41,11 +40,11 @@ dependencies = [
[project.optional-dependencies]
dev = [
"black>=26.5.1",
"flake8>=7.3.0",
"black>=26.10.1",
"flake8>=7.5.0",
"Flake8-pyproject>=1.2.4",
"isort>=9.0.1",
"pyright>=1.1.411",
"isort>=9.0.2",
"pyright>=1.1.414",
]
[project.scripts]
@@ -53,7 +52,7 @@ mdrs = "mdrsclient.__main__:main"
[tool.black]
line-length = 120
target-version = ['py310']
target-version = ['py311']
include = '\.pyi?$'
exclude = '''
/(
+45
View File
@@ -23,6 +23,7 @@ from mdrsclient.commands import (
VersionCommand,
WhoamiCommand,
)
from mdrsclient.exceptions import IllegalArgumentException, MissingConfigurationException
from mdrsclient.models import Folder, Laboratory
@@ -177,6 +178,7 @@ class TestCommands(unittest.TestCase):
mock_lab = Laboratory(id=1, name="mylab", pi_name="pi_name", full_name="full_name")
mock_client.resolve_folder.return_value = (mock_folder, mock_lab)
mock_client.project_names.return_value = {}
mock_file = MagicMock()
mock_file.name = "file.txt"
@@ -228,6 +230,7 @@ class TestCommands(unittest.TestCase):
mock_lab = Laboratory(id=1, name="mylab", pi_name="pi_name", full_name="full_name")
mock_client.resolve_folder.return_value = (mock_folder, mock_lab)
mock_client.project_names.return_value = {}
mock_client.connection.laboratories.find_by_id.return_value = mock_lab
mock_file = MagicMock()
@@ -316,6 +319,48 @@ class TestCommands(unittest.TestCase):
mock_client_class.from_remote.assert_called_once_with("myremote")
mock_client.chacl.assert_called_once_with("myremote:/mylab/", 1, True, "secret")
def test_chacl_command_project_level(self):
with patch("mdrsclient.client.MdrsClient") as mock_client_class:
mock_client = MagicMock()
mock_client_class.from_remote.return_value = mock_client
args = self.parse_args(ChaclCommand, ["chacl", "cbs_or_project0_open", "myremote:/mylab/"])
args.func(args)
mock_client.chacl.assert_called_once_with("myremote:/mylab/", 0x10100, False, None)
def test_chacl_command_invalid_level_lists_projects(self):
with patch("mdrsclient.client.MdrsClient") as mock_client_class:
mock_client = MagicMock()
mock_client_class.from_remote.return_value = mock_client
mock_client.project_names.return_value = {0: "Brain/MINDS 2.0"}
args = self.parse_args(ChaclCommand, ["chacl", "5kikan_open", "myremote:/mylab/"])
with self.assertRaises(IllegalArgumentException) as cm:
args.func(args)
self.assertIn("project<n>_open", str(cm.exception))
self.assertIn("project0_open: Brain/MINDS 2.0", str(cm.exception))
mock_client.chacl.assert_not_called()
def test_chacl_command_invalid_level_without_server(self):
with patch("mdrsclient.client.MdrsClient") as mock_client_class:
mock_client_class.from_remote.side_effect = MissingConfigurationException("no remote")
args = self.parse_args(ChaclCommand, ["chacl", "5kikan_open", "myremote:/mylab/"])
with self.assertRaises(IllegalArgumentException) as cm:
args.func(args)
self.assertNotIn("Projects you belong to", str(cm.exception))
def test_chacl_command_refuses_public_and_storage(self):
for level in ("public", "storage"):
with self.subTest(level=level), patch("mdrsclient.client.MdrsClient") as mock_client_class:
args = self.parse_args(ChaclCommand, ["chacl", level, "myremote:/mylab/"])
with self.assertRaises(IllegalArgumentException):
args.func(args)
mock_client_class.from_remote.assert_not_called()
def test_metadata_command(self):
with patch("mdrsclient.client.MdrsClient") as mock_client_class:
mock_client = MagicMock()
+154
View File
@@ -0,0 +1,154 @@
import unittest
from unittest.mock import MagicMock, patch
from mdrsclient.api import UsersApi
from mdrsclient.models import FolderSimple
from mdrsclient.models.folder import FolderAccessLevel
from mdrsclient.services import MdrsService
TIMESTAMP = "2026-01-01T00:00:00+09:00"
BM2 = {0: "Brain/MINDS 2.0"}
def make_folder(access_level: int) -> FolderSimple:
return FolderSimple(
id="f1",
pid=None,
name="data",
access_level=access_level,
lock=False,
size=0,
laboratory_id=1,
description="",
created_at=TIMESTAMP,
updated_at=TIMESTAMP,
restrict_opened_at=None,
)
def current_user(projects: list[dict]) -> dict:
return {
"id": 1,
"username": "user",
"first_name": "",
"last_name": "",
"email": "",
"orcid_id": "",
"laboratories": [],
"is_staff": False,
"is_active": True,
"is_superuser": False,
"is_reviewer": False,
"last_login": TIMESTAMP,
"date_joined": TIMESTAMP,
"projects": projects,
}
class TestId2Label(unittest.TestCase):
def test_single_levels(self) -> None:
cases = {
0x0000: "Storage",
0x0001: "Private",
0x0002: "Public",
0x0004: "PW Open",
0x0100: "CBS Open",
0x0104: "CBS or PW Open",
}
for id, label in cases.items():
with self.subTest(id=hex(id)):
self.assertEqual(FolderAccessLevel.id2label(id), label)
def test_restrict_slot_without_a_name_is_numbered(self) -> None:
self.assertEqual(FolderAccessLevel.id2label(0x0200), "Restrict2 Open")
self.assertEqual(FolderAccessLevel.id2label(0x8000), "Restrict8 Open")
def test_project_is_named_from_the_server(self) -> None:
self.assertEqual(FolderAccessLevel.id2label(0x10000, BM2), "Brain/MINDS 2.0 Open")
self.assertEqual(FolderAccessLevel.id2label(0x10004, BM2), "Brain/MINDS 2.0 or PW Open")
def test_project_not_named_gets_its_bit_position(self) -> None:
self.assertEqual(FolderAccessLevel.id2label(0x10000), "Project0 Open")
self.assertEqual(FolderAccessLevel.id2label(0x20000, BM2), "Project1 Open")
def test_parts_are_restrict_then_project_then_password(self) -> None:
self.assertEqual(
FolderAccessLevel.id2label(0x30304, BM2), "CBS or Restrict2 or Brain/MINDS 2.0 or Project1 or PW Open"
)
def test_meaningless_levels_are_unlabelled(self) -> None:
# Private or Public mixed with another bit, a reserved bit, a negative value.
for id in (0x0003, 0x0006, 0x0102, 0x10001, 0x0008, 0x0080, -1):
with self.subTest(id=hex(id)):
self.assertIsNone(FolderAccessLevel.id2label(id))
class TestKey2Id(unittest.TestCase):
def test_keys(self) -> None:
cases = {
"storage": 0x0000,
"private": 0x0001,
"public": 0x0002,
"pw_open": 0x0004,
"cbs_open": 0x0100,
"cbs_or_pw_open": 0x0104,
"restrict1_open": 0x0100,
"restrict2_open": 0x0200,
"project0_open": 0x10000,
"project1_or_pw_open": 0x20004,
}
for key, id in cases.items():
with self.subTest(key=key):
self.assertEqual(FolderAccessLevel.key2id(key), id)
def test_unknown_keys(self) -> None:
for key in ("5kikan_open", "restrict9_open", "restrict0_open", "project01_open", "cbs", "_open", "pw_or_open"):
with self.subTest(key=key):
self.assertIsNone(FolderAccessLevel.key2id(key))
class TestAccessLevelLabel(unittest.TestCase):
def test_known_level(self) -> None:
self.assertEqual(make_folder(0x0100).access_level_name, "CBS Open")
def test_project_names_are_used(self) -> None:
self.assertEqual(make_folder(0x10000).access_level_label(BM2), "Brain/MINDS 2.0 Open")
self.assertEqual(make_folder(0x10000).access_level_name, "Project0 Open")
def test_meaningless_level_shows_its_value(self) -> None:
self.assertEqual(make_folder(0x0008).access_level_name, "Unknown (0x8)")
class TestProjectNames(unittest.TestCase):
def test_anonymous_user_asks_nothing(self) -> None:
connection = MagicMock()
connection.token = None
with patch("mdrsclient.services.UsersApi") as users_api:
self.assertEqual(MdrsService(connection).project_names(), {})
users_api.assert_not_called()
def test_names_by_bit_position(self) -> None:
connection = MagicMock()
connection.get.return_value.json.return_value = current_user(
[
{"id": 1, "name": "Brain/MINDS 2.0", "bit_position": 0},
{"id": 2, "name": "B", "bit_position": 3},
{"id": 3, "name": "A", "bit_position": 3},
{"id": 4, "name": None, "bit_position": 5},
]
)
with patch.object(UsersApi, "_raise_response_error"):
names = UsersApi(connection).project_names()
self.assertEqual(names, {0: "Brain/MINDS 2.0", 3: "A, B"})
def test_server_without_projects(self) -> None:
connection = MagicMock()
user = current_user([])
del user["projects"]
connection.get.return_value.json.return_value = user
with patch.object(UsersApi, "_raise_response_error"):
self.assertEqual(UsersApi(connection).project_names(), {})
if __name__ == "__main__":
unittest.main()
+55 -6
View File
@@ -1,6 +1,8 @@
import unittest
from unittest.mock import MagicMock, patch
from mdrsclient.api import FoldersApi
from mdrsclient.exceptions import FolderLockedException
from mdrsclient.models import Doi, Folder, Laboratories, Laboratory
from mdrsclient.services import MdrsService
@@ -28,10 +30,43 @@ def make_locked_folder() -> Folder:
)
# What the server returns for a locked folder not yet opened with its password.
SHORT_RECORD = {
"id": "f1",
"pid": None,
"name": "root",
"access_level": 4,
"lock": True,
"size": 0,
"laboratory_id": 1,
"description": "",
"created_at": TIMESTAMP,
"updated_at": TIMESTAMP,
"restrict_opened_at": None,
"application": {"id": 0, "status": 0},
}
class TestRetrieveLockedFolder(unittest.TestCase):
def retrieve(self, data: dict) -> Folder:
connection = MagicMock()
connection.get.return_value.json.return_value = data
with patch.object(FoldersApi, "_raise_response_error"):
return FoldersApi(connection).retrieve("f1")
def test_short_record_means_locked(self):
with self.assertRaisesRegex(FolderLockedException, "Folder `root` is locked."):
self.retrieve(SHORT_RECORD)
def test_full_record_of_a_locked_folder_is_read(self):
full = dict(SHORT_RECORD, metadata=[], sub_folders=[], path="/root/")
self.assertEqual(self.retrieve(full).path, "/root/")
class TestFindFolderByDoi(unittest.TestCase):
"""The DOI response carries the folder id as a field, not as a nested object."""
def test_a_locked_doi_folder_is_unlocked_with_its_folder_id(self):
def find(self, retrieved: list, password: str | None) -> tuple[MagicMock, tuple]:
laboratories = Laboratories()
laboratories.append(LABORATORY)
@@ -41,14 +76,28 @@ class TestFindFolderByDoi(unittest.TestCase):
patch("mdrsclient.services.LaboratoriesApi") as laboratories_api,
):
doi_api.return_value.retrieve.return_value = Doi(id="20260429-001", doi=DOI, folder_id="f1")
folders_api.return_value.retrieve.return_value = make_locked_folder()
folders_api.return_value.retrieve.side_effect = retrieved
laboratories_api.return_value.list.return_value = laboratories
return folders_api.return_value, MdrsService(MagicMock()).find_folder_by_doi(DOI, password)
folder, laboratory = MdrsService(MagicMock()).find_folder_by_doi(DOI, "secret")
def test_a_locked_doi_folder_is_unlocked_with_its_folder_id(self):
api, (folder, laboratory) = self.find([FolderLockedException("locked"), make_locked_folder()], "secret")
api.auth.assert_called_once_with("f1", "secret")
self.assertEqual(api.retrieve.call_count, 2)
self.assertEqual(folder.id, "f1")
self.assertEqual(laboratory.name, "mylab")
folders_api.return_value.auth.assert_called_once_with("f1", "secret")
self.assertEqual(folder.id, "f1")
self.assertEqual(laboratory.name, "mylab")
def test_a_locked_doi_folder_without_a_password(self):
with self.assertRaisesRegex(FolderLockedException, f"Folder for DOI `{DOI}` is locked."):
self.find([FolderLockedException("locked")], None)
def test_a_full_record_still_asks_for_the_password(self):
# A locked folder whose record is readable anyway, e.g. its owner's.
api, (folder, _) = self.find([make_locked_folder()], "secret")
api.auth.assert_called_once_with("f1", "secret")
self.assertEqual(api.retrieve.call_count, 1)
with self.assertRaises(FolderLockedException):
self.find([make_locked_folder()], None)
if __name__ == "__main__":