Changeset View
Changeset View
Standalone View
Standalone View
swh/graphql/tests/functional/test_directory_entry.py
Show All 24 Lines | def test_get_directory_entry_missing_path(client): | ||||
directory = get_directories()[0] | directory = get_directories()[0] | ||||
path = "missing" | path = "missing" | ||||
query_str = """ | query_str = """ | ||||
query getDirEntry($swhid: SWHID!, $path: String!) { | query getDirEntry($swhid: SWHID!, $path: String!) { | ||||
directoryEntry(directorySWHID: $swhid, path: $path) { | directoryEntry(directorySWHID: $swhid, path: $path) { | ||||
name { | name { | ||||
text | text | ||||
} | } | ||||
targetType | |||||
target { | target { | ||||
type | |||||
identifier | |||||
node { | |||||
...on Content { | ...on Content { | ||||
swhid | swhid | ||||
} | } | ||||
} | } | ||||
} | } | ||||
} | } | ||||
} | |||||
""" | """ | ||||
utils.assert_missing_object( | utils.assert_missing_object( | ||||
client, | client, | ||||
query_str, | query_str, | ||||
"directoryEntry", | "directoryEntry", | ||||
swhid=str(directory.swhid()), | swhid=str(directory.swhid()), | ||||
path=path, | path=path, | ||||
) | ) | ||||
@pytest.mark.parametrize( | @pytest.mark.parametrize( | ||||
"directory", get_directories() + get_directories_with_nested_path() | "directory", get_directories() + get_directories_with_nested_path() | ||||
) | ) | ||||
def test_get_directory_entry(client, directory): | def test_get_directory_entry(client, directory): | ||||
storage = server.get_storage() | storage = server.get_storage() | ||||
query_str = """ | query_str = """ | ||||
query getDirEntry($swhid: SWHID!, $path: String!) { | query getDirEntry($swhid: SWHID!, $path: String!) { | ||||
directoryEntry(directorySWHID: $swhid, path: $path) { | directoryEntry(directorySWHID: $swhid, path: $path) { | ||||
name { | name { | ||||
text | text | ||||
} | } | ||||
targetType | |||||
target { | target { | ||||
type | |||||
node { | |||||
...on Content { | ...on Content { | ||||
swhid | swhid | ||||
} | } | ||||
...on Directory { | ...on Directory { | ||||
swhid | swhid | ||||
} | } | ||||
...on Revision { | ...on Revision { | ||||
swhid | swhid | ||||
} | } | ||||
} | } | ||||
} | } | ||||
} | } | ||||
} | |||||
""" | """ | ||||
for entry in storage.directory_ls(directory.id, recursive=True): | for entry in storage.directory_ls(directory.id, recursive=True): | ||||
data, _ = utils.get_query_response( | data, _ = utils.get_query_response( | ||||
client, | client, | ||||
query_str, | query_str, | ||||
swhid=str(directory.swhid()), | swhid=str(directory.swhid()), | ||||
path=entry["name"].decode(), | path=entry["name"].decode(), | ||||
) | ) | ||||
swhid = None | swhid = None | ||||
if entry["type"] == "file" and entry["sha1_git"] is not None: | if entry["type"] == "file" and entry["sha1_git"] is not None: | ||||
swhid = CoreSWHID( | swhid = CoreSWHID( | ||||
object_type=ObjectType.CONTENT, object_id=entry["sha1_git"] | object_type=ObjectType.CONTENT, object_id=entry["sha1_git"] | ||||
) | ) | ||||
elif entry["type"] == "dir" and entry["target"] is not None: | elif entry["type"] == "dir" and entry["target"] is not None: | ||||
swhid = CoreSWHID( | swhid = CoreSWHID( | ||||
object_type=ObjectType.DIRECTORY, object_id=entry["target"] | object_type=ObjectType.DIRECTORY, object_id=entry["target"] | ||||
) | ) | ||||
elif entry["type"] == "rev" and entry["target"] is not None: | elif entry["type"] == "rev" and entry["target"] is not None: | ||||
swhid = CoreSWHID( | swhid = CoreSWHID( | ||||
object_type=ObjectType.REVISION, object_id=entry["target"] | object_type=ObjectType.REVISION, object_id=entry["target"] | ||||
) | ) | ||||
assert data["directoryEntry"] == { | assert data["directoryEntry"] == { | ||||
"name": {"text": entry["name"].decode()}, | "name": {"text": entry["name"].decode()}, | ||||
"target": {"swhid": str(swhid)} if swhid else None, | "target": { | ||||
"targetType": get_target_type(entry["type"]), | "type": get_target_type(entry["type"]), | ||||
"node": {"swhid": str(swhid)}, | |||||
} | |||||
if swhid | |||||
else {"type": get_target_type(entry["type"]), "node": None}, | |||||
} | } | ||||
@pytest.mark.parametrize("directory", get_directories()) | @pytest.mark.parametrize("directory", get_directories()) | ||||
def test_get_directory_entry_connection(client, directory): | def test_get_directory_entry_connection(client, directory): | ||||
query_str = """ | query_str = """ | ||||
query getDirectory($swhid: SWHID!) { | query getDirectory($swhid: SWHID!) { | ||||
directory(swhid: $swhid) { | directory(swhid: $swhid) { | ||||
swhid | swhid | ||||
entries { | entries { | ||||
nodes { | nodes { | ||||
targetType | |||||
name { | name { | ||||
text | text | ||||
} | } | ||||
target { | |||||
type | |||||
} | |||||
} | } | ||||
} | } | ||||
} | } | ||||
} | } | ||||
""" | """ | ||||
data, _ = utils.get_query_response(client, query_str, swhid=str(directory.swhid())) | data, _ = utils.get_query_response(client, query_str, swhid=str(directory.swhid())) | ||||
directory_entries = data["directory"]["entries"]["nodes"] | directory_entries = data["directory"]["entries"]["nodes"] | ||||
assert len(directory_entries) == len(directory.entries) | assert len(directory_entries) == len(directory.entries) | ||||
output = [ | output = [ | ||||
{"name": {"text": de.name.decode()}, "targetType": get_target_type(de.type)} | { | ||||
"name": {"text": de.name.decode()}, | |||||
"target": {"type": get_target_type(de.type)}, | |||||
} | |||||
for de in directory.entries | for de in directory.entries | ||||
] | ] | ||||
for each_entry in output: | for each_entry in output: | ||||
assert each_entry in directory_entries | assert each_entry in directory_entries | ||||
@pytest.mark.parametrize("directory", get_directories()) | @pytest.mark.parametrize("directory", get_directories()) | ||||
def test_directory_entry_connection_filter_by_name(client, directory): | def test_directory_entry_connection_filter_by_name(client, directory): | ||||
storage = server.get_storage() | storage = server.get_storage() | ||||
for dir_entry in storage.directory_ls(directory.id): | for dir_entry in storage.directory_ls(directory.id): | ||||
name_include = dir_entry["name"][:-1].decode() | name_include = dir_entry["name"][:-1].decode() | ||||
query_str = """ | query_str = """ | ||||
query getDirectory($swhid: SWHID!, $nameInclude: String) { | query getDirectory($swhid: SWHID!, $nameInclude: String) { | ||||
directory(swhid: $swhid) { | directory(swhid: $swhid) { | ||||
swhid | swhid | ||||
entries(nameInclude: $nameInclude) { | entries(nameInclude: $nameInclude) { | ||||
nodes { | nodes { | ||||
targetType | |||||
name { | name { | ||||
text | text | ||||
} | } | ||||
target { | |||||
type | |||||
} | |||||
} | } | ||||
} | } | ||||
} | } | ||||
} | } | ||||
""" | """ | ||||
data, _ = utils.get_query_response( | data, _ = utils.get_query_response( | ||||
client, | client, | ||||
query_str, | query_str, | ||||
swhid=str(directory.swhid()), | swhid=str(directory.swhid()), | ||||
nameInclude=name_include, | nameInclude=name_include, | ||||
) | ) | ||||
for entry in data["directory"]["entries"]["nodes"]: | for entry in data["directory"]["entries"]["nodes"]: | ||||
assert name_include in entry["name"]["text"] | assert name_include in entry["name"]["text"] | ||||
assert entry["targetType"] == get_target_type(dir_entry["type"]) | assert entry["target"]["type"] == get_target_type(dir_entry["type"]) | ||||
def test_directory_entry_connection_filter_by_name_special_chars(client): | def test_directory_entry_connection_filter_by_name_special_chars(client): | ||||
directory = get_directories_with_special_name_entries()[0] | directory = get_directories_with_special_name_entries()[0] | ||||
query_str = """ | query_str = """ | ||||
query getDirectory($swhid: SWHID!, $nameInclude: String) { | query getDirectory($swhid: SWHID!, $nameInclude: String) { | ||||
directory(swhid: $swhid) { | directory(swhid: $swhid) { | ||||
entries(nameInclude: $nameInclude) { | entries(nameInclude: $nameInclude) { | ||||
nodes { | nodes { | ||||
targetType | |||||
name { | name { | ||||
text | text | ||||
} | } | ||||
} | } | ||||
} | } | ||||
} | } | ||||
} | } | ||||
""" | """ | ||||
data, _ = utils.get_query_response( | data, _ = utils.get_query_response( | ||||
client, | client, | ||||
query_str, | query_str, | ||||
swhid=str(directory.swhid()), | swhid=str(directory.swhid()), | ||||
nameInclude="ssSSé", | nameInclude="ssSSé", | ||||
) | ) | ||||
assert data["directory"]["entries"]["nodes"][0] == { | assert data["directory"]["entries"]["nodes"][0] == { | ||||
"name": {"text": "ßßétEÉt"}, | "name": {"text": "ßßétEÉt"}, | ||||
"targetType": "content", | |||||
} | } |