Page MenuHomeSoftware Heritage
Differential D2988 Diff 10855 swh/storage/tests/test_storage.py

Changeset View

Standalone View

View Options

swh/storage/tests/test_storage.py

Show First 20 Lines • Show All 3,164 Lines • ▼ Show 20 Lines def test_object_find_by_sha1_git(self, swh_storage):
expected[data.release["id"]] = [ expected[data.release["id"]] = [
{"sha1_git": data.release["id"], "type": "release",} {"sha1_git": data.release["id"], "type": "release",}
] ]
ret = swh_storage.object_find_by_sha1_git(sha1_gits) ret = swh_storage.object_find_by_sha1_git(sha1_gits)
assert expected == ret assert expected == ret
def test_tool_add(self, swh_storage): def test_metadata_fetcher_add_get(self, swh_storage):
tool = { actual_fetcher = swh_storage.metadata_fetcher_get(
"name": "some-unknown-tool", data.metadata_fetcher["name"], data.metadata_fetcher["version"]
"version": "some-version", )
"configuration": {"debian-package": "some-package"}, assert actual_fetcher is None # does not exist
}
actual_tool = swh_storage.tool_get(tool)
assert actual_tool is None # does not exist
# add it
actual_tools = swh_storage.tool_add([tool])
assert len(actual_tools) == 1
actual_tool = actual_tools[0]
assert actual_tool is not None # now it exists
new_id = actual_tool.pop("id")
assert actual_tool == tool
actual_tools2 = swh_storage.tool_add([tool])
actual_tool2 = actual_tools2[0]
assert actual_tool2 is not None # now it exists
new_id2 = actual_tool2.pop("id")
assert new_id == new_id2
assert actual_tool == actual_tool2
def test_tool_add_multiple(self, swh_storage):
tool = {
"name": "some-unknown-tool",
"version": "some-version",
"configuration": {"debian-package": "some-package"},
}
actual_tools = list(swh_storage.tool_add([tool]))
assert len(actual_tools) == 1
new_tools = [
tool,
{"name": "yet-another-tool", "version": "version", "configuration": {},},
]
actual_tools = swh_storage.tool_add(new_tools)
assert len(actual_tools) == 2
# order not guaranteed, so we iterate over results to check
for tool in actual_tools:
_id = tool.pop("id")
assert _id is not None
assert tool in new_tools
def test_tool_get_missing(self, swh_storage):
tool = {
"name": "unknown-tool",
"version": "3.1.0rc2-31-ga2cbb8c",
"configuration": {"command_line": "nomossa <filepath>"},
}
actual_tool = swh_storage.tool_get(tool)
assert actual_tool is None
def test_tool_metadata_get_missing_context(self, swh_storage):
tool = {
"name": "swh-metadata-translator",
"version": "0.0.1",
"configuration": {"context": "unknown-context"},
}
actual_tool = swh_storage.tool_get(tool)
assert actual_tool is None
def test_tool_metadata_get(self, swh_storage): swh_storage.metadata_fetcher_add(**data.metadata_fetcher)
tool = {
"name": "swh-metadata-translator",
"version": "0.0.1",
"configuration": {"type": "local", "context": "npm"},
}
expected_tool = swh_storage.tool_add([tool])[0]
# when res = swh_storage.metadata_fetcher_get(
actual_tool = swh_storage.tool_get(tool) data.metadata_fetcher["name"], data.metadata_fetcher["version"]
)
# then assert res is not data.metadata_fetcher
assert expected_tool == actual_tool assert res == data.metadata_fetcher
def test_metadata_provider_get(self, swh_storage): def test_metadata_authority_add_get(self, swh_storage):
# given actual_authority = swh_storage.metadata_authority_get(
no_provider = swh_storage.metadata_provider_get(6459456445615) data.metadata_authority["type"], data.metadata_authority["url"]
assert no_provider is None
# when
provider_id = swh_storage.metadata_provider_add(
data.provider["name"],
data.provider["type"],
data.provider["url"],
data.provider["metadata"],
) )
assert actual_authority is None # does not exist
actual_provider = swh_storage.metadata_provider_get(provider_id) swh_storage.metadata_authority_add(**data.metadata_authority)
expected_provider = {
"provider_name": data.provider["name"],
"provider_url": data.provider["url"],
}
# then
del actual_provider["id"]
assert actual_provider, expected_provider
def test_metadata_provider_get_by(self, swh_storage): res = swh_storage.metadata_authority_get(
# given data.metadata_authority["type"], data.metadata_authority["url"]
no_provider = swh_storage.metadata_provider_get_by(
{
"provider_name": data.provider["name"],
"provider_url": data.provider["url"],
}
)
assert no_provider is None
# when
provider_id = swh_storage.metadata_provider_add(
data.provider["name"],
data.provider["type"],
data.provider["url"],
data.provider["metadata"],
) )
actual_provider = swh_storage.metadata_provider_get_by( assert res is not data.metadata_authority
{ assert res == data.metadata_authority
"provider_name": data.provider["name"],
"provider_url": data.provider["url"],
}
)
# then
assert provider_id, actual_provider["id"]
def test_origin_metadata_add(self, swh_storage): def test_origin_metadata_add(self, swh_storage):
# given
origin = data.origin origin = data.origin
fetcher = data.metadata_fetcher
authority = data.metadata_authority
swh_storage.origin_add([origin])[0] swh_storage.origin_add([origin])[0]
tools = swh_storage.tool_add([data.metadata_tool]) swh_storage.metadata_fetcher_add(**fetcher)
tool = tools[0] swh_storage.metadata_authority_add(**authority)
swh_storage.metadata_provider_add( swh_storage.origin_metadata_add(**data.origin_metadata)
data.provider["name"], swh_storage.origin_metadata_add(**data.origin_metadata2)
data.provider["type"],
data.provider["url"],
data.provider["metadata"],
)
provider = swh_storage.metadata_provider_get_by(
{
"provider_name": data.provider["name"],
"provider_url": data.provider["url"],
}
)
# when adding for the same origin 2 metadatas swh_storage.origin_metadata_get(origin["url"], authority)
n_om = len(list(swh_storage.origin_metadata_get_by(origin["url"])))
swh_storage.origin_metadata_add( assert [data.origin_metadata, data.origin_metadata2] == list(
origin["url"], sorted(
data.origin_metadata["discovery_date"], swh_storage.origin_metadata_get(origin["url"], authority),
provider["id"], key=lambda x: x["discovery_date"],
tool["id"], )
data.origin_metadata["metadata"],
)
swh_storage.origin_metadata_add(
origin["url"],
"2015-01-01 23:00:00+00",
provider["id"],
tool["id"],
data.origin_metadata2["metadata"],
) )
n_actual_om = len(list(swh_storage.origin_metadata_get_by(origin["url"])))
# then
assert n_actual_om == n_om + 2
def test_origin_metadata_get(self, swh_storage): def test_origin_metadata_get(self, swh_storage):
# given authority = data.metadata_authority
origin_url = data.origin["url"] fetcher = data.metadata_fetcher
authority2 = data.metadata_authority2
fetcher2 = data.metadata_fetcher2
origin_url1 = data.origin["url"]
origin_url2 = data.origin2["url"] origin_url2 = data.origin2["url"]
swh_storage.origin_add([data.origin]) swh_storage.origin_add([data.origin])
swh_storage.origin_add([data.origin2]) swh_storage.origin_add([data.origin2])
swh_storage.metadata_provider_add( origin1_metadata1 = data.origin_metadata
data.provider["name"], origin1_metadata2 = data.origin_metadata2
data.provider["type"], origin1_metadata3 = data.origin_metadata3
data.provider["url"], origin2_metadata = {**data.origin_metadata2, "origin_url": origin_url2}
data.provider["metadata"],
) swh_storage.metadata_authority_add(**authority)
provider = swh_storage.metadata_provider_get_by( swh_storage.metadata_fetcher_add(**fetcher)
{ swh_storage.metadata_authority_add(**authority2)
"provider_name": data.provider["name"], swh_storage.metadata_fetcher_add(**fetcher2)
"provider_url": data.provider["url"],
} swh_storage.origin_metadata_add(**origin1_metadata1)
) swh_storage.origin_metadata_add(**origin1_metadata2)
tool = swh_storage.tool_add([data.metadata_tool])[0] swh_storage.origin_metadata_add(**origin1_metadata3)
# when adding for the same origin 2 metadatas swh_storage.origin_metadata_add(**origin2_metadata)
swh_storage.origin_metadata_add(
origin_url, assert [origin1_metadata1, origin1_metadata2] == list(
data.origin_metadata["discovery_date"],
provider["id"],
tool["id"],
data.origin_metadata["metadata"],
)
swh_storage.origin_metadata_add(
origin_url2,
data.origin_metadata2["discovery_date"],
provider["id"],
tool["id"],
data.origin_metadata2["metadata"],
)
swh_storage.origin_metadata_add(
origin_url,
data.origin_metadata2["discovery_date"],
provider["id"],
tool["id"],
data.origin_metadata2["metadata"],
)
all_metadatas = list(
sorted( sorted(
swh_storage.origin_metadata_get_by(origin_url), swh_storage.origin_metadata_get(origin_url1, authority),
key=lambda x: x["discovery_date"], key=lambda x: x["discovery_date"],
) )
) )
metadatas_for_origin2 = list(swh_storage.origin_metadata_get_by(origin_url2))
expected_results = [
{
"origin_url": origin_url,
"discovery_date": datetime.datetime(
2015, 1, 1, 23, 0, tzinfo=datetime.timezone.utc
),
"metadata": {"name": "test_origin_metadata", "version": "0.0.1"},
"provider_id": provider["id"],
"provider_name": "hal",
"provider_type": "deposit-client",
"provider_url": "http:///hal/inria",
"tool_id": tool["id"],
},
{
"origin_url": origin_url,
"discovery_date": datetime.datetime(
2017, 1, 1, 23, 0, tzinfo=datetime.timezone.utc
),
"metadata": {"name": "test_origin_metadata", "version": "0.0.1"},
"provider_id": provider["id"],
"provider_name": "hal",
"provider_type": "deposit-client",
"provider_url": "http:///hal/inria",
"tool_id": tool["id"],
},
]
# then
assert len(all_metadatas) == 2
assert len(metadatas_for_origin2) == 1
assert all_metadatas == expected_results
def test_metadata_provider_add(self, swh_storage):
provider = {
"provider_name": "swMATH",
"provider_type": "registry",
"provider_url": "http://www.swmath.org/",
"metadata": {
"email": "contact@swmath.org",
"license": "All rights reserved",
},
}
provider["id"] = provider_id = swh_storage.metadata_provider_add(**provider)
assert provider == swh_storage.metadata_provider_get_by(
{"provider_name": "swMATH", "provider_url": "http://www.swmath.org/"}
)
assert provider == swh_storage.metadata_provider_get(provider_id)
def test_origin_metadata_get_by_provider_type(self, swh_storage): assert [origin1_metadata3] == list(
# given sorted(
origin_url = data.origin["url"] swh_storage.origin_metadata_get(origin_url1, authority2),
origin_url2 = data.origin2["url"] key=lambda x: x["discovery_date"],
swh_storage.origin_add([data.origin])
swh_storage.origin_add([data.origin2])
provider1_id = swh_storage.metadata_provider_add(
data.provider["name"],
data.provider["type"],
data.provider["url"],
data.provider["metadata"],
) )
provider1 = swh_storage.metadata_provider_get_by(
{
"provider_name": data.provider["name"],
"provider_url": data.provider["url"],
}
) )
assert provider1 == swh_storage.metadata_provider_get(provider1_id)
provider2_id = swh_storage.metadata_provider_add( assert [origin2_metadata] == list(
"swMATH", swh_storage.origin_metadata_get(origin_url2, authority)
"registry",
"http://www.swmath.org/",
{"email": "contact@swmath.org", "license": "All rights reserved"},
)
provider2 = swh_storage.metadata_provider_get_by(
{"provider_name": "swMATH", "provider_url": "http://www.swmath.org/"}
) )
assert provider2 == swh_storage.metadata_provider_get(provider2_id)
# using the only tool now inserted in the data.sql, but for this
# provider should be a crawler tool (not yet implemented)
tool = swh_storage.tool_add([data.metadata_tool])[0]
# when adding for the same origin 2 metadatas
swh_storage.origin_metadata_add(
origin_url,
data.origin_metadata["discovery_date"],
provider1["id"],
tool["id"],
data.origin_metadata["metadata"],
)
swh_storage.origin_metadata_add(
origin_url2,
data.origin_metadata2["discovery_date"],
provider2["id"],
tool["id"],
data.origin_metadata2["metadata"],
)
provider_type = "registry"
m_by_provider = list(
swh_storage.origin_metadata_get_by(origin_url2, provider_type)
)
for item in m_by_provider:
if "id" in item:
del item["id"]
expected_results = [
{
"origin_url": origin_url2,
"discovery_date": datetime.datetime(
2017, 1, 1, 23, 0, tzinfo=datetime.timezone.utc
),
"metadata": {"name": "test_origin_metadata", "version": "0.0.1"},
"provider_id": provider2["id"],
"provider_name": "swMATH",
"provider_type": provider_type,
"provider_url": "http://www.swmath.org/",
"tool_id": tool["id"],
}
]
# then
assert len(m_by_provider) == 1
assert m_by_provider == expected_results
class TestStorageGeneratedData: class TestStorageGeneratedData:
def test_generate_content_get(self, swh_storage, swh_contents): def test_generate_content_get(self, swh_storage, swh_contents):
contents_with_data = [c for c in swh_contents if c["status"] != "absent"] contents_with_data = [c for c in swh_contents if c["status"] != "absent"]
# input the list of sha1s we want from storage # input the list of sha1s we want from storage
get_sha1s = [c["sha1"] for c in contents_with_data] get_sha1s = [c["sha1"] for c in contents_with_data]
▲ Show 20 Lines • Show All 518 Lines • Show Last 20 Lines
About · Developer information · Legal