Changeset View
Changeset View
Standalone View
Standalone View
swh/graph/tests/test_pid.py
# Copyright (C) 2019 The Software Heritage developers | # Copyright (C) 2019 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 General Public License version 3, or any later version | # License: GNU 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 | ||||
import os | import os | ||||
import shutil | import shutil | ||||
import tempfile | import tempfile | ||||
import unittest | import unittest | ||||
from itertools import islice | from itertools import islice | ||||
from swh.graph.pid import str_to_bytes, bytes_to_str | from swh.graph.pid import str_to_bytes, bytes_to_str | ||||
from swh.graph.pid import PidToIntMap, IntToPidMap | from swh.graph.pid import PidToIntMap, IntToPidMap | ||||
from swh.model.identifiers import PID_TYPES | |||||
class TestPidSerialization(unittest.TestCase): | class TestPidSerialization(unittest.TestCase): | ||||
pairs = [ | pairs = [ | ||||
('swh:1:cnt:94a9ed024d3859793618152ea559a168bbcbb5e2', | ('swh:1:cnt:94a9ed024d3859793618152ea559a168bbcbb5e2', | ||||
bytes.fromhex('01' + '01' + | bytes.fromhex('01' + '01' + | ||||
'94a9ed024d3859793618152ea559a168bbcbb5e2')), | '94a9ed024d3859793618152ea559a168bbcbb5e2')), | ||||
▲ Show 20 Lines • Show All 109 Lines • ▼ Show 20 Lines | def test_update(self): | ||||
map2 = PidToIntMap(fname2, mode='rb+') | map2 = PidToIntMap(fname2, mode='rb+') | ||||
for (pid, int) in islice(map2, 11): # update the first N items | for (pid, int) in islice(map2, 11): # update the first N items | ||||
new_int = int + 42 | new_int = int + 42 | ||||
map2[pid] = new_int | map2[pid] = new_int | ||||
self.assertEqual(map2[pid], new_int) # check updated value | self.assertEqual(map2[pid], new_int) # check updated value | ||||
os.unlink(fname2) # tmpdir will be cleaned even if we don't reach this | os.unlink(fname2) # tmpdir will be cleaned even if we don't reach this | ||||
def test_iter_type(self): | |||||
for t in PID_TYPES: | |||||
first_20 = list(islice(self.map.iter_type(t), 20)) | |||||
k = first_20[0][1] | |||||
expected = [('swh:1:{}:{:040x}'.format(t, i), i) | |||||
for i in range(k, k + 20)] | |||||
assert first_20 == expected | |||||
douardda: This lacks a (series of) test(s) for the iter_prefix() method.
Sure this is indirectly tested… | |||||
seirlAuthorUnsubmitted Done Inline ActionsThanks, this is fixed in master. I'm closing this diff. seirl: Thanks, this is fixed in master. I'm closing this diff. | |||||
class TestIntToPidMap(unittest.TestCase): | class TestIntToPidMap(unittest.TestCase): | ||||
@classmethod | @classmethod | ||||
def setUpClass(cls): | def setUpClass(cls): | ||||
"""create reasonably sized (~1 MB) int->PID map to test on-disk DB | """create reasonably sized (~1 MB) int->PID map to test on-disk DB | ||||
""" | """ | ||||
cls.tmpdir = tempfile.mkdtemp(prefix='swh.graph.test.') | cls.tmpdir = tempfile.mkdtemp(prefix='swh.graph.test.') | ||||
Show All 35 Lines |
This lacks a (series of) test(s) for the iter_prefix() method.
Sure this is indirectly tested vie this test, but it should be tested as is. One may find it clever to reimplement iter_type without using iter_prefix, for example.