Merge pull request #2003 from pbiering/sharing-properties-overlay
Sharing properties overlay
This commit is contained in:
17
SHARING.md
17
SHARING.md
@@ -28,6 +28,7 @@ Types of supported sharing configuration:
|
|||||||
* `HiddenByUser`: control by user
|
* `HiddenByUser`: control by user
|
||||||
* `TimestampCreated`: unixtime of creation
|
* `TimestampCreated`: unixtime of creation
|
||||||
* `TimestampUpdated`: unixtime of last update
|
* `TimestampUpdated`: unixtime of last update
|
||||||
|
* `Properties`: overlay properties (limited set whitelisted)
|
||||||
|
|
||||||
`Enabled*`: owner AND user have to enable a share to become usable
|
`Enabled*`: owner AND user have to enable a share to become usable
|
||||||
|
|
||||||
@@ -39,7 +40,9 @@ Types of supported sharing configuration:
|
|||||||
|
|
||||||
(_>= 3.7.0_)
|
(_>= 3.7.0_)
|
||||||
|
|
||||||
One CSV file containing one row per sharing config, separated by `,` and containing header with columns from above.
|
One CSV file containing one row per sharing config, separated by `;` and containing header with columns from above.
|
||||||
|
|
||||||
|
If given, properties are stored in JSON format in CSV.
|
||||||
|
|
||||||
#### Files
|
#### Files
|
||||||
|
|
||||||
@@ -120,6 +123,7 @@ Can be selected by `HTTP_ACCEPT`
|
|||||||
* `Permissions`: effective permission of the share
|
* `Permissions`: effective permission of the share
|
||||||
* `Enabled`: owner/user selected by authentication
|
* `Enabled`: owner/user selected by authentication
|
||||||
* `Hidden`: owner/user selected by authentication
|
* `Hidden`: owner/user selected by authentication
|
||||||
|
* `Properties`: properties to overlay
|
||||||
|
|
||||||
#### API Hooks
|
#### API Hooks
|
||||||
|
|
||||||
@@ -250,7 +254,9 @@ Update a share selected by `PathOrToken`
|
|||||||
| - | - | - |
|
| - | - | - |
|
||||||
| PathOrToken | yes | n/a |
|
| PathOrToken | yes | n/a |
|
||||||
| PathMapped | no | |
|
| PathMapped | no | |
|
||||||
|
| OwnerOrUser | yes | n/a |
|
||||||
| User | no | |
|
| User | no | |
|
||||||
|
| Properties | no | |
|
||||||
|
|
||||||
* Output: result status
|
* Output: result status
|
||||||
|
|
||||||
@@ -282,4 +288,13 @@ ApiVersion=1
|
|||||||
Status=success
|
Status=success
|
||||||
```
|
```
|
||||||
|
|
||||||
|
## Properties Overlay
|
||||||
|
|
||||||
|
Owner or user can define per share a set of properties to overlay on PROPFIND response during create or update via API.
|
||||||
|
|
||||||
|
Whitelisted ones are defined in `OVERLAY_PROPERTIES_WHITELIST` in `radicale/sharing/__init__.py`:
|
||||||
|
|
||||||
|
* `C:calendar-description` (_>= 3.7.0_)
|
||||||
|
* `ICAL:calendar-color` (_>= 3.7.0_)
|
||||||
|
* `CR:addressbook-description` (_>= 3.7.0_)
|
||||||
|
* `INF:addressbook-color` (_>= 3.7.0_)
|
||||||
|
|||||||
@@ -345,6 +345,10 @@ def xml_propfind_response(
|
|||||||
human_tag = xmlutils.make_human_tag(tag)
|
human_tag = xmlutils.make_human_tag(tag)
|
||||||
tag_text = collection.get_meta(human_tag)
|
tag_text = collection.get_meta(human_tag)
|
||||||
if tag_text is not None:
|
if tag_text is not None:
|
||||||
|
if sharing:
|
||||||
|
# map from overlay
|
||||||
|
if sharing['Properties'][human_tag] is not None:
|
||||||
|
tag_text = sharing['Properties'][human_tag]
|
||||||
element.text = tag_text
|
element.text = tag_text
|
||||||
else:
|
else:
|
||||||
is404 = True
|
is404 = True
|
||||||
|
|||||||
@@ -33,7 +33,7 @@ from radicale.log import logger
|
|||||||
|
|
||||||
INTERNAL_TYPES: Sequence[str] = ("csv", "files", "none")
|
INTERNAL_TYPES: Sequence[str] = ("csv", "files", "none")
|
||||||
|
|
||||||
DB_FIELDS_V1: Sequence[str] = ('ShareType', 'PathOrToken', 'PathMapped', 'Owner', 'User', 'Permissions', 'EnabledByOwner', 'EnabledByUser', 'HiddenByOwner', 'HiddenByUser', 'TimestampCreated', 'TimestampUpdated')
|
DB_FIELDS_V1: Sequence[str] = ('ShareType', 'PathOrToken', 'PathMapped', 'Owner', 'User', 'Permissions', 'EnabledByOwner', 'EnabledByUser', 'HiddenByOwner', 'HiddenByUser', 'TimestampCreated', 'TimestampUpdated', 'Properties')
|
||||||
DB_FIELDS_V1_BOOL: Sequence[str] = ('EnabledByOwner', 'EnabledByUser', 'HiddenByOwner', 'HiddenByUser')
|
DB_FIELDS_V1_BOOL: Sequence[str] = ('EnabledByOwner', 'EnabledByUser', 'HiddenByOwner', 'HiddenByUser')
|
||||||
DB_FIELDS_V1_INT: Sequence[str] = ('TimestampCreated', 'TimestampUpdated')
|
DB_FIELDS_V1_INT: Sequence[str] = ('TimestampCreated', 'TimestampUpdated')
|
||||||
# ShareType: <token|map>
|
# ShareType: <token|map>
|
||||||
@@ -76,6 +76,8 @@ PATH_PATTERN: str = "([a-zA-Z0-9/.\\-]+)" # TODO: extend or find better source
|
|||||||
|
|
||||||
USER_PATTERN: str = "([a-zA-Z0-9@]+)" # TODO: extend or find better source
|
USER_PATTERN: str = "([a-zA-Z0-9@]+)" # TODO: extend or find better source
|
||||||
|
|
||||||
|
OVERLAY_PROPERTIES_WHITELIST: Sequence[str] = ("C:calendar-description", "ICAL:calendar-color", "CR:addressbook-description", "INF:addressbook-color")
|
||||||
|
|
||||||
|
|
||||||
def load(configuration: "config.Configuration") -> "BaseSharing":
|
def load(configuration: "config.Configuration") -> "BaseSharing":
|
||||||
"""Load the sharing database module chosen in configuration."""
|
"""Load the sharing database module chosen in configuration."""
|
||||||
@@ -178,20 +180,22 @@ class BaseSharing:
|
|||||||
Permissions: str = "r",
|
Permissions: str = "r",
|
||||||
EnabledByOwner: bool = False, EnabledByUser: bool = False,
|
EnabledByOwner: bool = False, EnabledByUser: bool = False,
|
||||||
HiddenByOwner: bool = True, HiddenByUser: bool = True,
|
HiddenByOwner: bool = True, HiddenByUser: bool = True,
|
||||||
Timestamp: int = 0) -> dict:
|
Timestamp: int = 0,
|
||||||
|
Properties: Union[dict, None] = None) -> dict:
|
||||||
""" create sharing """
|
""" create sharing """
|
||||||
return {"status": "not-implemented"}
|
return {"status": "not-implemented"}
|
||||||
|
|
||||||
def update_sharing(self,
|
def update_sharing(self,
|
||||||
ShareType: str,
|
ShareType: str,
|
||||||
PathOrToken: str,
|
PathOrToken: str,
|
||||||
Owner: Union[str, None] = None,
|
OwnerOrUser: str,
|
||||||
User: Union[str, None] = None,
|
User: Union[str, None] = None,
|
||||||
PathMapped: Union[str, None] = None,
|
PathMapped: Union[str, None] = None,
|
||||||
Permissions: Union[str, None] = None,
|
Permissions: Union[str, None] = None,
|
||||||
EnabledByOwner: Union[bool, None] = None,
|
EnabledByOwner: Union[bool, None] = None,
|
||||||
HiddenByOwner: Union[bool, None] = None,
|
HiddenByOwner: Union[bool, None] = None,
|
||||||
Timestamp: int = 0) -> dict:
|
Timestamp: int = 0,
|
||||||
|
Properties: Union[dict, None] = None) -> dict:
|
||||||
""" update sharing """
|
""" update sharing """
|
||||||
return {"status": "not-implemented"}
|
return {"status": "not-implemented"}
|
||||||
|
|
||||||
@@ -391,6 +395,8 @@ class BaseSharing:
|
|||||||
PathOrToken: <path> (mandatory)
|
PathOrToken: <path> (mandatory)
|
||||||
User: <target_user> (mandatory)
|
User: <target_user> (mandatory)
|
||||||
|
|
||||||
|
action: (token|map)/update
|
||||||
|
|
||||||
action: (token|map)/(delete|disable|enable|hide|unhide)
|
action: (token|map)/(delete|disable|enable|hide|unhide)
|
||||||
PathOrToken: <path|token> (mandatory)
|
PathOrToken: <path|token> (mandatory)
|
||||||
|
|
||||||
@@ -408,7 +414,7 @@ class BaseSharing:
|
|||||||
Status in JSON/TEXT (TEXT can be parsed by shell)
|
Status in JSON/TEXT (TEXT can be parsed by shell)
|
||||||
|
|
||||||
"""
|
"""
|
||||||
if not self.sharing_collection_by_map and not self.sharing_collection_by_token:
|
if not self._enabled:
|
||||||
# API is not enabled
|
# API is not enabled
|
||||||
return httputils.NOT_FOUND
|
return httputils.NOT_FOUND
|
||||||
|
|
||||||
@@ -420,7 +426,7 @@ class BaseSharing:
|
|||||||
if not path.startswith("/.sharing/v1/"):
|
if not path.startswith("/.sharing/v1/"):
|
||||||
return httputils.NOT_FOUND
|
return httputils.NOT_FOUND
|
||||||
|
|
||||||
# split into ShareType and action or "info"
|
# split into ShareType and action
|
||||||
ShareType_action = path.removeprefix("/.sharing/v1/")
|
ShareType_action = path.removeprefix("/.sharing/v1/")
|
||||||
match = re.search('([a-z]+)/([a-z]+)$', ShareType_action)
|
match = re.search('([a-z]+)/([a-z]+)$', ShareType_action)
|
||||||
if not match:
|
if not match:
|
||||||
@@ -481,7 +487,21 @@ class BaseSharing:
|
|||||||
# convert arrays into single value
|
# convert arrays into single value
|
||||||
request_data = {}
|
request_data = {}
|
||||||
for key in request_parsed:
|
for key in request_parsed:
|
||||||
request_data[key] = request_parsed[key][0]
|
if key == "Properties":
|
||||||
|
# Properties key value parser
|
||||||
|
properties_dict: dict = {}
|
||||||
|
for entry in request_parsed[key]:
|
||||||
|
m = re.search('^([^=]+)=([^=]+)$', entry)
|
||||||
|
if not m:
|
||||||
|
return httputils.bad_request("Invalid properties format in form")
|
||||||
|
token = m.group(1).lstrip('"\'').rstrip('"\'')
|
||||||
|
value = m.group(2).lstrip('"\'').rstrip('"\'')
|
||||||
|
properties_dict[token] = value
|
||||||
|
if logger.isEnabledFor(logging.DEBUG):
|
||||||
|
logger.debug("TRACE/sharing/API: converted Properties from form into dict: %r", properties_dict)
|
||||||
|
request_data[key] = properties_dict
|
||||||
|
else:
|
||||||
|
request_data[key] = request_parsed[key][0]
|
||||||
if logger.isEnabledFor(logging.DEBUG):
|
if logger.isEnabledFor(logging.DEBUG):
|
||||||
logger.debug("TRACE/" + api_info + " (form): %r", f"{request_data}")
|
logger.debug("TRACE/" + api_info + " (form): %r", f"{request_data}")
|
||||||
else:
|
else:
|
||||||
@@ -518,6 +538,7 @@ class BaseSharing:
|
|||||||
HiddenByOwner: Union[bool, None] = None
|
HiddenByOwner: Union[bool, None] = None
|
||||||
EnabledByUser: Union[bool, None] = None
|
EnabledByUser: Union[bool, None] = None
|
||||||
HiddenByUser: Union[bool, None] = None
|
HiddenByUser: Union[bool, None] = None
|
||||||
|
Properties: Union[dict, None] = None
|
||||||
|
|
||||||
# parameters sanity check
|
# parameters sanity check
|
||||||
for key in request_data:
|
for key in request_data:
|
||||||
@@ -552,12 +573,9 @@ class BaseSharing:
|
|||||||
|
|
||||||
# check for mandatory parameters
|
# check for mandatory parameters
|
||||||
if 'PathMapped' not in request_data:
|
if 'PathMapped' not in request_data:
|
||||||
if action == 'info':
|
if action in ['info', 'list', 'update']:
|
||||||
# ignored
|
# ignored
|
||||||
pass
|
pass
|
||||||
elif action == "list":
|
|
||||||
# optional
|
|
||||||
pass
|
|
||||||
else:
|
else:
|
||||||
if ShareType == "token" and action != 'create':
|
if ShareType == "token" and action != 'create':
|
||||||
# optional
|
# optional
|
||||||
@@ -588,6 +606,13 @@ class BaseSharing:
|
|||||||
if 'Permissions' in request_data:
|
if 'Permissions' in request_data:
|
||||||
Permissions = request_data['Permissions']
|
Permissions = request_data['Permissions']
|
||||||
|
|
||||||
|
if 'Properties' in request_data:
|
||||||
|
# verify against whitelist
|
||||||
|
for entry in request_data['Properties']:
|
||||||
|
if entry not in OVERLAY_PROPERTIES_WHITELIST:
|
||||||
|
return httputils.bad_request("Property not supported to overlay: %r" % entry)
|
||||||
|
Properties = request_data['Properties']
|
||||||
|
|
||||||
if ShareType == "map":
|
if ShareType == "map":
|
||||||
if action == 'info':
|
if action == 'info':
|
||||||
# ignored
|
# ignored
|
||||||
@@ -609,6 +634,11 @@ class BaseSharing:
|
|||||||
answer['ApiVersion'] = 1
|
answer['ApiVersion'] = 1
|
||||||
Timestamp = int((datetime.now() - datetime(1970, 1, 1)).total_seconds())
|
Timestamp = int((datetime.now() - datetime(1970, 1, 1)).total_seconds())
|
||||||
|
|
||||||
|
if not self.sharing_collection_by_map and not self.sharing_collection_by_token:
|
||||||
|
if not action == 'info':
|
||||||
|
# API is not enabled
|
||||||
|
return httputils.NOT_FOUND
|
||||||
|
|
||||||
# action: list
|
# action: list
|
||||||
if action == "list":
|
if action == "list":
|
||||||
if logger.isEnabledFor(logging.DEBUG):
|
if logger.isEnabledFor(logging.DEBUG):
|
||||||
@@ -691,7 +721,8 @@ class BaseSharing:
|
|||||||
Owner=Owner, User=Owner,
|
Owner=Owner, User=Owner,
|
||||||
Permissions=str(Permissions), # mandantory
|
Permissions=str(Permissions), # mandantory
|
||||||
EnabledByOwner=EnabledByOwner, HiddenByOwner=HiddenByOwner,
|
EnabledByOwner=EnabledByOwner, HiddenByOwner=HiddenByOwner,
|
||||||
Timestamp=Timestamp)
|
Timestamp=Timestamp,
|
||||||
|
Properties=Properties)
|
||||||
if logger.isEnabledFor(logging.DEBUG):
|
if logger.isEnabledFor(logging.DEBUG):
|
||||||
logger.debug("TRACE/" + api_info + ": result=%r", result)
|
logger.debug("TRACE/" + api_info + ": result=%r", result)
|
||||||
|
|
||||||
@@ -747,7 +778,8 @@ class BaseSharing:
|
|||||||
Permissions=str(Permissions), # mandatory
|
Permissions=str(Permissions), # mandatory
|
||||||
EnabledByOwner=EnabledByOwner, HiddenByOwner=HiddenByOwner,
|
EnabledByOwner=EnabledByOwner, HiddenByOwner=HiddenByOwner,
|
||||||
EnabledByUser=EnabledByUser, HiddenByUser=HiddenByUser,
|
EnabledByUser=EnabledByUser, HiddenByUser=HiddenByUser,
|
||||||
Timestamp=Timestamp)
|
Timestamp=Timestamp,
|
||||||
|
Properties=Properties)
|
||||||
|
|
||||||
else:
|
else:
|
||||||
logger.error(api_info + ": unsupported for ShareType=%r", ShareType)
|
logger.error(api_info + ": unsupported for ShareType=%r", ShareType)
|
||||||
@@ -784,8 +816,10 @@ class BaseSharing:
|
|||||||
EnabledByOwner=EnabledByOwner,
|
EnabledByOwner=EnabledByOwner,
|
||||||
HiddenByOwner=HiddenByOwner,
|
HiddenByOwner=HiddenByOwner,
|
||||||
PathOrToken=str(PathOrToken), # verification above that it is not None
|
PathOrToken=str(PathOrToken), # verification above that it is not None
|
||||||
Owner=Owner,
|
OwnerOrUser=Owner,
|
||||||
Timestamp=Timestamp)
|
User=User,
|
||||||
|
Timestamp=Timestamp,
|
||||||
|
Properties=Properties)
|
||||||
|
|
||||||
elif ShareType == "map":
|
elif ShareType == "map":
|
||||||
result = self.update_sharing(
|
result = self.update_sharing(
|
||||||
@@ -795,8 +829,10 @@ class BaseSharing:
|
|||||||
EnabledByOwner=EnabledByOwner,
|
EnabledByOwner=EnabledByOwner,
|
||||||
HiddenByOwner=HiddenByOwner,
|
HiddenByOwner=HiddenByOwner,
|
||||||
PathOrToken=str(PathOrToken), # verification above that it is not None
|
PathOrToken=str(PathOrToken), # verification above that it is not None
|
||||||
Owner=Owner,
|
OwnerOrUser=Owner,
|
||||||
Timestamp=Timestamp)
|
User=User,
|
||||||
|
Timestamp=Timestamp,
|
||||||
|
Properties=Properties)
|
||||||
|
|
||||||
else:
|
else:
|
||||||
logger.error(api_info + ": unsupported for ShareType=%r", ShareType)
|
logger.error(api_info + ": unsupported for ShareType=%r", ShareType)
|
||||||
@@ -918,18 +954,19 @@ class BaseSharing:
|
|||||||
answer_array.append(key + '=' + str(answer[key]))
|
answer_array.append(key + '=' + str(answer[key]))
|
||||||
if 'Content' in answer and answer['Content'] is not None:
|
if 'Content' in answer and answer['Content'] is not None:
|
||||||
csv = io.StringIO()
|
csv = io.StringIO()
|
||||||
writer = DictWriter(csv, fieldnames=DB_FIELDS_V1)
|
writer = DictWriter(csv, fieldnames=DB_FIELDS_V1, delimiter=';')
|
||||||
if output_format == "csv":
|
if output_format == "csv":
|
||||||
writer.writeheader()
|
writer.writeheader()
|
||||||
for entry in answer['Content']:
|
for entry in answer['Content']:
|
||||||
writer.writerow(entry)
|
# TODO: Argument 1 to "writerow" of "DictWriter" has incompatible type "str"; expected "Mapping[str, Any]" [arg-type]
|
||||||
|
writer.writerow(entry) # type: ignore[arg-type]
|
||||||
if output_format == "csv":
|
if output_format == "csv":
|
||||||
answer_array.append(csv.getvalue())
|
answer_array.append(csv.getvalue())
|
||||||
else:
|
else:
|
||||||
index = 0
|
index = 0
|
||||||
for line in csv.getvalue().splitlines():
|
for line in csv.getvalue().splitlines():
|
||||||
# create a shell array with content lines
|
# create a shell array with content lines
|
||||||
answer_array.append('Content[' + str(index) + ']="' + line + '"')
|
answer_array.append('Content[' + str(index) + ']="' + line.replace('"', '\\"') + '"')
|
||||||
index += 1
|
index += 1
|
||||||
headers = {
|
headers = {
|
||||||
"Content-Type": "text/csv"
|
"Content-Type": "text/csv"
|
||||||
|
|||||||
@@ -128,8 +128,9 @@ class Sharing(sharing.BaseSharing):
|
|||||||
UserShare = row['User']
|
UserShare = row['User']
|
||||||
Permissions = row['Permissions']
|
Permissions = row['Permissions']
|
||||||
Hidden: bool = (row['HiddenByOwner'] or row['HiddenByUser'])
|
Hidden: bool = (row['HiddenByOwner'] or row['HiddenByUser'])
|
||||||
if logger.isEnabledFor(logging.DEBUG):
|
Properties: Union[dict, None] = None
|
||||||
logger.debug("TRACE/sharing: map %r to %r (Owner=%r User=%r Permissions=%r Hidden=%s)", PathOrToken, PathMapped, Owner, UserShare, Permissions, Hidden)
|
if 'Properties' in row:
|
||||||
|
Properties = row['Properties']
|
||||||
return {
|
return {
|
||||||
"mapped": True,
|
"mapped": True,
|
||||||
"PathOrToken": PathOrToken,
|
"PathOrToken": PathOrToken,
|
||||||
@@ -137,7 +138,8 @@ class Sharing(sharing.BaseSharing):
|
|||||||
"Owner": Owner,
|
"Owner": Owner,
|
||||||
"User": UserShare,
|
"User": UserShare,
|
||||||
"Hidden": Hidden,
|
"Hidden": Hidden,
|
||||||
"Permissions": Permissions}
|
"Permissions": Permissions,
|
||||||
|
"Properties": Properties}
|
||||||
return None
|
return None
|
||||||
|
|
||||||
def list_sharing(self,
|
def list_sharing(self,
|
||||||
@@ -205,7 +207,8 @@ class Sharing(sharing.BaseSharing):
|
|||||||
Permissions: str = "r",
|
Permissions: str = "r",
|
||||||
EnabledByOwner: bool = False, EnabledByUser: bool = False,
|
EnabledByOwner: bool = False, EnabledByUser: bool = False,
|
||||||
HiddenByOwner: bool = True, HiddenByUser: bool = True,
|
HiddenByOwner: bool = True, HiddenByUser: bool = True,
|
||||||
Timestamp: int = 0) -> dict:
|
Timestamp: int = 0,
|
||||||
|
Properties: Union[dict, None] = None) -> dict:
|
||||||
""" create sharing """
|
""" create sharing """
|
||||||
row: dict
|
row: dict
|
||||||
|
|
||||||
@@ -263,16 +266,17 @@ class Sharing(sharing.BaseSharing):
|
|||||||
def update_sharing(self,
|
def update_sharing(self,
|
||||||
ShareType: str,
|
ShareType: str,
|
||||||
PathOrToken: str,
|
PathOrToken: str,
|
||||||
Owner: Union[str, None] = None,
|
OwnerOrUser: str,
|
||||||
User: Union[str, None] = None,
|
User: Union[str, None] = None,
|
||||||
PathMapped: Union[str, None] = None,
|
PathMapped: Union[str, None] = None,
|
||||||
Permissions: Union[str, None] = None,
|
Permissions: Union[str, None] = None,
|
||||||
EnabledByOwner: Union[bool, None] = None,
|
EnabledByOwner: Union[bool, None] = None,
|
||||||
HiddenByOwner: Union[bool, None] = None,
|
HiddenByOwner: Union[bool, None] = None,
|
||||||
Timestamp: int = 0) -> dict:
|
Timestamp: int = 0,
|
||||||
|
Properties: Union[dict, None] = None) -> dict:
|
||||||
""" update sharing """
|
""" update sharing """
|
||||||
if logger.isEnabledFor(logging.DEBUG):
|
if logger.isEnabledFor(logging.DEBUG):
|
||||||
logger.debug("TRACE/sharing/%s/update: PathOrToken=%r Owner=%r PathMapped=%r", ShareType, PathOrToken, Owner, PathMapped)
|
logger.debug("TRACE/sharing/%s/update: PathOrToken=%r OwnerOrUser=%r PathMapped=%r Properties=%r", ShareType, PathOrToken, OwnerOrUser, PathMapped, Properties)
|
||||||
|
|
||||||
# lookup token
|
# lookup token
|
||||||
found = False
|
found = False
|
||||||
@@ -293,12 +297,23 @@ class Sharing(sharing.BaseSharing):
|
|||||||
if found:
|
if found:
|
||||||
if logger.isEnabledFor(logging.DEBUG):
|
if logger.isEnabledFor(logging.DEBUG):
|
||||||
logger.debug("TRACE/sharing/%s/update: found index=%d", ShareType, index)
|
logger.debug("TRACE/sharing/%s/update: found index=%d", ShareType, index)
|
||||||
if Owner is not None and row['Owner'] != Owner:
|
|
||||||
return {"status": "permission-denied"}
|
if row['Owner'] != OwnerOrUser:
|
||||||
|
if logger.isEnabledFor(logging.DEBUG):
|
||||||
|
logger.debug("TRACE/sharing/%s/update: OwnerOrUser=%r not matching Owner=%r -> check now for matching User=%r", ShareType, OwnerOrUser, row['Owner'], row['User'])
|
||||||
|
if row['User'] == OwnerOrUser and PathMapped is None and Permissions is None and EnabledByOwner is None and HiddenByOwner is None and Properties is not None:
|
||||||
|
# user is only permitted to update Properties
|
||||||
|
if logger.isEnabledFor(logging.DEBUG):
|
||||||
|
logger.debug("TRACE/sharing/%s/update: OwnerOrUser=%r PathOrToken=%r index=%d is permitted to update Properties", ShareType, OwnerOrUser, PathOrToken, index)
|
||||||
|
pass
|
||||||
|
else:
|
||||||
|
return {"status": "permission-denied"}
|
||||||
|
|
||||||
if User is not None and row['User'] != User:
|
if User is not None and row['User'] != User:
|
||||||
return {"status": "permission-denied"}
|
return {"status": "permission-denied"}
|
||||||
|
|
||||||
if logger.isEnabledFor(logging.DEBUG):
|
if logger.isEnabledFor(logging.DEBUG):
|
||||||
logger.debug("TRACE/sharing/%s/update: Owner=%r PathOrToken=%r index=%d", ShareType, Owner, PathOrToken, index)
|
logger.debug("TRACE/sharing/%s/update: OwnerOrUser=%r PathOrToken=%r index=%d", ShareType, OwnerOrUser, PathOrToken, index)
|
||||||
logger.debug("TRACE/sharing/%s/update: orig row=%r", ShareType, row)
|
logger.debug("TRACE/sharing/%s/update: orig row=%r", ShareType, row)
|
||||||
|
|
||||||
# CSV: remove+adjust+readd
|
# CSV: remove+adjust+readd
|
||||||
@@ -312,6 +327,8 @@ class Sharing(sharing.BaseSharing):
|
|||||||
row["EnabledByOwner"] = EnabledByOwner
|
row["EnabledByOwner"] = EnabledByOwner
|
||||||
if HiddenByOwner is not None:
|
if HiddenByOwner is not None:
|
||||||
row["HiddenByOwner"] = HiddenByOwner
|
row["HiddenByOwner"] = HiddenByOwner
|
||||||
|
if Properties is not None:
|
||||||
|
row["Properties"] = Properties
|
||||||
# update timestamp
|
# update timestamp
|
||||||
row["TimestampUpdated"] = Timestamp
|
row["TimestampUpdated"] = Timestamp
|
||||||
|
|
||||||
@@ -322,7 +339,7 @@ class Sharing(sharing.BaseSharing):
|
|||||||
self._sharing_cache.pop(index)
|
self._sharing_cache.pop(index)
|
||||||
self._sharing_cache.append(row)
|
self._sharing_cache.append(row)
|
||||||
|
|
||||||
with self._storage.acquire_lock("w", Owner, path=self._sharing_db_file):
|
with self._storage.acquire_lock("w", OwnerOrUser, path=self._sharing_db_file):
|
||||||
if self._write_csv(self._sharing_db_file):
|
if self._write_csv(self._sharing_db_file):
|
||||||
if logger.isEnabledFor(logging.DEBUG):
|
if logger.isEnabledFor(logging.DEBUG):
|
||||||
logger.debug("TRACE/sharing/%s/update: write CSV done", ShareType)
|
logger.debug("TRACE/sharing/%s/update: write CSV done", ShareType)
|
||||||
@@ -484,7 +501,7 @@ class Sharing(sharing.BaseSharing):
|
|||||||
def _create_empty_csv(self, file: str) -> bool:
|
def _create_empty_csv(self, file: str) -> bool:
|
||||||
with self._storage.acquire_lock("w", None, path=file):
|
with self._storage.acquire_lock("w", None, path=file):
|
||||||
with open(file, 'w', newline='') as csvfile:
|
with open(file, 'w', newline='') as csvfile:
|
||||||
writer = csv.DictWriter(csvfile, fieldnames=sharing.DB_FIELDS_V1)
|
writer = csv.DictWriter(csvfile, fieldnames=sharing.DB_FIELDS_V1, delimiter=';')
|
||||||
writer.writeheader()
|
writer.writeheader()
|
||||||
return True
|
return True
|
||||||
|
|
||||||
@@ -492,7 +509,7 @@ class Sharing(sharing.BaseSharing):
|
|||||||
logger.debug("sharing database load begin: %r", file)
|
logger.debug("sharing database load begin: %r", file)
|
||||||
with self._storage.acquire_lock("r", None):
|
with self._storage.acquire_lock("r", None):
|
||||||
with open(file, 'r', newline='') as csvfile:
|
with open(file, 'r', newline='') as csvfile:
|
||||||
reader = csv.DictReader(csvfile, fieldnames=sharing.DB_FIELDS_V1)
|
reader = csv.DictReader(csvfile, fieldnames=sharing.DB_FIELDS_V1, delimiter=';')
|
||||||
self._lines = 0
|
self._lines = 0
|
||||||
for row in reader:
|
for row in reader:
|
||||||
# logger.debug("sharing database load read: %r", row)
|
# logger.debug("sharing database load read: %r", row)
|
||||||
@@ -506,7 +523,10 @@ class Sharing(sharing.BaseSharing):
|
|||||||
# convert txt to bool
|
# convert txt to bool
|
||||||
if self._lines > 0:
|
if self._lines > 0:
|
||||||
for fieldname in sharing.DB_FIELDS_V1_BOOL:
|
for fieldname in sharing.DB_FIELDS_V1_BOOL:
|
||||||
row[fieldname] = config._convert_to_bool(row[fieldname])
|
try:
|
||||||
|
row[fieldname] = config._convert_to_bool(row[fieldname])
|
||||||
|
except Exception as e:
|
||||||
|
logger.error("sharing database row error fieldname=%r row=%r error: %r", fieldname, row, e)
|
||||||
for fieldname in sharing.DB_FIELDS_V1_INT:
|
for fieldname in sharing.DB_FIELDS_V1_INT:
|
||||||
row[fieldname] = int(row[fieldname])
|
row[fieldname] = int(row[fieldname])
|
||||||
# check for duplicates
|
# check for duplicates
|
||||||
@@ -525,6 +545,6 @@ class Sharing(sharing.BaseSharing):
|
|||||||
|
|
||||||
def _write_csv(self, file: str) -> bool:
|
def _write_csv(self, file: str) -> bool:
|
||||||
with open(file, 'w', newline='') as csvfile:
|
with open(file, 'w', newline='') as csvfile:
|
||||||
writer = csv.DictWriter(csvfile, fieldnames=sharing.DB_FIELDS_V1)
|
writer = csv.DictWriter(csvfile, fieldnames=sharing.DB_FIELDS_V1, delimiter=';')
|
||||||
writer.writerows(self._sharing_cache)
|
writer.writerows(self._sharing_cache)
|
||||||
return True
|
return True
|
||||||
|
|||||||
@@ -122,8 +122,11 @@ class Sharing(sharing.BaseSharing):
|
|||||||
UserShare = row['User']
|
UserShare = row['User']
|
||||||
Permissions = row['Permissions']
|
Permissions = row['Permissions']
|
||||||
Hidden: bool = (row['HiddenByOwner'] or row['HiddenByUser'])
|
Hidden: bool = (row['HiddenByOwner'] or row['HiddenByUser'])
|
||||||
|
Properties: Union[dict, None] = None
|
||||||
|
if 'Properties' in row:
|
||||||
|
Properties = row['Properties']
|
||||||
if logger.isEnabledFor(logging.DEBUG):
|
if logger.isEnabledFor(logging.DEBUG):
|
||||||
logger.debug("TRACE/sharing: map %r to %r (Owner=%r User=%r Permissions=%r Hidden=%s)", PathOrToken, PathMapped, Owner, UserShare, Permissions, Hidden)
|
logger.debug("TRACE/sharing: map %r to %r (Owner=%r User=%r Permissions=%r Hidden=%s Properties=%r)", PathOrToken, PathMapped, Owner, UserShare, Permissions, Hidden, Properties)
|
||||||
return {
|
return {
|
||||||
"mapped": True,
|
"mapped": True,
|
||||||
"PathOrToken": PathOrToken,
|
"PathOrToken": PathOrToken,
|
||||||
@@ -131,7 +134,8 @@ class Sharing(sharing.BaseSharing):
|
|||||||
"Owner": Owner,
|
"Owner": Owner,
|
||||||
"User": UserShare,
|
"User": UserShare,
|
||||||
"Hidden": Hidden,
|
"Hidden": Hidden,
|
||||||
"Permissions": Permissions}
|
"Permissions": Permissions,
|
||||||
|
"Properties": Properties}
|
||||||
|
|
||||||
return None
|
return None
|
||||||
|
|
||||||
@@ -214,7 +218,8 @@ class Sharing(sharing.BaseSharing):
|
|||||||
Permissions: str = "r",
|
Permissions: str = "r",
|
||||||
EnabledByOwner: bool = False, EnabledByUser: bool = False,
|
EnabledByOwner: bool = False, EnabledByUser: bool = False,
|
||||||
HiddenByOwner: bool = True, HiddenByUser: bool = True,
|
HiddenByOwner: bool = True, HiddenByUser: bool = True,
|
||||||
Timestamp: int = 0) -> dict:
|
Timestamp: int = 0,
|
||||||
|
Properties: Union[dict, None] = None) -> dict:
|
||||||
""" create sharing """
|
""" create sharing """
|
||||||
row: dict
|
row: dict
|
||||||
|
|
||||||
@@ -258,16 +263,17 @@ class Sharing(sharing.BaseSharing):
|
|||||||
def update_sharing(self,
|
def update_sharing(self,
|
||||||
ShareType: str,
|
ShareType: str,
|
||||||
PathOrToken: str,
|
PathOrToken: str,
|
||||||
Owner: Union[str, None] = None,
|
OwnerOrUser: str,
|
||||||
User: Union[str, None] = None,
|
User: Union[str, None] = None,
|
||||||
PathMapped: Union[str, None] = None,
|
PathMapped: Union[str, None] = None,
|
||||||
Permissions: Union[str, None] = None,
|
Permissions: Union[str, None] = None,
|
||||||
EnabledByOwner: Union[bool, None] = None,
|
EnabledByOwner: Union[bool, None] = None,
|
||||||
HiddenByOwner: Union[bool, None] = None,
|
HiddenByOwner: Union[bool, None] = None,
|
||||||
Timestamp: int = 0) -> dict:
|
Timestamp: int = 0,
|
||||||
|
Properties: Union[dict, None] = None) -> dict:
|
||||||
""" update sharing """
|
""" update sharing """
|
||||||
if logger.isEnabledFor(logging.DEBUG):
|
if logger.isEnabledFor(logging.DEBUG):
|
||||||
logger.debug("TRACE/sharing/%s/update: PathOrToken=%r Owner=%r User=%r", ShareType, PathOrToken, Owner, User)
|
logger.debug("TRACE/sharing/%s/update: PathOrToken=%r OwnerOrUser=%r User=%r Properties=%r", ShareType, PathOrToken, OwnerOrUser, User, Properties)
|
||||||
|
|
||||||
sharing_config_file = os.path.join(self._sharing_db_path_ShareType[ShareType], self._encode_path(PathOrToken))
|
sharing_config_file = os.path.join(self._sharing_db_path_ShareType[ShareType], self._encode_path(PathOrToken))
|
||||||
|
|
||||||
@@ -275,7 +281,7 @@ class Sharing(sharing.BaseSharing):
|
|||||||
return {"status": "not-found"}
|
return {"status": "not-found"}
|
||||||
|
|
||||||
# read content
|
# read content
|
||||||
with self._storage.acquire_lock("w", Owner, path=sharing_config_file):
|
with self._storage.acquire_lock("w", OwnerOrUser, path=sharing_config_file):
|
||||||
# read file
|
# read file
|
||||||
with open(sharing_config_file, "rb") as fb:
|
with open(sharing_config_file, "rb") as fb:
|
||||||
(version, row) = pickle.load(fb)
|
(version, row) = pickle.load(fb)
|
||||||
@@ -286,8 +292,17 @@ class Sharing(sharing.BaseSharing):
|
|||||||
if logger.isEnabledFor(logging.DEBUG):
|
if logger.isEnabledFor(logging.DEBUG):
|
||||||
logger.debug("TRACE/sharing/%s/update: check: %r", ShareType, row)
|
logger.debug("TRACE/sharing/%s/update: check: %r", ShareType, row)
|
||||||
|
|
||||||
if Owner is not None and row['Owner'] != Owner:
|
if row['Owner'] != OwnerOrUser:
|
||||||
return {"status": "permission-denied"}
|
if logger.isEnabledFor(logging.DEBUG):
|
||||||
|
logger.debug("TRACE/sharing/%s/update: OwnerOrUser=%r not matching Owner=%r -> check now for matching User=%r", ShareType, OwnerOrUser, row['Owner'], row['User'])
|
||||||
|
if row['User'] == OwnerOrUser and PathMapped is None and Permissions is None and EnabledByOwner is None and HiddenByOwner is None and Properties is not None:
|
||||||
|
# user is only permitted to update Properties
|
||||||
|
if logger.isEnabledFor(logging.DEBUG):
|
||||||
|
logger.debug("TRACE/sharing/%s/update: OwnerOrUser=%r PathOrToken=%r is permitted to update Properties", ShareType, OwnerOrUser, PathOrToken)
|
||||||
|
pass
|
||||||
|
else:
|
||||||
|
return {"status": "permission-denied"}
|
||||||
|
|
||||||
if User is not None and row['User'] != User:
|
if User is not None and row['User'] != User:
|
||||||
return {"status": "permission-denied"}
|
return {"status": "permission-denied"}
|
||||||
|
|
||||||
@@ -304,6 +319,8 @@ class Sharing(sharing.BaseSharing):
|
|||||||
row["EnabledByOwner"] = EnabledByOwner
|
row["EnabledByOwner"] = EnabledByOwner
|
||||||
if HiddenByOwner is not None:
|
if HiddenByOwner is not None:
|
||||||
row["HiddenByOwner"] = HiddenByOwner
|
row["HiddenByOwner"] = HiddenByOwner
|
||||||
|
if Properties is not None:
|
||||||
|
row["Properties"] = Properties
|
||||||
# update timestamp
|
# update timestamp
|
||||||
row["TimestampUpdated"] = Timestamp
|
row["TimestampUpdated"] = Timestamp
|
||||||
|
|
||||||
|
|||||||
@@ -76,25 +76,56 @@ class TestSharingApiSanity(BaseTest):
|
|||||||
# disabled
|
# disabled
|
||||||
for path in ["/.sharing", "/.sharing/"]:
|
for path in ["/.sharing", "/.sharing/"]:
|
||||||
_, headers, _ = self.request("POST", path, check=404)
|
_, headers, _ = self.request("POST", path, check=404)
|
||||||
# enabled (permutations)
|
|
||||||
|
path = "/.sharing/"
|
||||||
|
|
||||||
|
# no database is active
|
||||||
|
logging.info("\n*** check API hook base: map=True token=False")
|
||||||
self.configure({"sharing": {
|
self.configure({"sharing": {
|
||||||
"collection_by_map": "True",
|
"collection_by_map": "True",
|
||||||
"collection_by_token": "False"}
|
"collection_by_token": "False"}
|
||||||
})
|
})
|
||||||
path = "/.sharing/"
|
_, headers, _ = self.request("POST", path, check=404)
|
||||||
_, headers, _ = self.request("POST", path, check=401)
|
|
||||||
|
logging.info("\n*** check API hook base: map=False token=True")
|
||||||
self.configure({"sharing": {
|
self.configure({"sharing": {
|
||||||
"collection_by_map": "False",
|
"collection_by_map": "False",
|
||||||
"collection_by_token": "True"}
|
"collection_by_token": "True"}
|
||||||
})
|
})
|
||||||
path = "/.sharing/"
|
_, headers, _ = self.request("POST", path, check=404)
|
||||||
_, headers, _ = self.request("POST", path, check=401)
|
|
||||||
|
logging.info("\n*** check API hook base: map=True token=True")
|
||||||
self.configure({"sharing": {
|
self.configure({"sharing": {
|
||||||
"collection_by_map": "True",
|
"collection_by_map": "True",
|
||||||
"collection_by_token": "True"}
|
"collection_by_token": "True"}
|
||||||
})
|
})
|
||||||
path = "/.sharing/"
|
_, headers, _ = self.request("POST", path, check=404)
|
||||||
_, headers, _ = self.request("POST", path, check=401)
|
|
||||||
|
for db_type in list(filter(lambda item: item != "none", sharing.INTERNAL_TYPES)):
|
||||||
|
logging.info("\n*** test: %s", db_type)
|
||||||
|
self.configure({"sharing": {"type": db_type}})
|
||||||
|
|
||||||
|
# no database is active
|
||||||
|
logging.info("\n*** check API hook base: map=True token=False")
|
||||||
|
self.configure({"sharing": {
|
||||||
|
"collection_by_map": "True",
|
||||||
|
"collection_by_token": "False"}
|
||||||
|
})
|
||||||
|
_, headers, _ = self.request("POST", path, check=401)
|
||||||
|
|
||||||
|
logging.info("\n*** check API hook base: map=False token=True")
|
||||||
|
self.configure({"sharing": {
|
||||||
|
"collection_by_map": "False",
|
||||||
|
"collection_by_token": "True"}
|
||||||
|
})
|
||||||
|
_, headers, _ = self.request("POST", path, check=401)
|
||||||
|
|
||||||
|
logging.info("\n*** check API hook base: map=True token=True")
|
||||||
|
self.configure({"sharing": {
|
||||||
|
"collection_by_map": "True",
|
||||||
|
"collection_by_token": "True"}
|
||||||
|
})
|
||||||
|
_, headers, _ = self.request("POST", path, check=401)
|
||||||
|
|
||||||
def test_sharing_api_base_with_auth(self) -> None:
|
def test_sharing_api_base_with_auth(self) -> None:
|
||||||
"""POST request at '/.sharing' with authentication."""
|
"""POST request at '/.sharing' with authentication."""
|
||||||
@@ -108,75 +139,82 @@ class TestSharingApiSanity(BaseTest):
|
|||||||
|
|
||||||
json_dict: dict
|
json_dict: dict
|
||||||
|
|
||||||
# path with no valid API hook
|
for db_type in list(filter(lambda item: item != "none", sharing.INTERNAL_TYPES)):
|
||||||
for path in ["/.sharing/", "/.sharing/v9/"]:
|
logging.info("\n*** test: %s", db_type)
|
||||||
_, headers, _ = self.request("POST", path, check=404, login="owner:ownerpw")
|
self.configure({"sharing": {"type": db_type}})
|
||||||
|
|
||||||
# path with valid API but no hook
|
for path in ["/.sharing/", "/.sharing/v9/"]:
|
||||||
for path in ["/.sharing/v1/"]:
|
logging.info("\n*** check invalid API URI: %r", path)
|
||||||
_, headers, _ = self.request("POST", path, check=404, login="owner:ownerpw")
|
_, headers, _ = self.request("POST", path, check=404, login="owner:ownerpw")
|
||||||
|
|
||||||
# path with valid API and hook but not enabled "map"
|
# path with valid API but no hook
|
||||||
self.configure({"sharing": {
|
for path in ["/.sharing/v1/"]:
|
||||||
"collection_by_map": "False",
|
logging.info("\n*** check valid API URI without hook: %r", path)
|
||||||
"collection_by_token": "True"}
|
_, headers, _ = self.request("POST", path, check=404, login="owner:ownerpw")
|
||||||
})
|
|
||||||
sharetype = "map"
|
|
||||||
for action in sharing.API_HOOKS_V1:
|
|
||||||
path = "/.sharing/v1/" + sharetype + "/" + action
|
|
||||||
_, headers, _ = self.request("POST", path, check=404, login="owner:ownerpw")
|
|
||||||
|
|
||||||
# path with valid API and hook but not enabled "token"
|
# path with valid API and hook but not enabled "map"
|
||||||
self.configure({"sharing": {
|
self.configure({"sharing": {
|
||||||
"collection_by_map": "True",
|
"collection_by_map": "False",
|
||||||
"collection_by_token": "False"}
|
"collection_by_token": "True"}
|
||||||
})
|
})
|
||||||
sharetype = "token"
|
sharetype = "map"
|
||||||
for action in sharing.API_HOOKS_V1:
|
for action in sharing.API_HOOKS_V1:
|
||||||
path = "/.sharing/v1/" + sharetype + "/" + action
|
path = "/.sharing/v1/" + sharetype + "/" + action
|
||||||
_, headers, _ = self.request("POST", path, check=404, login="owner:ownerpw")
|
logging.info("\n*** check valid API URI hook (but not enabled): %r", path)
|
||||||
|
_, headers, _ = self.request("POST", path, check=404, login="owner:ownerpw")
|
||||||
|
|
||||||
# check info hook
|
# path with valid API and hook but not enabled "token"
|
||||||
logging.info("\n*** check API hook: info/all")
|
self.configure({"sharing": {
|
||||||
json_dict = {}
|
"collection_by_map": "True",
|
||||||
_, headers, answer = self._sharing_api_json("all", "info", check=200, login="owner:ownerpw", json_dict=json_dict)
|
"collection_by_token": "False"}
|
||||||
answer_dict = json.loads(answer)
|
})
|
||||||
assert answer_dict['FeatureEnabledCollectionByMap'] is True
|
sharetype = "token"
|
||||||
assert answer_dict['FeatureEnabledCollectionByToken'] is False
|
for action in sharing.API_HOOKS_V1:
|
||||||
assert answer_dict['PermittedCreateCollectionByMap'] is True
|
path = "/.sharing/v1/" + sharetype + "/" + action
|
||||||
assert answer_dict['PermittedCreateCollectionByToken'] is True
|
logging.info("\n*** check valid API URI hook (but not enabled): %r", path)
|
||||||
|
_, headers, _ = self.request("POST", path, check=404, login="owner:ownerpw")
|
||||||
|
|
||||||
logging.info("\n*** check API hook: info/map")
|
# check info hook
|
||||||
json_dict = {}
|
logging.info("\n*** check API hook: info/all")
|
||||||
_, headers, answer = self._sharing_api_json("map", "info", check=200, login="owner:ownerpw", json_dict=json_dict)
|
json_dict = {}
|
||||||
answer_dict = json.loads(answer)
|
_, headers, answer = self._sharing_api_json("all", "info", check=200, login="owner:ownerpw", json_dict=json_dict)
|
||||||
assert answer_dict['FeatureEnabledCollectionByMap'] is True
|
answer_dict = json.loads(answer)
|
||||||
assert 'FeatureEnabledCollectionByToken' not in answer_dict
|
assert answer_dict['FeatureEnabledCollectionByMap'] is True
|
||||||
assert 'PermittedCreateCollectionByToken' not in answer_dict
|
assert answer_dict['FeatureEnabledCollectionByToken'] is False
|
||||||
|
assert answer_dict['PermittedCreateCollectionByMap'] is True
|
||||||
|
assert answer_dict['PermittedCreateCollectionByToken'] is True
|
||||||
|
|
||||||
logging.info("\n*** check API hook: info/token -> 404 (not enabled)")
|
logging.info("\n*** check API hook: info/map")
|
||||||
json_dict = {}
|
json_dict = {}
|
||||||
_, headers, answer = self._sharing_api_json("token", "info", check=404, login="owner:ownerpw", json_dict=json_dict)
|
_, headers, answer = self._sharing_api_json("map", "info", check=200, login="owner:ownerpw", json_dict=json_dict)
|
||||||
|
answer_dict = json.loads(answer)
|
||||||
|
assert answer_dict['FeatureEnabledCollectionByMap'] is True
|
||||||
|
assert 'FeatureEnabledCollectionByToken' not in answer_dict
|
||||||
|
assert 'PermittedCreateCollectionByToken' not in answer_dict
|
||||||
|
|
||||||
# path with valid API and hook and all enabled
|
logging.info("\n*** check API hook: info/token -> 404 (not enabled)")
|
||||||
self.configure({"sharing": {
|
json_dict = {}
|
||||||
"collection_by_map": "True",
|
_, headers, answer = self._sharing_api_json("token", "info", check=404, login="owner:ownerpw", json_dict=json_dict)
|
||||||
"collection_by_token": "True"}
|
|
||||||
})
|
|
||||||
for sharetype in sharing.SHARE_TYPES:
|
|
||||||
path = "/.sharing/v1/" + sharetype + "/" + action
|
|
||||||
# invalid API
|
|
||||||
_, headers, _ = self.request("POST", path + "NA", check=404, login="owner:ownerpw")
|
|
||||||
# valid API
|
|
||||||
_, headers, _ = self.request("POST", path, check=400, login="owner:ownerpw")
|
|
||||||
|
|
||||||
logging.info("\n*** check API hook: info/token -> 200")
|
# path with valid API and hook and all enabled
|
||||||
json_dict = {}
|
self.configure({"sharing": {
|
||||||
_, headers, answer = self._sharing_api_json("token", "info", check=200, login="owner:ownerpw", json_dict=json_dict)
|
"collection_by_map": "True",
|
||||||
answer_dict = json.loads(answer)
|
"collection_by_token": "True"}
|
||||||
assert answer_dict['FeatureEnabledCollectionByToken'] is True
|
})
|
||||||
assert 'FeatureEnabledCollectionByMap' not in answer_dict
|
for sharetype in sharing.SHARE_TYPES:
|
||||||
assert 'PermittedCreateCollectionByMap' not in answer_dict
|
path = "/.sharing/v1/" + sharetype + "/" + action
|
||||||
|
# invalid API
|
||||||
|
_, headers, _ = self.request("POST", path + "NA", check=404, login="owner:ownerpw")
|
||||||
|
# valid API
|
||||||
|
_, headers, _ = self.request("POST", path, check=400, login="owner:ownerpw")
|
||||||
|
|
||||||
|
logging.info("\n*** check API hook: info/token -> 200")
|
||||||
|
json_dict = {}
|
||||||
|
_, headers, answer = self._sharing_api_json("token", "info", check=200, login="owner:ownerpw", json_dict=json_dict)
|
||||||
|
answer_dict = json.loads(answer)
|
||||||
|
assert answer_dict['FeatureEnabledCollectionByToken'] is True
|
||||||
|
assert 'FeatureEnabledCollectionByMap' not in answer_dict
|
||||||
|
assert 'PermittedCreateCollectionByMap' not in answer_dict
|
||||||
|
|
||||||
def test_sharing_api_list_with_auth(self) -> None:
|
def test_sharing_api_list_with_auth(self) -> None:
|
||||||
"""POST/list with authentication."""
|
"""POST/list with authentication."""
|
||||||
@@ -196,9 +234,7 @@ class TestSharingApiSanity(BaseTest):
|
|||||||
form_array: Sequence[str]
|
form_array: Sequence[str]
|
||||||
json_dict: dict
|
json_dict: dict
|
||||||
|
|
||||||
for db_type in sharing.INTERNAL_TYPES:
|
for db_type in list(filter(lambda item: item != "none", sharing.INTERNAL_TYPES)):
|
||||||
if db_type == "none":
|
|
||||||
continue
|
|
||||||
logging.info("\n*** test: %s", db_type)
|
logging.info("\n*** test: %s", db_type)
|
||||||
self.configure({"sharing": {"type": db_type}})
|
self.configure({"sharing": {"type": db_type}})
|
||||||
|
|
||||||
@@ -286,9 +322,7 @@ class TestSharingApiSanity(BaseTest):
|
|||||||
form_array: Sequence[str]
|
form_array: Sequence[str]
|
||||||
json_dict: dict
|
json_dict: dict
|
||||||
|
|
||||||
for db_type in sharing.INTERNAL_TYPES:
|
for db_type in list(filter(lambda item: item != "none", sharing.INTERNAL_TYPES)):
|
||||||
if db_type == "none":
|
|
||||||
continue
|
|
||||||
logging.info("\n*** test: %s", db_type)
|
logging.info("\n*** test: %s", db_type)
|
||||||
self.configure({"sharing": {"type": db_type}})
|
self.configure({"sharing": {"type": db_type}})
|
||||||
|
|
||||||
@@ -361,7 +395,7 @@ class TestSharingApiSanity(BaseTest):
|
|||||||
_, headers, answer = self._sharing_api_form("token", "list", check=200, login="owner:ownerpw", form_array=form_array, accept="text/csv")
|
_, headers, answer = self._sharing_api_form("token", "list", check=200, login="owner:ownerpw", form_array=form_array, accept="text/csv")
|
||||||
assert "Status=success" not in answer
|
assert "Status=success" not in answer
|
||||||
assert "Lines=2" not in answer
|
assert "Lines=2" not in answer
|
||||||
assert ",".join(sharing.DB_FIELDS_V1) in answer
|
assert ";".join(sharing.DB_FIELDS_V1) in answer
|
||||||
assert "/owner/collection1/" in answer
|
assert "/owner/collection1/" in answer
|
||||||
assert "/owner/collection2/" in answer
|
assert "/owner/collection2/" in answer
|
||||||
|
|
||||||
@@ -408,7 +442,7 @@ class TestSharingApiSanity(BaseTest):
|
|||||||
_, headers, answer = self._sharing_api_form("token", "list", check=200, login="owner:ownerpw", form_array=form_array)
|
_, headers, answer = self._sharing_api_form("token", "list", check=200, login="owner:ownerpw", form_array=form_array)
|
||||||
assert "Status=success" in answer
|
assert "Status=success" in answer
|
||||||
assert "Lines=1" in answer
|
assert "Lines=1" in answer
|
||||||
assert "True,True,True,True" in answer
|
assert "True;True;True;True" in answer
|
||||||
|
|
||||||
logging.info("\n*** hide token#2 (form->text)")
|
logging.info("\n*** hide token#2 (form->text)")
|
||||||
form_array = []
|
form_array = []
|
||||||
@@ -422,7 +456,7 @@ class TestSharingApiSanity(BaseTest):
|
|||||||
_, headers, answer = self._sharing_api_form("token", "list", check=200, login="owner:ownerpw", form_array=form_array)
|
_, headers, answer = self._sharing_api_form("token", "list", check=200, login="owner:ownerpw", form_array=form_array)
|
||||||
assert "Status=success" in answer
|
assert "Status=success" in answer
|
||||||
assert "Lines=1" in answer
|
assert "Lines=1" in answer
|
||||||
assert "True,True,True,True" in answer
|
assert "True;True;True;True" in answer
|
||||||
|
|
||||||
logging.info("\n*** unhide token#2 (json->json)")
|
logging.info("\n*** unhide token#2 (json->json)")
|
||||||
json_dict = {}
|
json_dict = {}
|
||||||
@@ -484,9 +518,7 @@ class TestSharingApiSanity(BaseTest):
|
|||||||
path = path_base + "/event1.ics"
|
path = path_base + "/event1.ics"
|
||||||
self.put(path, event, login="owner:ownerpw")
|
self.put(path, event, login="owner:ownerpw")
|
||||||
|
|
||||||
for db_type in sharing.INTERNAL_TYPES:
|
for db_type in list(filter(lambda item: item != "none", sharing.INTERNAL_TYPES)):
|
||||||
if db_type == "none":
|
|
||||||
continue
|
|
||||||
logging.info("\n*** test: %s", db_type)
|
logging.info("\n*** test: %s", db_type)
|
||||||
self.configure({"sharing": {"type": db_type}})
|
self.configure({"sharing": {"type": db_type}})
|
||||||
|
|
||||||
@@ -592,9 +624,7 @@ class TestSharingApiSanity(BaseTest):
|
|||||||
|
|
||||||
json_dict: dict
|
json_dict: dict
|
||||||
|
|
||||||
for db_type in sharing.INTERNAL_TYPES:
|
for db_type in list(filter(lambda item: item != "none", sharing.INTERNAL_TYPES)):
|
||||||
if db_type == "none":
|
|
||||||
continue
|
|
||||||
logging.info("\n*** test: %s", db_type)
|
logging.info("\n*** test: %s", db_type)
|
||||||
self.configure({"sharing": {"type": db_type}})
|
self.configure({"sharing": {"type": db_type}})
|
||||||
|
|
||||||
@@ -645,9 +675,7 @@ class TestSharingApiSanity(BaseTest):
|
|||||||
event = get_file_content(file_item2)
|
event = get_file_content(file_item2)
|
||||||
self.put(path_mapped_item2, event, check=201, login="owner:ownerpw")
|
self.put(path_mapped_item2, event, check=201, login="owner:ownerpw")
|
||||||
|
|
||||||
for db_type in sharing.INTERNAL_TYPES:
|
for db_type in list(filter(lambda item: item != "none", sharing.INTERNAL_TYPES)):
|
||||||
if db_type == "none":
|
|
||||||
continue
|
|
||||||
logging.info("\n*** test: %s", db_type)
|
logging.info("\n*** test: %s", db_type)
|
||||||
self.configure({"sharing": {"type": db_type}})
|
self.configure({"sharing": {"type": db_type}})
|
||||||
|
|
||||||
@@ -834,9 +862,7 @@ class TestSharingApiSanity(BaseTest):
|
|||||||
path = path_mapped2 + "/event1.ics"
|
path = path_mapped2 + "/event1.ics"
|
||||||
self.put(path, event, login="%s:%s" % ("owner2", "owner2pw"))
|
self.put(path, event, login="%s:%s" % ("owner2", "owner2pw"))
|
||||||
|
|
||||||
for db_type in sharing.INTERNAL_TYPES:
|
for db_type in list(filter(lambda item: item != "none", sharing.INTERNAL_TYPES)):
|
||||||
if db_type == "none":
|
|
||||||
continue
|
|
||||||
logging.info("\n*** test: %s", db_type)
|
logging.info("\n*** test: %s", db_type)
|
||||||
self.configure({"sharing": {"type": db_type}})
|
self.configure({"sharing": {"type": db_type}})
|
||||||
|
|
||||||
@@ -923,9 +949,7 @@ class TestSharingApiSanity(BaseTest):
|
|||||||
path = path_mapped + "/event1.ics"
|
path = path_mapped + "/event1.ics"
|
||||||
self.put(path, event, login="owner:ownerpw")
|
self.put(path, event, login="owner:ownerpw")
|
||||||
|
|
||||||
for db_type in sharing.INTERNAL_TYPES:
|
for db_type in list(filter(lambda item: item != "none", sharing.INTERNAL_TYPES)):
|
||||||
if db_type == "none":
|
|
||||||
continue
|
|
||||||
logging.info("\n*** test: %s", db_type)
|
logging.info("\n*** test: %s", db_type)
|
||||||
self.configure({"sharing": {"type": db_type}})
|
self.configure({"sharing": {"type": db_type}})
|
||||||
|
|
||||||
@@ -1117,9 +1141,7 @@ class TestSharingApiSanity(BaseTest):
|
|||||||
event = get_file_content("event1.ics")
|
event = get_file_content("event1.ics")
|
||||||
self.put(path_mapped_item, event, login="owner:ownerpw")
|
self.put(path_mapped_item, event, login="owner:ownerpw")
|
||||||
|
|
||||||
for db_type in sharing.INTERNAL_TYPES:
|
for db_type in list(filter(lambda item: item != "none", sharing.INTERNAL_TYPES)):
|
||||||
if db_type == "none":
|
|
||||||
continue
|
|
||||||
logging.info("\n*** test: %s", db_type)
|
logging.info("\n*** test: %s", db_type)
|
||||||
self.configure({"sharing": {"type": db_type}})
|
self.configure({"sharing": {"type": db_type}})
|
||||||
|
|
||||||
@@ -1228,9 +1250,7 @@ class TestSharingApiSanity(BaseTest):
|
|||||||
event = get_file_content("event2.ics")
|
event = get_file_content("event2.ics")
|
||||||
self.put(path_user_item, event, login="user:userpw")
|
self.put(path_user_item, event, login="user:userpw")
|
||||||
|
|
||||||
for db_type in sharing.INTERNAL_TYPES:
|
for db_type in list(filter(lambda item: item != "none", sharing.INTERNAL_TYPES)):
|
||||||
if db_type == "none":
|
|
||||||
continue
|
|
||||||
logging.info("\n*** test: %s", db_type)
|
logging.info("\n*** test: %s", db_type)
|
||||||
self.configure({"sharing": {"type": db_type}})
|
self.configure({"sharing": {"type": db_type}})
|
||||||
|
|
||||||
@@ -1364,9 +1384,7 @@ class TestSharingApiSanity(BaseTest):
|
|||||||
path = os.path.join(path_mapped, "event1.ics")
|
path = os.path.join(path_mapped, "event1.ics")
|
||||||
self.put(path, event, login="owner:ownerpw")
|
self.put(path, event, login="owner:ownerpw")
|
||||||
|
|
||||||
for db_type in sharing.INTERNAL_TYPES:
|
for db_type in list(filter(lambda item: item != "none", sharing.INTERNAL_TYPES)):
|
||||||
if db_type == "none":
|
|
||||||
continue
|
|
||||||
logging.info("\n*** test: %s", db_type)
|
logging.info("\n*** test: %s", db_type)
|
||||||
self.configure({"sharing": {"type": db_type}})
|
self.configure({"sharing": {"type": db_type}})
|
||||||
|
|
||||||
@@ -1439,7 +1457,7 @@ class TestSharingApiSanity(BaseTest):
|
|||||||
element = prop.find(xmlutils.make_clark("D:href"))
|
element = prop.find(xmlutils.make_clark("D:href"))
|
||||||
assert element is not None and element.text == "/user/"
|
assert element is not None and element.text == "/user/"
|
||||||
|
|
||||||
def test_sharing_api_map_proppatch(self) -> None:
|
def test_sharing_api_map_proppatch_acl(self) -> None:
|
||||||
"""share-by-map API usage tests related to report."""
|
"""share-by-map API usage tests related to report."""
|
||||||
self.configure({"auth": {"type": "htpasswd",
|
self.configure({"auth": {"type": "htpasswd",
|
||||||
"htpasswd_filename": self.htpasswd_file_path,
|
"htpasswd_filename": self.htpasswd_file_path,
|
||||||
@@ -1468,9 +1486,7 @@ class TestSharingApiSanity(BaseTest):
|
|||||||
path = os.path.join(path_mapped, "event1.ics")
|
path = os.path.join(path_mapped, "event1.ics")
|
||||||
self.put(path, event, login="owner:ownerpw")
|
self.put(path, event, login="owner:ownerpw")
|
||||||
|
|
||||||
for db_type in sharing.INTERNAL_TYPES:
|
for db_type in list(filter(lambda item: item != "none", sharing.INTERNAL_TYPES)):
|
||||||
if db_type == "none":
|
|
||||||
continue
|
|
||||||
logging.info("\n*** test: %s", db_type)
|
logging.info("\n*** test: %s", db_type)
|
||||||
self.configure({"sharing": {"type": db_type}})
|
self.configure({"sharing": {"type": db_type}})
|
||||||
|
|
||||||
@@ -1647,9 +1663,7 @@ class TestSharingApiSanity(BaseTest):
|
|||||||
event = get_file_content("event3.ics")
|
event = get_file_content("event3.ics")
|
||||||
self.put(os.path.join(path_user, "event3.ics"), event, login="user:userpw")
|
self.put(os.path.join(path_user, "event3.ics"), event, login="user:userpw")
|
||||||
|
|
||||||
for db_type in sharing.INTERNAL_TYPES:
|
for db_type in list(filter(lambda item: item != "none", sharing.INTERNAL_TYPES)):
|
||||||
if db_type == "none":
|
|
||||||
continue
|
|
||||||
logging.info("\n*** test: %s", db_type)
|
logging.info("\n*** test: %s", db_type)
|
||||||
self.configure({"sharing": {"type": db_type}})
|
self.configure({"sharing": {"type": db_type}})
|
||||||
|
|
||||||
@@ -1837,9 +1851,7 @@ class TestSharingApiSanity(BaseTest):
|
|||||||
event = get_file_content("event1.ics")
|
event = get_file_content("event1.ics")
|
||||||
self.put(os.path.join(path_mapped1, "event1.ics"), event, login="owner:ownerpw")
|
self.put(os.path.join(path_mapped1, "event1.ics"), event, login="owner:ownerpw")
|
||||||
|
|
||||||
for db_type in sharing.INTERNAL_TYPES:
|
for db_type in list(filter(lambda item: item != "none", sharing.INTERNAL_TYPES)):
|
||||||
if db_type == "none":
|
|
||||||
continue
|
|
||||||
logging.info("\n*** test: %s", db_type)
|
logging.info("\n*** test: %s", db_type)
|
||||||
self.configure({"sharing": {"type": db_type}})
|
self.configure({"sharing": {"type": db_type}})
|
||||||
|
|
||||||
@@ -1959,9 +1971,7 @@ class TestSharingApiSanity(BaseTest):
|
|||||||
self.mkcalendar(path_user1, login="user1:user1pw")
|
self.mkcalendar(path_user1, login="user1:user1pw")
|
||||||
self.mkcalendar(path_user2, login="user2:user2pw")
|
self.mkcalendar(path_user2, login="user2:user2pw")
|
||||||
|
|
||||||
for db_type in sharing.INTERNAL_TYPES:
|
for db_type in list(filter(lambda item: item != "none", sharing.INTERNAL_TYPES)):
|
||||||
if db_type == "none":
|
|
||||||
continue
|
|
||||||
logging.info("\n*** test: %s", db_type)
|
logging.info("\n*** test: %s", db_type)
|
||||||
self.configure({"sharing": {"type": db_type}})
|
self.configure({"sharing": {"type": db_type}})
|
||||||
|
|
||||||
@@ -2107,9 +2117,7 @@ class TestSharingApiSanity(BaseTest):
|
|||||||
logging.info("\n*** mkcalendar user2 -> conflict")
|
logging.info("\n*** mkcalendar user2 -> conflict")
|
||||||
self.mkcalendar(path_user2, login="user2:user2pw", check=409)
|
self.mkcalendar(path_user2, login="user2:user2pw", check=409)
|
||||||
|
|
||||||
for db_type in sharing.INTERNAL_TYPES:
|
for db_type in list(filter(lambda item: item != "none", sharing.INTERNAL_TYPES)):
|
||||||
if db_type == "none":
|
|
||||||
continue
|
|
||||||
logging.info("\n*** test: %s", db_type)
|
logging.info("\n*** test: %s", db_type)
|
||||||
self.configure({"sharing": {"type": db_type}})
|
self.configure({"sharing": {"type": db_type}})
|
||||||
|
|
||||||
@@ -2179,9 +2187,7 @@ class TestSharingApiSanity(BaseTest):
|
|||||||
logging.info("\n*** prepare")
|
logging.info("\n*** prepare")
|
||||||
self.mkcalendar(path_owner1, login="owner1:owner1pw")
|
self.mkcalendar(path_owner1, login="owner1:owner1pw")
|
||||||
|
|
||||||
for db_type in sharing.INTERNAL_TYPES:
|
for db_type in list(filter(lambda item: item != "none", sharing.INTERNAL_TYPES)):
|
||||||
if db_type == "none":
|
|
||||||
continue
|
|
||||||
logging.info("\n*** test: %s", db_type)
|
logging.info("\n*** test: %s", db_type)
|
||||||
self.configure({"sharing": {"type": db_type}})
|
self.configure({"sharing": {"type": db_type}})
|
||||||
|
|
||||||
@@ -2272,9 +2278,7 @@ permissions: RrWw""")
|
|||||||
self.mkcalendar(path_owner1_M, login="owner1:owner1pw")
|
self.mkcalendar(path_owner1_M, login="owner1:owner1pw")
|
||||||
self.mkcalendar(path_owner1_m, login="owner1:owner1pw")
|
self.mkcalendar(path_owner1_m, login="owner1:owner1pw")
|
||||||
|
|
||||||
for db_type in sharing.INTERNAL_TYPES:
|
for db_type in list(filter(lambda item: item != "none", sharing.INTERNAL_TYPES)):
|
||||||
if db_type == "none":
|
|
||||||
continue
|
|
||||||
logging.info("\n*** test: %s", db_type)
|
logging.info("\n*** test: %s", db_type)
|
||||||
self.configure({"sharing": {"type": db_type}})
|
self.configure({"sharing": {"type": db_type}})
|
||||||
|
|
||||||
@@ -2394,9 +2398,7 @@ permissions: RrWw""")
|
|||||||
logging.info("\n*** prepare")
|
logging.info("\n*** prepare")
|
||||||
self.mkcalendar(path_owner1, login="owner1:owner1pw")
|
self.mkcalendar(path_owner1, login="owner1:owner1pw")
|
||||||
|
|
||||||
for db_type in sharing.INTERNAL_TYPES:
|
for db_type in list(filter(lambda item: item != "none", sharing.INTERNAL_TYPES)):
|
||||||
if db_type == "none":
|
|
||||||
continue
|
|
||||||
logging.info("\n*** test: %s", db_type)
|
logging.info("\n*** test: %s", db_type)
|
||||||
self.configure({"sharing": {"type": db_type}})
|
self.configure({"sharing": {"type": db_type}})
|
||||||
|
|
||||||
@@ -2492,3 +2494,199 @@ permissions: RrWw""")
|
|||||||
assert answer_dict['Status'] == "success"
|
assert answer_dict['Status'] == "success"
|
||||||
assert answer_dict['Lines'] == 1
|
assert answer_dict['Lines'] == 1
|
||||||
assert answer_dict['Content'][0]['Permissions'] == "RrWw"
|
assert answer_dict['Content'][0]['Permissions'] == "RrWw"
|
||||||
|
|
||||||
|
def test_sharing_api_map_propfind_overlay(self) -> None:
|
||||||
|
"""share-by-map API usage tests related to proppatch."""
|
||||||
|
self.configure({"auth": {"type": "htpasswd",
|
||||||
|
"htpasswd_filename": self.htpasswd_file_path,
|
||||||
|
"htpasswd_encryption": "plain"},
|
||||||
|
"sharing": {
|
||||||
|
"type": "csv",
|
||||||
|
"permit_create_map": True,
|
||||||
|
"permit_create_token": True,
|
||||||
|
"collection_by_map": "True",
|
||||||
|
"collection_by_token": "True"},
|
||||||
|
"logging": {"request_header_on_debug": "False",
|
||||||
|
"response_content_on_debug": "True",
|
||||||
|
"request_content_on_debug": "True"},
|
||||||
|
"rights": {"type": "owner_only"}})
|
||||||
|
|
||||||
|
form_array: Sequence[str]
|
||||||
|
json_dict: dict
|
||||||
|
|
||||||
|
path_mapped = "/owner/calendarPFO.ics/"
|
||||||
|
path_shared_r = "/user/calendarPFO-shared-by-owner-r.ics/"
|
||||||
|
|
||||||
|
logging.info("\n*** prepare and test access")
|
||||||
|
self.mkcalendar(path_mapped, login="owner:ownerpw")
|
||||||
|
|
||||||
|
for db_type in list(filter(lambda item: item != "none", sharing.INTERNAL_TYPES)):
|
||||||
|
logging.info("\n*** test: %s", db_type)
|
||||||
|
self.configure({"sharing": {"type": db_type}})
|
||||||
|
|
||||||
|
# check PROPFIND as owner
|
||||||
|
logging.info("\n*** PROPFIND collection owner -> ok")
|
||||||
|
_, responses = self.propfind(path_mapped, """\
|
||||||
|
<?xml version="1.0" encoding="utf-8"?>
|
||||||
|
<propfind xmlns="DAV:">
|
||||||
|
<prop>
|
||||||
|
<current-user-principal />
|
||||||
|
</prop>
|
||||||
|
</propfind>""", login="owner:ownerpw")
|
||||||
|
logging.info("response: %r", responses)
|
||||||
|
response = responses[path_mapped]
|
||||||
|
assert not isinstance(response, int) and len(response) == 1
|
||||||
|
status, prop = response["D:current-user-principal"]
|
||||||
|
assert status == 200 and len(prop) == 1
|
||||||
|
element = prop.find(xmlutils.make_clark("D:href"))
|
||||||
|
assert element is not None and element.text == "/owner/"
|
||||||
|
|
||||||
|
# execute PROPPATCH as owner
|
||||||
|
logging.info("\n*** PROPPATCH collection owner -> ok")
|
||||||
|
_, responses = self.proppatch(path_mapped, """\
|
||||||
|
<?xml version="1.0" encoding="utf-8"?>
|
||||||
|
<D:propertyupdate xmlns:D="DAV:">
|
||||||
|
<D:set>
|
||||||
|
<D:prop>
|
||||||
|
<I:calendar-color xmlns:I="http://apple.com/ns/ical/">#AAAAAA</I:calendar-color>
|
||||||
|
<C:calendar-description xmlns:C="urn:ietf:params:xml:ns:caldav">ICAL-OWNER</C:calendar-description>
|
||||||
|
</D:prop>
|
||||||
|
</D:set>
|
||||||
|
</D:propertyupdate>""", login="owner:ownerpw")
|
||||||
|
logging.info("response: %r", responses)
|
||||||
|
response = responses[path_mapped]
|
||||||
|
assert not isinstance(response, int) and len(response) == 2
|
||||||
|
status, prop = response["ICAL:calendar-color"]
|
||||||
|
assert status == 200 and not prop.text
|
||||||
|
status, prop = response["C:calendar-description"]
|
||||||
|
assert status == 200 and not prop.text
|
||||||
|
|
||||||
|
# verify PROPPATCH by owner
|
||||||
|
logging.info("\n*** PROPFIND collection owner -> ok")
|
||||||
|
propfind_calendar_color = get_file_content("propfind_multiple.xml")
|
||||||
|
_, responses = self.propfind(path_mapped, propfind_calendar_color, login="owner:ownerpw")
|
||||||
|
logging.info("response: %r", responses)
|
||||||
|
response = responses[path_mapped]
|
||||||
|
assert not isinstance(response, int)
|
||||||
|
status, prop = response["C:calendar-description"]
|
||||||
|
logging.debug("calendar-description: %r", prop.text)
|
||||||
|
assert status == 200 and prop.text == "ICAL-OWNER"
|
||||||
|
status, prop = response["ICAL:calendar-color"]
|
||||||
|
logging.debug("calendar-color: %r", prop.text)
|
||||||
|
assert status == 200 and prop.text == "#AAAAAA"
|
||||||
|
|
||||||
|
# create map
|
||||||
|
logging.info("\n*** create map user/owner:r -> ok")
|
||||||
|
json_dict = {}
|
||||||
|
json_dict['User'] = "user"
|
||||||
|
json_dict['PathMapped'] = path_mapped
|
||||||
|
json_dict['PathOrToken'] = path_shared_r
|
||||||
|
json_dict['Permissions'] = "r"
|
||||||
|
json_dict['Enabled'] = "True"
|
||||||
|
json_dict['Hidden'] = "False"
|
||||||
|
_, headers, answer = self._sharing_api_json("map", "create", check=200, login="owner:ownerpw", json_dict=json_dict)
|
||||||
|
answer_dict = json.loads(answer)
|
||||||
|
assert answer_dict['Status'] == "success"
|
||||||
|
|
||||||
|
# enable map by user
|
||||||
|
logging.info("\n*** enable map by user")
|
||||||
|
json_dict = {}
|
||||||
|
json_dict['User'] = "user"
|
||||||
|
json_dict['PathMapped'] = path_mapped
|
||||||
|
json_dict['PathOrToken'] = path_shared_r
|
||||||
|
_, headers, answer = self._sharing_api_json("map", "enable", check=200, login="user:userpw", json_dict=json_dict)
|
||||||
|
|
||||||
|
# verify PROPPATCH as user
|
||||||
|
logging.info("\n*** PROPFIND collection user -> ok")
|
||||||
|
propfind_calendar_color = get_file_content("propfind_multiple.xml")
|
||||||
|
_, responses = self.propfind(path_mapped, propfind_calendar_color, login="owner:ownerpw")
|
||||||
|
logging.info("response: %r", responses)
|
||||||
|
response = responses[path_mapped]
|
||||||
|
assert not isinstance(response, int)
|
||||||
|
status, prop = response["C:calendar-description"]
|
||||||
|
logging.debug("calendar-description: %r", prop.text)
|
||||||
|
assert status == 200 and prop.text == "ICAL-OWNER"
|
||||||
|
status, prop = response["ICAL:calendar-color"]
|
||||||
|
logging.debug("calendar-color: %r", prop.text)
|
||||||
|
assert status == 200 and prop.text == "#AAAAAA"
|
||||||
|
|
||||||
|
# update map by user
|
||||||
|
logging.info("\n*** update map by user (json)")
|
||||||
|
json_dict = {}
|
||||||
|
json_dict['User'] = "user"
|
||||||
|
json_dict['PathOrToken'] = path_shared_r
|
||||||
|
json_dict['Properties'] = {"C:calendar-description": "ICAL-USER", "ICAL:calendar-color": "#BBBBBB"}
|
||||||
|
_, headers, answer = self._sharing_api_json("map", "update", check=200, login="user:userpw", json_dict=json_dict)
|
||||||
|
|
||||||
|
logging.info("\n*** list (json->json)")
|
||||||
|
json_dict['PathOrToken'] = path_shared_r
|
||||||
|
_, headers, answer = self._sharing_api_json("map", "list", check=200, login="owner:ownerpw", json_dict=json_dict)
|
||||||
|
answer_dict = json.loads(answer)
|
||||||
|
assert answer_dict['Status'] == "success"
|
||||||
|
assert answer_dict['Lines'] == 1
|
||||||
|
|
||||||
|
logging.info("\n*** list (json->csv)")
|
||||||
|
json_dict['PathOrToken'] = path_shared_r
|
||||||
|
_, headers, answer = self._sharing_api_json("map", "list", check=200, login="owner:ownerpw", json_dict=json_dict, accept="text/csv")
|
||||||
|
|
||||||
|
logging.info("\n*** list (json->txt)")
|
||||||
|
json_dict['PathOrToken'] = path_shared_r
|
||||||
|
_, headers, answer = self._sharing_api_json("map", "list", check=200, login="owner:ownerpw", json_dict=json_dict, accept="text/plain")
|
||||||
|
|
||||||
|
# verify overlay as user
|
||||||
|
logging.info("\n*** PROPFIND collection user (overlay) -> ok")
|
||||||
|
propfind_calendar_color = get_file_content("propfind_multiple.xml")
|
||||||
|
_, responses = self.propfind(path_shared_r, propfind_calendar_color, login="user:userpw")
|
||||||
|
logging.info("response: %r", responses)
|
||||||
|
response = responses[path_shared_r]
|
||||||
|
assert not isinstance(response, int)
|
||||||
|
status, prop = response["C:calendar-description"]
|
||||||
|
logging.debug("calendar-description: %r", prop.text)
|
||||||
|
assert status == 200 and prop.text == "ICAL-USER"
|
||||||
|
status, prop = response["ICAL:calendar-color"]
|
||||||
|
logging.debug("calendar-color: %r", prop.text)
|
||||||
|
assert status == 200 and prop.text == "#BBBBBB"
|
||||||
|
|
||||||
|
# verify overlay not visible by owner
|
||||||
|
logging.info("\n*** PROPFIND collection owner -> ok")
|
||||||
|
propfind_calendar_color = get_file_content("propfind_multiple.xml")
|
||||||
|
_, responses = self.propfind(path_mapped, propfind_calendar_color, login="owner:ownerpw")
|
||||||
|
logging.info("response: %r", responses)
|
||||||
|
response = responses[path_mapped]
|
||||||
|
assert not isinstance(response, int)
|
||||||
|
status, prop = response["C:calendar-description"]
|
||||||
|
logging.debug("calendar-description: %r", prop.text)
|
||||||
|
assert status == 200 and prop.text == "ICAL-OWNER"
|
||||||
|
status, prop = response["ICAL:calendar-color"]
|
||||||
|
logging.debug("calendar-color: %r", prop.text)
|
||||||
|
assert status == 200 and prop.text == "#AAAAAA"
|
||||||
|
|
||||||
|
# update map by user
|
||||||
|
logging.info("\n*** update map by user (form)")
|
||||||
|
form_array = ["User=" + "user"]
|
||||||
|
form_array.append("PathOrToken=" + path_shared_r)
|
||||||
|
form_array.append("Properties='C:calendar-description'='ICAL-USER-NEW'")
|
||||||
|
form_array.append("Properties='ICAL:calendar-color'='#CCCCCC'")
|
||||||
|
_, headers, answer = self._sharing_api_form("map", "update", check=200, login="user:userpw", form_array=form_array)
|
||||||
|
assert "Status=success" in answer
|
||||||
|
|
||||||
|
# verify overlay as user
|
||||||
|
logging.info("\n*** PROPFIND collection user (overlay) -> ok")
|
||||||
|
propfind_calendar_color = get_file_content("propfind_multiple.xml")
|
||||||
|
_, responses = self.propfind(path_shared_r, propfind_calendar_color, login="user:userpw")
|
||||||
|
logging.info("response: %r", responses)
|
||||||
|
response = responses[path_shared_r]
|
||||||
|
assert not isinstance(response, int)
|
||||||
|
status, prop = response["C:calendar-description"]
|
||||||
|
logging.debug("calendar-description: %r", prop.text)
|
||||||
|
assert status == 200 and prop.text == "ICAL-USER-NEW"
|
||||||
|
status, prop = response["ICAL:calendar-color"]
|
||||||
|
logging.debug("calendar-color: %r", prop.text)
|
||||||
|
assert status == 200 and prop.text == "#CCCCCC"
|
||||||
|
|
||||||
|
# update map by user
|
||||||
|
logging.info("\n*** update map by user (form)")
|
||||||
|
form_array = ["User=" + "user"]
|
||||||
|
form_array.append("PathOrToken=" + path_shared_r)
|
||||||
|
form_array.append("Properties=BUGGYENTRY=BUGGYVALUE")
|
||||||
|
_, headers, answer = self._sharing_api_form("map", "update", check=400, login="user:userpw", form_array=form_array)
|
||||||
|
|||||||
@@ -385,7 +385,7 @@ def limit_str(content: str, limit: int) -> str:
|
|||||||
return content
|
return content
|
||||||
|
|
||||||
|
|
||||||
def textwrap_str(content: str, limit: int = 2000) -> str:
|
def textwrap_str(content: str, limit: int = 3000) -> str:
|
||||||
# TODO: add support for config option and prefix
|
# TODO: add support for config option and prefix
|
||||||
return textwrap.indent(limit_str(content, limit), " ", lambda line: True)
|
return textwrap.indent(limit_str(content, limit), " ", lambda line: True)
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user