860 lines
45 KiB
Python
860 lines
45 KiB
Python
import contextlib
|
|
import io
|
|
import json
|
|
import os
|
|
import stat
|
|
import struct
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
import import_audit
|
|
import nid_names
|
|
|
|
GLOBAL_FUNCTION = 0x12
|
|
LOCAL_FUNCTION = 0x02
|
|
STUB_SOURCE = "int APS5_VABI {name}(int value) {{ NotImplemented_nid_no_patch(__func__); return 0; }}\n"
|
|
DONE_SOURCE = "int APS5_VABI {name}(int value) {{ return value; }}\n"
|
|
MESSAGE_NID = "wTpfglkmv34"
|
|
MESSAGE_NAME = "sceMsgDialogProgressBarSetValue"
|
|
|
|
|
|
def build_elf(symbols, needed=()):
|
|
strings = b"\0"
|
|
entries = [struct.pack("<IBBHQQ", 0, 0, 0, 0, 0, 0)]
|
|
for name, index, info in symbols:
|
|
entries.append(struct.pack("<IBBHQQ", len(strings), info, 0, index, 0, 0))
|
|
strings += name.encode() + b"\0"
|
|
libraries = []
|
|
for library in needed:
|
|
libraries.append(len(strings))
|
|
strings += library.encode() + b"\0"
|
|
table = b"".join(entries)
|
|
table_offset = 64 + len(strings)
|
|
table_offset += -table_offset % 8
|
|
dynamic_offset = table_offset + len(table)
|
|
dynamic_offset += -dynamic_offset % 8
|
|
pairs = [(1, offset) for offset in libraries]
|
|
if pairs:
|
|
pairs.append((0, 0))
|
|
dynamic = b"".join(struct.pack("<qQ", tag, value) for tag, value in pairs)
|
|
sections_offset = dynamic_offset + len(dynamic)
|
|
sections_offset += -sections_offset % 8
|
|
image = bytearray(sections_offset + (4 if dynamic else 3) * 64)
|
|
image[64:64 + len(strings)] = strings
|
|
image[table_offset:table_offset + len(table)] = table
|
|
if dynamic:
|
|
image[dynamic_offset:dynamic_offset + len(dynamic)] = dynamic
|
|
image[sections_offset + 192:sections_offset + 256] = struct.pack("<IIQQQQIIQQ", 0, 6, 2, 0, dynamic_offset, len(dynamic), 2, 0, 8, 16)
|
|
image[sections_offset + 64:sections_offset + 128] = struct.pack("<IIQQQQIIQQ", 0, 11, 2, 0, table_offset, len(table), 2, 1, 8, 24)
|
|
image[sections_offset + 128:sections_offset + 192] = struct.pack("<IIQQQQIIQQ", 0, 3, 2, 0, 64, len(strings), 0, 0, 1, 0)
|
|
image[:64] = struct.pack("<16sHHIQQQIHHHHHH", b"\x7fELF\x02\x01\x01" + bytes(9), 3, 0x3e, 1, 0, 0, sections_offset, 0, 64, 0, 0, 64, 4 if dynamic else 3, 0)
|
|
return bytes(image)
|
|
|
|
|
|
def build_sectionless_elf(symbols, needed=(), gnu_hash=False):
|
|
strings = b"\0"
|
|
entries = [struct.pack("<IBBHQQ", 0, 0, 0, 0, 0, 0)]
|
|
for name, index, info in symbols:
|
|
entries.append(struct.pack("<IBBHQQ", len(strings), info, 0, index, 0, 0))
|
|
strings += name.encode() + b"\0"
|
|
libraries = []
|
|
for library in needed:
|
|
libraries.append(len(strings))
|
|
strings += library.encode() + b"\0"
|
|
program_offset = 64
|
|
content_offset = program_offset + 2 * 56
|
|
strings_offset = content_offset
|
|
symbols_offset = (strings_offset + len(strings) + 7) & ~7
|
|
symbol_table = b"".join(entries)
|
|
hash_offset = (symbols_offset + len(symbol_table) + 3) & ~3
|
|
if gnu_hash:
|
|
hash_table = struct.pack("<IIII", 1, 1, 1, 0) + bytes(8)
|
|
hash_table += struct.pack("<I", 1) + b"".join(struct.pack("<I", 1 if index == len(entries) - 1 else 0) for index in range(1, len(entries)))
|
|
hash_tag = 0x6ffffef5
|
|
else:
|
|
hash_table = struct.pack("<II", 1, len(entries)) + struct.pack("<I", 0) + bytes(4 * len(entries))
|
|
hash_tag = 4
|
|
dynamic_offset = (hash_offset + len(hash_table) + 7) & ~7
|
|
dynamic = b"".join(struct.pack("<qQ", 1, value) for value in libraries)
|
|
dynamic += b"".join(struct.pack("<qQ", tag, value) for tag, value in (
|
|
(5, 0x400000 + strings_offset),
|
|
(10, len(strings)),
|
|
(6, 0x400000 + symbols_offset),
|
|
(11, 24),
|
|
(hash_tag, 0x400000 + hash_offset),
|
|
(0, 0),
|
|
))
|
|
image = bytearray(dynamic_offset + len(dynamic))
|
|
image[strings_offset:strings_offset + len(strings)] = strings
|
|
image[symbols_offset:symbols_offset + len(symbol_table)] = symbol_table
|
|
image[hash_offset:hash_offset + len(hash_table)] = hash_table
|
|
image[dynamic_offset:dynamic_offset + len(dynamic)] = dynamic
|
|
image[program_offset:program_offset + 56] = struct.pack("<IIQQQQQQ", 1, 7, 0, 0x400000, 0x400000, len(image), len(image), 0x1000)
|
|
image[program_offset + 56:program_offset + 112] = struct.pack("<IIQQQQQQ", 2, 6, dynamic_offset, 0x400000 + dynamic_offset, 0x400000 + dynamic_offset, len(dynamic), len(dynamic), 8)
|
|
image[:64] = struct.pack("<16sHHIQQQIHHHHHH", b"\x7fELF\x02\x01\x01" + bytes(9), 3, 0x3e, 1, 0, program_offset, 0, 0, 64, 56, 2, 64, 0, 0)
|
|
return bytes(image)
|
|
|
|
|
|
def exporting(*names):
|
|
return build_elf([(name, 1, GLOBAL_FUNCTION) for name in names])
|
|
|
|
|
|
def build_pe(names):
|
|
section_rva, raw_offset = 0x1000, 0x200
|
|
strings, rvas = b"", []
|
|
start = 40 + 10 * len(names)
|
|
for name in names:
|
|
rvas.append(section_rva + start + len(strings))
|
|
strings += name.encode() + b"\0"
|
|
raw = struct.pack("<IIHHIIIIIII", 0, 0, 0, 0, 0, 1, len(names), len(names), section_rva + 40, section_rva + 40 + 4 * len(names), section_rva + 40 + 8 * len(names))
|
|
raw += b"".join(struct.pack("<I", 0x2000 + 16 * index) for index in range(len(names)))
|
|
raw += b"".join(struct.pack("<I", rva) for rva in rvas)
|
|
raw += b"".join(struct.pack("<H", index) for index in range(len(names))) + strings
|
|
image = bytearray(raw_offset)
|
|
image[0:2] = b"MZ"
|
|
image[0x3c:0x40] = struct.pack("<I", 0x40)
|
|
image[0x40:0x44] = b"PE\0\0"
|
|
image[0x44:0x58] = struct.pack("<HHIIIHH", 0x8664, 2, 0, 0, 0, 240, 0x2022)
|
|
image[0x58:0x5a] = struct.pack("<H", 0x20b)
|
|
image[0x58 + 60:0x58 + 64] = struct.pack("<I", raw_offset)
|
|
image[0x58 + 108:0x58 + 112] = struct.pack("<I", 16)
|
|
image[0x58 + 112:0x58 + 120] = struct.pack("<II", section_rva, len(raw))
|
|
image[0x58 + 240:0x58 + 280] = struct.pack("<8sIIIIIIHHI", b".edata", len(raw), section_rva, len(raw), raw_offset, 0, 0, 0, 0, 0x40000040)
|
|
image[0x58 + 280:0x58 + 320] = struct.pack("<8sIIIIIIHHI", b".text", 16 * max(1, len(names)), 0x2000, 16 * max(1, len(names)), raw_offset + len(raw), 0, 0, 0, 0, 0x60000020)
|
|
return bytes(image) + raw + b"\xc3" * (16 * max(1, len(names)))
|
|
|
|
|
|
def write_source(root, library, text):
|
|
directory = Path(root) / library
|
|
directory.mkdir(parents=True, exist_ok=True)
|
|
(directory / "Export.cpp").write_text(text)
|
|
|
|
|
|
def write_registry(path, entries, encoding="utf-8"):
|
|
path.write_text(json.dumps([{"nid": nid, "library": library, "relocationType": "R_X86_64_GLOB_DAT"} for nid, library in entries]), encoding=encoding)
|
|
|
|
|
|
class Workspace:
|
|
def __init__(self):
|
|
self._directory = tempfile.TemporaryDirectory()
|
|
self.root = Path(self._directory.name)
|
|
self.libs = self.root / "libs"
|
|
self.modules = self.root / "modules"
|
|
self.source = self.root / "prx"
|
|
for path in (self.libs, self.modules, self.source):
|
|
path.mkdir()
|
|
|
|
def close(self):
|
|
self._directory.cleanup()
|
|
|
|
def run(self, registry, *extra):
|
|
out, err = io.StringIO(), io.StringIO()
|
|
arguments = [str(registry), "--libs", str(self.libs), "--modules", str(self.modules), "--source", str(self.source), *extra]
|
|
with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err):
|
|
code = import_audit.main(arguments)
|
|
return code, out.getvalue(), err.getvalue()
|
|
|
|
|
|
class ImportAuditTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.work = Workspace()
|
|
self.addCleanup(self.work.close)
|
|
|
|
def test_failed_json_write_preserves_existing_report_and_leaves_no_partial_report(self):
|
|
work = self.work
|
|
(work.libs / "libSceA.prx").write_bytes(exporting("AAAAAAAAAAA"))
|
|
registry = work.root / "game.json"
|
|
write_registry(registry, [("AAAAAAAAAAA", "libSceA.prx")])
|
|
output = work.root / "audit.json"
|
|
previous = b'{"previous": "complete report"}\n'
|
|
|
|
def disk_full(path, content, **kwargs):
|
|
with path.open("w", encoding="utf-8") as handle:
|
|
handle.write(content[:24])
|
|
raise OSError(28, "No space left on device", str(path))
|
|
|
|
for exists in (True, False):
|
|
with self.subTest(existing_report=exists):
|
|
if exists:
|
|
output.write_bytes(previous)
|
|
else:
|
|
output.unlink(missing_ok=True)
|
|
files = set(work.root.iterdir())
|
|
with patch.object(Path, "write_text", disk_full):
|
|
code, out, err = work.run(registry, "--json", str(output))
|
|
self.assertEqual((code, out), (2, ""))
|
|
self.assertIn("No space left on device", err)
|
|
self.assertEqual(set(work.root.iterdir()), files)
|
|
if exists:
|
|
self.assertEqual(output.read_bytes(), previous)
|
|
else:
|
|
self.assertFalse(output.exists())
|
|
|
|
def test_failed_json_replacement_preserves_report_and_cleans_staging(self):
|
|
work = self.work
|
|
(work.libs / "libSceA.prx").write_bytes(exporting("AAAAAAAAAAA"))
|
|
registry = work.root / "game.json"
|
|
write_registry(registry, [("AAAAAAAAAAA", "libSceA.prx")])
|
|
output = work.root / "audit.json"
|
|
previous = b'{"previous": "complete report"}\n'
|
|
output.write_bytes(previous)
|
|
files = set(work.root.iterdir())
|
|
with patch.object(Path, "replace", side_effect=PermissionError(13, "Permission denied", str(output))):
|
|
code, out, err = work.run(registry, "--json", str(output))
|
|
self.assertEqual((code, out), (2, ""))
|
|
self.assertIn("Permission denied", err)
|
|
self.assertEqual(output.read_bytes(), previous)
|
|
self.assertEqual(set(work.root.iterdir()), files)
|
|
|
|
def test_successful_json_write_replaces_complete_report_without_staging_files(self):
|
|
work = self.work
|
|
(work.libs / "libSceA.prx").write_bytes(exporting("AAAAAAAAAAA"))
|
|
registry = work.root / "game.json"
|
|
write_registry(registry, [("AAAAAAAAAAA", "libSceA.prx")])
|
|
output = work.root / "audit.json"
|
|
output.write_text("old report", encoding="utf-8")
|
|
files = set(work.root.iterdir())
|
|
code, _, err = work.run(registry, "--json", str(output))
|
|
self.assertEqual((code, err), (0, ""))
|
|
self.assertEqual(json.loads(output.read_text(encoding="utf-8"))["unique_by_class"]["implemented"], 1)
|
|
self.assertEqual(set(work.root.iterdir()), files)
|
|
|
|
@unittest.skipUnless(sys.platform == "win32", "Windows read-only replacement behavior")
|
|
def test_read_only_json_report_stays_intact_without_a_read_only_temporary(self):
|
|
work = self.work
|
|
(work.libs / "libSceA.prx").write_bytes(exporting("AAAAAAAAAAA"))
|
|
registry = work.root / "game.json"
|
|
write_registry(registry, [("AAAAAAAAAAA", "libSceA.prx")])
|
|
output = work.root / "audit.json"
|
|
previous = b'{"previous": "complete report"}\n'
|
|
output.write_bytes(previous)
|
|
files = set(work.root.iterdir())
|
|
output.chmod(0o444)
|
|
try:
|
|
code, out, err = work.run(registry, "--json", str(output))
|
|
self.assertEqual((code, out), (2, ""))
|
|
self.assertIn("FAIL:", err)
|
|
self.assertEqual(output.read_bytes(), previous)
|
|
self.assertEqual(set(work.root.iterdir()), files)
|
|
finally:
|
|
output.chmod(0o666)
|
|
|
|
def test_json_report_symlink_keeps_its_target(self):
|
|
work = self.work
|
|
(work.libs / "libSceA.prx").write_bytes(exporting("AAAAAAAAAAA"))
|
|
registry = work.root / "game.json"
|
|
write_registry(registry, [("AAAAAAAAAAA", "libSceA.prx")])
|
|
target = work.root / "actual-report.json"
|
|
target.write_text("old report", encoding="utf-8")
|
|
output = work.root / "audit.json"
|
|
try:
|
|
output.symlink_to(target.name)
|
|
except (OSError, NotImplementedError) as error:
|
|
self.skipTest(str(error))
|
|
files = set(work.root.iterdir())
|
|
code, _, err = work.run(registry, "--json", str(output))
|
|
self.assertEqual((code, err), (0, ""))
|
|
self.assertTrue(output.is_symlink())
|
|
self.assertEqual(json.loads(target.read_text(encoding="utf-8"))["unique_imports"], 1)
|
|
self.assertEqual(set(work.root.iterdir()), files)
|
|
|
|
def test_json_directory_error_preserves_the_supplied_path(self):
|
|
work = self.work
|
|
(work.libs / "libSceA.prx").write_bytes(exporting("AAAAAAAAAAA"))
|
|
registry = work.root / "game.json"
|
|
write_registry(registry, [("AAAAAAAAAAA", "libSceA.prx")])
|
|
output = work.modules / ".." / work.modules.name
|
|
self.assertNotEqual(str(output), str(output.resolve()))
|
|
files = set(work.root.iterdir())
|
|
code, out, err = work.run(registry, "--json", str(output))
|
|
self.assertEqual((code, out), (2, ""))
|
|
self.assertEqual(err, f"FAIL: {output}: JSON output is a directory\n")
|
|
self.assertEqual(set(work.root.iterdir()), files)
|
|
|
|
@unittest.skipUnless(os.name == "posix", "POSIX report permissions")
|
|
def test_new_json_report_permissions_follow_the_process_umask(self):
|
|
script = "\n".join([
|
|
"import os, sys",
|
|
"from pathlib import Path",
|
|
"sys.path.insert(0, sys.argv[1])",
|
|
"import import_audit",
|
|
"os.umask(int(sys.argv[4], 8))",
|
|
"Path(sys.argv[3]).write_text('{}\\n', encoding='utf-8')",
|
|
"import_audit.write_json(Path(sys.argv[2]), {'complete': True})",
|
|
])
|
|
for mask, mode in ((0o022, 0o644), (0o002, 0o664), (0o077, 0o600)):
|
|
with self.subTest(umask=oct(mask)):
|
|
output = self.work.root / f"audit-{mask:o}.json"
|
|
reference = self.work.root / f"plain-{mask:o}.json"
|
|
files = set(self.work.root.iterdir())
|
|
completed = subprocess.run([sys.executable, "-c", script, str(import_audit.ROOT / "tools"), str(output), str(reference), oct(mask)], capture_output=True, text=True)
|
|
self.assertEqual(completed.returncode, 0, completed.stderr)
|
|
self.assertEqual(stat.S_IMODE(output.stat().st_mode), mode)
|
|
self.assertEqual(stat.S_IMODE(output.stat().st_mode), stat.S_IMODE(reference.stat().st_mode))
|
|
self.assertEqual(json.loads(output.read_text(encoding="utf-8")), {"complete": True})
|
|
self.assertEqual(set(self.work.root.iterdir()), files | {output, reference})
|
|
|
|
@unittest.skipUnless(os.name == "posix", "POSIX report permissions")
|
|
def test_existing_json_report_permissions_are_preserved(self):
|
|
output = self.work.root / "audit.json"
|
|
for mode in (0o600, 0o640, 0o664):
|
|
with self.subTest(mode=oct(mode)):
|
|
output.write_text("old report", encoding="utf-8")
|
|
output.chmod(mode)
|
|
files = set(self.work.root.iterdir())
|
|
import_audit.write_json(output, {"complete": True})
|
|
self.assertEqual(stat.S_IMODE(output.stat().st_mode), mode)
|
|
self.assertEqual(json.loads(output.read_text(encoding="utf-8")), {"complete": True})
|
|
self.assertEqual(set(self.work.root.iterdir()), files)
|
|
|
|
def test_classification_accounts_for_every_reference(self):
|
|
work = self.work
|
|
(work.libs / "libSceMsgDialog.prx").write_bytes(build_elf([
|
|
(MESSAGE_NID, 1, GLOBAL_FUNCTION), ("AAAAAAAAAAA", 1, GLOBAL_FUNCTION), ("BBBBBBBBBBB", 1, GLOBAL_FUNCTION),
|
|
("UNDEFINED11", 0, GLOBAL_FUNCTION), ("LOCALLOCAL1", 1, LOCAL_FUNCTION)]))
|
|
(work.libs / "libc.prx").write_bytes(exporting("CCCCCCCCCCC"))
|
|
(work.modules / "libSceNpCppWebApi.prx").write_bytes(exporting())
|
|
write_source(work.source, "libSceMsgDialog", STUB_SOURCE.format(name=MESSAGE_NAME) + 'APS5_EXPORT("BBBBBBBBBBB", sceFooUnknown_B);\n' + STUB_SOURCE.format(name="sceFooUnknown_B"))
|
|
registry = work.root / "game.registry.json"
|
|
write_registry(registry, [
|
|
(MESSAGE_NID + "#A#B", "libSceMsgDialog.prx"), (MESSAGE_NID, "libSceMsgDialog.prx"),
|
|
("AAAAAAAAAAA", "libSceMsgDialog.prx"), ("BBBBBBBBBBB", "libSceMsgDialog.prx"),
|
|
("DDDDDDDDDDD", "libSceMsgDialog.prx"), ("EEEEEEEEEEE", "libSceNpCppWebApi.prx"),
|
|
("CCCCCCCCCCC", "libSceFoo.prx"), ("AAAAAAAAAAA", "libSceOther.prx"),
|
|
("UNDEFINED11", "libSceMsgDialog.prx"), ("LOCALLOCAL1", "libSceMsgDialog.prx")])
|
|
result = work.root / "audit.json"
|
|
code, out, err = work.run(registry, "--json", str(result))
|
|
self.assertEqual((code, err), (1, ""))
|
|
data = json.loads(result.read_text())
|
|
self.assertEqual(data["references"], 10)
|
|
self.assertEqual(data["unique_imports"], 9)
|
|
self.assertEqual(data["unique_by_class"], {"implemented": 3, "stub": 2, "absent": 3, "module": 1})
|
|
self.assertEqual(data["references_by_class"], {"implemented": 3, "stub": 3, "absent": 3, "module": 1})
|
|
self.assertEqual(data["missing_libraries"], {"libSceFoo.prx": 1, "libSceOther.prx": 1})
|
|
self.assertEqual(data["library_mismatch"], 1)
|
|
kinds = {(record["nid"], record["library"]): record["class"] for record in data["imports"]}
|
|
self.assertEqual(kinds[(MESSAGE_NID, "libSceMsgDialog.prx")], "stub")
|
|
self.assertEqual(kinds[("BBBBBBBBBBB", "libSceMsgDialog.prx")], "stub")
|
|
self.assertEqual(kinds[("UNDEFINED11", "libSceMsgDialog.prx")], "absent")
|
|
self.assertEqual(kinds[("LOCALLOCAL1", "libSceMsgDialog.prx")], "absent")
|
|
self.assertEqual(kinds[("EEEEEEEEEEE", "libSceNpCppWebApi.prx")], "module")
|
|
self.assertIn(f"{MESSAGE_NID} {MESSAGE_NAME}", out)
|
|
self.assertIn("DDDDDDDDDDD", out)
|
|
|
|
def test_module_imports_are_audited(self):
|
|
work = self.work
|
|
stub = nid_names.compute_nid("sceModuleStub")
|
|
(work.libs / "libSceA.prx").write_bytes(exporting("AAAAAAAAAAA"))
|
|
(work.libs / "libSceB.prx").write_bytes(exporting(stub))
|
|
write_source(work.source, "libSceB", STUB_SOURCE.format(name="sceModuleStub"))
|
|
(work.modules / "sibling.prx").write_bytes(exporting("BBBBBBBBBBB"))
|
|
(work.modules / "main.prx").write_bytes(build_elf([
|
|
("AAAAAAAAAAA", 0, GLOBAL_FUNCTION), ("BBBBBBBBBBB", 0, GLOBAL_FUNCTION),
|
|
(stub, 0, GLOBAL_FUNCTION), ("EEEEEEEEEEE", 0, GLOBAL_FUNCTION)], ["libSceA.prx", "sibling.prx"]))
|
|
registry = work.root / "game.json"
|
|
write_registry(registry, [])
|
|
result = work.root / "audit.json"
|
|
code, out, err = work.run(registry, "--json", str(result))
|
|
self.assertEqual((code, err), (1, ""))
|
|
data = json.loads(result.read_text())
|
|
self.assertEqual(data["references"], 4)
|
|
self.assertEqual(data["unique_by_class"], {"implemented": 2, "stub": 1, "absent": 1, "module": 0})
|
|
self.assertEqual(data["missing_libraries"], {})
|
|
records = {(record["nid"], record["module"]): record for record in data["imports"]}
|
|
self.assertEqual(records[("AAAAAAAAAAA", "main.prx")]["class"], "implemented")
|
|
self.assertEqual(records[("AAAAAAAAAAA", "main.prx")]["providers"], ["libSceA.prx"])
|
|
self.assertEqual(records[("BBBBBBBBBBB", "main.prx")]["class"], "implemented")
|
|
self.assertEqual(records[("BBBBBBBBBBB", "main.prx")]["providers"], ["sibling.prx"])
|
|
self.assertEqual(records[(stub, "main.prx")]["class"], "stub")
|
|
self.assertEqual(records[(stub, "main.prx")]["name"], "sceModuleStub")
|
|
self.assertEqual(records[("EEEEEEEEEEE", "main.prx")]["class"], "absent")
|
|
self.assertIn("[library not named]", out)
|
|
self.assertIn("EEEEEEEEEEE (imported by main.prx)", out)
|
|
|
|
def test_module_needed_library_without_a_file_is_missing(self):
|
|
work = self.work
|
|
(work.libs / "libc.prx").write_bytes(exporting())
|
|
(work.modules / "main.prx").write_bytes(build_elf([], ["libSceMissing.prx"]))
|
|
registry = work.root / "game.json"
|
|
write_registry(registry, [])
|
|
result = work.root / "audit.json"
|
|
code, out, err = work.run(registry, "--json", str(result))
|
|
self.assertEqual((code, err), (1, ""))
|
|
data = json.loads(result.read_text())
|
|
self.assertEqual((data["references"], data["imports"]), (0, []))
|
|
self.assertEqual(data["missing_libraries"], {"libSceMissing.prx": 1})
|
|
self.assertIn("libSceMissing.prx", out)
|
|
|
|
def test_module_files_must_be_elf(self):
|
|
work = self.work
|
|
(work.libs / "libc.prx").write_bytes(exporting())
|
|
(work.modules / "guest.prx").write_bytes(build_pe(["alpha"]))
|
|
registry = work.root / "game.json"
|
|
write_registry(registry, [])
|
|
code, out, err = work.run(registry)
|
|
self.assertEqual((code, out), (2, ""))
|
|
self.assertIn("guest.prx", err)
|
|
self.assertIn("ELF", err)
|
|
|
|
def test_sectionless_module_imports_and_exports_are_audited(self):
|
|
work = self.work
|
|
(work.libs / "libSceA.prx").write_bytes(exporting("AAAAAAAAAAA"))
|
|
(work.modules / "stripped.prx").write_bytes(build_sectionless_elf([
|
|
("AAAAAAAAAAA", 0, GLOBAL_FUNCTION), ("BBBBBBBBBBB", 1, GLOBAL_FUNCTION)], ["libSceA.prx"]))
|
|
(work.modules / "main.prx").write_bytes(build_elf([("BBBBBBBBBBB", 0, GLOBAL_FUNCTION)], []))
|
|
registry = work.root / "game.json"
|
|
write_registry(registry, [])
|
|
result = work.root / "audit.json"
|
|
code, out, err = work.run(registry, "--json", str(result))
|
|
self.assertEqual((code, err), (0, ""))
|
|
data = json.loads(result.read_text())
|
|
self.assertEqual(data["references"], 2)
|
|
records = {(record["nid"], record["module"]): record for record in data["imports"]}
|
|
self.assertEqual(records[("AAAAAAAAAAA", "stripped.prx")]["class"], "implemented")
|
|
self.assertEqual(records[("BBBBBBBBBBB", "main.prx")]["class"], "implemented")
|
|
|
|
def test_sectionless_elf_gnu_hash(self):
|
|
image = build_sectionless_elf([
|
|
("AAAAAAAAAAA", 0, GLOBAL_FUNCTION), ("BBBBBBBBBBB", 1, GLOBAL_FUNCTION)],
|
|
["libSceA.prx"], gnu_hash=True)
|
|
self.assertEqual(import_audit.elf_program_dynamic_symbols(Path("module.prx"), image),
|
|
({"AAAAAAAAAAA"}, {"BBBBBBBBBBB"}, {"libSceA.prx"}))
|
|
|
|
def test_module_symbol_suffix_is_stripped(self):
|
|
work = self.work
|
|
exported = nid_names.compute_nid("sceRealName")
|
|
(work.libs / "libSceA.prx").write_bytes(exporting(exported))
|
|
(work.modules / "main.prx").write_bytes(build_elf([
|
|
(exported + "#lib#mod", 0, GLOBAL_FUNCTION)], []))
|
|
registry = work.root / "game.json"
|
|
write_registry(registry, [])
|
|
result = work.root / "audit.json"
|
|
code, out, err = work.run(registry, "--json", str(result))
|
|
self.assertEqual((code, err), (0, ""))
|
|
data = json.loads(result.read_text())
|
|
self.assertEqual(data["unique_by_class"], {"implemented": 1, "stub": 0, "absent": 0, "module": 0})
|
|
self.assertEqual(data["references"], 1)
|
|
|
|
def test_clean_audit_exits_zero_and_empty_registry_is_empty(self):
|
|
work = self.work
|
|
(work.libs / "libSceA.prx").write_bytes(exporting("AAAAAAAAAAA"))
|
|
registry = work.root / "clean.json"
|
|
write_registry(registry, [("AAAAAAAAAAA", "libSceA.prx")])
|
|
self.assertEqual(work.run(registry)[0], 0)
|
|
write_registry(registry, [])
|
|
result = work.root / "empty.json"
|
|
code, out, _ = work.run(registry, "--json", str(result))
|
|
data = json.loads(result.read_text())
|
|
self.assertEqual((code, data["references"], data["unique_imports"], data["imports"]), (0, 0, 0, []))
|
|
self.assertIn("0 references, 0 unique imports", out)
|
|
|
|
def test_current_stop_is_the_first_unresolved_import(self):
|
|
work = self.work
|
|
(work.libs / "libSceA.prx").write_bytes(exporting("AAAAAAAAAAA"))
|
|
registry = work.root / "game.json"
|
|
write_registry(registry, [("AAAAAAAAAAA", "libSceA.prx"), ("BBBBBBBBBBB", "libSceA.prx"), ("CCCCCCCCCCC", "libSceA.prx")])
|
|
result = work.root / "audit.json"
|
|
code, out, err = work.run(registry, "--json", str(result))
|
|
self.assertEqual((code, err), (1, ""))
|
|
data = json.loads(result.read_text())
|
|
self.assertEqual(data["stops"], [{"registry": str(registry), "nid": "BBBBBBBBBBB", "library": "libSceA.prx", "name": None}])
|
|
self.assertIn("Current stop (first unresolved import, in registry order)", out)
|
|
self.assertIn("BBBBBBBBBBB", out)
|
|
|
|
def test_current_stop_is_none_when_every_import_resolves(self):
|
|
work = self.work
|
|
(work.libs / "libSceA.prx").write_bytes(exporting("AAAAAAAAAAA"))
|
|
registry = work.root / "clean.json"
|
|
write_registry(registry, [("AAAAAAAAAAA", "libSceA.prx")])
|
|
result = work.root / "audit.json"
|
|
code, out, _ = work.run(registry, "--json", str(result))
|
|
self.assertEqual(code, 0)
|
|
data = json.loads(result.read_text())
|
|
self.assertEqual(data["stops"], [{"registry": str(registry), "nid": None, "library": None, "name": None}])
|
|
self.assertIn(f" {registry}: none", out)
|
|
|
|
def test_removing_an_export_flips_the_class(self):
|
|
work = self.work
|
|
registry = work.root / "game.json"
|
|
write_registry(registry, [("AAAAAAAAAAA", "libSceA.prx")])
|
|
(work.libs / "libSceA.prx").write_bytes(exporting("AAAAAAAAAAA"))
|
|
self.assertEqual(work.run(registry)[0], 0)
|
|
(work.libs / "libSceA.prx").write_bytes(exporting("ZZZZZZZZZZZ"))
|
|
code, out, _ = work.run(registry)
|
|
self.assertEqual(code, 1)
|
|
self.assertIn("AAAAAAAAAAA", out)
|
|
|
|
def test_source_stubs(self):
|
|
source = self.work.source
|
|
write_source(source, "libSceTwin", STUB_SOURCE.format(name="sceTwinCall") + STUB_SOURCE.format(name="printf_nid_postfix"))
|
|
write_source(source, "libSceTwin.native", DONE_SOURCE.format(name="sceTwinCall"))
|
|
write_source(source, "libSceOnly", STUB_SOURCE.format(name="sceOnlyPending"))
|
|
stubs = import_audit.source_stubs(source)
|
|
twin = nid_names.compute_nid("sceTwinCall")
|
|
self.assertEqual(stubs, {
|
|
("libSceTwin.prx", twin): "sceTwinCall",
|
|
("libSceTwin.prx", nid_names.compute_nid("printf")): "printf_nid_postfix",
|
|
("libSceOnly.prx", nid_names.compute_nid("sceOnlyPending")): "sceOnlyPending"})
|
|
self.assertNotIn(("libSceTwin.native.prx", twin), stubs)
|
|
self.assertNotEqual(nid_names.compute_nid("printf"), nid_names.compute_nid("printf_nid_postfix"))
|
|
|
|
def test_stub_in_one_library_is_not_hidden_by_another(self):
|
|
source = self.work.source
|
|
write_source(source, "libA", DONE_SOURCE.format(name="sceTwinCall"))
|
|
write_source(source, "libB", STUB_SOURCE.format(name="sceTwinCall"))
|
|
nid = nid_names.compute_nid("sceTwinCall")
|
|
stubs = import_audit.source_stubs(source)
|
|
exports = {nid: ["libA.prx", "libB.prx"]}
|
|
records, _ = import_audit.audit([(nid, "libB.prx")], {nid: ["libB.prx"]}, stubs, {"libB.prx"}, set())
|
|
self.assertEqual((records[0]["class"], records[0]["name"]), ("stub", "sceTwinCall"))
|
|
records, _ = import_audit.audit([(nid, "libA.prx")], {nid: ["libA.prx"]}, stubs, {"libA.prx"}, set())
|
|
self.assertEqual((records[0]["class"], records[0]["name"]), ("implemented", None))
|
|
records, _ = import_audit.audit([(nid, "libB.prx")], exports, stubs, {"libA.prx", "libB.prx"}, set())
|
|
self.assertEqual(records[0]["class"], "implemented")
|
|
|
|
def test_formatted_export_aliases_remain_stubs(self):
|
|
work = self.work
|
|
nid = "AAAAAAAAAAA"
|
|
(work.libs / "libSceAlias.prx").write_bytes(exporting(nid))
|
|
registry = work.root / "game.json"
|
|
result = work.root / "audit.json"
|
|
write_registry(registry, [(nid, "libSceAlias.prx")])
|
|
aliases = [
|
|
'APS5_EXPORT (\n "AAAAAAAAAAA",\n NAME\n);\n',
|
|
'APS5_EXPORT(/* nid */ "AAAAAAAAAAA", /* name */ NAME);\n',
|
|
'APS5_EX\\\nPORT("AAAAAAAAAAA", NAME);\n',
|
|
]
|
|
for name in ("sceAliasUnknown", "sceAliasedCall"):
|
|
for alias in aliases:
|
|
with self.subTest(name=name, alias=alias):
|
|
write_source(work.source, "libSceAlias", alias.replace("NAME", name) + STUB_SOURCE.format(name=name))
|
|
code, _, err = work.run(registry, "--json", str(result))
|
|
self.assertEqual((code, err), (0, ""))
|
|
record, = json.loads(result.read_text())["imports"]
|
|
self.assertEqual((record["class"], record["name"]), ("stub", name))
|
|
|
|
def test_comments_and_literals_do_not_override_export_aliases(self):
|
|
work = self.work
|
|
nid, name = "AAAAAAAAAAA", "sceAliasUnknown"
|
|
(work.libs / "libSceAlias.prx").write_bytes(exporting(nid))
|
|
registry = work.root / "game.json"
|
|
result = work.root / "audit.json"
|
|
write_registry(registry, [(nid, "libSceAlias.prx")])
|
|
source = 'APS5_EXPORT("AAAAAAAAAAA", sceAliasUnknown);\n' + STUB_SOURCE.format(name=name)
|
|
decoys = [
|
|
'// APS5_EXPORT("BBBBBBBBBBB", sceAliasUnknown);\n',
|
|
'/* APS5_EXPORT("BBBBBBBBBBB", sceAliasUnknown); */\n',
|
|
'auto example = R"text(APS5_EXPORT("BBBBBBBBBBB", sceAliasUnknown);)text";\n',
|
|
]
|
|
for decoy in decoys:
|
|
with self.subTest(decoy=decoy):
|
|
write_source(work.source, "libSceAlias", source + decoy)
|
|
code, _, err = work.run(registry, "--json", str(result))
|
|
self.assertEqual((code, err), (0, ""))
|
|
record, = json.loads(result.read_text())["imports"]
|
|
self.assertEqual((record["class"], record["name"]), ("stub", name))
|
|
|
|
@unittest.skipUnless((import_audit.ROOT / ".git").exists(), "needs a git checkout")
|
|
def test_json_is_stamped_with_the_source_commit(self):
|
|
work = self.work
|
|
import_audit.subprocess.run(["git", "-C", str(work.source), "init", "-q"], check=True)
|
|
import_audit.subprocess.run(["git", "-C", str(work.source), "-c", "user.name=Audit test", "-c", "user.email=audit@example.invalid", "commit", "--allow-empty", "-qm", "Source stamp fixture"], check=True)
|
|
commit, dirty = import_audit.git_stamp(work.source)
|
|
(work.libs / "libSceA.prx").write_bytes(exporting("AAAAAAAAAAA"))
|
|
registry = work.root / "game.json"
|
|
write_registry(registry, [("AAAAAAAAAAA", "libSceA.prx")])
|
|
result = work.root / "stamped.json"
|
|
work.run(registry, "--json", str(result))
|
|
data = json.loads(result.read_text())
|
|
self.assertRegex(data["source_commit"], r"^[0-9a-f]{40}$")
|
|
self.assertEqual((data["source_commit"], data["source_dirty"]), (commit, dirty))
|
|
self.assertEqual(data["registries"], [str(registry)])
|
|
|
|
def test_nid_anchor_matches_the_known_pair(self):
|
|
self.assertEqual(nid_names.compute_nid(MESSAGE_NAME), MESSAGE_NID)
|
|
|
|
def test_elf_reader(self):
|
|
image = build_elf([("one", 1, GLOBAL_FUNCTION), ("weak", 1, 0x22), ("undefined", 0, GLOBAL_FUNCTION), ("local", 1, LOCAL_FUNCTION)])
|
|
path = Path("lib.prx")
|
|
self.assertEqual(import_audit.elf_exports(path, image), {"one", "weak"})
|
|
self.assertEqual(import_audit.elf_exports(path, exporting()), set())
|
|
for cut in (10, 64, len(image) - 5):
|
|
with self.assertRaisesRegex(import_audit.AuditError, "lib.prx"):
|
|
import_audit.elf_exports(path, image[:cut])
|
|
|
|
def test_elf_imports_reader(self):
|
|
image = build_elf([("one", 1, GLOBAL_FUNCTION), ("two", 0, GLOBAL_FUNCTION), ("three", 0, 0x22)], ["libc.prx", "libSceZip.prx"])
|
|
path = Path("mod.prx")
|
|
self.assertEqual(import_audit.elf_imports(path, image), ({"two", "three"}, {"libc.prx", "libSceZip.prx"}))
|
|
self.assertEqual(import_audit.elf_imports(path, exporting()), (set(), set()))
|
|
for cut in (10, 64, len(image) - 5):
|
|
with self.assertRaisesRegex(import_audit.AuditError, "mod.prx"):
|
|
import_audit.elf_imports(path, image[:cut])
|
|
|
|
def test_pe_reader(self):
|
|
path = Path("lib.prx")
|
|
self.assertEqual(import_audit.pe_exports(path, build_pe(["alpha", "beta"])), {"alpha", "beta"})
|
|
self.assertEqual(import_audit.pe_exports(path, build_pe([])), set())
|
|
image = build_pe(["alpha"])
|
|
for cut in (0x3c, 0x60, 0x210):
|
|
with self.assertRaisesRegex(import_audit.AuditError, "lib.prx"):
|
|
import_audit.pe_exports(path, image[:cut])
|
|
|
|
def test_pe_ignores_bytes_after_the_declared_directory_count(self):
|
|
image = bytearray(build_pe(["alpha"]))
|
|
struct.pack_into("<I", image, 0x58 + 108, 0)
|
|
with self.assertRaises(import_audit.AuditError):
|
|
import_audit.pe_exports(Path("lib.prx"), image)
|
|
|
|
def test_pe_null_export_address_is_not_an_export(self):
|
|
image = bytearray(build_pe(["alpha"]))
|
|
struct.pack_into("<I", image, 0x200 + 40, 0)
|
|
self.assertEqual(import_audit.pe_exports(Path("lib.prx"), image), set())
|
|
|
|
def test_pe_export_ordinal_must_index_the_address_table(self):
|
|
image = bytearray(build_pe(["alpha"]))
|
|
struct.pack_into("<H", image, 0x200 + 48, 1)
|
|
with self.assertRaises(import_audit.AuditError):
|
|
import_audit.pe_exports(Path("lib.prx"), image)
|
|
|
|
def test_pe_export_name_cannot_terminate_in_unmapped_overlay(self):
|
|
image = bytearray(build_pe(["alpha"]))
|
|
image[0x200 + 55] = ord("x")
|
|
image[0x238:] = b"y" * (len(image) - 0x238)
|
|
image += b"\0"
|
|
with self.assertRaises(import_audit.AuditError):
|
|
import_audit.pe_exports(Path("lib.prx"), image)
|
|
|
|
def test_pe_exports_can_reside_in_headers(self):
|
|
image = bytearray(build_pe(["alpha"]))
|
|
raw = bytes(image[0x200:0x238])
|
|
image[0x200:0x200] = bytes(0x200)
|
|
image[0x200:0x238] = raw
|
|
struct.pack_into("<I", image, 0x58 + 60, 0x400)
|
|
struct.pack_into("<I", image, 0x58 + 112, 0x200)
|
|
struct.pack_into("<III", image, 0x200 + 28, 0x228, 0x22c, 0x230)
|
|
struct.pack_into("<I", image, 0x22c, 0x232)
|
|
struct.pack_into("<I", image, 0x58 + 240 + 20, 0x400)
|
|
struct.pack_into("<I", image, 0x58 + 280 + 20, 0x438)
|
|
self.assertEqual(import_audit.pe_exports(Path("lib.prx"), image), {"alpha"})
|
|
|
|
def test_elf_extended_section_count_keeps_exports(self):
|
|
image = bytearray(exporting("alpha"))
|
|
section_offset = struct.unpack_from("<Q", image, 40)[0]
|
|
image.extend(bytes((0xff00 - 3) * 64))
|
|
struct.pack_into("<H", image, 60, 0)
|
|
struct.pack_into("<Q", image, section_offset + 32, 0xff00)
|
|
self.assertEqual(import_audit.elf_exports(Path("lib.prx"), image), {"alpha"})
|
|
|
|
def test_elf_extended_undefined_index_is_not_an_export(self):
|
|
image = bytearray(exporting("alpha"))
|
|
section_offset = struct.unpack_from("<Q", image, 40)[0]
|
|
symbol_offset = struct.unpack_from("<Q", image, section_offset + 64 + 24)[0]
|
|
index_offset = len(image) + 64
|
|
image.extend(struct.pack("<IIQQQQIIQQ", 0, 18, 0, 0, index_offset, 8, 1, 0, 4, 4))
|
|
image.extend(struct.pack("<II", 0, 0))
|
|
struct.pack_into("<H", image, 60, 4)
|
|
struct.pack_into("<H", image, symbol_offset + 24 + 6, 0xffff)
|
|
self.assertEqual(import_audit.elf_exports(Path("lib.prx"), image), set())
|
|
struct.pack_into("<I", image, index_offset + 4, 1)
|
|
self.assertEqual(import_audit.elf_exports(Path("lib.prx"), image), {"alpha"})
|
|
|
|
def test_pe_export_aliases_share_an_address(self):
|
|
image = bytearray(build_pe(["alpha", "beta"]))
|
|
struct.pack_into("<H", image, 0x200 + 56 + 2, 0)
|
|
self.assertEqual(import_audit.pe_exports(Path("lib.prx"), image), {"alpha", "beta"})
|
|
|
|
def test_pe32_uses_its_own_directory_layout(self):
|
|
image = bytearray(build_pe(["alpha"]))
|
|
struct.pack_into("<H", image, 0x58, 0x10b)
|
|
struct.pack_into("<I", image, 0x58 + 92, 16)
|
|
struct.pack_into("<II", image, 0x58 + 96, 0x1000, 56)
|
|
self.assertEqual(import_audit.pe_exports(Path("lib.prx"), image), {"alpha"})
|
|
|
|
|
|
def test_elf_unnamed_defined_symbol_is_not_an_export(self):
|
|
image = bytearray(exporting("alpha"))
|
|
section_offset = struct.unpack_from("<Q", image, 40)[0]
|
|
symbol_offset = struct.unpack_from("<Q", image, section_offset + 64 + 24)[0]
|
|
struct.pack_into("<I", image, symbol_offset + 24, 0)
|
|
self.assertEqual(import_audit.elf_exports(Path("lib.prx"), image), set())
|
|
|
|
def test_elf_preserves_non_utf8_symbol_bytes(self):
|
|
image = bytearray(exporting("alpha"))
|
|
image[65] = 0xff
|
|
self.assertEqual(import_audit.elf_exports(Path("lib.prx"), image), {"\udcfflpha"})
|
|
|
|
def test_elf_rejects_missing_regular_symbol_section(self):
|
|
image = build_elf([("alpha", 3, GLOBAL_FUNCTION)])
|
|
with self.assertRaisesRegex(import_audit.AuditError, "section index"):
|
|
import_audit.elf_exports(Path("lib.prx"), image)
|
|
self.assertEqual(import_audit.elf_exports(Path("lib.prx"), build_elf([("absolute", 0xfff1, GLOBAL_FUNCTION)])), {"absolute"})
|
|
|
|
def test_pe_forwarder_cannot_end_outside_export_directory(self):
|
|
image = bytearray(build_pe(["alpha"]))
|
|
struct.pack_into("<I", image, 0x200 + 40, 0x1032)
|
|
struct.pack_into("<II", image, 0x58 + 112, 0x1000, 55)
|
|
with self.assertRaisesRegex(import_audit.AuditError, "forwarder"):
|
|
import_audit.pe_exports(Path("lib.prx"), image)
|
|
|
|
def test_pe_forwarder_inside_export_directory_keeps_public_name(self):
|
|
image = bytearray(build_pe(["alpha"]))
|
|
target = b"NTDLL.#27\0"
|
|
image[0x238:0x238] = target
|
|
struct.pack_into("<I", image, 0x200 + 40, 0x1038)
|
|
struct.pack_into("<II", image, 0x58 + 112, 0x1000, 56 + len(target))
|
|
struct.pack_into("<I", image, 0x58 + 240 + 8, 56 + len(target))
|
|
struct.pack_into("<I", image, 0x58 + 240 + 16, 56 + len(target))
|
|
struct.pack_into("<I", image, 0x58 + 280 + 20, 0x238 + len(target))
|
|
self.assertEqual(import_audit.pe_exports(Path("lib.prx"), image), {"alpha"})
|
|
|
|
def test_pe_export_name_pointer_table_must_be_sorted(self):
|
|
image = build_pe(["beta", "alpha"])
|
|
with self.assertRaisesRegex(import_audit.AuditError, "sorted"):
|
|
import_audit.pe_exports(Path("lib.prx"), image)
|
|
|
|
def test_pe_export_name_must_be_ascii(self):
|
|
image = bytearray(build_pe(["alpha"]))
|
|
image[0x200 + 50:0x200 + 52] = b"\xc3\xa9"
|
|
with self.assertRaisesRegex(import_audit.AuditError, "ASCII"):
|
|
import_audit.pe_exports(Path("lib.prx"), image)
|
|
|
|
def test_libraries_with_conflicting_basenames_are_not_merged(self):
|
|
first, second = self.work.libs, self.work.root / "other-libs"
|
|
second.mkdir()
|
|
(first / "libSceA.prx").write_bytes(exporting("AAAAAAAAAAA"))
|
|
(second / "libSceA.prx").write_bytes(exporting("BBBBBBBBBBB"))
|
|
with self.assertRaisesRegex(import_audit.AuditError, "ambiguous"):
|
|
import_audit.built_libraries([first, second])
|
|
(second / "libSceA.prx").write_bytes(exporting("AAAAAAAAAAA"))
|
|
self.assertEqual(import_audit.built_libraries([first, second, first]), ({"AAAAAAAAAAA": ["libSceA.prx"]}, {"libSceA.prx"}))
|
|
|
|
def test_json_output_cannot_overwrite_registry_or_names(self):
|
|
work = self.work
|
|
(work.libs / "libSceA.prx").write_bytes(exporting("AAAAAAAAAAA"))
|
|
registry = work.root / "game.json"
|
|
write_registry(registry, [("AAAAAAAAAAA", "libSceA.prx")])
|
|
before = registry.read_bytes()
|
|
code, out, err = work.run(registry, "--json", str(registry))
|
|
self.assertEqual((code, out), (2, ""))
|
|
self.assertIn("input", err)
|
|
self.assertEqual(registry.read_bytes(), before)
|
|
names = work.root / "names.csv"
|
|
names.write_text("AAAAAAAAAAA name\n")
|
|
code, out, err = work.run(registry, "--json", str(names), "--names", str(names))
|
|
self.assertEqual((code, out), (2, ""))
|
|
self.assertEqual(names.read_text(), "AAAAAAAAAAA name\n")
|
|
|
|
def test_json_output_cannot_overwrite_hard_linked_input(self):
|
|
work = self.work
|
|
(work.libs / "libSceA.prx").write_bytes(exporting("AAAAAAAAAAA"))
|
|
registry = work.root / "game.json"
|
|
write_registry(registry, [("AAAAAAAAAAA", "libSceA.prx")])
|
|
output = work.root / "hard-link.json"
|
|
os.link(registry, output)
|
|
before = registry.read_bytes()
|
|
code, out, err = work.run(registry, "--json", str(output))
|
|
self.assertEqual((code, out), (2, ""))
|
|
self.assertIn("input", err)
|
|
self.assertEqual(registry.read_bytes(), before)
|
|
|
|
def test_git_stamp_uses_selected_source_tree(self):
|
|
completed = type("Completed", (), {"stdout": "abcdef\n"})()
|
|
with patch.object(import_audit.subprocess, "run", return_value=completed) as run:
|
|
self.assertEqual(import_audit.git_stamp(self.work.source), ("abcdef", True))
|
|
self.assertEqual([call.args[0][2] for call in run.call_args_list], [str(self.work.source), str(self.work.source)])
|
|
|
|
def test_library_directories(self):
|
|
work = self.work
|
|
(work.libs / "libSceA.prx").write_bytes(build_pe(["AAAAAAAAAAA"]))
|
|
(work.libs / "libSceB.prx").write_bytes(exporting("AAAAAAAAAAA", "BBBBBBBBBBB"))
|
|
exports, libraries = import_audit.built_libraries([work.libs])
|
|
self.assertEqual(exports, {"AAAAAAAAAAA": ["libSceA.prx", "libSceB.prx"], "BBBBBBBBBBB": ["libSceB.prx"]})
|
|
self.assertEqual(libraries, {"libSceA.prx", "libSceB.prx"})
|
|
(work.libs / "libSceC.prx").write_bytes(b"text")
|
|
with self.assertRaisesRegex(import_audit.AuditError, r"libSceC.prx: neither ELF nor PE"):
|
|
import_audit.built_libraries([work.libs])
|
|
with self.assertRaisesRegex(import_audit.AuditError, "no .prx files"):
|
|
import_audit.built_libraries([work.modules])
|
|
with self.assertRaisesRegex(import_audit.AuditError, "not a directory"):
|
|
import_audit.built_libraries([work.root / "absent"])
|
|
|
|
def test_registry_errors(self):
|
|
path = self.work.root / "registry.json"
|
|
path.write_text("{}")
|
|
with self.assertRaisesRegex(import_audit.AuditError, "JSON list"):
|
|
import_audit.read_registry(path)
|
|
path.write_text("[1")
|
|
with self.assertRaisesRegex(import_audit.AuditError, "cannot read registry"):
|
|
import_audit.read_registry(path)
|
|
path.write_bytes(b"\xff\xfe\x00")
|
|
with self.assertRaisesRegex(import_audit.AuditError, "cannot read registry"):
|
|
import_audit.read_registry(path)
|
|
path.write_text('[{"nid": "AAAAAAAAAAA"}]')
|
|
with self.assertRaisesRegex(import_audit.AuditError, r"entry 0 needs string fields"):
|
|
import_audit.read_registry(path)
|
|
write_registry(path, [("AAAAAAAAAAA", ""), ("bad\x01nid", "")])
|
|
with self.assertRaisesRegex(import_audit.AuditError, r"entry 1 has an invalid NID 'bad\\x01nid'"):
|
|
import_audit.read_registry(path)
|
|
write_registry(path, [("AAAAAAAAAAA#A#B", "libSceA.prx")], encoding="utf-8-sig")
|
|
self.assertEqual(import_audit.read_registry(path), [("AAAAAAAAAAA", "libSceA.prx")])
|
|
|
|
def test_names_never_download(self):
|
|
work = self.work
|
|
(work.libs / "libSceA.prx").write_bytes(exporting("AAAAAAAAAAA"))
|
|
registry = work.root / "game.json"
|
|
write_registry(registry, [(MESSAGE_NID, "libSceA.prx"), ("DDDDDDDDDDD", "libSceA.prx")])
|
|
with patch.object(nid_names.urllib.request, "urlretrieve", side_effect=AssertionError("download attempted")):
|
|
code, _, err = work.run(registry, "--names", str(work.root / "missing.csv"))
|
|
self.assertEqual(code, 2)
|
|
self.assertIn("never downloads", err)
|
|
names = work.root / "aerolib.csv"
|
|
names.write_text(f"{MESSAGE_NID} {MESSAGE_NAME}\nDDDDDDDDDDD wrongname\n")
|
|
result = work.root / "named.json"
|
|
work.run(registry, "--names", str(names), "--json", str(result))
|
|
named = {record["nid"]: record for record in json.loads(result.read_text())["imports"]}
|
|
self.assertEqual((named[MESSAGE_NID]["name"], named[MESSAGE_NID]["name_verified"]), (MESSAGE_NAME, True))
|
|
self.assertEqual((named["DDDDDDDDDDD"]["name"], named["DDDDDDDDDDD"]["name_verified"]), ("wrongname", False))
|
|
|
|
def test_conservation_check_fires(self):
|
|
records = [{"references": 2, "class": "absent"}, {"references": 1, "class": "stub"}]
|
|
import_audit.check_conservation(3, records)
|
|
with self.assertRaisesRegex(import_audit.AuditError, "3 classified"):
|
|
import_audit.check_conservation(4, records)
|
|
with self.assertRaisesRegex(import_audit.AuditError, "unknown classes"):
|
|
import_audit.check_conservation(1, [{"references": 1, "class": "lost"}])
|
|
|
|
def test_errors_exit_two_with_the_failing_value(self):
|
|
work = self.work
|
|
(work.libs / "libSceA.prx").write_bytes(exporting("AAAAAAAAAAA"))
|
|
registry = work.root / "game.json"
|
|
write_registry(registry, [("short", "libSceA.prx")])
|
|
code, out, err = work.run(registry)
|
|
self.assertEqual((code, out), (2, ""))
|
|
self.assertIn("game.json: entry 0 has an invalid NID 'short'", err)
|
|
|
|
def test_unreadable_library_and_unwritable_json_exit_two(self):
|
|
work = self.work
|
|
library = work.libs / "libSceA.prx"
|
|
library.write_bytes(exporting("AAAAAAAAAAA"))
|
|
registry = work.root / "game.json"
|
|
write_registry(registry, [("AAAAAAAAAAA", "libSceA.prx")])
|
|
code, out, err = work.run(registry, "--json", str(work.modules))
|
|
self.assertEqual((code, out), (2, ""))
|
|
self.assertIn("FAIL: ", err)
|
|
self.assertIn(str(work.modules), err)
|
|
denied = PermissionError(13, "Permission denied", str(library))
|
|
with patch.object(Path, "read_bytes", side_effect=denied):
|
|
code, out, err = work.run(registry)
|
|
self.assertEqual((code, out), (2, ""))
|
|
self.assertIn(f"FAIL: Permission denied: {library}", err)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|