test: add test cases for woc.local

tests/fixtures/create_fixtures.py

@@ -99,7 +99,7 @@ def get_values_raw(self, map_name, key):
             print('reading large', _map["larges"][_hex], 'key', key, 'type', _map["dtypes"][1])
             return read_large(_map["larges"][_hex], _map["dtypes"][1])
         else:
-            print('reading from tch', key, _map["shards"], _map["sharding_bits"], _map["dtypes"][0] != 'h')
+            print('reading from tch', key, _map["sharding_bits"], _map["dtypes"][0] != 'h')
             return get_from_tch(key, _map["shards"], _map["sharding_bits"], _map["dtypes"][0] != 'h')
 
     def show_content_raw(
@@ -112,7 +112,7 @@ def show_content_raw(
 
         if obj == 'tree':
             _map_obj = self.config['objects']['tree.tch']
-            print('reading from tch', key, _map_obj['shards'], _map_obj['sharding_bits'])
+            print('reading from tch', key, _map_obj['sharding_bits'])
             v = get_from_tch(key, 
                 shards=_map_obj['shards'],
                 sharding_bits=_map_obj['sharding_bits'],
@@ -122,7 +122,7 @@ def show_content_raw(
 
         elif obj == 'commit':
             _map_obj = self.config['objects']['commit.tch']
-            print('reading from tch', key, _map_obj['shards'], _map_obj['sharding_bits'])
+            print('reading from tch', key, _map_obj['sharding_bits'])
             v = get_from_tch(key, 
                 shards=_map_obj['shards'],
                 sharding_bits=_map_obj['sharding_bits'],
@@ -132,7 +132,7 @@ def show_content_raw(
 
         elif obj == 'blob':
             _map_obj = self.config['objects']['sha1.blob.tch']
-            print('reading from tch', key, _map_obj['shards'], _map_obj['sharding_bits'])
+            print('reading from tch', key, _map_obj['sharding_bits'])
             v = get_from_tch(key, 
                 shards=_map_obj['shards'],
                 sharding_bits=_map_obj['sharding_bits'],
@@ -180,7 +180,7 @@ def copy_values(self, map_name, key):
             return write_large(_map["larges"][_hex], key, value, _map["dtypes"][1])
         else:
             # use fnv hash as shading idx if key is not a git sha
-            print('writing to tch', key, _map["shards"], _map["sharding_bits"], _map["dtypes"][0] != 'h')
+            print('writing to tch', key, _map["sharding_bits"], _map["dtypes"][0] != 'h')
             return write_to_tch(key, value, _map["shards"], _map["sharding_bits"], _map["dtypes"][0] != 'h')
 
     def copy_content(self, obj: str, key: Union[bytes, str]):
@@ -189,31 +189,39 @@ def copy_content(self, obj: str, key: Union[bytes, str]):
 
         if obj == 'tree':
             _map_obj = self.config2['objects']['tree.tch']
-            print('writing to tch', key, _map_obj["shards"], _map_obj["sharding_bits"])
-            write_to_tch(key.encode(), value, _map_obj['shards'], _map_obj['sharding_bits'], use_fnv_keys=False)
+            print('writing to tch', key, _map_obj["sharding_bits"])
+            write_to_tch(bytes.fromhex(key), value, _map_obj['shards'], _map_obj['sharding_bits'], use_fnv_keys=True)
 
         elif obj == 'commit':
             _map_obj = self.config2['objects']['commit.tch']
-            print('writing to tch', key, _map_obj["shards"], _map_obj["sharding_bits"])
-            write_to_tch(key.encode(), value, _map_obj['shards'], _map_obj['sharding_bits'], use_fnv_keys=False)
+            print('writing to tch', key ,_map_obj["sharding_bits"])
+            write_to_tch(bytes.fromhex(key), value, _map_obj['shards'], _map_obj['sharding_bits'], use_fnv_keys=True)
 
         elif obj == 'blob':
             _map_obj = self.config2['objects']['sha1.blob.tch']
             _idx, _v = value
             offset, length = unber(_idx)
             _idx = ber(0, length)
-            print('writing to tch', key, _map_obj["shards"], _map_obj["sharding_bits"])
-            write_to_tch(key.encode(), _idx, _map_obj['shards'], _map_obj['sharding_bits'], use_fnv_keys=False)
+            print('writing to tch', key, _map_obj["sharding_bits"])
+            write_to_tch(bytes.fromhex(key), _idx, _map_obj['shards'], _map_obj['sharding_bits'], use_fnv_keys=False)
             _map_obj = self.config2['objects']['blob.bin']
-            print('writing to file', _map_obj['shards'][0], length)
-            with open(_map_obj['shards'][0], "ab") as f:
+            shard = get_shard(bytes.fromhex(key), _map_obj['sharding_bits'], use_fnv_keys=False)
+            print('writing to file', _map_obj['shards'][shard], length)
+            with open(_map_obj['shards'][shard], "ab") as f:
                 f.write(_v)
 
         else:
             raise ValueError(f'Unsupported object type: {obj}, expected one of tree, blob, commit')
 
 
 if __name__ == '__main__':
+    import glob
+    import os
+
+    for f in glob.glob('./tests/fixtures/*.tch*') + glob.glob('./tests/fixtures/*.bin'):
+        print('removing', f)
+        os.remove(f)
+
     cp = WocMapsCopier('./wocprofile.json', './tests/test_profile.json')
     cp.copy_values('c2p', 'e4af89166a17785c1d741b8b1d5775f3223f510f')
     cp.copy_values('c2dat', 'e4af89166a17785c1d741b8b1d5775f3223f510f')

tests/test_local.py

@@ -0,0 +1,91 @@
+import pytest
+import os
+
+# Import the TCHashDB class
+from woc.local import WocMapsLocal
+
+@pytest.fixture
+def woc():
+    _test_pr = os.path.join(os.path.dirname(__file__), 'test_profile.json')
+    woc = WocMapsLocal(_test_pr)
+    yield woc
+
+def test_c2p(woc):
+    res = woc.get_values('c2p', 'e4af89166a17785c1d741b8b1d5775f3223f510f')
+    assert res[0] == 'W4D3_news' 
+
+def test_c2dat(woc):
+    res = woc.get_values('c2dat', 'e4af89166a17785c1d741b8b1d5775f3223f510f')
+    assert res[0] == '1410029988'
+
+def test_b2tac(woc):
+    res = woc.get_values('b2tac', '05fe634ca4c8386349ac519f899145c75fff4169')
+    assert res[0] == ('1410029988', 'Audris Mockus <[email protected]>', 'e4af89166a17785c1d741b8b1d5775f3223f510f')
+
+def test_p2a(woc):
+    res = woc.get_values('p2a', 'ArtiiQ_PocketMine-MP')
+    assert res[0] == '0929hitoshi <[email protected]>'
+
+def test_b2c(woc):
+    res = woc.get_values('b2c', '05fe634ca4c8386349ac519f899145c75fff4169')
+    assert res[0] == 'e4af89166a17785c1d741b8b1d5775f3223f510f'  
+
+def test_b2c_large(woc):
+    res = woc.get_values('b2c', '3f2eca18f1bc0f3117748e2cea9251e5182db2f7')
+    assert res[0] == '00003a69db53b45a67f76632f33a93691da77197'  
+
+def test_a2c(woc):
+    res = woc.get_values('a2c', 'Audris Mockus <[email protected]>')
+    assert res[0] == '001ec7302de3b07f32669a1f1faed74585c8a8dc'  
+
+def test_c2cc_null_filename(woc):  # file name is null
+    with pytest.raises(AssertionError):
+        woc.get_values('c2cc', 'e4af89166a17785c1d741b8b1d5775f3223f510f')
+
+def test_a2f(woc):
+    res = woc.get_values('a2f', 'Audris Mockus <[email protected]>')
+    assert res[0] == '.#analyze.sh'  
+
+def test_c2f(woc):
+    res = woc.get_values('c2f', 'e4af89166a17785c1d741b8b1d5775f3223f510f')
+    assert res[0] == 'README.md'  
+
+def test_c2b(woc):
+    res = woc.get_values('c2b', 'e4af89166a17785c1d741b8b1d5775f3223f510f')
+    assert res[0] == '05fe634ca4c8386349ac519f899145c75fff4169'  
+
+def test_p2c(woc):
+    res = woc.get_values('p2c', 'ArtiiQ_PocketMine-MP')
+    assert res[0] == '0000000bab11354f9a759332065be5f066c3398f'  
+
+def test_f2a(woc):
+    res = woc.get_values('f2a', 'youtube-statistics-analysis.pdf')
+    assert res[0] == 'Audris Mockus <[email protected]>'  
+
+def test_b2f(woc):
+    res = woc.get_values('b2f', '05fe634ca4c8386349ac519f899145c75fff4169')
+    assert res[0] == 'README.md'  
+
+def test_c2r(woc):
+    res = woc.get_values('c2r', 'e4af89166a17785c1d741b8b1d5775f3223f510f')
+    assert res[0] == '9531fc286ef1f4753ca4be9a3bf76274b929cdeb'  
+
+def test_b2fa(woc):
+    res = woc.get_values('b2fa', '05fe634ca4c8386349ac519f899145c75fff4169')
+    assert res[0] == '1410029988'  
+
+def test_tree(woc):
+    res = woc.show_content('tree', 'f1b66dcca490b5c4455af319bc961a34f69c72c2')
+    assert len(res) == 2 
+
+def test_commit(woc):
+    res = woc.show_content('commit', 'e4af89166a17785c1d741b8b1d5775f3223f510f')
+    assert len(res) == 222 
+
+def test_blob_1(woc):
+    res = woc.show_content('blob', '05fe634ca4c8386349ac519f899145c75fff4169')
+    assert len(res) == 14194
+
+def test_blob_2(woc):
+    res = woc.show_content('blob', '46aaf071f1b859c5bf452733c2583c70d92cd0c8')
+    assert len(res) == 1236