diff --git a/.github/pull_request_template.md b/.github/pull_request_template.md new file mode 100644 index 0000000..3bba7d4 --- /dev/null +++ b/.github/pull_request_template.md @@ -0,0 +1,16 @@ +## What this changes + + + +## For a new or changed artifact + + + +- [ ] Ran the tool against a real extraction and confirmed the row counts, not just that it imports. +- [ ] Ran `python admin/scripts/check_artifact_output.py ` on that report and **fixed or documented every finding**. An empty or constant column is often a real result (no group chats, coarse location denied); the fix for those is to say so in the artifact's `notes`, which is also what stops the checker reporting them. +- [ ] Checked it against a second app data directory where the platform provides one (a second user, account, or container). `--compare ` reads the scaling for you: it should be exactly double. +- [ ] `notes`, `description` and `sample_data` say only what the data shows, and the numbers were re-derived from the finished run. + +## Anything reviewers should know + + diff --git a/.github/workflows/request_test_data.yml b/.github/workflows/request_test_data.yml new file mode 100644 index 0000000..318f681 --- /dev/null +++ b/.github/workflows/request_test_data.yml @@ -0,0 +1,42 @@ +name: Request Test Data + +# Asks external contributors for test data when a PR changes artifact modules +# without any. pull_request_target so the comment can be posted on fork PRs; +# safe here because the job checks out and runs only the base branch's code. +# The contributor's files are fetched as text by the script and never executed. + +on: + pull_request_target: + types: [opened, synchronize, reopened] + paths: + - 'scripts/artifacts/**' + +permissions: + contents: read + issues: write + pull-requests: write + +jobs: + request-test-data: + runs-on: ubuntu-latest + steps: + - name: Checkout base branch + uses: actions/checkout@v4 + with: + # Blob-less fetch; the script needs only admin/ sources, not the + # committed test fixtures under admin/test/cases/data. + sparse-checkout: | + /* + !/admin/test/cases/data + sparse-checkout-cone-mode: false + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: '3.x' + + - name: Check the PR for test data + env: + GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} + PR_NUMBER: ${{ github.event.pull_request.number }} + run: python admin/scripts/check_pr_test_data.py diff --git a/.github/workflows/test_cases.yml b/.github/workflows/test_cases.yml new file mode 100644 index 0000000..93620c1 --- /dev/null +++ b/.github/workflows/test_cases.yml @@ -0,0 +1,51 @@ +name: Test Cases + +# Runs every committed test case (admin/test/cases) and compares the output +# against the recorded baselines (admin/test/results). Baselines are ordinary +# committed files: a PR that deliberately changes a parser's output re-records +# the affected snapshot in the same PR, so the reviewer sees the row-level +# diff next to the code change. Units excluded from gating are listed with +# reasons in admin/test/cases/known_failures.json. +# +# This job needs the committed fixtures, so it does NOT use the sparse +# checkout the other PR workflows use. + +on: + pull_request: + paths: + - 'scripts/**' + - 'admin/test/cases/**' + - 'admin/test/results/**' + - 'admin/test/scripts/**' + - 'requirements.txt' + - '.github/workflows/test_cases.yml' + push: + branches: + - main + paths: + - 'scripts/**' + - 'admin/test/cases/**' + - 'admin/test/results/**' + - 'admin/test/scripts/**' + workflow_dispatch: + +permissions: + contents: read + +jobs: + test-cases: + runs-on: ubuntu-latest + steps: + - name: Checkout code + uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: '3.x' + + - name: Install runtime dependencies + run: python -m pip install -r requirements.txt + + - name: Run test cases against recorded baselines + run: python admin/test/scripts/run_test_cases.py diff --git a/admin/docs/testing/create_module_test_cases.md b/admin/docs/testing/create_module_test_cases.md new file mode 100644 index 0000000..9867732 --- /dev/null +++ b/admin/docs/testing/create_module_test_cases.md @@ -0,0 +1,191 @@ +# Creating Module Test Cases + +This document describes the process of creating test cases for LEAPP modules using the `make_test_data.py` script. + +## Overview + +The `make_test_data.py` script is designed to generate test data for LEAPP modules. It processes input files (zip, tar, or tar.gz) to extract relevant files based on the module's artifact patterns and creates structured test cases. + +## Usage + +To create test cases for a module, use one of the following commands: + +```bash +python make_test_data.py --image +python make_test_data.py --case --input +python make_test_data.py --image-prompt +``` + +Arguments: +- ``: Name of the module (e.g., keyboard or keyboard.py) +- `--image `: Name of the image from the manifest +- `--case `: Case number for the test data +- `--input `: Path to the input file (zip, tar, or tar.gz) +- `--image-prompt`: Prompt for image selection from the manifest + +## Process + +1. The script imports the specified module and retrieves artifact information. +2. It creates or updates a JSON file with test case metadata. +3. The script processes the input archive file, searching for files matching the artifact patterns. +4. For each artifact, it creates a zip file containing the matching files. +5. The JSON file is updated with information about the created test data. + +## Output + +The script generates the following outputs: + +1. A JSON file containing test case metadata. + - `admin/test/cases/testdata..json` +2. Zip files for each artifact, containing the relevant test data files. + - `admin/test/cases/data//testdata....zip` + +## Example + +```bash +python make_test_data.py keyboard --image ios_15_image +``` + +This command will create test data for the keyboard module, using the specified image from the manifest. + +## Notes + +- The script supports zip, tar, and tar.gz input files. +- Test data is stored in the `admin/test/cases/data` directory. +- JSON metadata files are stored in the `admin/test/cases` directory. +- Always review and update the generated JSON file with additional test case details as needed. + +## Test Case JSON File Structure + +The script generates a JSON file (e.g., `testdata..json`) that contains metadata and information about the test cases. This file is crucial for defining the scenarios that `test_module.py` will execute. Here's an explanation of its structure: + +```json +{ + "case_ios12_basic": { + "description": "Tests basic data extraction for iOS 12.x.", + "maker": "Your Name", + "os_name": "iOS", + "os_version": "12.5.5", + "make_data": { + "input_data_path": "/path/to/iOS12_image.tar.gz", + "os": "macOS-14.0-...", + "timestamp": "2024-10-14T10:17:49.432528", + "last_commit": { + "hash": "abcdef123...", + } + }, + "artifacts": { + "get_photosMetadata": { + "search_patterns": [ + "*/mobile/Media/PhotoData/Photos.sqlite*" + ], + "file_count": 1 + }, + "another_artifact_function_if_any": { + } + } + } +} +``` + +- `case_ios12_basic`: A unique identifier for each test case. This key is used by `test_module.py` to find the corresponding input data ZIP and golden file. + - `description`: A brief description of what this test case covers (to be filled in manually). + - `maker`: The person who created or last verified the test case (to be filled in manually). + - `os_name`: Include the name of the operating system this case data originated from (iOS, Android, etc) + - `os_version`: The specific OS version string (e.g., "12.5.5", "14.1") that this test case represents. `test_module.py` will use this to mock `iOS.get_version()`. + - `make_data`: Information about the `make_test_data.py` run that generated this case's input data. + - `input_data_path`: The path to the original full image/archive used. + - `os`: The operating system on which `make_test_data.py` was run. + - `timestamp`: The date and time when the input data ZIP was created. + - `last_commit`: Information about the module's Git commit at the time of data creation. + - `artifacts`: A dictionary where keys are artifact function names from the module. + - `artifact_name` (e.g., "get_photosMetadata"): + - `search_patterns`: The file patterns from the module's `__artifacts_v2__` block used to find relevant files. + - `file_count`: The number of files found and included in the input ZIP for this artifact and case. + +Multiple test cases (e.g., "case_ios12_basic", "case_ios14_advanced") can be included in a single `testdata..json` file. It's recommended to use descriptive case keys. + +## Image Manifest + +The script uses an `image_manifest.json` file located in the `admin` directory. This manifest contains information about available test images, including their names, descriptions, and local paths. + +## Known Issues and Limitations + +### Dynamic File Searching in Modules + +Some modules, such as `sms.py`, use a two-step process for file searching that can limit the effectiveness of automated test case creation: + +1. The module provides an initial search pattern to locate a primary database file. +2. After processing the database, the module performs additional file searches based on data extracted from the database. + +For example, the `sms.py` module: +1. First searches for the SMS database file. +2. Then searches for attachment files based on paths stored in the database. + +This approach presents challenges for automated test case creation: + +- The `make_test_data.py` script only uses the initial search patterns defined in the module. +- It cannot anticipate or include files that would be found by secondary searches within the module. + +#### Impact on Test Coverage + +This limitation may result in incomplete test data for modules that employ this dynamic searching technique. The created test cases might not include all the files that the module would process in a real-world scenario. + +#### Future Improvements + +To address this issue and improve test coverage, we need to consider the following approaches: + +1. Enhance the `make_test_data.py` script to simulate the module's dynamic file searching behavior. +2. Modify the module structure to separate file searching from data processing, allowing for more comprehensive initial search patterns. +3. Implement a two-pass system in the test case creation process to capture files found by secondary searches. + +Until these improvements are implemented, be aware that modules using dynamic file searching may require manual intervention to ensure comprehensive test data. + +## Updating the JSON File + +After creating test cases with `make_test_data.py`, it's important to manually update the `testdata..json` file with: + +1. A meaningful `description` for each test case. +2. Your name as the `maker`. +3. The relevant `os_name` and `os_version` string so `test_module.py` can simulate it. + +This information, along with the golden files generated by `test_module.py`, is crucial for understanding test coverage and validating results. + +## Baselines and CI + +Every committed case runs in CI against a recorded snapshot. The workflow +`.github/workflows/test_cases.yml` calls `admin/test/scripts/run_test_cases.py`, +which re-runs each artifact against its case zip and fails when the output no +longer matches the latest snapshot under `admin/test/results//`. + +Baselines are ordinary committed files. When a PR deliberately changes a +parser's output, re-record the affected snapshot in the same PR: + +```bash +python admin/test/scripts/test_module.py -a all -c all +``` + +Commit the new snapshot and delete the superseded one, so the reviewer sees the +row-level diff next to the code change that caused it. Rows are compared as +unordered sets, so a change in result order alone does not fail the check. + +Units that cannot gate yet are listed with a reason per unit in +`admin/test/cases/known_failures.json`. They still run and report, but do not +fail CI. When one of them passes again, the run says so; remove its entry in +the same PR that fixed it. + +## Requirements + +### __artifacts_v2__ Block + +The test case creation process requires modules to use the `__artifacts_v2__` block for defining artifacts. This is because the v1 artifact definition has some limitations that can cause issues with the test harness. + +If you want to apply this testing process to a v1 script before converting the entire code, you can update the artifact block to v2 format first to capture the results. + +## Git Integration + +The script now includes git integration to capture information about the last commit that modified the module file. This information is stored in the JSON metadata file for each test case. + +## Performance Considerations + +The script includes performance optimizations and progress indicators for processing large archive files. It also provides timing information for various stages of the test data creation process. diff --git a/admin/scripts/check_pr_test_data.py b/admin/scripts/check_pr_test_data.py new file mode 100644 index 0000000..f0e242e --- /dev/null +++ b/admin/scripts/check_pr_test_data.py @@ -0,0 +1,309 @@ +#!/usr/bin/env python3 +"""Comments on pull requests that change artifact modules without test data. + +Runs from .github/workflows/request_test_data.yml on pull_request_target, so +the copy of this script that executes is always the one on the base branch, +never the contributor's. The pull request's own code is fetched as text for +ast parsing and is never imported or executed. + +Decision per changed artifact module: + +- the PR also touches admin/test/cases/testdata..json or anything + under admin/test/cases/data// -> covered, nothing to ask +- the module's __artifacts_v2__ sample_data (read at the PR head) cites a + corpus key present in admin/image_manifest.json -> a maintainer can + generate the fixture from the public image (fixture-needed label) +- otherwise -> ask the contributor for a fixture (needs-test-data label) + +Authors with write or admin permission on the repository, and bot accounts, +are skipped. The comment is sticky: one marker comment per PR, edited in +place, including flipping to a resolved note once data arrives. This is a +request, not a gate: the job succeeds whatever the outcome. + +Environment: GITHUB_TOKEN, GITHUB_REPOSITORY, PR_NUMBER. Set DRY_RUN=1 to +print intended writes instead of performing them (reads still happen). +""" +import ast +import json +import os +import urllib.error +import urllib.request +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[2] +MANIFEST_PATH = REPO_ROOT / "admin" / "image_manifest.json" +MARKER = "" +LABEL_ASK = "needs-test-data" +LABEL_FIXTURE = "fixture-needed" +LABELS = { + LABEL_ASK: ("d93f0b", "Artifact PR without test data for the changed modules"), + LABEL_FIXTURE: ("c5def5", "Cites a public image; a maintainer can generate the fixture"), +} +API = "https://api.github.com" + + +def api_request(token, url, method="GET", body=None, accept="application/vnd.github+json"): + """One GitHub API call; returns parsed JSON, raw text, or None on 404.""" + data = json.dumps(body).encode() if body is not None else None + req = urllib.request.Request(url, data=data, method=method, headers={ + "Authorization": f"Bearer {token}", + "Accept": accept, + "X-GitHub-Api-Version": "2022-11-28", + }) + try: + with urllib.request.urlopen(req) as resp: + raw = resp.read().decode() + except urllib.error.HTTPError as ex: + if ex.code == 404: + return None + raise RuntimeError(f"{method} {url} -> HTTP {ex.code}: {ex.read().decode()[:300]}") from ex + if accept.endswith("raw+json") or accept.endswith(".raw"): + return raw + return json.loads(raw) if raw else {} + + +def paginate(token, url): + """Yields items from a paginated list endpoint.""" + page = 1 + sep = "&" if "?" in url else "?" + while True: + batch = api_request(token, f"{url}{sep}per_page=100&page={page}") + if not batch: + return + yield from batch + if len(batch) < 100: + return + page += 1 + + +def artifact_modules(files): + """{module_name: status} for artifact files the PR adds or changes.""" + modules = {} + for f in files: + path, status = f["filename"], f["status"] + if status == "removed": + continue + parts = path.split("/") + if parts[:2] == ["scripts", "artifacts"] and len(parts) == 3 and path.endswith(".py"): + modules[parts[2][:-3]] = status + return modules + + +def covered_modules(files, modules): + """Modules whose test data the same PR touches.""" + covered = set() + for f in files: + if f["status"] == "removed": + continue + path = f["filename"] + for module in modules: + if (path == f"admin/test/cases/testdata.{module}.json" + or path.startswith(f"admin/test/cases/data/{module}/")): + covered.add(module) + return covered + + +def sample_data_keys_from_source(source_text): + """Corpus keys cited in a module's __artifacts_v2__ sample_data blocks. + + Parsed with ast only; nothing is imported or executed. A module whose + metadata is not a literal yields no keys, which routes it to the full ask. + """ + keys = set() + try: + tree = ast.parse(source_text) + except SyntaxError: + return keys + for node in tree.body: + if not isinstance(node, ast.Assign): + continue + if not any(isinstance(t, ast.Name) and t.id == "__artifacts_v2__" for t in node.targets): + continue + try: + block = ast.literal_eval(node.value) + except ValueError: + continue + if isinstance(block, dict): + for info in block.values(): + sample_data = info.get("sample_data") if isinstance(info, dict) else None + if isinstance(sample_data, dict): + keys.update(k for k in sample_data if isinstance(k, str)) + return keys + + +def manifest_keys(manifest_path=MANIFEST_PATH): + """Every name the image manifest answers to (public images only). + + A repo without a manifest gets an empty set, which routes every module to + the full ask rather than the maintainer-can-generate note. + """ + if not Path(manifest_path).exists(): + return set() + with open(manifest_path, encoding="utf-8") as f: + entries = json.load(f)["images"] + names = set() + for e in entries: + names.update(filter(None, (e.get("image_name"), e.get("sample_data_key")))) + return names + + +def classify(modules, covered, cited_by_module, public): + """Splits uncovered modules into (fixture_modules, ask_modules).""" + fixture, ask = {}, [] + for module in sorted(modules): + if module in covered: + continue + cited = sorted(cited_by_module.get(module, set()) & public) + if cited: + fixture[module] = cited + else: + ask.append(module) + return fixture, ask + + +def render_comment(repo, fixture, ask): + """The sticky comment body for the current state.""" + doc = f"https://github.com/{repo}/blob/main/admin/docs/testing" + if not fixture and not ask: + return (f"{MARKER}\nTest data is now included for every changed artifact module. " + "Thank you!") + lines = [MARKER, "Thanks for the contribution!", ""] + if ask: + lines += [ + "This PR changes artifact modules without test data for them. A small fixture " + "with each artifact change lets reviewers run the module against real data, and " + "the committed case keeps guarding the module after merge.", ""] + else: + lines += [ + "The changed artifact modules cite public research images in their `sample_data`, " + "so a maintainer can generate the test fixtures from those images. Nothing is " + "needed from you, though you are welcome to add the fixtures yourself with " + "`admin/test/scripts/make_test_data.py`.", ""] + for module, cited in fixture.items(): + cited_text = ", ".join(f"`{k}`" for k in cited) + lines.append(f"- `{module}.py`: cites {cited_text}; a maintainer can generate the fixture.") + for module in ask: + lines.append(f"- `{module}.py`: please include a fixture with this PR.") + if ask: + lines += [ + "", "**Adding a fixture**", "", + "Generate it from your extraction with the helper (details in " + f"[create_module_test_cases.md]({doc}/create_module_test_cases.md)):", "", + " python admin/test/scripts/make_test_data.py --case " + "--input ", "", + "It writes `admin/test/cases/testdata..json` and one zip per artifact " + "under `admin/test/cases/data//`.", "", + "Size rules:", "", + "- Under 10 MB per zip: commit the files in this PR.", + "- 10 to 25 MB: commit the case JSON in the PR and attach the zip to a comment here.", + "- Over 25 MB: say so here and a maintainer will arrange a handoff.", "", + "If your extraction cannot be shared:", "", + "- If the app appears on a public research image, generate the fixture from that " + f"instead. [public_corpus_images.md]({doc}/public_corpus_images.md) lists the " + "images and where to download them.", + "- Or sanitize the real file in place: keep the file the app wrote and overwrite " + "only the personal values, which keeps the format honest.", + "- Or script a known session: install the app on a test device with a throwaway " + "account, perform documented actions, and extract that.", "", + "If none of those fit, say so here and we will work it out. The PR can still be " + "reviewed and merged with the gap recorded in the artifact's `notes`.", "", + "This is a request, not a gate. Nothing here blocks review."] + return "\n".join(lines) + + +def desired_labels(fixture, ask): + labels = set() + if fixture: + labels.add(LABEL_FIXTURE) + if ask: + labels.add(LABEL_ASK) + return labels + + +def find_marker_comment(token, repo, pr_number): + for comment in paginate(token, f"{API}/repos/{repo}/issues/{pr_number}/comments"): + if MARKER in comment.get("body", ""): + return comment + return None + + +def author_permission(token, repo, login): + """Repo permission for a user: admin, write, read, or none.""" + result = api_request(token, f"{API}/repos/{repo}/collaborators/{login}/permission") + return (result or {}).get("permission", "none") + + +def apply_state(token, repo, pr_number, body, labels, dry_run): + """Upserts the sticky comment and reconciles our two labels.""" + existing = find_marker_comment(token, repo, pr_number) + resolved = not labels + if dry_run: + print(f"DRY_RUN: would {'edit' if existing else 'create'} comment; labels -> {sorted(labels)}") + print("---- comment body ----") + print(body) + return + if existing: + if existing["body"] != body: + api_request(token, f"{API}/repos/{repo}/issues/comments/{existing['id']}", + method="PATCH", body={"body": body}) + elif not resolved: + # Never open a resolved-state comment on a PR that was never asked. + api_request(token, f"{API}/repos/{repo}/issues/{pr_number}/comments", + method="POST", body={"body": body}) + current = {l["name"] for l in api_request(token, f"{API}/repos/{repo}/issues/{pr_number}") ["labels"]} + for name in labels - current: + if api_request(token, f"{API}/repos/{repo}/labels/{name}") is None: + color, description = LABELS[name] + api_request(token, f"{API}/repos/{repo}/labels", method="POST", + body={"name": name, "color": color, "description": description}) + api_request(token, f"{API}/repos/{repo}/issues/{pr_number}/labels", + method="POST", body={"labels": [name]}) + for name in (current & set(LABELS)) - labels: + api_request(token, f"{API}/repos/{repo}/issues/{pr_number}/labels/{name}", method="DELETE") + + +def main(): + token = os.environ["GITHUB_TOKEN"] + repo = os.environ["GITHUB_REPOSITORY"] + pr_number = int(os.environ["PR_NUMBER"]) + dry_run = os.environ.get("DRY_RUN") == "1" + + pr = api_request(token, f"{API}/repos/{repo}/pulls/{pr_number}") + login = pr["user"]["login"] + if login.endswith("[bot]"): + print(f"Author {login} is a bot; skipping.") + return + permission = author_permission(token, repo, login) + if permission in ("admin", "write"): + print(f"Author {login} has {permission} access; skipping.") + if not dry_run: + return + print("DRY_RUN: continuing anyway to show the decision.") + + files = list(paginate(token, f"{API}/repos/{repo}/pulls/{pr_number}/files")) + modules = artifact_modules(files) + if not modules: + print("No artifact modules added or changed; nothing to do.") + return + covered = covered_modules(files, modules) + + cited_by_module = {} + contents_by_module = {f["filename"].split("/")[2][:-3]: f.get("contents_url") + for f in files if f["filename"].startswith("scripts/artifacts/")} + for module in modules: + if module in covered: + continue + url = contents_by_module.get(module) + source = api_request(token, url, accept="application/vnd.github.raw+json") if url else None + cited_by_module[module] = sample_data_keys_from_source(source or "") + + fixture, ask = classify(modules, covered, cited_by_module, manifest_keys()) + print(f"modules: {sorted(modules)} covered: {sorted(covered)} " + f"fixture: {sorted(fixture)} ask: {ask}") + body = render_comment(repo, fixture, ask) + apply_state(token, repo, pr_number, body, desired_labels(fixture, ask), dry_run) + print("Done.") + + +if __name__ == "__main__": + main() diff --git a/admin/test/scripts/make_test_data.py b/admin/test/scripts/make_test_data.py new file mode 100644 index 0000000..67110dc --- /dev/null +++ b/admin/test/scripts/make_test_data.py @@ -0,0 +1,702 @@ +"""Creates test data for iLEAPP modules. + +This script is a command-line tool used to generate test data sets for iLEAPP +artifact modules. It can extract relevant files from a forensic image +(in .zip, .tar, or .tar.gz format) based on the file path patterns +defined within an artifact module. + +The script generates a .zip file for each artifact containing the matched files +and creates or updates a JSON metadata file that describes the test case. +This allows for standardized and repeatable testing of artifact parsing logic. + +Usage: + python make_test_data.py [--image | --case --input | --image-prompt] +""" +import os +import sys +import json +import zipfile +import tarfile +import fnmatch +import argparse +import time +import platform +import subprocess +from datetime import datetime +from collections import defaultdict +# from io import BytesIO +import csv +from io import StringIO +import textwrap +# import glob + +# Add the correct path to the system path +repo_root = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..', '..')) +sys.path.append(repo_root) + +def get_artifact_info(module_name): + """Dynamically imports an artifact module and retrieves its artifact definitions. + + Args: + module_name (str): The name of the artifact module (e.g., 'callHistory'). + + Returns: + dict: The __artifacts_v2__ dictionary from the specified module. + + Raises: + SystemExit: If the module cannot be imported. + """ + try: + module = __import__(f'scripts.artifacts.{module_name}', fromlist=['__artifacts_v2__']) + return module.__artifacts_v2__ + except ImportError: + print(f"Error: Could not import module 'scripts.artifacts.{module_name}'") + sys.exit(1) + +def get_last_commit_info(file_path): + """Retrieves the last git commit information for a given file. + + Args: + file_path (str): The path to the file. + + Returns: + dict or None: A dictionary containing the commit hash, author, date, and + message, or None if the git command fails. + """ + try: + # Get the last commit hash + git_log = subprocess.check_output(['git', 'log', '-n', '1', '--pretty=format:%H|%an|%ae|%ad|%s', '--', file_path], + universal_newlines=True).strip() + if not git_log: + # File is not yet in git history + return { + 'hash': 'Uncommitted', + 'author_name': 'N/A', + 'author_email': 'N/A', + 'date': datetime.now().isoformat(), + 'message': 'File not yet committed to git' + } + commit_hash, author_name, author_email, commit_date, commit_message = git_log.split('|') + + # Convert the commit date to ISO format + commit_date = datetime.strptime(commit_date, '%a %b %d %H:%M:%S %Y %z').isoformat() + + return { + 'hash': commit_hash, + 'author_name': author_name, + 'author_email': author_email, + 'date': commit_date, + 'message': commit_message + } + except subprocess.CalledProcessError: + return None + +def load_image_manifest(): + """Loads the image manifest JSON file. + + The manifest contains metadata about available test images. + + Returns: + list: A list of image dictionaries from the manifest. + """ + manifest_path = os.path.join(repo_root, 'admin', 'image_manifest.json') + if not os.path.exists(manifest_path): + sys.exit("This repo has no admin/image_manifest.json yet; use --case with --input, " + "or add a manifest (see admin/docs/testing/guide_adding_images.md in iLEAPP).") + with open(manifest_path, 'r', encoding='utf-8') as f: + return json.load(f)['images'] + +def expand_user_path(path): + """Expands a path containing a tilde (~) to the user's home directory. + + Args: + path (str): The path to expand. + + Returns: + str: The expanded path. + """ + return os.path.expanduser(path) + +LOCAL_CONFIG_FILENAME = 'image_manifest.local.json' + +def load_local_image_config(config_path=None): + """Loads the per-machine image location file, if present. + + The manifest itself carries only machine-independent identity; where an + image lives on a given machine is recorded in a git-ignored + admin/image_manifest.local.json: + + { + "image_paths": {"": "/path/to/image.zip"}, + "search_roots": ["/path/to/a/folder/of/images"] + } + + Args: + config_path (str): Override for the config location (used by tests). + + Returns: + dict: The parsed config, or an empty dict when the file is absent. + """ + if config_path is None: + config_path = os.path.join(repo_root, 'admin', LOCAL_CONFIG_FILENAME) + if not os.path.exists(config_path): + return {} + with open(config_path, 'r', encoding='utf-8') as f: + return json.load(f) + +def resolve_image_path(image_data, local_config): + """Finds an existing local copy of a manifest image. + + Tried in order: an explicit "image_paths" mapping in the local config + (keyed by image_name or sample_data_key), the entry's legacy + local_image_paths list, then each local config "search_roots" directory + walked for a file named exactly like the entry's published_file. + + Args: + image_data (dict): One entry from the image manifest. + local_config (dict): Result of load_local_image_config(). + + Returns: + str: An existing path, or None when nothing resolves. + """ + mapped = local_config.get('image_paths', {}) + for key in (image_data.get('image_name'), image_data.get('sample_data_key')): + if key and key in mapped: + candidate = expand_user_path(mapped[key]) + if os.path.exists(candidate): + return candidate + print(f"Warning: {LOCAL_CONFIG_FILENAME} maps '{key}' to a missing path: {candidate}") + + for path in image_data.get('local_image_paths', []): + expanded_path = expand_user_path(path) + if os.path.exists(expanded_path): + return expanded_path + + published = image_data.get('published_file') + if published: + for root in local_config.get('search_roots', []): + for dirpath, _dirnames, filenames in os.walk(expand_user_path(root)): + if published in filenames: + return os.path.join(dirpath, published) + return None + +def get_image_info(image_name, config_path=None): + """Finds an image in the manifest and resolves its local path. + + Args: + image_name (str): The image_name or sample_data_key to look up. + config_path (str): Override for the local config location (tests). + + Returns: + dict: The image's metadata dictionary from the manifest, with an + 'input_file' key added for the resolved local path. + + Raises: + ValueError: If the image is not found in the manifest. + FileNotFoundError: If no local copy of the image can be found. + """ + manifest = load_image_manifest() + image_data = next((img for img in manifest + if image_name in (img.get('image_name'), img.get('sample_data_key'))), None) + if not image_data: + raise ValueError(f"Image '{image_name}' not found in manifest") + + resolved = resolve_image_path(image_data, load_local_image_config(config_path)) + if resolved: + image_data['input_file'] = resolved + return image_data + + raise FileNotFoundError( + f"No local copy found for image '{image_name}'. Map it in admin/{LOCAL_CONFIG_FILENAME} " + "(see admin/docs/testing/guide_adding_images.md).") + +def read_filepath_list(file_path_list): + """Reads a zipped CSV file containing a list of file paths from an image. + + This is used to speed up file searching by using a pre-compiled list + of all paths in a forensic image. + + Args: + file_path_list (str): The path to the .zip file containing the CSV. + + Returns: + list: A list of file paths extracted from the CSV. + """ + filepath_list = [] + with zipfile.ZipFile(file_path_list, 'r') as zip_ref: + csv_filename = zip_ref.namelist()[0] # Assume the first file in the zip is the CSV + with zip_ref.open(csv_filename) as csvfile: + csv_content = csvfile.read().decode('utf-8') + csv_reader = csv.DictReader(StringIO(csv_content)) + for row in csv_reader: + filepath_list.append(row['path']) + return filepath_list + +def match_files_from_list(filepath_list, patterns): + """Filters a list of file paths against a set of glob patterns. + + Args: + filepath_list (list): A list of file paths to search through. + patterns (str or tuple): A single glob pattern or a tuple/list of patterns. + + Returns: + list: A list of file paths that match the given patterns. + """ + matching_files = [] + # Ensure patterns is always a tuple or list + if isinstance(patterns, str): + patterns = (patterns,) + for filepath in filepath_list: + if any(fnmatch.fnmatch(filepath, pattern) for pattern in patterns): + matching_files.append(filepath) + return matching_files + +def process_archive(input_file, all_patterns, filepath_list=None): + """Searches an archive for files matching patterns defined by artifacts. + + This function can either iterate through a live archive (.zip, .tar, .tar.gz) + or search through a pre-compiled list of file paths for efficiency. + + Args: + input_file (str): The path to the archive file. + all_patterns (dict): A dictionary where keys are artifact names and + values are lists of glob patterns. + filepath_list (list, optional): A pre-compiled list of file paths + within the archive. Defaults to None. + + Returns: + defaultdict: A dictionary where keys are artifact names. + If `filepath_list` is provided, values are lists of matching file paths. + Otherwise, values are dictionaries mapping file paths to file content. + + Raises: + ValueError: If the input file is not a supported archive format. + """ + matching_files = defaultdict(list if filepath_list else dict) + file_count = 0 + start_time = time.time() + local_start_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + print(f"Search started at: {local_start_time}") + + if filepath_list: + print("Using pre-compiled file path list") + file_count = len(filepath_list) + for artifact_name, patterns in all_patterns.items(): + matching_files[artifact_name] = match_files_from_list(filepath_list, patterns) + else: + print("Searching for files matching all patterns") + + if input_file.endswith('.zip'): + with zipfile.ZipFile(input_file, 'r') as zip_ref: + for file in zip_ref.namelist(): + file_count += 1 + if file_count % 10000 == 0: + print(".", end="", flush=True) + for artifact, patterns in all_patterns.items(): + for pattern in patterns: + if fnmatch.fnmatch(file, pattern): + matching_files[artifact][file] = zip_ref.read(file) + break + elif input_file.endswith('.tar.gz') or input_file.endswith('.tgz'): + print("Processing tar.gz file") + with tarfile.open(input_file, 'r:gz') as tar_ref: + print("Searching files\n") + for member in tar_ref.getmembers(): + file_count += 1 + if file_count % 10000 == 0: + print(".", end="", flush=True) + for artifact, patterns in all_patterns.items(): + for pattern in patterns: + if fnmatch.fnmatch(member.name, pattern): + matching_files[artifact][member.name] = tar_ref.extractfile(member).read() + break + elif input_file.endswith('.tar'): + with tarfile.open(input_file, 'r') as tar_ref: + for member in tar_ref.getmembers(): + file_count += 1 + if file_count % 10000 == 0: + print(".", end="", flush=True) + for artifact, patterns in all_patterns.items(): + for pattern in patterns: + if fnmatch.fnmatch(member.name, pattern): + matching_files[artifact][member.name] = tar_ref.extractfile(member).read() + break + else: + raise ValueError("Unsupported file format. Please use .zip, tar, or .tar.gz") + + print() # New line after progress dots + end_time = time.time() + local_end_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + print(f"Search completed at: {local_end_time}") + total_matches = sum(len(files) for files in matching_files.values()) + print(f"Searched {file_count} files, found {total_matches} matching files in {end_time - start_time:.2f} seconds") + return matching_files + +def find_file_in_tar(tar_archive, target_path, tar_filename): + """Finds a file in a tar archive, handling optional path prefixes. + + Some tar archives have all their contents inside a single top-level + directory. This function attempts to find the file with and without + a prefix derived from the archive's name. + + Args: + tar_archive (tarfile.TarFile): An open tarfile object. + target_path (str): The path of the file to find within the archive. + tar_filename (str): The filename of the .tar archive. + + Returns: + str or None: The full, correct path of the file within the tar archive, + or None if the file cannot be found. + """ + # Remove the extension from the tar filename + tar_prefix = os.path.splitext(os.path.basename(tar_filename))[0] + if tar_prefix.endswith('.tar'): + tar_prefix = os.path.splitext(tar_prefix)[0] + + try: + prefixed_path = f"{tar_prefix}/{target_path}" + if tar_archive.getmember(prefixed_path): + return prefixed_path + except KeyError: + pass + + try: + if tar_archive.getmember(target_path): + return target_path + except KeyError: + pass + + # If both attempts fail, return None + return None + +def create_test_data(module_name, image_name=None, case_number=None, input_file=None, image_metadata=None): + """Orchestrates the creation of test data for a specific module. + + This function performs the end-to-end process of generating test data: + 1. Gathers artifact and version control information. + 2. Reads or creates a JSON file to store test case metadata. + 3. Processes an input archive to find files matching artifact patterns. + 4. Creates a .zip file for each artifact containing the found files. + 5. Updates the JSON metadata file with the results. + + Args: + module_name (str): The name of the artifact module. + image_name (str, optional): The name of the image from the manifest. + Used as the case key. Defaults to None. + case_number (str, optional): The case number for the test data. + Used if `image_name` is not provided. + Defaults to None. + input_file (str, optional): The path to the input archive. Required + if `case_number` is used. Defaults to None. + image_metadata (dict, optional): A dictionary of metadata about the + image. Defaults to None. + """ + overall_start_time = time.time() + print(f"Processing module: {module_name}") + artifacts = get_artifact_info(module_name) + + # Get git information for the module + module_path = os.path.join(repo_root, 'scripts', 'artifacts', f'{module_name}.py') + last_commit_info = get_last_commit_info(module_path) + + # Update paths for new folder structure + cases_dir = os.path.join(repo_root, 'admin', 'test', 'cases') + data_dir = os.path.join(cases_dir, 'data', module_name) # Add module_name to the path + os.makedirs(data_dir, exist_ok=True) + + json_file = os.path.join(cases_dir, f"testdata.{module_name}.json") + + # Check if JSON file exists and is not empty + if os.path.exists(json_file) and os.path.getsize(json_file) > 0: + try: + with open(json_file, 'r', encoding='utf-8') as f: + json_data = json.load(f) + print(f"Updating existing JSON file: {json_file}") + except json.JSONDecodeError: + print("Existing JSON file is invalid.") + create_new = input("Do you want to create new JSON data? This will overwrite the existing file. (y/n): ") + if create_new.lower() != 'y': + print("Aborting operation.") + sys.exit(1) + json_data = {} + else: + json_data = {} + print(f"Creating new JSON file: {json_file}") + + # Use image_name as case_key if provided, otherwise use the given case_number + case_key = image_name if image_name else f"case{case_number}" + + if case_key in json_data: + overwrite = input(f"Case {case_key} already exists. Do you want to overwrite it? (y/n): ") + if overwrite.lower() != 'y': + print("Aborting operation.") + sys.exit(1) + + # Create or update case entry + json_data[case_key] = { + "description": "", + "maker": "", + "make_data": { + "input_data_path": os.path.abspath(input_file), + "os": platform.platform(), + "timestamp": datetime.now().isoformat(), + "last_commit": last_commit_info + }, + "artifacts": {} + } + + if image_name: + json_data[case_key]["image_name"] = image_name + + if image_metadata: + json_data[case_key]["image_info"] = image_metadata + + # Collect all patterns + all_patterns = {artifact_name: artifact_info['paths'] for artifact_name, artifact_info in artifacts.items()} + + # Check if file_path_list exists in the image info + filepath_list = None + if image_name: + image_info = get_image_info(image_name) + if 'file_path_list' in image_info: + filepath_list_path = os.path.join(repo_root, image_info['file_path_list']) + if os.path.exists(filepath_list_path): + print(f"Using file path list: {filepath_list_path}") + filepath_list = read_filepath_list(filepath_list_path) + else: + print(f"Warning: file_path_list specified in manifest does not exist: {filepath_list_path}") + + # Process archive and get matching files for all artifacts at once + matching_files = process_archive(input_file, all_patterns, filepath_list) + + # Open the source archive once + source_archive = None + try: + if input_file.endswith('.zip'): + source_archive = zipfile.ZipFile(input_file, 'r') + elif input_file.endswith('.tar.gz') or input_file.endswith('.tgz'): + source_archive = tarfile.open(input_file, 'r:gz') + elif input_file.endswith('.tar'): + source_archive = tarfile.open(input_file, 'r') + else: + raise ValueError("Unsupported file format. Please use .zip, .tar, or .tar.gz") + + for artifact_name, artifact_info in artifacts.items(): + # artifact_start_time = time.time() + local_artifact_start_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + print(f"\nProcessing artifact: {artifact_name}") + print(f"Artifact processing started at: {local_artifact_start_time}") + + matching_file_count = len(matching_files[artifact_name]) + + json_data[case_key]["artifacts"][artifact_name] = { + "search_patterns": artifact_info['paths'], + "file_count": matching_file_count, + } + + if matching_file_count == 0: + print(f"No responsive files found for artifact: {artifact_name}") + json_data[case_key]["artifacts"][artifact_name]["note"] = "No responsive files found for this artifact" + continue + + # Create file name in the new data directory + file_name = os.path.join(data_dir, f"testdata.{module_name}.{artifact_name}.{case_key}.zip") + + # Create zip file with matching files + # zip_start_time = time.time() + with zipfile.ZipFile(file_name, 'w') as zip_file: + if isinstance(matching_files[artifact_name], list): + # If using filepath_list, we only have file paths + for file_path in matching_files[artifact_name]: + if isinstance(source_archive, zipfile.ZipFile): + if file_path in source_archive.namelist(): + file_content = source_archive.read(file_path) + zip_file.writestr(file_path, file_content) + else: + print(f"Warning: File not found in zip: {file_path}") + else: # tarfile + tar_path = find_file_in_tar(source_archive, file_path, input_file) + if tar_path: + file_content = source_archive.extractfile(tar_path).read() + zip_file.writestr(file_path, file_content) + else: + print(f"Warning: File not found in tar: {file_path}") + else: + # If processing archive directly, we have file contents + for file_path, file_content in matching_files[artifact_name].items(): + zip_file.writestr(file_path, file_content) + # zip_end_time = time.time() + + json_data[case_key]["artifacts"][artifact_name]["expected_output"] = { + "headers": [], + "data": [] + } + + print(f"Test data created: {file_name}") + print(f"Added {matching_file_count} files to the test data") + # artifact_end_time = time.time() + # local_artifact_end_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + + if all(artifact.get("file_count", 0) == 0 for artifact in json_data[case_key]["artifacts"].values()): + print(f"\nNo responsive files found for any artifacts in case {case_key}.") + json_data[case_key]["note"] = "No responsive files found for any artifacts" + + # Write updated JSON data + json_start_time = time.time() + with open(json_file, 'w', encoding='utf-8') as f: + json.dump(json_data, f, indent=2) + json_end_time = time.time() + + print(f"\nJSON file updated: {json_file}") + print(f"JSON file update took {json_end_time - json_start_time:.2f} seconds") + print("Please update the JSON file with test case details.") + + finally: + if source_archive: + source_archive.close() + + overall_end_time = time.time() + local_overall_end_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + print(f"\nTotal processing time: {overall_end_time - overall_start_time:.2f} seconds") + print(f"Script completed at: {local_overall_end_time}") + +def get_case_data_summary(module_name, image_name): + """ + Checks for existing test data for a given module and image, and returns + a summary. + + Args: + module_name (str): The name of the artifact module. + image_name (str): The name of the image, used as the case key. + + Returns: + dict: A dictionary with 'exists' (bool), and if it exists, + 'file_count' (int) and 'zip_size' (int). + """ + json_file_path = os.path.join(repo_root, 'admin', 'test', 'cases', + f'testdata.{module_name}.json') + + if not os.path.exists(json_file_path): + return {'exists': False} + + with open(json_file_path, 'r', encoding='utf-8') as f: + try: + all_cases_data = json.load(f) + except json.JSONDecodeError: + return {'exists': False} # Treat invalid JSON as no data + + case_data = all_cases_data.get(image_name) + if not case_data: + return {'exists': False} + + artifacts = case_data.get('artifacts', {}) + total_file_count = 0 + total_zip_size = 0 + data_dir = os.path.join(repo_root, 'admin', 'test', 'cases', 'data', module_name) + + for artifact_name, artifact_data in artifacts.items(): + file_count = artifact_data.get('file_count', 0) + total_file_count += file_count + if file_count > 0: + zip_path = os.path.join(data_dir, f"testdata.{module_name}.{artifact_name}.{image_name}.zip") + if os.path.exists(zip_path): + total_zip_size += os.path.getsize(zip_path) + + return {'exists': True, 'file_count': total_file_count, 'zip_size': total_zip_size} + +def prompt_for_image(module_name): + """Displays a menu of available images and prompts the user to select one. + + It checks for existing test data for the given module and displays a + status marker next to each image name. + + Args: + module_name (str): The name of the module being tested, used to + check for existing data. + + Returns: + str: The name of the image selected by the user. + """ + manifest = load_image_manifest() + print("\nAvailable images:") + for i, image in enumerate(manifest, 1): + image_name = image['image_name'] + summary = get_case_data_summary(module_name, image_name) + + if not summary['exists']: + case_data_status = "✗ No case data" + elif summary['file_count'] == 0: + case_data_status = "✓ Case logged (0 files)" + else: + size_kb = summary['zip_size'] / 1024 + case_data_status = (f"✓ Has data ({summary['file_count']} files, " + f"{size_kb:.2f} KB)") + + print(f"{i}. {image_name} [{case_data_status}]") + if 'description' in image: + wrapped_description = textwrap.wrap(image['description'], width=70, initial_indent=' ', subsequent_indent=' ') + print('\n'.join(wrapped_description)) + print() + + while True: + try: + choice = int(input("Enter the number of the image you want to use: ")) + if 1 <= choice <= len(manifest): + return manifest[choice - 1]['image_name'] + else: + print("Invalid choice. Please enter a number from the list.") + except ValueError: + print("Invalid input. Please enter a number.") + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description="Create test data for artifacts") + parser.add_argument("module_name", help="Name of the module (e.g., keyboard or keyboard.py)") + group = parser.add_mutually_exclusive_group(required=True) + group.add_argument("--image", help="Name of the image from the manifest") + group.add_argument("--case", help="Case number for the test data") + group.add_argument("--image-prompt", action="store_true", help="Prompt for image selection from the manifest") + parser.add_argument("--input", help="Path to the input file (zip, tar, or tar.gz)", required='--case' in sys.argv) + + args = parser.parse_args() + + # Remove .py extension if present + input_module_name = args.module_name[:-3] if args.module_name.endswith('.py') else args.module_name + + script_start_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + print(f"Starting test data creation for module: {input_module_name}") + print(f"Script started at: {script_start_time}") + + if args.image_prompt: + selected_image = prompt_for_image(input_module_name) + args.image = selected_image + + if args.image: + try: + input_image_info = get_image_info(args.image) + print(f"Using image: {args.image}") + print(f"Image path: {input_image_info['input_file']}") + + # Check if case data already exists + case_summary = get_case_data_summary(input_module_name, args.image) + if case_summary['exists']: + print(f"Note: Case data already exists for module '{input_module_name}' using image '{args.image}'.") + proceed = input("Do you want to proceed and potentially overwrite existing data? (y/n): ") + if proceed.lower() != 'y': + print("Operation aborted.") + sys.exit(0) + + create_test_data(input_module_name, + image_name=args.image, + input_file=input_image_info['input_file'], + image_metadata=input_image_info.get('image_info')) + except (ValueError, FileNotFoundError) as e: + print(f"Error: {e}") + sys.exit(1) + else: + if not args.input: + parser.error("--input is required when --case is used") + print(f"Using case number: {args.case}") + print(f"Input file: {args.input}") + create_test_data(input_module_name, case_number=args.case, input_file=args.input) + + print("\nTest data creation completed.") diff --git a/admin/test/scripts/run_test_cases.py b/admin/test/scripts/run_test_cases.py new file mode 100644 index 0000000..9317259 --- /dev/null +++ b/admin/test/scripts/run_test_cases.py @@ -0,0 +1,189 @@ +#!/usr/bin/env python3 +"""Runs every committed test case and compares the output to its recorded baseline. + +The interactive recorder is admin/test/scripts/test_module.py: it runs a module's +artifacts against the case zips and writes timestamped snapshots (headers plus +full rows) under admin/test/results//. This script is the other half: +it re-runs the same artifacts through the same machinery and fails when the +output no longer matches the latest snapshot, so the committed cases act as +regression tests in CI. + +Baselines are ordinary committed files. To accept a new baseline after a +deliberate parser change, re-record it in the same PR: + + python admin/test/scripts/test_module.py -a all -c all + +then commit the new snapshot (and delete the superseded one), so the reviewer +sees the row-level diff next to the code that caused it. + +Comparison notes: +- Rows are compared as unordered multisets: SQLite result order without an + ORDER BY is not stable across platforms and a reorder is not a regression. +- Both sides are normalized before comparison: values are passed through a + JSON round-trip (matching how snapshots were serialized) and the per-run + extraction directory admin/test/temp/extract__ is replaced by + a fixed token, since its epoch differs on every run by construction. + +Units listed in admin/test/cases/known_failures.json (unit -> reason) run and +report but do not gate, so a unit can be excluded deliberately, with a stated +reason, instead of blocking every PR while it is being repaired. A known +failure that passes again is flagged so the entry gets removed. + +Exit status is 1 if any non-excluded unit failed, errored, or has no +baseline; 0 otherwise. +""" +import argparse +import json +import os +import re +import sys +from pathlib import Path + +SCRIPT_DIR = Path(__file__).resolve().parent +REPO_ROOT = SCRIPT_DIR.parents[2] +sys.path.insert(0, str(SCRIPT_DIR)) +sys.path.insert(0, str(REPO_ROOT)) + +CASES_DIR = REPO_ROOT / "admin" / "test" / "cases" +RESULTS_DIR = REPO_ROOT / "admin" / "test" / "results" +TEMP_EXTRACT_RE = re.compile(r"admin/test/temp/extract_[A-Za-z0-9_]+_\d+") +TEMP_TOKEN = "" +KNOWN_FAILURES_PATH = CASES_DIR / "known_failures.json" +COMPLETION_MARKER = "Test case comparison complete" + + +def discover_modules(): + """Module names that have a committed case file.""" + return sorted(p.name[len("testdata."):-len(".json")] + for p in CASES_DIR.glob("testdata.*.json")) + + +def latest_baseline(module, artifact, case): + """Path of the newest snapshot for one (module, artifact, case), or None.""" + pattern = f"{module}.{artifact}.{case}.*.json" + candidates = sorted((RESULTS_DIR / module).glob(pattern)) + return candidates[-1] if candidates else None + + +def normalize_rows(rows): + """Rows as a sorted list of JSON strings, with per-run paths tokenized.""" + normalized = [] + for row in rows: + text = json.dumps(row, default=str, ensure_ascii=False) + normalized.append(TEMP_EXTRACT_RE.sub(TEMP_TOKEN, text)) + return sorted(normalized) + + +def normalize_headers(headers): + return json.loads(json.dumps(headers, default=str)) + + +def compare(fresh_headers, fresh_rows, baseline): + """Returns a list of difference descriptions; empty means match.""" + problems = [] + base_headers = normalize_headers(baseline.get("headers", [])) + if normalize_headers(fresh_headers) != base_headers: + problems.append(f"headers differ: now {normalize_headers(fresh_headers)!r}, " + f"recorded {base_headers!r}") + now = normalize_rows(fresh_rows) + recorded = normalize_rows(baseline.get("data", [])) + if now != recorded: + now_set, rec_set = set(now), set(recorded) + gained = sorted(now_set - rec_set) + lost = sorted(rec_set - now_set) + problems.append(f"rows differ: now {len(now)}, recorded {len(recorded)}; " + f"{len(gained)} new, {len(lost)} missing") + for label, rows in (("new", gained), ("missing", lost)): + for row in rows[:2]: + problems.append(f" {label}: {row[:160]}") + return problems + + +def run_one(test_module, module, artifact, case, case_data): + """Runs one artifact against one case zip; returns (status, detail).""" + zip_path = CASES_DIR / "data" / module / f"testdata.{module}.{artifact}.{case}.zip" + if not zip_path.exists(): + return "BROKEN", f"case declares files but zip is missing: {zip_path}" + baseline_path = latest_baseline(module, artifact, case) + if baseline_path is None: + return "NO_BASELINE", ("no recorded snapshot; record one with " + f"test_module.py {module} -a {artifact} -c {case}") + try: + os_version = case_data.get("image_info", {}).get("os_version") + headers, rows, _elapsed, _commit, _media, _embedded = test_module.process_artifact( + zip_path, module, artifact, case_data["artifacts"][artifact], + target_os_version=os_version) + headers, rows = test_module.process_data(headers, rows) + except Exception as ex: # pylint: disable=broad-except + return "ERROR", f"{type(ex).__name__}: {ex}" + with open(baseline_path, encoding="utf-8") as f: + baseline = json.load(f) + problems = compare(headers, rows, baseline) + if problems: + return "FAIL", "\n ".join([f"vs {baseline_path.name}"] + problems) + return "PASS", f"{len(rows)} rows match {baseline_path.name}" + + +def main(argv=None): + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument("--module", action="append", + help="limit to this module (repeatable; default: all)") + parser.add_argument("--list", action="store_true", help="list runnable units and exit") + parser.add_argument("--strict", action="store_true", + help="ignore known_failures.json and gate on everything") + args = parser.parse_args(argv) + + known = {} + if KNOWN_FAILURES_PATH.exists() and not args.strict: + with open(KNOWN_FAILURES_PATH, encoding="utf-8") as f: + known = json.load(f) + + os.chdir(REPO_ROOT) + import test_module # noqa: E402 imported late so sys.path is set + + modules = args.module or discover_modules() + counts = {} + failures = [] + for module in modules: + cases_file = CASES_DIR / f"testdata.{module}.json" + if not cases_file.exists(): + print(f"{module}: no case file", flush=True) + failures.append((module, "-", "-", "BROKEN", "case file missing")) + continue + with open(cases_file, encoding="utf-8") as f: + cases = json.load(f) + for case, case_data in sorted(cases.items()): + for artifact, artifact_data in sorted(case_data.get("artifacts", {}).items()): + if artifact_data.get("file_count", 0) == 0: + continue + if args.list: + print(f"{module} {artifact} {case}") + continue + status, detail = run_one(test_module, module, artifact, case, case_data) + unit = f"{module}.{artifact}.{case}" + if unit in known and status != "PASS": + print(f"[KNOWN_FAIL ] {unit}: {status}; excluded: {known[unit]}", flush=True) + counts["KNOWN_FAIL"] = counts.get("KNOWN_FAIL", 0) + 1 + continue + counts[status] = counts.get(status, 0) + 1 + if unit in known and status == "PASS": + detail += " (listed in known_failures.json; remove its entry)" + print(f"[{status:11s}] {unit}: {detail}", flush=True) + if status != "PASS": + failures.append((module, artifact, case, status, detail)) + if args.list: + return 0 + + print() + print(f"{COMPLETION_MARKER}: " + + ", ".join(f"{k}={v}" for k, v in sorted(counts.items())) if counts else "nothing ran") + if failures: + print(f"\n{len(failures)} unit(s) need attention:") + for module, artifact, case, status, _detail in failures: + print(f" {status:11s} {module}.{artifact}.{case}") + return 1 + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/admin/test/scripts/test_check_pr_test_data.py b/admin/test/scripts/test_check_pr_test_data.py new file mode 100644 index 0000000..2d6abdd --- /dev/null +++ b/admin/test/scripts/test_check_pr_test_data.py @@ -0,0 +1,144 @@ +"""Tests for the pure logic in admin/scripts/check_pr_test_data.py. + +The GitHub API layer is not exercised here; these cover file classification, +sample_data extraction from source text, the decision matrix, and the comment +rendering, all against synthetic inputs. +""" +import os +import sys +import unittest + +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..', 'scripts'))) + +import check_pr_test_data as bot # noqa: E402 pylint: disable=wrong-import-position + + +def _files(*pairs): + return [{'filename': name, 'status': status} for name, status in pairs] + + +class ArtifactModulesTests(unittest.TestCase): + def test_added_and_modified_artifacts_counted(self): + files = _files(('scripts/artifacts/foo.py', 'added'), + ('scripts/artifacts/bar.py', 'modified'), + ('scripts/ilapfuncs.py', 'modified'), + ('README.md', 'modified')) + self.assertEqual(bot.artifact_modules(files), {'foo': 'added', 'bar': 'modified'}) + + def test_removed_artifacts_ignored(self): + files = _files(('scripts/artifacts/gone.py', 'removed')) + self.assertEqual(bot.artifact_modules(files), {}) + + def test_non_python_and_nested_paths_ignored(self): + files = _files(('scripts/artifacts/notes.md', 'added'), + ('scripts/artifacts/sub/x.py', 'added')) + self.assertEqual(bot.artifact_modules(files), {}) + + +class CoveredModulesTests(unittest.TestCase): + def test_case_json_covers(self): + files = _files(('admin/test/cases/testdata.foo.json', 'added')) + self.assertEqual(bot.covered_modules(files, {'foo': 'added'}), {'foo'}) + + def test_data_dir_covers(self): + files = _files(('admin/test/cases/data/foo/testdata.foo.a.img.zip', 'added')) + self.assertEqual(bot.covered_modules(files, {'foo': 'added'}), {'foo'}) + + def test_other_modules_data_does_not_cover(self): + files = _files(('admin/test/cases/data/bar/testdata.bar.a.img.zip', 'added')) + self.assertEqual(bot.covered_modules(files, {'foo': 'added'}), set()) + + def test_removed_data_does_not_cover(self): + files = _files(('admin/test/cases/testdata.foo.json', 'removed')) + self.assertEqual(bot.covered_modules(files, {'foo': 'modified'}), set()) + + +class SampleDataKeysTests(unittest.TestCase): + def test_keys_collected_across_artifacts(self): + source = ( + '__artifacts_v2__ = {\n' + ' "a": {"name": "A", "sample_data": {"hickman_ios15": "5 rows"}},\n' + ' "b": {"name": "B", "sample_data": {"dexter_ios18": "1 row",\n' + ' "hickman_ios15": "2 rows"}},\n' + '}\n') + self.assertEqual(bot.sample_data_keys_from_source(source), + {'hickman_ios15', 'dexter_ios18'}) + + def test_no_sample_data_yields_empty(self): + source = '__artifacts_v2__ = {"a": {"name": "A"}}\n' + self.assertEqual(bot.sample_data_keys_from_source(source), set()) + + def test_non_literal_metadata_yields_empty(self): + source = 'V = "x"\n__artifacts_v2__ = {"a": {"name": V}}\n' + self.assertEqual(bot.sample_data_keys_from_source(source), set()) + + def test_syntax_error_yields_empty(self): + self.assertEqual(bot.sample_data_keys_from_source('def broken(:\n'), set()) + + +class ClassifyTests(unittest.TestCase): + PUBLIC = {'hickman_ios15', 'dexter_ios18'} + + def test_covered_module_needs_nothing(self): + fixture, ask = bot.classify({'foo': 'added'}, {'foo'}, {}, self.PUBLIC) + self.assertEqual((fixture, ask), ({}, [])) + + def test_public_key_routes_to_fixture(self): + fixture, ask = bot.classify({'foo': 'added'}, set(), + {'foo': {'hickman_ios15', 'private_img'}}, self.PUBLIC) + self.assertEqual(fixture, {'foo': ['hickman_ios15']}) + self.assertEqual(ask, []) + + def test_no_public_key_routes_to_ask(self): + fixture, ask = bot.classify({'foo': 'added'}, set(), + {'foo': {'private_img'}}, self.PUBLIC) + self.assertEqual((fixture, ask), ({}, ['foo'])) + + def test_mixed_pr(self): + fixture, ask = bot.classify( + {'a': 'added', 'b': 'added', 'c': 'added'}, {'c'}, + {'a': {'dexter_ios18'}, 'b': set()}, self.PUBLIC) + self.assertEqual(fixture, {'a': ['dexter_ios18']}) + self.assertEqual(ask, ['b']) + + +class RenderCommentTests(unittest.TestCase): + REPO = 'abrignoni/iLEAPP' + + def test_ask_comment_has_marker_module_and_ladder(self): + body = bot.render_comment(self.REPO, {}, ['foo']) + self.assertIn(bot.MARKER, body) + self.assertIn('`foo.py`', body) + self.assertIn('Under 10 MB', body) + self.assertIn('not a gate', body) + self.assertIn('admin/test/cases/data//', body) + + def test_fixture_only_comment_asks_nothing(self): + body = bot.render_comment(self.REPO, {'foo': ['hickman_ios15']}, []) + self.assertIn('`hickman_ios15`', body) + self.assertIn('Nothing is needed from you', body) + self.assertNotIn('Size rules', body) + + def test_resolved_comment(self): + body = bot.render_comment(self.REPO, {}, []) + self.assertIn(bot.MARKER, body) + self.assertIn('Thank you', body) + + def test_no_em_dashes_anywhere(self): + for body in (bot.render_comment(self.REPO, {}, ['foo']), + bot.render_comment(self.REPO, {'a': ['k']}, []), + bot.render_comment(self.REPO, {}, [])): + self.assertNotIn('—', body) + + +class DesiredLabelsTests(unittest.TestCase): + def test_matrix(self): + self.assertEqual(bot.desired_labels({}, []), set()) + self.assertEqual(bot.desired_labels({'a': ['k']}, []), {bot.LABEL_FIXTURE}) + self.assertEqual(bot.desired_labels({}, ['b']), {bot.LABEL_ASK}) + self.assertEqual(bot.desired_labels({'a': ['k']}, ['b']), + {bot.LABEL_FIXTURE, bot.LABEL_ASK}) + + +if __name__ == '__main__': + unittest.main() diff --git a/admin/test/scripts/test_module.py b/admin/test/scripts/test_module.py new file mode 100644 index 0000000..f245b14 --- /dev/null +++ b/admin/test/scripts/test_module.py @@ -0,0 +1,623 @@ +""" +This script provides a testing framework for iLEAPP artifact modules. +It allows for running individual artifacts or entire modules against pre-defined +test cases (stored as .zip files containing forensic evidence) and comparing +the output against expected results. + +Design Goals: +- Standalone Execution: Can be run from the command line for manual testing. +- IDE Integration: Designed to work as a task within the VS Code IDE for a + streamlined developer experience. +- CI/CD Ready: Structured to be eventually integrated into continuous integration + pipelines for automated regression testing. +""" + +import sys +import os +import zipfile +import importlib +import json +from unittest.mock import MagicMock, patch +from contextlib import ExitStack +from pathlib import Path +from datetime import datetime, timezone, date +import time +import shutil +import subprocess +import argparse +import inspect + +# Adjust import paths as necessary +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..', '..'))) +# import scripts.ilapfuncs as ilapfuncs +from scripts.context import Context + + +def mock_logdevinfo(message): + """ + Mocks the logdevinfo function to print to the console. + + Args: + message (str): The message to log. + """ + print(f"[LOGDEVINFO] {message}") + + +def mock_logfunc(message): + """ + Mocks the logfunc function to print to the console. + + Args: + message (str): The message to log. + """ + print(f"[LOGFUNC] {message}") + + +def process_artifact(zip_path, module_name, artifact_name, _artifact_data, target_os_version=None): + """ + Processes a specific artifact from a given zip file. + + Args: + zip_path (Path): Path to the zip file containing test data. + module_name (str): Name of the artifact module. + artifact_name (str): Name of the artifact function to test. + _artifact_data (dict): Metadata about the artifact from the test case (unused). + target_os_version (str, optional): OS version to mock for the test. + + Returns: + tuple: (data_headers, data_list, run_time, last_commit_info, + check_in_media_call_count, check_in_media_embedded_call_count) + """ + module = importlib.import_module(f'scripts.artifacts.{module_name}') + + # Get the function to test + func_to_test = getattr(module, artifact_name) + + # Extract the original function from the decorated one + original_func = func_to_test + while hasattr(original_func, '__wrapped__'): + original_func = original_func.__wrapped__ + + # Prepare mock objects + # mock_report_folder = 'mock_report_folder' + mock_seeker = MagicMock() + mock_wrap_text = MagicMock() + timezone_offset = 'UTC' + + # <<< NEW COUNTERS >>> + check_in_media_call_count = 0 + check_in_media_embedded_call_count = 0 + + # Capture original functions before patching if we need to call them + # original_check_in_media = ilapfuncs.check_in_media + # original_check_in_media_embedded = ilapfuncs.check_in_embedded_media + + # Configure mock_seeker.file_infos.get() to return a mock + # with a dynamic source_path based on the input extraction_path. + def mock_file_infos_get_side_effect(key_extraction_path): + mock_file_info = MagicMock() + # Use the unique extraction_path (path in temp dir for the test) + # as the source_path for hashing purposes. In a real run, + # source_path is the original path in the evidence. + mock_file_info.source_path = key_extraction_path + # Provide default datetime objects for creation and modification dates + mock_file_info.creation_date = datetime.now(timezone.utc) + mock_file_info.modification_date = datetime.now(timezone.utc) + return mock_file_info + + mock_seeker.file_infos.get.side_effect = mock_file_infos_get_side_effect + + all_files = [] + + # Create the base temp directory if it doesn't exist + base_temp_dir = Path('admin/test/temp') + base_temp_dir.mkdir(parents=True, exist_ok=True) + + # Create a unique temporary directory within the base temp directory + temp_dir = base_temp_dir / f'extract_{module_name}_{artifact_name}_{int(time.time())}' + + # Define mock_report_folder path within the temp_dir and create it + mock_report_folder_path = temp_dir / 'mock_reports' + mock_report_folder_path.mkdir(parents=True, exist_ok=True) + + # Get the module file path + module_file_path = module.__file__ + + last_commit_info = get_last_commit_info(module_file_path) + + try: + # Extract all files from the zip + with zipfile.ZipFile(zip_path, 'r') as zip_ref: + zip_ref.extractall(temp_dir) + + # Recursively get all files + for root, _, files in os.walk(temp_dir): + for file in files: + all_files.append(os.path.join(root, file)) + + # Mock for lava_db connection and cursor + mock_lava_cursor_instance = MagicMock() + mock_lava_cursor_instance.execute.return_value = mock_lava_cursor_instance + mock_lava_cursor_instance.fetchone.return_value = None + mock_lava_cursor_instance.fetchall.return_value = [] + + mock_lava_db_instance = MagicMock() + mock_lava_db_instance.cursor.return_value = mock_lava_cursor_instance + mock_lava_db_instance.commit.return_value = None + + # <<< NEW MOCKS FOR CHECK_IN_MEDIA >>> + def mocked_check_in_media(file_path, *_args, **_kwargs): + nonlocal check_in_media_call_count + check_in_media_call_count += 1 + # Simplified return for counter, avoiding deep side effects of original if problematic + return f"mock_hash_for_{os.path.basename(str(file_path))}" + + + def mocked_check_in_embedded_media(*_args, **_kwargs): + nonlocal check_in_media_embedded_call_count + check_in_media_embedded_call_count += 1 + # Similar to above, call original or return dummy + return "mock_embedded_html_path" + + def mocked_lava_get_full_media_info(_media_ref_id): + # Return a dictionary to support string indexing used in artifacts and ilapfuncs + # The last element should be a Unix timestamp for the modification date. + # Using Jan 24, 1984 - the day the first Mac went on sale. + mac_bday_ts = 443750400 + return { + 'media_ref_id': 'mock_media_ref_id', + 'media_item_id': 'mock_media_item_id', + 'module_name': 'mock_module_name', + 'artifact_name': 'mock_artifact_name', + 'name': 'mock_name', + 'source_path': 'mock_source_path', + 'extraction_path': 'mock_extraction_path', + 'type': 'image/png', + 'metadata': 'mock_metadata', + 'created_at': mac_bday_ts, + 'updated_at': mac_bday_ts, + 'is_embedded': 0 + } + + patches = [ + patch('scripts.ilapfuncs.logdevinfo', mock_logdevinfo, create=True), + patch(f'scripts.artifacts.{module_name}.logdevinfo', mock_logdevinfo, create=True), + patch(f'scripts.artifacts.{module_name}.logfunc', mock_logfunc, create=True), + patch('scripts.lavafuncs.lava_db', mock_lava_db_instance), + # <<< ADD PATCHES FOR CHECK_IN_MEDIA FUNCTIONS >>> + # Patch it in ilapfuncs (where it's defined) + patch('scripts.ilapfuncs.check_in_media', mocked_check_in_media), + patch('scripts.ilapfuncs.check_in_embedded_media', mocked_check_in_embedded_media), + # Also patch it directly in the artifact module's namespace if it's imported there like: + # from scripts.ilapfuncs import check_in_media + patch(f'scripts.artifacts.{module_name}.check_in_media', mocked_check_in_media, create=True), + patch(f'scripts.artifacts.{module_name}.check_in_embedded_media', mocked_check_in_embedded_media, create=True), + patch(f'scripts.artifacts.{module_name}.lava_get_full_media_info', mocked_lava_get_full_media_info, create=True) + + ] + + # If a target OS version is provided, mock iOS.get_version() where the + # core defines it (iLEAPP); other cores have no such class to mock. + import scripts.ilapfuncs as _ilapfuncs + if target_os_version and hasattr(_ilapfuncs, 'iOS'): + mock_ios_get_version = MagicMock(return_value=target_os_version) + patches.append(patch('scripts.ilapfuncs.iOS.get_version', mock_ios_get_version)) + + with ExitStack() as stack: + for p in patches: + stack.enter_context(p) + + all_artifacts_info = getattr(module, '__artifacts_v2__', {}) + artifact_info = all_artifacts_info.get(artifact_name, {}) + + Context.set_report_folder(str(mock_report_folder_path)) + Context.set_seeker(mock_seeker) + Context.set_files_found(all_files) + Context.set_artifact_info(artifact_info) + Context.set_module_name(module_name) + Context.set_module_file_path(module_file_path) + Context.set_artifact_name(artifact_name) + + start_time = time.time() + try: + sig = inspect.signature(original_func) + if len(sig.parameters) == 1: + data_headers, data_list, _ = original_func(Context) + else: + data_headers, data_list, _ = original_func(all_files, + str(mock_report_folder_path), + mock_seeker, + mock_wrap_text, + timezone_offset) + finally: + Context.clear() + + end_time = time.time() + + return data_headers, data_list, end_time - start_time, last_commit_info, \ + check_in_media_call_count, check_in_media_embedded_call_count + + finally: + if temp_dir.exists(): + shutil.rmtree(temp_dir, ignore_errors=True) + + +def calculate_data_size(data_list): + """ + Calculates the total size of data in a list of rows. + + Args: + data_list (list): A list of rows, where each row is a list or tuple of items. + + Returns: + int: Total size in bytes when encoded as UTF-8. + """ + return sum(len(str(item).encode('utf-8')) for row in data_list for item in row) + + +def load_test_cases(module_name): + """ + Loads test cases for a module from its JSON metadata file. + + Args: + module_name (str): Name of the module. + + Returns: + dict: Test cases data. + """ + cases_file = Path(f'admin/test/cases/testdata.{module_name}.json') + with open(cases_file, 'r', encoding='utf-8') as f: + return json.load(f) + + +def get_artifact_names(_module_name, test_cases): + """ + Retrieves all artifact names defined in the test cases for a module. + + Args: + module_name (str): Name of the module. + test_cases (dict): Test cases data. + + Returns: + list: List of artifact names. + """ + artifact_names = set() + for case in test_cases.values(): + artifact_names.update(case['artifacts'].keys()) + return list(artifact_names) + + +def select_case(test_cases): + """ + Prompts the user to select a test case from the available ones. + + Args: + test_cases (dict): Test cases data. + + Returns: + str: The selected case name, or 'all' for all cases. + """ + sorted_cases = sorted(test_cases.keys()) + valid_cases = [] + invalid_cases = [] + + for case_num in sorted_cases: + case_data = test_cases[case_num] + has_files = any(artifact.get('file_count', 0) > 0 + for artifact in case_data['artifacts'].values()) + if has_files: + valid_cases.append(case_num) + else: + invalid_cases.append(case_num) + + if not valid_cases: + print("No valid test cases with responsive files found for this module.") + return None + + if invalid_cases: + print("Test cases with no responsive files:") + for case_num in invalid_cases: + print(f"- {case_num}") + print() + + print("Available test cases:") + for i, case_num in enumerate(valid_cases, 1): + case_data = test_cases[case_num] + input_path = case_data.get('make_data', {}).get('input_data_path', 'N/A') + input_filename = os.path.basename(input_path) + description = case_data.get('description', 'No description') + print(f"{i}. {case_num}") + print(f" Input: {input_filename}") + print(f" Description: {description}") + print() + + print("\nEnter case number, name, or press Enter for all cases (Ctrl+C to exit):") + try: + case_choice = input().strip().lower() + if case_choice == '' or case_choice == 'all': + return 'all' + try: + index = int(case_choice) - 1 + if 0 <= index < len(valid_cases): + return valid_cases[index] + except ValueError: + if case_choice in valid_cases: + return case_choice + print("Invalid choice. Please try again.") + return select_case(test_cases) + except KeyboardInterrupt: + print("\nExiting...") + sys.exit(0) + + +def select_artifact(artifact_names, test_cases): + """ + Prompts the user to select an artifact to test. + + Args: + artifact_names (list): List of all artifact names in the module. + test_cases (dict): Test cases data. + + Returns: + str: The selected artifact name, or 'all' for all artifacts. + """ + artifacts_with_data = [] + artifacts_without_data = [] + sorted_artifacts = sorted(artifact_names) + + for name in sorted_artifacts: + has_data = False + for case_data in test_cases.values(): + if name in case_data['artifacts'] and \ + case_data['artifacts'][name].get('file_count', 0) > 0: + has_data = True + break + if has_data: + artifacts_with_data.append(name) + else: + artifacts_without_data.append(name) + + if artifacts_without_data: + print("Artifacts with no test data available:") + for name in artifacts_without_data: + print(f"- {name}") + print() + + if not artifacts_with_data: + print("No test data found for any artifacts in this module. Exiting.") + sys.exit(0) + + print("Available artifacts with test data:") + for i, name in enumerate(artifacts_with_data, 1): + print(f"{i}. {name}") + + print("\nEnter artifact number, name, or press Enter for all artifacts " + "(Ctrl+C to exit):") + try: + artifact_choice = input().strip().lower() + if artifact_choice == '' or artifact_choice == 'all': + return 'all' + try: + index = int(artifact_choice) - 1 + if 0 <= index < len(artifacts_with_data): + return artifacts_with_data[index] + except ValueError: + if artifact_choice in artifacts_with_data: + return artifact_choice + print("Invalid choice. Please try again.") + return select_artifact(artifact_names, test_cases) + except KeyboardInterrupt: + print("\nExiting...") + sys.exit(0) + + +def convert_to_unix_time(value): + """ + Converts a datetime or date object to a Unix timestamp. + + Args: + value: The object to convert. + + Returns: + int or object: Unix timestamp if input was datetime/date, otherwise original value. + """ + if isinstance(value, (datetime, date)): + return int(value.timestamp()) + return value + + +def process_data(headers, data): + """ + Processes artifact output data for comparison, handling datetime conversions. + + Args: + headers (list): Artifact output headers. + data (list): Artifact output data rows. + + Returns: + tuple: (processed_headers, processed_data) + """ + datetime_indices = [i for i, header in enumerate(headers) if isinstance(header, tuple) + and header[1].lower() in ['datetime', 'date']] + + processed_headers = [header[0] if isinstance(header, tuple) else header for header in headers] + processed_data = [] + + for row in data: + processed_row = [] + for i, value in enumerate(row): + if i in datetime_indices: + processed_row.append(convert_to_unix_time(value)) + elif isinstance(value, (datetime, date)): + processed_row.append(convert_to_unix_time(value)) + else: + processed_row.append(value) + processed_data.append(processed_row) + + return processed_headers, processed_data + + +def main(module_name, artifact_name=None, case_number=None): + """ + Main entry point for testing module artifacts. + + Args: + module_name (str): Name of the module to test. + artifact_name (str, optional): Name of the artifact to test. + case_number (str, optional): Case number to test. + """ + try: + test_cases = load_test_cases(module_name) + artifact_names = get_artifact_names(module_name, test_cases) + + if artifact_name is None: + artifact_name = select_artifact(artifact_names, test_cases) + elif artifact_name.lower() == 'all': + artifact_name = 'all' + + if case_number is None: + case_number = select_case(test_cases) + elif case_number.lower() == 'all': + case_number = 'all' + + if case_number is None: + print("No valid test cases available. Exiting.") + return + + cases_to_process = [case_number] if case_number != 'all' \ + else [case for case in test_cases.keys() + if any(artifact.get('file_count', 0) > 0 + for artifact in test_cases[case]['artifacts'].values())] + artifacts_to_process = [artifact_name] if artifact_name != 'all' else artifact_names + + module = importlib.import_module(f'scripts.artifacts.{module_name}') + artifacts_info = getattr(module, '__artifacts_v2__', {}) + + for case in cases_to_process: + case_data = test_cases[case] + for artifact in artifacts_to_process: + if artifact in case_data['artifacts']: + artifact_data_for_function = case_data['artifacts'][artifact] # Renamed to avoid conflict + + if artifact_data_for_function.get('file_count', 0) == 0: + print(f"\nSkipping artifact: {artifact} for case: {case} (no files found)") + continue + + print(f"\nTesting artifact: {artifact} for case: {case}") + zip_path = Path('admin/test/cases/data') / module_name / f"testdata.{module_name}.{artifact}.{case}.zip" + artifact_info_v2 = artifacts_info.get(artifact, {}) # Renamed to avoid conflict + start_datetime = datetime.now(timezone.utc) + + # Extract os_version from case_data + image_info = case_data.get("image_info", {}) + current_os_version = image_info.get("os_version") + if current_os_version: + print(f"Using OS version from test case data for {case}: {current_os_version}") + else: + print(f"Warning: 'os_version' not found in image_info for {case}. " + "iOS.get_version() will not be specifically mocked.") + + headers, data, run_time, last_commit_info, media_checkins_count, media_embedded_checkins_count = \ + process_artifact(zip_path, module_name, artifact, artifact_data_for_function, + target_os_version=current_os_version) + + processed_headers, processed_data = process_data(headers, data) + + end_datetime = datetime.now(timezone.utc) + + result = { + "metadata": { + "module_name": module_name, + "artifact_name": artifact_info_v2.get('name', artifact), + "function_name": artifact, + "case_number": case, + "number_of_columns": len(processed_headers), + "number_of_rows": len(processed_data), + "total_data_size_bytes": calculate_data_size(processed_data), + "media_checkins": media_checkins_count, + "media_embedded_checkins": media_embedded_checkins_count, + "input_zip_path": str(zip_path), + "start_time": start_datetime.isoformat(), + "end_time": end_datetime.isoformat(), + "run_time_seconds": run_time, + "last_commit": last_commit_info + }, + "headers": processed_headers, + "data": processed_data + } + + output_dir = Path('admin/test/results') / module_name + output_dir.mkdir(parents=True, exist_ok=True) + output_file = output_dir / f"{module_name}.{artifact}.{case}.{start_datetime.strftime('%Y%m%d%H%M%S')}.json" + + with open(output_file, 'w', encoding='utf-8') as f: + json.dump(result, f, indent=2, default=str) + + print(f"Test results for {module_name} - {artifact} - Case {case} saved to {output_file}") + print(f"Processed {len(processed_data)} rows in {run_time:.2f} seconds. " + f"Media Checkins: {media_checkins_count}. " + f"Media Embedded Checkins: {media_embedded_checkins_count}.") + else: + print(f"\nSkipping artifact: {artifact} for case: {case} (not found in test data)") + + print("\nTesting completed.") + + except KeyboardInterrupt: + print("\nExiting...") + sys.exit(0) + + +def get_last_commit_info(file_path): + """ + Retrieves the last git commit information for a given file. + + Args: + file_path (str): Path to the file. + + Returns: + dict: Dictionary containing commit hash, author, date, and message. + """ + try: + # Get the last commit hash + git_log = subprocess.check_output( + ['git', 'log', '-n', '1', '--pretty=format:%H|%an|%ae|%ad|%s', '--', file_path], + universal_newlines=True).strip() + if not git_log: + # File is not yet in git history + return { + 'hash': 'Uncommitted', + 'author_name': 'N/A', + 'author_email': 'N/A', + 'date': datetime.now().isoformat(), + 'message': 'File not yet committed to git' + } + commit_hash, author_name, author_email, commit_date, commit_message = git_log.split('|') + + # Convert the commit date to ISO format + commit_date = datetime.strptime( + commit_date, '%a %b %d %H:%M:%S %Y %z').isoformat() + + return { + 'hash': commit_hash, + 'author_name': author_name, + 'author_email': author_email, + 'date': commit_date, + 'message': commit_message + } + except subprocess.CalledProcessError: + return None + + +if __name__ == '__main__': + parser = argparse.ArgumentParser(description="Test module artifacts") + parser.add_argument("module_name", help="Name of the module to test") + parser.add_argument("-a", "--artifact", + help="Name of the artifact to test (or 'all' for all artifacts)", + default=None) + parser.add_argument("-c", "--case", + help="Case number to test (or 'all' for all cases)", + default=None) + + args = parser.parse_args() + + main(args.module_name, args.artifact, args.case) diff --git a/admin/test/scripts/test_run_test_cases.py b/admin/test/scripts/test_run_test_cases.py new file mode 100644 index 0000000..f00bae8 --- /dev/null +++ b/admin/test/scripts/test_run_test_cases.py @@ -0,0 +1,91 @@ +"""Tests for the pure logic in run_test_cases.py. + +The artifact-execution path is exercised by the runner itself in CI; these +cover baseline selection, normalization, and the comparison verdicts against +synthetic inputs only. +""" +import json +import os +import sys +import tempfile +import unittest +from pathlib import Path + +sys.path.append(os.path.dirname(__file__)) + +import run_test_cases as rtc # noqa: E402 pylint: disable=wrong-import-position + + +class NormalizeRowsTests(unittest.TestCase): + def test_temp_extract_path_is_tokenized(self): + rows = [["admin/test/temp/extract_foo_bar_1750464339/x/y.plist", 1]] + other = [["admin/test/temp/extract_foo_bar_1999999999/x/y.plist", 1]] + self.assertEqual(rtc.normalize_rows(rows), rtc.normalize_rows(other)) + + def test_row_order_does_not_matter(self): + self.assertEqual(rtc.normalize_rows([[1, "a"], [2, "b"]]), + rtc.normalize_rows([[2, "b"], [1, "a"]])) + + def test_content_change_is_detected(self): + self.assertNotEqual(rtc.normalize_rows([[1, "a"]]), + rtc.normalize_rows([[1, "b"]])) + + def test_tuple_and_list_rows_normalize_alike(self): + self.assertEqual(rtc.normalize_rows([(1, "a")]), rtc.normalize_rows([[1, "a"]])) + + +class CompareTests(unittest.TestCase): + BASE = {"headers": ["A", "B"], "data": [[1, "x"], [2, "y"]]} + + def test_match_returns_no_problems(self): + self.assertEqual(rtc.compare(["A", "B"], [[2, "y"], [1, "x"]], self.BASE), []) + + def test_header_change_reported(self): + problems = rtc.compare(["A", "C"], [[1, "x"], [2, "y"]], self.BASE) + self.assertTrue(any("headers differ" in p for p in problems)) + + def test_row_change_reported_with_counts(self): + problems = rtc.compare(["A", "B"], [[1, "x"], [2, "z"]], self.BASE) + joined = "\n".join(problems) + self.assertIn("rows differ", joined) + self.assertIn("1 new", joined) + self.assertIn("1 missing", joined) + + +class LatestBaselineTests(unittest.TestCase): + def test_picks_newest_snapshot(self): + with tempfile.TemporaryDirectory() as tmp: + mod_dir = Path(tmp) / "results" / "m" + mod_dir.mkdir(parents=True) + for stamp in ("20240101000000", "20250101000000", "20230101000000"): + (mod_dir / f"m.a.case1.{stamp}.json").write_text("{}", encoding="utf-8") + (mod_dir / "m.a.case2.20990101000000.json").write_text("{}", encoding="utf-8") + old_results = rtc.RESULTS_DIR + rtc.RESULTS_DIR = Path(tmp) / "results" + try: + chosen = rtc.latest_baseline("m", "a", "case1") + finally: + rtc.RESULTS_DIR = old_results + self.assertEqual(chosen.name, "m.a.case1.20250101000000.json") + + def test_none_when_absent(self): + with tempfile.TemporaryDirectory() as tmp: + old_results = rtc.RESULTS_DIR + rtc.RESULTS_DIR = Path(tmp) + try: + self.assertIsNone(rtc.latest_baseline("m", "a", "c")) + finally: + rtc.RESULTS_DIR = old_results + + +class BaselineJsonRoundTripTests(unittest.TestCase): + def test_fresh_python_values_match_their_json_serialized_form(self): + from datetime import datetime, timezone + ts = int(datetime(2024, 5, 1, tzinfo=timezone.utc).timestamp()) + fresh = [[ts, b"bytes-value"]] + recorded = json.loads(json.dumps({"data": fresh}, default=str))["data"] + self.assertEqual(rtc.normalize_rows(fresh), rtc.normalize_rows(recorded)) + + +if __name__ == "__main__": + unittest.main()