diff --git a/.github/workflows/generate-hacs-data.yml b/.github/workflows/generate-hacs-data.yml index 4117d1c5a5a..ed536eb4ff3 100644 --- a/.github/workflows/generate-hacs-data.yml +++ b/.github/workflows/generate-hacs-data.yml @@ -42,26 +42,50 @@ jobs: runs-on: ubuntu-latest if: github.repository == 'hacs/integration' outputs: + shards: ${{ steps.set-matrix.outputs.shards }} categories: ${{ steps.set-matrix.outputs.categories }} steps: - id: set-matrix + env: + EVENT_NAME: ${{ github.event_name }} + INPUT_CATEGORY: ${{ inputs.category }} run: | - if [[ "${{ github.event_name }}" == "workflow_dispatch" ]] && [[ "${{ inputs.category }}" != "None" ]] && [[ "${{ inputs.category }}" != "" ]]; then - echo "categories=['${{ inputs.category }}']" >> $GITHUB_OUTPUT + # Number of shards each category is split into (default 1). + declare -A shard_counts=( [integration]=3 [plugin]=2 ) + + if [[ "$EVENT_NAME" == "workflow_dispatch" ]] && [[ "$INPUT_CATEGORY" != "None" ]] && [[ -n "$INPUT_CATEGORY" ]]; then + categories=("$INPUT_CATEGORY") else - echo "categories=['appdaemon','integration','plugin','python_script','template','theme']" >> $GITHUB_OUTPUT + categories=(appdaemon integration plugin python_script template theme) fi + shards_json="[]" + for category in "${categories[@]}"; do + count=${shard_counts[$category]:-1} + for ((shard=1; shard<=count; shard++)); do + shards_json=$(jq -c \ + --arg c "$category" --argjson s "$shard" --argjson t "$count" \ + '. + [{category: $c, shard: $s, shards: $t}]' <<< "$shards_json") + done + done + + categories_json=$(printf '%s\n' "${categories[@]}" | jq -R . | jq -cs .) + + echo "shards=$shards_json" >> $GITHUB_OUTPUT + echo "categories=$categories_json" >> $GITHUB_OUTPUT + echo "shards: $shards_json" + echo "categories: $categories_json" + category-data: runs-on: ubuntu-latest needs: generate-matrix if: github.repository == 'hacs/integration' - name: Generate ${{ matrix.category }} data + name: Generate ${{ matrix.category }} data (shard ${{ matrix.shard }}/${{ matrix.shards }}) strategy: fail-fast: false max-parallel: 1 matrix: - category: ${{ fromJSON( needs.generate-matrix.outputs.categories )}} + include: ${{ fromJSON( needs.generate-matrix.outputs.shards )}} steps: - name: Checkout the repository uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0 @@ -83,12 +107,64 @@ jobs: scripts/install/frontend scripts/install/pip_packages --requirement requirements_generate_data.txt - - name: Generate ${{ matrix.category }} data - run: python3 -m scripts.data.generate_category_data ${{ matrix.category }} + - name: Generate ${{ matrix.category }} data (shard ${{ matrix.shard }}/${{ matrix.shards }}) + run: python3 -m scripts.data.generate_category_data ${{ matrix.category }} --shard ${{ matrix.shard }}/${{ matrix.shards }} env: DATA_GENERATOR_TOKEN: ${{ secrets.DATA_GENERATOR_TOKEN }} FORCE_REPOSITORY_UPDATE: ${{ inputs.forceRepositoryUpdate }} + - name: Upload shard artifacts + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 + with: + name: shard-${{ matrix.category }}-${{ matrix.shard }} + path: outputdata/_shards/${{ matrix.category }}/${{ matrix.shard }} + if-no-files-found: error + retention-days: 1 + + merge: + runs-on: ubuntu-latest + needs: [generate-matrix, category-data] + if: github.repository == 'hacs/integration' + name: Merge ${{ matrix.category }} data + strategy: + fail-fast: false + matrix: + category: ${{ fromJSON( needs.generate-matrix.outputs.categories )}} + steps: + - name: Checkout the repository + uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0 + with: + persist-credentials: false + + - name: Set up Python + uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6.3.0 + id: python + with: + python-version: "3.14" + cache: 'pip' + cache-dependency-path: | + requirements_base.txt + requirements_generate_data.txt + + - name: Install dependencies + run: | + scripts/install/frontend + scripts/install/pip_packages --requirement requirements_generate_data.txt + + # Each shard artifact contains data.json + stored.json at its root. Without + # merge-multiple, download-artifact nests each artifact in its own + # name-scoped subdirectory, so the per-shard files never collide. + - name: Download shard artifacts + uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8.0.1 + with: + pattern: shard-${{ matrix.category }}-* + path: outputdata/_shards/${{ matrix.category }} + + - name: Merge ${{ matrix.category }} shards + run: python3 -m scripts.data.merge_category_data ${{ matrix.category }} + env: + DATA_GENERATOR_TOKEN: ${{ secrets.DATA_GENERATOR_TOKEN }} + - name: Validate output with JQ run: | jq -c . outputdata/${{ matrix.category }}/data.json @@ -129,7 +205,7 @@ jobs: summarize: name: Summarize runs-on: ubuntu-latest - needs: category-data + needs: merge if: ${{ always() && github.repository == 'hacs/integration' }} outputs: changedCategories: ${{ steps.combined.outputs.changedCategories }} @@ -159,6 +235,7 @@ jobs: const subDirectories = fs.readdirSync("outputdata", { withFileTypes: true }) .filter(entry => entry.isDirectory()) .map(entry => entry.name) + .filter(directory => fs.existsSync(`outputdata/${directory}/summary.json`)) for (const directory of subDirectories) { let changedPctTarget = Number(process.env.HACS_CHANGED_PCT_TARGET) @@ -287,7 +364,7 @@ jobs: notify_on_failure: runs-on: ubuntu-latest name: Trigger Discord notification when jobs fail - needs: ["generate-matrix", "category-data", "summarize", "publish"] + needs: ["generate-matrix", "category-data", "merge", "summarize", "publish"] if: ${{ always() && github.repository == 'hacs/integration' && contains(join(needs.*.result, ','), 'failure') && github.event_name == 'schedule' }} steps: - name: Send notification diff --git a/scripts/data/generate_category_data.py b/scripts/data/generate_category_data.py index fe04c61df21..f9463a6d714 100644 --- a/scripts/data/generate_category_data.py +++ b/scripts/data/generate_category_data.py @@ -2,8 +2,10 @@ from __future__ import annotations +import argparse import asyncio from datetime import datetime +import hashlib import json import logging import os @@ -56,9 +58,37 @@ log_handler.addHandler(stream_handler) OUTPUT_DIR = os.path.join(os.getcwd(), "outputdata") +SHARDS_DIR = os.path.join(OUTPUT_DIR, "_shards") COMPARE_IGNORE = {"etag_releases", "etag_repository", "last_fetched"} +def shard_for(full_name: str, shards: int) -> int: + """Return the (0-based) shard a repository belongs to. + + Uses a stable, salt-free hash so the assignment is identical across + processes and runs (unlike the built-in ``hash``). + """ + if shards <= 1: + return 0 + digest = hashlib.md5(full_name.lower().encode()).digest() + return int.from_bytes(digest[:8], "big") % shards + + +def _slice_by_shard( + current_data: dict[str, dict[str, Any]], + index: int, + shards: int, +) -> dict[str, dict[str, Any]]: + """Return the subset of ``current_data`` that belongs to the given shard.""" + if shards <= 1: + return current_data + return { + key: value + for key, value in current_data.items() + if shard_for(value["full_name"], shards) == index + } + + def jsonprint(data: any): print( json.dumps( @@ -304,8 +334,15 @@ async def generate_data_for_category( repository_name: str | None, current_data: dict[str, dict[str, Any]], force: bool, + index: int = 0, + shards: int = 1, ) -> dict[str, dict[str, Any]]: - """Generate data for category.""" + """Generate data for category. + + When ``shards`` > 1 only the repositories belonging to shard ``index`` + (0-based) are processed and stored, so the result is a disjoint slice of + the full category data. + """ removed = ( [] if repository_name is not None @@ -313,11 +350,11 @@ async def generate_data_for_category( ) await self.data.register_base_data( category, - {} if force else current_data, + {} if force else _slice_by_shard(current_data, index, shards), removed, ) self.queue.clear() - await self.get_category_repositories(category, repository_name, removed) + await self.get_category_repositories(category, repository_name, removed, index, shards) async def _handle_queue(): if not self.queue.pending_tasks: @@ -357,6 +394,8 @@ async def get_category_repositories( category: str, repository_name: str | None, removed: list[str], + index: int = 0, + shards: int = 1, ) -> None: """Get repositories from category.""" repositories = ( @@ -371,6 +410,11 @@ async def get_category_repositories( # hacs/integration i not in the default file, but it's still needed repositories.append("hacs/integration") + if shards > 1: + repositories = [ + repo for repo in repositories if shard_for(repo, shards) == index + ] + for repo in repositories: if repo in removed: self.log.warning("Skipping %s as it's removed from HACS", repo) @@ -453,13 +497,183 @@ async def async_github_get_hacs_default_file(self, filename: str) -> list: return json_loads(decode_content(response.data.content)) -async def generate_category_data(category: str, repository_name: str = None): - """Generate data.""" +def _dump_diff(updated: dict[str, dict[str, Any]], data_file) -> None: + """Write a diff-friendly (COMPARE_IGNORE stripped) JSON dump.""" + json.dump( + { + i: {k: v for k, v in d.items() if k not in COMPARE_IGNORE} + for i, d in updated.items() + }, + data_file, + cls=JSONEncoder, + sort_keys=True, + indent=2, + ) + + +async def finalize_category_output( + hacs: AdjustedHacs, + category: str, + stored_data: dict[str, dict[str, Any]], + current_data: dict[str, dict[str, Any]], + updated_data: dict[str, dict[str, Any]], +) -> None: + """Summarize, validate and write the final per-category output files. + + Runs once on the full (merged) dataset, so the summary, diff and validation + are computed against the complete category data. + """ + os.makedirs(os.path.join(OUTPUT_DIR, category), exist_ok=True) + os.makedirs(os.path.join(OUTPUT_DIR, "diff"), exist_ok=True) + + summary = await hacs.summarize_data(current_data, updated_data) + with open( + os.path.join(OUTPUT_DIR, "summary.json"), + mode="w", + encoding="utf-8", + ) as data_file: + json.dump( + summary, + data_file, + cls=JSONEncoder, + sort_keys=True, + indent=2, + ) + + did_raise = False + if ( + not updated_data + or len(updated_data) == 0 + or not isinstance(updated_data, dict) + ): + print_error_and_exit("Updated data is empty", category) + did_raise = True + + try: + VALIDATE_GENERATED_V2_REPO_DATA[category](updated_data) + except vol.Invalid as error: + did_raise = True + errors = expand_and_humanize_error(updated_data, error) + if isinstance(errors, list): + for err in errors: + print(f"::error::{err}") + sys.exit(1) + + print_error_and_exit(f"Invalid data: {errors}", category) + + if did_raise: + print_error_and_exit( + "Validation did raise but did not exit!", category) + sys.exit(1) # Fallback, should not be reached + + # All three files are published to R2, and under sharding the merged + # insertion order depends on shard-assembly order. Sort so the output is + # deterministic regardless of how the data was produced. (sort_keys sorts + # the repo-id keys of the dicts; repositories.json is an explicit list that + # sort_keys would not touch, so it is sorted directly.) + with open( + os.path.join(OUTPUT_DIR, category, "stored.json"), + mode="w", + encoding="utf-8", + ) as data_file: + json.dump( + stored_data, + data_file, + cls=JSONEncoder, + separators=(",", ":"), + sort_keys=True, + ) + with open( + os.path.join(OUTPUT_DIR, category, "data.json"), + mode="w", + encoding="utf-8", + ) as data_file: + json.dump( + updated_data, + data_file, + cls=JSONEncoder, + separators=(",", ":"), + sort_keys=True, + ) + with open( + os.path.join(OUTPUT_DIR, category, "repositories.json"), + mode="w", + encoding="utf-8", + ) as repositories_file: + json.dump( + sorted(v["full_name"] for v in updated_data.values()), + repositories_file, + separators=(",", ":"), + ) + + with open( + os.path.join(OUTPUT_DIR, "diff", f"{category}_before.json"), + mode="w", + encoding="utf-8", + ) as data_file: + _dump_diff(current_data, data_file) + + with open( + os.path.join(OUTPUT_DIR, "diff", f"{category}_after.json"), + mode="w", + encoding="utf-8", + ) as data_file: + _dump_diff(updated_data, data_file) + + +def write_shard_output( + category: str, + shard_number: int, + updated_data: dict[str, dict[str, Any]], + stored_slice: dict[str, dict[str, Any]], +) -> None: + """Write a shard's partial data to a shard-scoped path for later merging.""" + shard_dir = os.path.join(SHARDS_DIR, category, str(shard_number)) + os.makedirs(shard_dir, exist_ok=True) + with open( + os.path.join(shard_dir, "data.json"), + mode="w", + encoding="utf-8", + ) as data_file: + json.dump( + updated_data, + data_file, + cls=JSONEncoder, + separators=(",", ":"), + ) + with open( + os.path.join(shard_dir, "stored.json"), + mode="w", + encoding="utf-8", + ) as data_file: + json.dump( + stored_slice, + data_file, + cls=JSONEncoder, + separators=(",", ":"), + ) + + +async def generate_category_data( + category: str, + repository_name: str | None = None, + shard: tuple[int, int] | None = None, +) -> None: + """Generate data. + + ``shard`` is a 1-based ``(number, total)`` pair, or ``None`` for an + unsharded run. When ``shard`` is ``None`` the full category output (summary, + diff and validation) is produced, matching the original behaviour. When a + shard is given only that shard's disjoint slice is generated and written as + a partial artifact for a later merge step -- this holds even for ``1/1``, so + every sharded invocation goes through the merge step uniformly. + """ + sharded = shard is not None + shard_number, shards = shard if sharded else (1, 1) + index = shard_number - 1 async with ClientSession() as session: hacs = AdjustedHacs( session=session, token=os.getenv("DATA_GENERATOR_TOKEN")) - os.makedirs(os.path.join(OUTPUT_DIR, category), exist_ok=True) - os.makedirs(os.path.join(OUTPUT_DIR, "diff"), exist_ok=True) force = os.environ.get("FORCE_REPOSITORY_UPDATE") == "True" stored_data = await hacs.data_client.get_data(category, validate=False) current_data = ( @@ -480,125 +694,66 @@ async def generate_category_data(category: str, repository_name: str = None): repository_name, current_data, force=force, + index=index, + shards=shards, ) - summary = await hacs.summarize_data(current_data, updated_data) - with open( - os.path.join(OUTPUT_DIR, "summary.json"), - mode="w", - encoding="utf-8", - ) as data_file: - json.dump( - summary, - data_file, - cls=JSONEncoder, - sort_keys=True, - indent=2, - ) - - did_raise = False - if ( - not updated_data - or len(updated_data) == 0 - or not isinstance(updated_data, dict) - ): - print_error_and_exit("Updated data is empty", category) - did_raise = True - - try: - VALIDATE_GENERATED_V2_REPO_DATA[category](updated_data) - except vol.Invalid as error: - did_raise = True - errors = expand_and_humanize_error(updated_data, error) - if isinstance(errors, list): - for err in errors: - print(f"::error::{err}") - sys.exit(1) - - print_error_and_exit(f"Invalid data: {errors}", category) - - if did_raise: - print_error_and_exit( - "Validation did raise but did not exit!", category) - sys.exit(1) # Fallback, should not be reached - - with open( - os.path.join(OUTPUT_DIR, category, "stored.json"), - mode="w", - encoding="utf-8", - ) as data_file: - json.dump( - stored_data, - data_file, - cls=JSONEncoder, - separators=(",", ":"), - ) - with open( - os.path.join(OUTPUT_DIR, category, "data.json"), - mode="w", - encoding="utf-8", - ) as data_file: - json.dump( + if sharded: + write_shard_output( + category, + shard_number, updated_data, - data_file, - cls=JSONEncoder, - separators=(",", ":"), - ) - with open( - os.path.join(OUTPUT_DIR, category, "repositories.json"), - mode="w", - encoding="utf-8", - ) as repositories_file: - json.dump( - [v["full_name"] for v in updated_data.values()], - repositories_file, - separators=(",", ":"), - sort_keys=True, + _slice_by_shard(stored_data, index, shards), ) + return - with open( - os.path.join(OUTPUT_DIR, "diff", f"{category}_before.json"), - mode="w", - encoding="utf-8", - ) as data_file: - json.dump( - { - i: { - k: v - for k, v in d.items() if k not in COMPARE_IGNORE - } - for i, d in current_data.items() - }, - data_file, - cls=JSONEncoder, - sort_keys=True, - indent=2, - ) + await finalize_category_output( + hacs, + category, + stored_data, + current_data, + updated_data, + ) - with open( - os.path.join(OUTPUT_DIR, "diff", f"{category}_after.json"), - mode="w", - encoding="utf-8", - ) as data_file: - json.dump( - { - i: { - k: v - for k, v in d.items() if k not in COMPARE_IGNORE - } - for i, d in updated_data.items() - }, - data_file, - cls=JSONEncoder, - sort_keys=True, - indent=2, - ) + +def _parse_shard(value: str) -> tuple[int, int]: + """Parse a ``x/y`` (1-based) shard argument into an ``(x, y)`` tuple.""" + parts = value.split("/") + if len(parts) != 2: + raise argparse.ArgumentTypeError( + f"Invalid shard '{value}', expected format 'x/y' (e.g. 1/3)" + ) + try: + number, total = int(parts[0]), int(parts[1]) + except ValueError: + raise argparse.ArgumentTypeError( + f"Invalid shard '{value}', expected format 'x/y' (e.g. 1/3)" + ) from None + if total < 1 or number < 1 or number > total: + raise argparse.ArgumentTypeError( + f"Invalid shard '{value}', require 1 <= x <= y" + ) + return (number, total) if __name__ == "__main__": + parser = argparse.ArgumentParser(description="Generate HACS compliant data.") + parser.add_argument("category") + parser.add_argument("repository_name", nargs="?", default=None) + parser.add_argument( + "--shard", + type=_parse_shard, + default=None, + help=( + "Shard to generate as 'x/y' (1-based), e.g. --shard 1/3. " + "Omit for a full, unsharded run." + ), + ) + args = parser.parse_args() asyncio.run( generate_category_data( - sys.argv[1], # category - sys.argv[2] if len(sys.argv) > 2 else None, # repository_name + args.category, + args.repository_name, + shard=args.shard, ) ) diff --git a/scripts/data/merge_category_data.py b/scripts/data/merge_category_data.py new file mode 100644 index 00000000000..44d6de615b3 --- /dev/null +++ b/scripts/data/merge_category_data.py @@ -0,0 +1,70 @@ +"""Merge sharded HACS data into a single per-category output. + +Each shard of a category produces a partial ``data.json`` (its slice of the +generated data) and a partial ``stored.json`` (its slice of the currently +published data) under ``outputdata/_shards///``. This script +concatenates those partials back into the full dataset and runs the shared +``finalize_category_output`` step, which computes the summary and diff and +validates the complete category data. +""" + +from __future__ import annotations + +import argparse +import asyncio +import glob +import json +import os +from typing import Any + +from aiohttp import ClientSession + +from .common import print_error_and_exit +from .generate_category_data import ( + SHARDS_DIR, + AdjustedHacs, + finalize_category_output, +) + + +def _load_shard_files(category: str, name: str) -> dict[str, dict[str, Any]]: + """Merge a given partial file across all shards of a category.""" + merged: dict[str, dict[str, Any]] = {} + paths = sorted(glob.glob(os.path.join(SHARDS_DIR, category, "*", name))) + if not paths: + print_error_and_exit(f"No shard '{name}' files found", category) + for path in paths: + with open(path, encoding="utf-8") as shard_file: + shard_data = json.load(shard_file) + overlap = merged.keys() & shard_data.keys() + if overlap: + print_error_and_exit( + f"Duplicate keys across shards in '{name}': {sorted(overlap)[:5]}", + category, + ) + merged.update(shard_data) + return merged + + +async def merge_category_data(category: str) -> None: + """Merge all shards for a category and write the final output.""" + updated_data = _load_shard_files(category, "data.json") + stored_data = _load_shard_files(category, "stored.json") + + async with ClientSession() as session: + hacs = AdjustedHacs( + session=session, token=os.getenv("DATA_GENERATOR_TOKEN")) + await finalize_category_output( + hacs, + category, + stored_data, + stored_data, + updated_data, + ) + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description="Merge sharded HACS data.") + parser.add_argument("category") + args = parser.parse_args() + asyncio.run(merge_category_data(args.category)) diff --git a/tests/scripts/data/test_generate_category_data.py b/tests/scripts/data/test_generate_category_data.py index 6b602d9feb8..d7bfcd2498b 100644 --- a/tests/scripts/data/test_generate_category_data.py +++ b/tests/scripts/data/test_generate_category_data.py @@ -1,14 +1,25 @@ """Test generate category data.""" +import argparse import asyncio import json import os +import shutil from typing import Any from homeassistant.core import HomeAssistant import pytest -from scripts.data.generate_category_data import OUTPUT_DIR, generate_category_data +from scripts.data.generate_category_data import ( + OUTPUT_DIR, + SHARDS_DIR, + _parse_shard, + _slice_by_shard, + generate_category_data, + shard_for, + write_shard_output, +) +from scripts.data.merge_category_data import _load_shard_files from tests.common import ( FIXTURES_PATH, @@ -311,3 +322,99 @@ async def test_generate_category_data_with_30plus_prereleases( f"scripts/data/test_generate_category_data_with_30plus_prereleases/{ category_test_data['category']}.json", ) + + +@pytest.mark.parametrize( + ("full_name", "shards", "expected"), + [ + ("hacs/integration", 3, 2), + ("hacs/integration", 2, 1), + ("hacs-test-org/integration-basic", 3, 1), + ("octocat/Hello-World", 3, 1), + ("octocat/Hello-World", 2, 0), + ("anything", 1, 0), + ("anything", 0, 0), + ], +) +def test_shard_for(full_name: str, shards: int, expected: int): + """Shard assignment is stable and matches precomputed values.""" + assert shard_for(full_name, shards) == expected + + +def test_shard_for_is_case_insensitive(): + """Repository casing must not change the shard assignment.""" + assert shard_for("HACS/Integration", 3) == shard_for("hacs/integration", 3) + + +def test_shard_for_partitions_all_buckets(): + """Every shard is used and assignments stay within range.""" + names = [f"user{i}/repo{i}" for i in range(500)] + for shards in (2, 3): + assignments = [shard_for(name, shards) for name in names] + assert set(assignments) == set(range(shards)) + + +@pytest.mark.parametrize( + ("value", "expected"), + [("1/3", (1, 3)), ("3/3", (3, 3)), ("1/1", (1, 1))], +) +def test_parse_shard_valid(value: str, expected: tuple[int, int]): + assert _parse_shard(value) == expected + + +@pytest.mark.parametrize("value", ("0/3", "4/3", "3/0", "abc", "1/2/3", "1", "-1/3")) +def test_parse_shard_invalid(value: str): + with pytest.raises(argparse.ArgumentTypeError): + _parse_shard(value) + + +def test_slice_by_shard_is_disjoint_and_complete(): + """Slices are disjoint, cover everything, and match the shard function.""" + data = {str(i): {"full_name": f"user{i}/repo{i}"} for i in range(200)} + shards = 3 + slices = [_slice_by_shard(data, index, shards) for index in range(shards)] + + keys = [set(entry) for entry in slices] + assert set.union(*keys) == set(data) + assert sum(len(entry) for entry in slices) == len(data) + for first in range(shards): + for second in range(first + 1, shards): + assert keys[first].isdisjoint(keys[second]) + + for index, entries in enumerate(slices): + for value in entries.values(): + assert shard_for(value["full_name"], shards) == index + + +def test_slice_by_shard_single_shard_returns_all(): + data = {"1": {"full_name": "a/b"}} + assert _slice_by_shard(data, 0, 1) == data + + +@pytest.mark.parametrize("shards", (1, 2, 3)) +def test_write_and_merge_shard_partials_roundtrip(shards: int): + """Writing disjoint shard partials and merging reassembles the full set. + + This mirrors the production shard -> merge data flow (``write_shard_output`` + followed by ``_load_shard_files``) without touching the network, so it needs + no API-usage snapshots. + """ + category = "integration" + full = {str(i): {"full_name": f"user{i}/repo{i}"} for i in range(30)} + + shard_root = os.path.join(SHARDS_DIR, category) + shutil.rmtree(shard_root, ignore_errors=True) + + written = 0 + for index in range(shards): + shard_slice = _slice_by_shard(full, index, shards) + written += len(shard_slice) + write_shard_output(category, index + 1, shard_slice, shard_slice) + + # The shards partition the input: disjoint and complete. + assert written == len(full) + + assert _load_shard_files(category, "data.json") == full + assert _load_shard_files(category, "stored.json") == full + + shutil.rmtree(shard_root, ignore_errors=True)