#!/usr/bin/env python3 # Copyright (c) 2019-2026 Andreas T Jonsson # # This software is provided 'as-is', without any express or implied # warranty. In no event will the authors be held liable for any damages # arising from the use of this software. # # Permission is granted to anyone to use this software for any purpose, # including commercial applications, and to alter it and redistribute it # freely, subject to the following restrictions: # # 1. The origin of this software must not be misrepresented; you must not # claim that you wrote the original software. If you use this software # in a product, an acknowledgment in the product documentation would be # appreciated but is not required. # # 2. Altered source versions must be plainly marked as such, and must not be # misrepresented as being the original software. # # 3. This notice may not be removed or altered from any source # distribution. """fetch.py -- Download 8088 CPU test data (MOO binary files). Downloads metadata.json from the 8088 test repository, then fetches and decompresses .MOO.gz binary test files for each opcode. Already-downloaded files are skipped (cached). A masks.txt file is generated alongside the MOO files containing per-opcode flag masks extracted from the metadata. Usage: python3 tools/testdata/fetch.py """ import gzip import json import os import shutil import subprocess import sys # --------------------------------------------------------------------------- # Configuration # --------------------------------------------------------------------------- TESTDATA_DIR = "src/tests/testdata" TESTDATA_V20_DIR = os.path.join(TESTDATA_DIR, "v20") GIT_CACHE_DIR = os.environ.get("TESTDATA_CACHE", "tools/testdata/.cache") # Test repositories are hosted on Codeberg. # 8088 tests REPO_8088 = "https://codeberg.org/virtualxt/8088" META_PATH_8088 = "v2/metadata.json" TEST_PATH_8088 = "v2_binary" # V20 tests (used to verify 186 instructions). REPO_V20 = "https://codeberg.org/virtualxt/v20" META_PATH_V20 = "v1_native/metadata.json" TEST_PATH_V20 = "v1_native_binary" DEFAULT_FLAG_MASK = 0x00D5 # VXT_VALID_LOW_FLAGS # --------------------------------------------------------------------------- # Skip lists # --------------------------------------------------------------------------- # Simple opcodes to skip entirely. # 0x0F (POP CS): implemented in cpu.c but no upstream MOO test data available. SKIP_OPCODES = {0x0F, 0x9B, 0xF4} # (opcode, reg) tuples to skip. # FE /2-/7 are implemented in cpu.c as undocumented aliases of FF /2-/7. SKIP_TUPLES = { (0xFE, 2), (0xFE, 3), (0xFE, 4), (0xFE, 5), (0xFE, 6), (0xFE, 7), } # Opcodes whose metadata has a "reg" dict but whose MOO files on the server # are stored as flat (non-register) files (e.g. 8F.MOO.gz, not 8F.0.MOO.gz). FLATTEN_OPCODES = {"8F", "C6", "C7"} # Statuses that mean "not a real instruction test". SKIP_STATUSES = {"prefix", "fpu"} # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def git_clone_or_update(repo_url, dest_dir, sparse_paths): """Clone a repo with sparse checkout, or pull if it already exists.""" if os.path.exists(dest_dir): print(f" Updating {repo_url}") result = subprocess.run( ["git", "-C", dest_dir, "pull", "--ff-only"], capture_output=True, text=True ) if result.returncode != 0: print(f" WARNING: git pull failed: {result.stderr.strip()}", file=sys.stderr) # Re-apply sparse paths in case the repo was cloned without them # or the paths have changed. subprocess.run( ["git", "-C", dest_dir, "sparse-checkout", "set"] + sparse_paths, check=True, capture_output=True ) return os.makedirs(dest_dir, exist_ok=True) print(f" Cloning {repo_url}") subprocess.run( ["git", "clone", "--depth", "1", "--filter=blob:none", "--sparse", repo_url, dest_dir], check=True, capture_output=True ) subprocess.run( ["git", "-C", dest_dir, "sparse-checkout", "set"] + sparse_paths, check=True, capture_output=True ) def copy_and_decompress(src_gz, dest_moo): """Copy and decompress a .MOO.gz file to a .MOO file.""" try: with gzip.open(src_gz, "rb") as f_in, open(dest_moo, "wb") as f_out: f_out.write(f_in.read()) return True except Exception as e: print(f" ERROR decompressing {src_gz}: {e}", file=sys.stderr) if os.path.exists(dest_moo): os.remove(dest_moo) return False def fetch_moo(name, test_path, dest_dir): """Copy and decompress .MOO.gz into dest_dir/.MOO. Skips if the .MOO file already exists and is newer than the source .gz (cache hit). Re-extracts when the source is newer. Returns True on success or cache hit, False on failure. """ moo_path = os.path.join(dest_dir, f"{name}.MOO") gz_path = os.path.join(test_path, f"{name}.MOO.gz") if not os.path.exists(gz_path): print(f" ERROR: {gz_path} not found", file=sys.stderr) return False if os.path.exists(moo_path): # Cache hit only when the extracted file is newer than the source. if os.path.getmtime(moo_path) >= os.path.getmtime(gz_path): return True print(f" Re-extracting {name}.MOO.gz (source newer)") else: print(f" Extracting {name}.MOO.gz") return copy_and_decompress(gz_path, moo_path) def should_skip(op_val, reg, status): """Return True if this opcode/reg combination should be skipped.""" if status in SKIP_STATUSES: return True if op_val in SKIP_OPCODES: return True if reg is not None and (op_val, reg) in SKIP_TUPLES: return True return False # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- def process_metadata(metadata, test_path, dest_dir, masks, arch_filter=None): """Process metadata and extract MOO files from local git clone. arch_filter: if not None, only fetch opcodes whose arch field is in this set (e.g. {"86", "186"}). Used to select V20-compatible tests. Returns (downloaded, cached, errors) counters. """ downloaded = 0 cached = 0 errors = 0 opcodes = metadata["opcodes"] for op_hex, op_data in sorted(opcodes.items()): op_val = int(op_hex, 16) # Some opcodes have a "reg" structure in metadata but their MOO # files are stored flat (without the register suffix). Collapse # them so they are treated as simple opcodes below. if op_hex in FLATTEN_OPCODES and "reg" in op_data: op_data = op_data["reg"]["0"] if "reg" in op_data: # Opcode has sub-register variants (e.g. FF.0, FF.1, ...). for reg_key, reg_data in sorted(op_data["reg"].items(), key=lambda kv: int(kv[0])): reg_num = int(reg_key) status = reg_data.get("status", "") if should_skip(op_val, reg_num, status): continue # Apply arch filter if specified. if arch_filter is not None: arch = reg_data.get("arch", "") if arch not in arch_filter: continue name = f"{op_hex}.{reg_key}" mask = reg_data.get("flags-mask", DEFAULT_FLAG_MASK) was_cached = os.path.exists( os.path.join(dest_dir, f"{name}.MOO")) if fetch_moo(name, test_path, dest_dir): masks.append((f"{name}.MOO", int(mask))) if was_cached: cached += 1 else: downloaded += 1 else: errors += 1 else: # Simple opcode (no sub-registers). status = op_data.get("status", "") if should_skip(op_val, None, status): continue # Apply arch filter if specified. if arch_filter is not None: arch = op_data.get("arch", "") if arch not in arch_filter: continue mask = op_data.get("flags-mask", DEFAULT_FLAG_MASK) was_cached = os.path.exists( os.path.join(dest_dir, f"{op_hex}.MOO")) if fetch_moo(op_hex, test_path, dest_dir): masks.append((f"{op_hex}.MOO", int(mask))) if was_cached: cached += 1 else: downloaded += 1 else: errors += 1 return downloaded, cached, errors def fetch_suite(repo_url, meta_path_in_repo, test_path_in_repo, dest_dir, arch_filter=None): """Clone repo and extract metadata and MOO files for one test suite. arch_filter: if not None, only fetch opcodes whose arch field is in this set. Returns (downloaded, cached, errors, masks). """ os.makedirs(dest_dir, exist_ok=True) repo_name = repo_url.rstrip("/").split("/")[-1] cache_dir = os.path.join(GIT_CACHE_DIR, repo_name) sparse_paths = [meta_path_in_repo.split("/")[0], test_path_in_repo] suite_name = str(arch_filter) if arch_filter else "8088" print(f"=== Cloning/updating {suite_name} repo ===") git_clone_or_update(repo_url, cache_dir, sparse_paths) meta_src = os.path.join(cache_dir, meta_path_in_repo) meta_path = os.path.join(dest_dir, "metadata.json") print(f"=== Copying {suite_name} metadata ===") if not os.path.exists(meta_src): print(f"ERROR: could not find {meta_src}", file=sys.stderr) return 0, 0, 1, [] shutil.copy2(meta_src, meta_path) with open(meta_path, "r") as f: metadata = json.load(f) test_path = os.path.join(cache_dir, test_path_in_repo) masks = [] print(f"=== Fetching {suite_name} test files ===") d, c, e = process_metadata(metadata, test_path, dest_dir, masks, arch_filter) return d, c, e, masks def write_masks(masks, dest_dir): masks_path = os.path.join(dest_dir, "masks.txt") with open(masks_path, "w") as f: for filename, mask in sorted(masks): f.write(f"{filename} {mask}\n") return masks_path def main(): # Fetch 8088 tests (all opcodes, no arch filtering). d1, c1, e1, masks1 = fetch_suite( REPO_8088, META_PATH_8088, TEST_PATH_8088, TESTDATA_DIR, arch_filter=None) # Fetch V20 tests: both 8088-compatible (arch: "86") and 186-specific # (arch: "186") opcodes. Skip V30-only ("v30") and 286 ("286") tests. d2, c2, e2, masks2 = fetch_suite( REPO_V20, META_PATH_V20, TEST_PATH_V20, TESTDATA_V20_DIR, arch_filter={"86", "186"}) # Write masks files. path1 = write_masks(masks1, TESTDATA_DIR) path2 = write_masks(masks2, TESTDATA_V20_DIR) total_d = d1 + d2 total_c = c1 + c2 total_e = e1 + e2 total_m = len(masks1) + len(masks2) print(f"\n=== Summary ===") print(f"8088: {d1} downloaded, {c1} cached, {e1} errors, {len(masks1)} total") print(f"V20: {d2} downloaded, {c2} cached, {e2} errors, {len(masks2)} total") print(f"Total: {total_d} downloaded, {total_c} cached, {total_e} errors, {total_m} total") print(f"Generated: {path1}") print(f"Generated: {path2}") return 1 if total_e > 0 else 0 if __name__ == "__main__": sys.exit(main())