initial commit

src/archipelago_tester/pipeline/batch/runner.py

@@ -0,0 +1,86 @@
+"""Running the tests the cache could not answer."""
+
+import threading
+from concurrent.futures import ThreadPoolExecutor, as_completed
+
+from archipelago_tester.pipeline.batch.record import TestRecord
+from archipelago_tester.pipeline.generation.tester import WorldTester
+
+
+class Runner:
+    """Tests each outstanding world and persists what it did."""
+
+    def __init__(self, config, paths, store, options, state, tests, tag,
+                 image, roms_fingerprint):
+        self.config = config
+        self.paths = paths
+        self.store = store
+        self.options = options
+        self.state = state
+        self.tests = tests
+        self.tag = tag
+        self.tester = WorldTester(config, paths, image)
+        self.roms_fingerprint = roms_fingerprint
+        self.lock = threading.Lock()
+
+    @staticmethod
+    def promotion_name(report, game_name):
+        """The name Archipelago itself registered this world under.
+
+        Recorded for promotion's collision check, while the report's
+        own "game" becomes the sheet folder name for display. A world
+        that never loaded has no real name, so it falls back to a
+        marked sheet name - otherwise every such entry would collide
+        under one None key and none would promote.
+        """
+        return report.get("game") or f"unverified:{game_name}"
+
+    def store_record(self, key, record):
+        """Persist after every test, not just at the end, so a killed
+        run does not lose the tests it already completed.
+        """
+        with self.lock:
+            self.tests[key] = record
+            self.store.save(self.state)
+
+    def test_one(self, game_name, apworld_path, key, current_hash):
+        options = self.options
+        report = self.tester.test(
+            apworld_path=apworld_path,
+            game=None if apworld_path else game_name,
+            output_dir=options["output_directory"],
+            timeout=options["timeout"],
+            companion_range=options["companion_range"],
+            spoiler=options["spoiler"],
+            repeats=options["repeats"],
+            companion_pool=options["pool"],
+            random_repeats=options["random_repeats"],
+        )
+        record = TestRecord(game_name, key,
+                            is_core=apworld_path is None)
+        self.store_record(key, record.from_report(
+            report=report,
+            current_hash=current_hash,
+            tag=self.tag,
+            internal_game=self.promotion_name(report, game_name),
+            roms_fingerprint=self.roms_fingerprint,
+        ))
+        report["game"] = game_name
+        report["apworld"] = key
+        report["skipped"] = False
+        return report
+
+    def run(self, to_test, total, completed, on_progress):
+        """Run the outstanding tests, reporting as they land."""
+        results = []
+        reporting = threading.Lock()
+        with ThreadPoolExecutor(max_workers=self.options["jobs"]) as pool:
+            futures = [pool.submit(self.test_one, *args) for args in to_test]
+            for future in as_completed(futures):
+                report = future.result()
+                results.append(report)
+                with reporting:
+                    completed += 1
+                    if on_progress:
+                        on_progress(completed, total, report)
+        return results