[patched] test - unpack to temporary file

.travis.yml

@@ -8,7 +8,7 @@ matrix:
     dist: xenial          # required for Python >= 3.7
     cache: pip
     install:
-        - pip install --upgrade pip setuptools pytest wheel twine
+        - pip install --upgrade pip setuptools pytest-xdist wheel twine
         - pip install -e .
     deploy:
       provider: pypi
@@ -24,18 +24,18 @@ matrix:
     osx_image: xcode10.2  # Python 3.7.2 running on macOS 10.14.3
     language: shell       # 'language: python' is an error on Travis CI macOS
     install:
-        - pip3 install --upgrade pip setuptools pytest wheel twine
+        - pip3 install --upgrade pip setuptools pytest-xdist wheel twine
         - pip3 install -e .
   - name: Python 3.7.3 on Windows
     os: windows           # Windows 10.0.17134 N/A Build 17134
     language: shell       # 'language: python' is an error on Travis CI Windows
     before_install: choco install python make
     env: PATH=/c/Python37:/c/Python37/Scripts:$PATH
     install:
-        - pip3 install --user --upgrade pip setuptools pytest wheel twine
+        - pip3 install --user --upgrade pip setuptools pytest-xdist wheel twine
         - pip3 install --user -e .
   allow_failures:
   - os: windows
 
 script:
-- python3 -m pytest -s || python -m pytest -s
+- python3 -m pytest -s -n $((($(getconf _NPROCESSORS_ONLN)+1)/2)) || python -m pytest -s -n $((($(getconf _NPROCESSORS_ONLN)+1)/2))

Tests/test_unpacker.py

@@ -3,13 +3,14 @@
 import os
 import sys
 import threading
+from pathlib import Path
+from tempfile import TemporaryDirectory
 from unittest import TestCase
 
 from colorama import Fore
+
 from unipacker.core import Sample, SimpleClient, UnpackerEngine
-from unipacker.unpackers import get_unpacker
 from unipacker.utils import RepeatedTimer
-from pathlib import Path
 
 
 def calc_md5(sample):
@@ -74,29 +75,30 @@ def test_integrity(self):
                                         f"Tested file: {relative_test_path}. Expected: {self.hashes[relative_test_path]}, got: {calc_md5(test_path).hexdigest()}")
                         print(f"Tested:{relative_test_path}, MD5: {calc_md5(test_path).hexdigest()}")
 
+
 def _unpack(t):
-    file, unpacked_file = t
-    unpacked = file + '.unpacked'
-    # if os.path.exists(unpacked):
-    #     os.remove(unpacked)
-    sample = Sample(file)
-    event = threading.Event()
-    client = SimpleClient(event)
-    heartbeat = RepeatedTimer(120, print, "- still running -", file=sys.stderr)
-
-    engine = UnpackerEngine(sample, unpacked)
-    engine.register_client(client)
-    heartbeat.start()
-    threading.Thread(target=engine.emu).start()
-    event.wait()
-    heartbeat.stop()
-    engine.stop()
-    assert os.path.exists(unpacked)
-    assert not os.path.exists(sample.unpacker.dumper.brokenimport_dump_file)
-    if os.path.exists(unpacked):
-        return file, calc_md5(unpacked).hexdigest(), calc_md5(unpacked_file).hexdigest()
-    else:
-        return file, '', calc_md5(unpacked_file) 
+    with TemporaryDirectory() as unpack_dir:
+        file, unpacked_file = t
+        unpacked = f"{unpack_dir}/unpacked.exe"
+        sample = Sample(file)
+        event = threading.Event()
+        client = SimpleClient(event)
+        heartbeat = RepeatedTimer(120, print, "- still running -", file=sys.stderr)
+
+        engine = UnpackerEngine(sample, unpacked)
+        engine.register_client(client)
+        heartbeat.start()
+        threading.Thread(target=engine.emu).start()
+        event.wait()
+        heartbeat.stop()
+        engine.stop()
+        assert os.path.exists(unpacked)
+        assert not os.path.exists(sample.unpacker.dumper.brokenimport_dump_file)
+        if os.path.exists(unpacked):
+            return file, calc_md5(unpacked).hexdigest(), calc_md5(unpacked_file).hexdigest()
+        else:
+            return file, '', calc_md5(unpacked_file)
+
 
 class EngineTest(TestCase):
 
@@ -114,7 +116,7 @@ def perform_test(self, packer, ignore):
         tuples = list(zip(files, files_unpacked))
 
         with multiprocessing.Pool(
-            processes=min(multiprocessing.cpu_count(), 12)) as pool:
+                processes=min(multiprocessing.cpu_count(), 12)) as pool:
             for file, new_md5, old_md5 in pool.imap_unordered(_unpack, tuples):
                 hashes.append((file, old_md5, new_md5))