|
| 1 | +""" |
| 2 | +TestCases for 'EssenceDriveFS not respecting // separator' |
| 3 | +https://github.com/MAK-Relic-Tool/Issue-Tracker/issues/39 |
| 4 | +""" |
| 5 | +import zlib |
| 6 | +from contextlib import contextmanager |
| 7 | + |
| 8 | +import fs |
| 9 | +from fs.base import FS |
| 10 | +from fs.memoryfs import MemoryFS |
| 11 | + |
| 12 | +from relic.sga.core import StorageType |
| 13 | +from relic.sga.core.filesystem import EssenceFS |
| 14 | + |
| 15 | + |
| 16 | +@contextmanager |
| 17 | +def _generate_fake_osfs() -> FS: |
| 18 | + raw_text = b"""Ready to unleash 11 barrels of lead. |
| 19 | +Where's that artillery?!?! |
| 20 | +Orks are da biggust and da strongest. |
| 21 | +Fix bayonets! |
| 22 | +Fear me, but follow! |
| 23 | +Call for an earth-shaker? |
| 24 | +My mind is too weary to fight on... |
| 25 | +We'll be off as soon as the fuel arrives. |
| 26 | +Where are those tech priests. |
| 27 | +Fire until they see the glow of our barrels!""" |
| 28 | + |
| 29 | + comp_text = zlib.compress(raw_text) |
| 30 | + |
| 31 | + with MemoryFS() as fs: |
| 32 | + with fs.makedir("/samples") as samples_folder: |
| 33 | + with samples_folder.makedir("/strings") as strings_folders: |
| 34 | + with strings_folders.openbin("buffer.txt", "wb") as file: |
| 35 | + file.write(comp_text) |
| 36 | + with strings_folders.openbin("stream.txt", "wb") as file: |
| 37 | + file.write(comp_text) |
| 38 | + with strings_folders.openbin("store.txt", "wb") as file: |
| 39 | + file.write(raw_text) |
| 40 | + yield fs |
| 41 | + |
| 42 | + |
| 43 | +_CHUNK_SIZE = 1024 * 1024 * 16 # 16 MiB |
| 44 | + |
| 45 | + |
| 46 | +def _pack_fake_osfs(osfs: FS, name: str) -> EssenceFS: |
| 47 | + # Create 'SGA' |
| 48 | + sga = EssenceFS() |
| 49 | + sga.setmeta( |
| 50 | + { |
| 51 | + "name": name, # Specify name of archive |
| 52 | + "header_md5": "0" |
| 53 | + * 16, # Must be present due to a bug, recalculated when packed |
| 54 | + "file_md5": "0" |
| 55 | + * 16, # Must be present due to a bug, recalculated when packed |
| 56 | + }, |
| 57 | + "essence", |
| 58 | + ) |
| 59 | + |
| 60 | + alias = "test" |
| 61 | + sga_drive = None # sga.create_drive(alias) |
| 62 | + for path in osfs.walk.files(): |
| 63 | + if ( |
| 64 | + sga_drive is None |
| 65 | + ): # Lazily create drive, to avoid empty drives from being created |
| 66 | + sga_drive = sga.create_drive(alias) |
| 67 | + |
| 68 | + if "stream" in path: |
| 69 | + storage = StorageType.STREAM_COMPRESS |
| 70 | + elif "buffer" in path: |
| 71 | + storage = StorageType.BUFFER_COMPRESS |
| 72 | + else: |
| 73 | + storage = StorageType.STORE |
| 74 | + |
| 75 | + with osfs.openbin(path, "r") as unpacked_file: |
| 76 | + parent, file = fs.path.split(path) |
| 77 | + with sga_drive.makedirs(parent, recreate=True) as folder: |
| 78 | + with folder.openbin(file, "w") as packed_file: |
| 79 | + while True: |
| 80 | + buffer = unpacked_file.read(_CHUNK_SIZE) |
| 81 | + if len(buffer) == 0: |
| 82 | + break |
| 83 | + packed_file.write(buffer) |
| 84 | + sga_drive.setinfo(path, {"essence": {"storage_type": storage}}) |
| 85 | + return sga |
| 86 | + |
| 87 | + |
| 88 | +def _check_path(sga: EssenceFS, path: str): |
| 89 | + left_sep = path.replace("\\", "/") |
| 90 | + right_sep = path.replace("/", "\\") |
| 91 | + |
| 92 | + info = sga.getinfo(path) |
| 93 | + l_info = sga.getinfo(left_sep) |
| 94 | + r_info = sga.getinfo(right_sep) |
| 95 | + |
| 96 | + assert info == l_info |
| 97 | + assert l_info == r_info |
| 98 | + |
| 99 | + |
| 100 | +def test_fix_39(): |
| 101 | + with _generate_fake_osfs() as osfs: |
| 102 | + sga = _pack_fake_osfs(osfs, "Test Archive") |
| 103 | + for root, folders, files in sga.walk(): |
| 104 | + _check_path(sga, root) |
| 105 | + # for folder in folders |
| 106 | + # folders are checked when we walk into them |
| 107 | + for file in files: |
| 108 | + full_path = fs.path.join(root, file.name) |
| 109 | + _check_path(sga, full_path) |
0 commit comments