Changeset View
Changeset View
Standalone View
Standalone View
swh/dataset/test/test_graph.py
Show First 20 Lines • Show All 62 Lines • ▼ Show 20 Lines | TEST_ORIGIN_VISIT = { | ||||
"status": "ongoing", # TODO | "status": "ongoing", # TODO | ||||
"metadata": {"foo": "bar"}, | "metadata": {"foo": "bar"}, | ||||
"type": "git", | "type": "git", | ||||
} | } | ||||
@pytest.fixture | @pytest.fixture | ||||
def exporter(): | def exporter(): | ||||
def wrapped(messages, config=None) -> Tuple[Mock, Mock]: | def wrapped(messages, config=None, nodes_already_there=False) -> Tuple[Mock, Mock]: | ||||
if config is None: | if config is None: | ||||
config = {} | config = {} | ||||
node_writer = Mock() | node_writer = Mock() | ||||
edge_writer = Mock() | edge_writer = Mock() | ||||
node_set = Mock() | |||||
node_set.add.return_value = not nodes_already_there | |||||
process_messages( | process_messages( | ||||
messages, config=config, node_writer=node_writer, edge_writer=edge_writer, | messages, | ||||
config=config, | |||||
node_writer=node_writer, | |||||
edge_writer=edge_writer, | |||||
node_set=node_set, | |||||
) | ) | ||||
return node_writer.write, edge_writer.write | return node_writer.write, edge_writer.write | ||||
return wrapped | return wrapped | ||||
def binhash(s): | def binhash(s): | ||||
return hashlib.sha1(s.encode()).digest() | return hashlib.sha1(s.encode()).digest() | ||||
▲ Show 20 Lines • Show All 252 Lines • ▼ Show 20 Lines | def test_export_content(exporter): | ||||
) | ) | ||||
assert node_writer.mock_calls == [ | assert node_writer.mock_calls == [ | ||||
call(f"swh:1:cnt:{hexhash('cnt1')}\n"), | call(f"swh:1:cnt:{hexhash('cnt1')}\n"), | ||||
call(f"swh:1:cnt:{hexhash('cnt2')}\n"), | call(f"swh:1:cnt:{hexhash('cnt2')}\n"), | ||||
] | ] | ||||
assert edge_writer.mock_calls == [] | assert edge_writer.mock_calls == [] | ||||
def test_export_already_there_nodes(exporter): | |||||
node_writer, edge_writer = exporter( | |||||
{ | |||||
"content": [{**TEST_CONTENT, "sha1_git": binhash("cnt1")}], | |||||
"directory": [{"id": binhash("dir2"), "entries": []}], | |||||
}, | |||||
nodes_already_there=True, | |||||
) | |||||
assert node_writer.mock_calls == [] | |||||
assert edge_writer.mock_calls == [] | |||||
def zstwrite(fp, lines): | def zstwrite(fp, lines): | ||||
with ZSTFile(fp, "w") as writer: | with ZSTFile(fp, "w") as writer: | ||||
for l in lines: | for l in lines: | ||||
writer.write(l + "\n") | writer.write(l + "\n") | ||||
def zstread(fp): | def zstread(fp): | ||||
with ZSTFile(fp, "r") as reader: | with ZSTFile(fp, "r") as reader: | ||||
▲ Show 20 Lines • Show All 67 Lines • Show Last 20 Lines |