Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 14 additions & 2 deletions keepercommander/commands/nested_share_folder/folder_commands.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@
command_error_handler, check_result,
check_folder_edit_permission, check_folder_share_permission, check_folder_delete_permission,
classify_share_recipient,
ensure_nested_share_folder,
ensure_nested_share_folder, is_nested_share_folder,
)
from .parsers import (
nested_share_folder_mkdir_parser,
Expand Down Expand Up @@ -69,7 +69,7 @@ def execute(self, params, **kwargs):

for idx, segment in enumerate(segments):
is_leaf = (idx == last_idx)
existing_uid = self._find_existing_child(params, segment, parent_uid)
existing_uid = self._resolve_segment(params, segment, parent_uid)
if existing_uid:
if is_leaf:
logging.warning('nsf-mkdir: Folder "%s" already exists', segment)
Expand Down Expand Up @@ -134,6 +134,18 @@ def _cache_new_folder(params, folder_uid, name, parent_uid, folder_key=None):
entry['folder_key_unencrypted'] = folder_key
nsf[folder_uid] = entry

@staticmethod
def _resolve_segment(params, segment, parent_uid):
"""Resolve *segment* to an existing NSF folder UID.

UID segments are resolved globally, matching legacy ``mkdir`` path
behavior. Name segments are matched as children of *parent_uid*.
"""
if is_nested_share_folder(params, segment):
return segment
return NestedShareFolderMkdirCommand._find_existing_child(
params, segment, parent_uid)

@staticmethod
def _find_existing_child(params, folder_name, parent_uid):
"""Find an existing NSF folder named *folder_name* whose parent matches
Expand Down
3 changes: 2 additions & 1 deletion keepercommander/commands/nested_share_folder/parsers.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,8 @@ def _make_parser(prog, description):
'nsf-mkdir', 'Create a new Nested Share Folder using v3 API')
nested_share_folder_mkdir_parser.add_argument(
'folder', type=str,
help='Folder name or path (e.g. "Parent/Child/Grand") to create. '
help='Folder name or path (e.g. "Parent/Child" or "ParentUID/Child"). '
'Parent segments may be an existing folder name or UID. '
'Intermediate folders are created automatically. '
'Use "//" to embed a literal "/" in a segment name.')
nested_share_folder_mkdir_parser.add_argument(
Expand Down
135 changes: 72 additions & 63 deletions keepercommander/commands/record.py
Original file line number Diff line number Diff line change
Expand Up @@ -290,6 +290,66 @@ class RecordGetUidCommand(Command):
def get_parser(self):
return get_info_parser

@staticmethod
def _build_folder_json(params, f):
folder_type = 'nested_share_folder' if f.type == BaseFolderNode.NestedShareFolderType else 'classic_folder'
parent_uid = f.parent_uid or None
fo = {
'folder_uid': f.uid,
'type': folder_type,
'name': f.name,
'path': get_folder_path(params, f.uid),
'parent_uid': parent_uid,
'folder': {
'uid': parent_uid,
'path': get_folder_path(params, parent_uid) if parent_uid else '/'
}
}
if isinstance(f, subfolder.SharedFolderFolderNode):
fo['shared_folder_uid'] = f.shared_folder_uid
if f.type == BaseFolderNode.UserFolderType:
record_uids = params.subfolder_record_cache.get(f.uid, set())
records_list = []
for r_uid in record_uids:
entry = {'record_uid': r_uid}
rec = vault.KeeperRecord.load(params, r_uid)
if rec:
entry['record_name'] = rec.title
records_list.append(entry)
fo['records'] = records_list
elif f.type == BaseFolderNode.NestedShareFolderType:
from .nested_share_folder.helpers import collect_records_in_folder
record_uids = collect_records_in_folder(params, f.uid, recursive=False)
records_list = []
for r_uid in record_uids:
entry = {'record_uid': r_uid}
rec = vault.KeeperRecord.load(params, r_uid)
if rec:
entry['record_name'] = rec.title
records_list.append(entry)
fo['records'] = records_list
return fo

@staticmethod
def _print_folder_detail(params, f):
fo = RecordGetUidCommand._build_folder_json(params, f)
uid_label = 'Nested Share Folder UID' if f.type == BaseFolderNode.NestedShareFolderType else 'Folder UID'
print('')
print('{0:>25s}: {1:<20s}'.format(uid_label, f.uid))
print('{0:>25s}: {1:<20s}'.format('Folder Type', f.get_folder_type()))
print('{0:>25s}: {1}'.format('Name', f.name))
print('{0:>25s}: {1}'.format('Path', fo['path']))
if f.parent_uid:
print('{0:>25s}: {1:<20s}'.format('Parent Folder UID', f.parent_uid))
if isinstance(f, subfolder.SharedFolderFolderNode):
print('{0:>25s}: {1:<20s}'.format('Shared Folder UID', f.shared_folder_uid))
records = fo.get('records') or []
if records:
print('{0:>25s}:'.format('Records'))
for entry in records:
name = entry.get('record_name') or ''
print(' {0:>23s} {1}'.format(entry['record_uid'], name))

def execute(self, params, **kwargs):
uid = kwargs.get('uid')
if not uid:
Expand Down Expand Up @@ -372,45 +432,9 @@ def _format_expiration(expiration_value):
if uid in params.folder_cache:
f = params.folder_cache[uid]
if fmt == 'json':
folder_type = 'nested_share_folder' if f.type == BaseFolderNode.NestedShareFolderType else 'classic_folder'
parent_uid = f.parent_uid or None
fo = {
'folder_uid': f.uid,
'type': folder_type,
'name': f.name,
'path': get_folder_path(params, f.uid),
'parent_uid': parent_uid,
'folder': {
'uid': parent_uid,
'path': get_folder_path(params, parent_uid) if parent_uid else '/'
}
}
if isinstance(f, subfolder.SharedFolderFolderNode):
fo['shared_folder_uid'] = f.shared_folder_uid
if f.type == BaseFolderNode.UserFolderType:
record_uids = params.subfolder_record_cache.get(f.uid, set())
records_list = []
for r_uid in record_uids:
entry = {'record_uid': r_uid}
rec = vault.KeeperRecord.load(params, r_uid)
if rec:
entry['record_name'] = rec.title
records_list.append(entry)
fo['records'] = records_list
elif f.type == BaseFolderNode.NestedShareFolderType:
from .nested_share_folder.helpers import collect_records_in_folder
record_uids = collect_records_in_folder(params, f.uid, recursive=False)
records_list = []
for r_uid in record_uids:
entry = {'record_uid': r_uid}
rec = vault.KeeperRecord.load(params, r_uid)
if rec:
entry['record_name'] = rec.title
records_list.append(entry)
fo['records'] = records_list
print(json.dumps(fo, indent=2))
print(json.dumps(self._build_folder_json(params, f), indent=2))
else:
f.display()
self._print_folder_detail(params, f)
if f.type == BaseFolderNode.NestedShareFolderType:
try:
from .nested_share_folder.display_commands import NestedShareGetCommand
Expand Down Expand Up @@ -850,33 +874,18 @@ def _format_expiration(expiration_value):
include_dag=kwargs.get('include_dag')
)
elif len(matched_folders) == 1:
uid = matched_folders[0].uid
f = params.folder_cache[uid]
f = matched_folders[0]
sf_uid = f.uid if isinstance(f, subfolder.SharedFolderNode) else \
(f.shared_folder_uid if isinstance(f, subfolder.SharedFolderFolderNode) else None)
if sf_uid and api.is_shared_folder(params, sf_uid):
return self.execute(
params,
uid=sf_uid,
format=fmt,
unmask=kwargs.get('unmask'),
legacy=kwargs.get('legacy'),
include_dag=kwargs.get('include_dag')
)
if fmt == 'json':
fo = {
'folder_uid': f.uid,
'type': f.type,
'name': f.name
}
if sf_uid:
fo['shared_folder_uid'] = sf_uid
if f.parent_uid:
fo['parent_folder_uid'] = f.parent_uid
print(json.dumps(fo, indent=2))
else:
f.display()
return
resolve_uid = sf_uid if (sf_uid and api.is_shared_folder(params, sf_uid)) else f.uid
return self.execute(
params,
uid=resolve_uid,
format=fmt,
unmask=kwargs.get('unmask'),
legacy=kwargs.get('legacy'),
include_dag=kwargs.get('include_dag')
)
elif len(matched_teams) == 1:
team_uid = matched_teams[0].get('team_uid')
if api.is_team(params, team_uid):
Expand Down
19 changes: 19 additions & 0 deletions unit-tests/test_command_record.py
Original file line number Diff line number Diff line change
Expand Up @@ -304,6 +304,25 @@ def test_get_shared_folder_uid(self):
cmd.execute(params, uid=shared_folder_uid)
cmd.execute(params, format='json', uid=shared_folder_uid)

def test_get_user_folder_json_consistent_by_name_and_uid(self):
params = get_synced_params()
cmd = record.RecordGetUidCommand()
user_folder = next(
x for x in params.folder_cache.values()
if x.type == 'user_folder')
captured = []
with mock.patch('builtins.print', side_effect=captured.append):
cmd.execute(params, uid=user_folder.uid, format='json')
by_uid = json.loads(captured[-1])
cmd.execute(params, uid=user_folder.name, format='json')
by_name = json.loads(captured[-1])
self.assertEqual(by_name, by_uid)
self.assertEqual(by_uid['type'], 'classic_folder')
self.assertIn('path', by_uid)
self.assertIn('parent_uid', by_uid)
self.assertIn('folder', by_uid)
self.assertIn('records', by_uid)

def test_get_team_uid(self):
params = get_synced_params()
cmd = record.RecordGetUidCommand()
Expand Down
84 changes: 83 additions & 1 deletion unit-tests/test_nested_share_folder.py
Original file line number Diff line number Diff line change
Expand Up @@ -231,7 +231,7 @@ def setUp(self):
def tearDown(self):
mock.patch.stopall()

@patch('keepercommander.nested_share_folder.folder_api.create_folder_v3')
@patch('keepercommander.commands.nested_share_folder.folder_commands._nsf.create_folder_v3')
def test_mkdir(self, mock_create):
from keepercommander.commands.nested_share_folder import NestedShareFolderMkdirCommand
mock_create.return_value = {
Expand All @@ -243,6 +243,88 @@ def test_mkdir(self, mock_create):
cmd.execute(_make_params(), folder='NewFolder')
mock_create.assert_called_once()

@patch('keepercommander.commands.nested_share_folder.folder_commands._nsf.create_folder_v3')
def test_mkdir_resolves_parent_uid_in_path(self, mock_create):
from keepercommander.commands.nested_share_folder import NestedShareFolderMkdirCommand
parent_uid = 'tY6D-RanxY252zzBY_xU4A'
child_uid = utils.generate_uid()
mock_create.return_value = {
'folder_uid': child_uid, 'status': 'SUCCESS',
'message': '', 'success': True,
}
parent_fuid, parent_fobj = _make_folder(
folder_uid=parent_uid, name='Real Parent')
cmd = NestedShareFolderMkdirCommand()
with mock.patch('builtins.print'):
cmd.execute(
_make_params(nested_share_folders={parent_uid: parent_fobj}),
folder=f'{parent_uid}/My Child Folder',
)
mock_create.assert_called_once_with(
params=mock.ANY,
folder_name='My Child Folder',
parent_uid=parent_uid,
color=None,
inherit_permissions=True,
)

@patch('keepercommander.commands.nested_share_folder.folder_commands._nsf.create_folder_v3')
def test_mkdir_resolves_parent_name_in_path(self, mock_create):
from keepercommander.commands.nested_share_folder import NestedShareFolderMkdirCommand
parent_fuid, parent_fobj = _make_folder(name='Engineering')
child_uid = utils.generate_uid()
mock_create.return_value = {
'folder_uid': child_uid, 'status': 'SUCCESS',
'message': '', 'success': True,
}
cmd = NestedShareFolderMkdirCommand()
with mock.patch('builtins.print'):
cmd.execute(
_make_params(nested_share_folders={parent_fuid: parent_fobj}),
folder='Engineering/New Folder KD 1 June',
)
mock_create.assert_called_once_with(
params=mock.ANY,
folder_name='New Folder KD 1 June',
parent_uid=parent_fuid,
color=None,
inherit_permissions=True,
)

@patch('keepercommander.commands.nested_share_folder.folder_commands._nsf.create_folder_v3')
def test_mkdir_creates_intermediate_name_segments(self, mock_create):
from keepercommander.commands.nested_share_folder import NestedShareFolderMkdirCommand
eng_uid = utils.generate_uid()
child_uid = utils.generate_uid()
mock_create.side_effect = [
{
'folder_uid': eng_uid, 'status': 'SUCCESS',
'message': '', 'success': True,
},
{
'folder_uid': child_uid, 'status': 'SUCCESS',
'message': '', 'success': True,
},
]
cmd = NestedShareFolderMkdirCommand()
with mock.patch('builtins.print'):
cmd.execute(_make_params(), folder='Engineering/New Folder KD 1 June')
self.assertEqual(mock_create.call_count, 2)
mock_create.assert_any_call(
params=mock.ANY,
folder_name='Engineering',
parent_uid=None,
color=None,
inherit_permissions=True,
)
mock_create.assert_any_call(
params=mock.ANY,
folder_name='New Folder KD 1 June',
parent_uid=eng_uid,
color=None,
inherit_permissions=True,
)

@patch('keepercommander.nested_share_folder.folder_api.update_folder_v3')
def test_update_folder(self, mock_update):
from keepercommander.commands.nested_share_folder import NestedShareFolderUpdateCommand
Expand Down
Loading