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 | |||||
| swhid | |||||
| 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", | |||||
| } | } | ||||