Files
chia-blockchain/chia/_tests/plot_sync/test_delta.py

102 lines
3.3 KiB
Python

from __future__ import annotations
import logging
import pytest
from chia_rs import G1Element
from chia_rs.sized_bytes import bytes32
from chia_rs.sized_ints import uint8, uint64
from chia.plot_sync.delta import Delta, DeltaType, PathListDelta, PlotListDelta
from chia.protocols.harvester_protocol import Plot
log = logging.getLogger(__name__)
def dummy_plot(path: str) -> Plot:
return Plot(
filename=path,
size=uint8(32),
plot_id=bytes32(b"\00" * 32),
pool_public_key=G1Element(),
pool_contract_puzzle_hash=None,
plot_public_key=G1Element(),
file_size=uint64(0),
time_modified=uint64(0),
compression_level=uint8(0),
)
@pytest.mark.parametrize(
["delta"],
[
pytest.param(PathListDelta(), id="path list"),
pytest.param(PlotListDelta(), id="plot list"),
],
)
def test_list_delta(delta: DeltaType) -> None:
assert delta.empty()
if type(delta) is PathListDelta:
assert delta.additions == []
elif type(delta) is PlotListDelta:
assert delta.additions == {}
else:
assert False
assert delta.removals == []
assert delta.empty()
if type(delta) is PathListDelta:
delta.additions.append("0")
elif type(delta) is PlotListDelta:
delta.additions["0"] = dummy_plot("0")
else:
assert False, "Invalid delta type"
assert not delta.empty()
delta.removals.append("0")
assert not delta.empty()
delta.additions.clear()
assert not delta.empty()
delta.clear()
assert delta.empty()
@pytest.mark.parametrize(
["old", "new", "result"],
[
[[], [], PathListDelta()],
[["1"], ["0"], PathListDelta(["0"], ["1"])],
[["1", "2", "3"], ["1", "2", "3"], PathListDelta([], [])],
[["2", "1", "3"], ["2", "3", "1"], PathListDelta([], [])],
[["2"], ["2", "3", "1"], PathListDelta(["3", "1"], [])],
[["2"], ["1", "3"], PathListDelta(["1", "3"], ["2"])],
[["1"], ["1", "2", "3"], PathListDelta(["2", "3"], [])],
[[], ["1", "2", "3"], PathListDelta(["1", "2", "3"], [])],
[["-1"], ["1", "2", "3"], PathListDelta(["1", "2", "3"], ["-1"])],
[["-1", "1"], ["2", "3"], PathListDelta(["2", "3"], ["-1", "1"])],
[["-1", "1", "2"], ["2", "3"], PathListDelta(["3"], ["-1", "1"])],
[["-1", "2", "3"], ["2", "3"], PathListDelta([], ["-1"])],
[["-1", "2", "3", "-2"], ["2", "3"], PathListDelta([], ["-1", "-2"])],
[["-2", "2", "3", "-1"], ["2", "3"], PathListDelta([], ["-2", "-1"])],
],
)
def test_path_list_delta_from_lists(old: list[str], new: list[str], result: PathListDelta) -> None:
assert PathListDelta.from_lists(old, new) == result
def test_delta_empty() -> None:
delta: Delta = Delta()
all_deltas: list[DeltaType] = [delta.valid, delta.invalid, delta.keys_missing, delta.duplicates]
assert delta.empty()
for d1 in all_deltas:
delta.valid.additions["0"] = dummy_plot("0")
delta.invalid.additions.append("0")
delta.keys_missing.additions.append("0")
delta.duplicates.additions.append("0")
assert not delta.empty()
for d2 in all_deltas:
if d2 is not d1:
d2.clear()
assert not delta.empty()
assert not delta.empty()
d1.clear()
assert delta.empty()