Page MenuHomeSoftware Heritage
Differential D4931 Diff 17723 swh/clearlydefined/tests/test_mapping.py

Changeset View

Standalone View

View Options

swh/clearlydefined/tests/test_mapping.py

# Copyright (C) 2017-2021 The Software Heritage developers # Copyright (C) 2017-2021 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution # See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version # License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information # See top-level LICENSE file for more information
from swh.clearlydefined.mapping_utils import map_sha1_with_swhid from swh.clearlydefined.mapping_utils import map_sha1_with_swhid
import psycopg2 from swh.clearlydefined.mapping_utils import map_row
from swh.clearlydefined.mapping_utils import map_harvest
from swh.clearlydefined.mapping_utils import map_definition
import gzip
from swh.model import from_disk
from swh.model.model import MetadataTargetType
from swh.model.model import Origin
from swh.model.model import Content
from swh.model.model import Revision
from swh.model.model import Person
from swh.model.model import TimestampWithTimezone
from swh.model.model import Timestamp
from swh.model.model import RevisionType
vlorentzUnsubmitted
Done
Inline Actions

merge these imports

vlorentz: merge these imports
from swh.model.model import Directory
from swh.model.model import DirectoryEntry
from swh.model.hashutil import hash_to_bytes
from swh.storage import get_storage
def add_data(dsn: str):
write_connection = psycopg2.connect(dsn=dsn) path = "swh/clearlydefined/tests/data/"
cur = write_connection.cursor()
vlorentzUnsubmitted
Done
Inline Actions

is this still used?

vlorentz: is this still used?
def make_row(id, text):
row = []
row.append(id)
vlorentzUnsubmitted
Done
Inline Actions

missing type and documentation

vlorentz: missing type and documentation
row.append(gzip.compress(text.encode()))
row = tuple(row)
return row
def file_data(file_name):
with open(file_name) as file:
data = file.read()
return data
def add_content_data(swh_storage_backend_config):
storage = get_storage(**swh_storage_backend_config)
data = [ data = [
{ Content(
"sha1": "\\xa6628c8a4fbc08c29b8472e2222975e5b9918131", data=b"42\n",
"sha1_git": "\\xe103b11cbfecbc116dacbb1f9ab2a02176092a32", length=717,
"sha256": ( sha1=hash_to_bytes("a6628c8a4fbc08c29b8472e2222975e5b9918131"),
"\\x6ac599151a7aaa8ca5d38dc5bb61b49193a3cadc1ed33de5a57e4d1ecc53c846" sha1_git=hash_to_bytes("e103b11cbfecbc116dacbb1f9ab2a02176092a32"),
sha256=hash_to_bytes(
"6ac599151a7aaa8ca5d38dc5bb61b49193a3cadc1ed33de5a57e4d1ecc53c846"
), ),
"blake2s256": ( blake2s256=hash_to_bytes(
"\\xc509b320abede3580bb1de75a0efa09ba7416db9c8de845a4e1b4317c0b8a8d9" "c509b320abede3580bb1de75a0efa09ba7416db9c8de845a4e1b4317c0b8a8d9"
), ),
"length": 717, status="visible",
},
{
"sha1": "\\xd1ece3dbe3e78a6648f37206f996e202acb3926b",
"sha1_git": "\\x095b80e14c3ea6254f57e94761f2313e32b1d58a",
"sha256": (
"\\x2a24791d738e4380d55e1c990dd0bd2bcdc98240d9a25c488804abfd814b8c96"
), ),
"blake2s256": ( Content(
"\\xa18463fe94e1b4ee191815c2fa48948c44ffbdffcda49fad6591ac4c93f19aef" data=b"4242\n",
length=3138,
sha1=hash_to_bytes("d1ece3dbe3e78a6648f37206f996e202acb3926b"),
sha1_git=hash_to_bytes("095b80e14c3ea6254f57e94761f2313e32b1d58a"),
sha256=hash_to_bytes(
"859f0b154fdb2d630f45e1ecae4a862915435e663248bb8461d914696fc047cd"
),
blake2s256=hash_to_bytes(
"849c20fad132b7c2d62c15de310adfe87be94a379941bed295e8141c6219810d"
),
status="visible",
), ),
"length": 3138,
},
] ]
vlorentzUnsubmitted
Done
Inline Actions

simpler way to do this:

data = [
    Content.from_data(b"42\n"),
    Content.from_data(b"4242\n"),
]

and make it a global variable so you can access it in the other functions.

vlorentz: simpler way to do this: ``` data = [ Content.from_data(b"42\n"), Content.from_data…
for row in data: storage.content_add(data)
cur.execute(
"INSERT INTO content (sha1, sha1_git, sha256, blake2s256, length) VALUES \
(%s,%s,%s,%s,%s);", def add_revision_data(swh_storage_backend_config):
( storage = get_storage(**swh_storage_backend_config)
row["sha1"], content = Content(
row["sha1_git"], data=b"42\n",
row["sha256"], length=717,
row["blake2s256"], sha1=hash_to_bytes("a6628c8a4fbc08c29b8472e2222975e5b9918131"),
row["length"], sha1_git=hash_to_bytes("e103b11cbfecbc116dacbb1f9ab2a02176092a32"),
sha256=hash_to_bytes(
"6ac599151a7aaa8ca5d38dc5bb61b49193a3cadc1ed33de5a57e4d1ecc53c846"
),
blake2s256=hash_to_bytes(
"c509b320abede3580bb1de75a0efa09ba7416db9c8de845a4e1b4317c0b8a8d9"
), ),
status="visible",
vlorentzUnsubmitted
Done
Inline Actions

duplicate

vlorentz: duplicate
) )
write_connection.commit() directory = Directory(
id=hash_to_bytes("5256e856a0a0898966d6ba14feb4388b8b82d302"),
entries=tuple(
[
DirectoryEntry(
name=b"foo",
type="file",
target=content.sha1_git,
perms=from_disk.DentryPerms.content,
),
],
),
)
data = [
Revision(
id=hash_to_bytes("4c66129b968ab8122964823d1d77677f50884cf6"),
message=b"hello",
author=Person(
name=b"Nicolas Dandrimont",
email=b"nicolas@example.com",
fullname=b"Nicolas Dandrimont <nicolas@example.com> ",
),
date=TimestampWithTimezone(
timestamp=Timestamp(seconds=1234567890, microseconds=0),
offset=120,
negative_utc=False,
),
committer=Person(
name=b"St\xc3fano Zacchiroli",
email=b"stefano@example.com",
fullname=b"St\xc3fano Zacchiroli <stefano@example.com>",
),
committer_date=TimestampWithTimezone(
timestamp=Timestamp(seconds=1123456789, microseconds=0),
offset=120,
negative_utc=False,
),
parents=(),
type=RevisionType.GIT,
directory=directory.id,
metadata={
"checksums": {
"sha1": "tarball-sha1",
"sha256": "tarball-sha256",
},
"signed-off-by": "some-dude",
},
extra_headers=(
(b"gpgsig", b"test123"),
(b"mergetag", b"foo\\bar"),
(b"mergetag", b"\x22\xaf\x89\x80\x01\x00"),
),
synthetic=True,
),
Revision(
id=hash_to_bytes("3c66129b968ab8122964823d1d77677f50884cf6"),
message=b"hello again",
author=Person(
name=b"Roberto Dicosmo",
email=b"roberto@example.com",
fullname=b"Roberto Dicosmo <roberto@example.com>",
),
date=TimestampWithTimezone(
timestamp=Timestamp(
seconds=1234567843,
microseconds=220000,
),
offset=-720,
negative_utc=False,
),
committer=Person(
name=b"tony",
email=b"ar@dumont.fr",
fullname=b"tony <ar@dumont.fr>",
),
committer_date=TimestampWithTimezone(
timestamp=Timestamp(
seconds=1123456789,
microseconds=220000,
),
offset=0,
negative_utc=False,
),
parents=(),
type=RevisionType.GIT,
directory=directory.id,
metadata=None,
extra_headers=(),
synthetic=False,
),
vlorentzUnsubmitted
Done
Inline Actions

also make these globals (the function is way too long otherwise)

vlorentz: also make these globals (the function is way too long otherwise)
]
storage.revision_add(data)
def test_mapping(swh_storage_backend_config): def test_mapping_sha1_with_swhID(swh_storage_backend_config):
dsn = swh_storage_backend_config["db"] dsn = swh_storage_backend_config["db"]
add_data(dsn=dsn) add_content_data(swh_storage_backend_config)
sha1 = "a6628c8a4fbc08c29b8472e2222975e5b9918131" sha1 = "a6628c8a4fbc08c29b8472e2222975e5b9918131"
assert "swh:1:cnt:e103b11cbfecbc116dacbb1f9ab2a02176092a32" == map_sha1_with_swhid( assert "swh:1:cnt:e103b11cbfecbc116dacbb1f9ab2a02176092a32" == map_sha1_with_swhid(
sha1=sha1, dsn=dsn sha1=sha1, dsn=dsn
) )
def test_mapping_with_empty_sha1(swh_storage_backend_config): def test_mapping_with_empty_sha1(swh_storage_backend_config):
dsn = swh_storage_backend_config["db"] dsn = swh_storage_backend_config["db"]
add_data(dsn=dsn) add_content_data(swh_storage_backend_config)
sha1 = "" sha1 = ""
assert map_sha1_with_swhid(sha1=sha1, dsn=dsn) is None assert map_sha1_with_swhid(sha1=sha1, dsn=dsn) is None
def test_mapping_with_wrong_sha1(swh_storage_backend_config): def test_mapping_with_wrong_sha1(swh_storage_backend_config):
dsn = swh_storage_backend_config["db"] dsn = swh_storage_backend_config["db"]
add_data(dsn=dsn) add_content_data(swh_storage_backend_config)
sha1 = "6ac599151a7aaa8ca5d38dc5bb61b49193a3cadc1ed33de5a57e4d1ecc53c846" sha1 = "6ac599151a7aaa8ca5d38dc5bb61b49193a3cadc1ed33de5a57e4d1ecc53c846"
assert map_sha1_with_swhid(sha1=sha1, dsn=dsn) is None assert map_sha1_with_swhid(sha1=sha1, dsn=dsn) is None
def test_map_row_for_definitions_with_sha1(swh_storage_backend_config):
dsn = swh_storage_backend_config["db"]
add_content_data(swh_storage_backend_config)
row = make_row(
id="maven/mavencentral/za.co.absa.cobrix/cobol-parser/revision/0.4.0.json",
text=file_data(f"{path}definitions.json"),
)
assert (
True,
[
(
"swh:1:cnt:e103b11cbfecbc116dacbb1f9ab2a02176092a32",
MetadataTargetType.CONTENT,
Origin(
url="http://central.maven.org/maven2/za/co/absa/cobrix/"
"cobol-parser/0.4.0/cobol-parser-0.4.0-sources.jar"
),
)
],
) == map_row(row, dsn)
def test_map_row_for_definitions_with_gitsha1(swh_storage_backend_config):
vlorentzUnsubmitted
Done
Inline Actions
text=file_data(os.path.join(datadir, "definitions.json")),
)
- assert (
+ expected = (
True,
[
(
"swh:1:cnt:e103b11cbfecbc116dacbb1f9ab2a02176092a32",
MetadataTargetType.CONTENT,
Origin(
url="http://central.maven.org/maven2/za/co/absa/cobrix/"
"cobol-parser/0.4.0/cobol-parser-0.4.0-sources.jar"
),
)
],
- ) == map_row(row=row, swh_storage_backend_config=swh_storage_backend_config)
+ )
+ assert map_row(row=row, swh_storage_backend_config=swh_storage_backend_config) == expected
def test_map_row_for_definitions_with_gitsha1(swh_storage_backend_config, datadir):

It's simpler this way.

(the line is too long, but you will rename it to a shorter argument name when you apply my other comments)

same comment for the functions below

vlorentz: It's simpler this way. (the line is too long, but you will rename it to a shorter argument…
dsn = swh_storage_backend_config["db"]
row = []
add_revision_data(swh_storage_backend_config)
row = make_row(
id="maven/mavencentral/za.co.absa.cobrix/cobol-parser/revision/0.4.0.json",
text=file_data(f"{path}definitions_sha1git.json"),
)
assert (
True,
[
(
"swh:1:rev:4c66129b968ab8122964823d1d77677f50884cf6",
MetadataTargetType.REVISION,
Origin(
url="http://central.maven.org/maven2/za/co/absa/cobrix/"
"cobol-parser/0.4.0/cobol-parser-0.4.0-sources.jar"
),
)
],
) == map_row(row, dsn)
def test_map_row_for_scancode(swh_storage_backend_config):
dsn = swh_storage_backend_config["db"]
add_content_data(swh_storage_backend_config)
row = make_row(
id="npm/npmjs/@ngtools/webpack/revision/10.2.1/tool/scancode/3.2.2.json",
text=file_data(f"{path}scancode.json"),
)
assert (
False,
[
(
"swh:1:cnt:e103b11cbfecbc116dacbb1f9ab2a02176092a32",
MetadataTargetType.CONTENT,
None,
)
],
) == map_row(row, dsn)
def test_map_row_for_licensee(swh_storage_backend_config):
dsn = swh_storage_backend_config["db"]
add_content_data(swh_storage_backend_config)
row = make_row(
id="npm/npmjs/@fluidframework/replay-driver/revision/0.31.0/tool/licensee/\
9.13.0.json",
text=file_data(f"{path}licensee.json"),
)
assert (
False,
[
(
"swh:1:cnt:095b80e14c3ea6254f57e94761f2313e32b1d58a",
MetadataTargetType.CONTENT,
None,
)
],
) == map_row(row, dsn)
def test_map_row_for_clearlydefined(swh_storage_backend_config):
dsn = swh_storage_backend_config["db"]
row = []
add_content_data(swh_storage_backend_config)
row = make_row(
id="npm/npmjs/@pixi/mesh-extras/revision/5.3.5/tool/clearlydefined/1.3.4.json",
text=file_data(f"{path}clearlydefined.json"),
)
assert (
False,
[
(
"swh:1:cnt:095b80e14c3ea6254f57e94761f2313e32b1d58a",
MetadataTargetType.CONTENT,
None,
)
],
) == map_row(row, dsn)
def test_sha1git_not_in_revision(swh_storage_backend_config):
dsn = swh_storage_backend_config["db"]
add_revision_data(swh_storage_backend_config)
assert (
map_definition(
metadata_string=file_data(f"{path}definitions_not_mapped_sha1_git.json"),
dsn=dsn,
)
is None
)
def test_sha1_not_in_content(swh_storage_backend_config):
dsn = swh_storage_backend_config["db"]
add_content_data(swh_storage_backend_config)
assert (
map_definition(
metadata_string=file_data(f"{path}definitions_not_mapped.json"),
dsn=dsn,
)
is None
)
def test_map_harvest_with_wrong_tool():
assert (False, []) == map_harvest(tool="abc", metadata_string="", dsn="")
def test_map_definition_with_wrong_metadata():
vlorentzUnsubmitted
Done
Inline Actions
def test_map_harvest_with_wrong_tool(swh_storage_backend_config):
- assert (False, []) == map_harvest(
+ assert map_harvest(
tool="abc",
metadata_string="",
swh_storage_backend_config=swh_storage_backend_config,
- )
+ ) == (False, [])
def test_map_definition_with_wrong_metadata(swh_storage_backend_config, datadir):
vlorentz:
assert (
map_definition(metadata_string=file_data(f"{path}licensee.json"), dsn="")
is None
)
def test_map_row_with_invalid_ID():
row = make_row(id="maven/mavencentral/cobol-parser/revision/0.4.0.json", text="abc")
assert map_row(row=row, swh_dsn="") is None
def test_map_row_with_empty_metadata_string():
row = make_row(
id="maven/mavencentral/za.co.absa.cobrix/cobol-parser/revision/0.4.0.json",
text="",
)
assert map_row(row=row, swh_dsn="") is None
def test_map_row_with_invalid_ID_without_revision():
row = make_row(
id="maven/mavencentral/za.co.absa.cobrix/cobol-parser/abc/0.4.0.json",
text="abc",
)
assert map_row(row=row, swh_dsn="") is None
def test_map_row_with_invalid_ID_without_json_extension():
row = make_row(
id="maven/mavencentral/za.co.absa.cobrix/cobol-parser/revision/0.4.0.txt",
text="abc",
)
assert map_row(row=row, swh_dsn="") is None
def test_map_row_with_invalid_ID_without_6_or_9_length():
row = make_row(
id="npm/npmjs/@ngtools/webpack/revision/10.2.1/tool/3.2.2.json", text="abc"
)
assert map_row(row=row, swh_dsn="") is None
def test_map_row_with_invalid_tool():
row = make_row(
id="npm/npmjs/@ngtools/webpack/revision/10.2.1/tool/abc/3.2.2.json", text="abc"
)
assert map_row(row=row, swh_dsn="") is None
def test_map_row_with_invalid_harvest_ID():
row = make_row(
id="npm/npmjs/@ngtools/webpack/revision/10.2.1/abc/scancode/3.2.2.json",
text="abc",
)
assert map_row(row=row, swh_dsn="") is None
About · Developer information · Legal