Files
gyxx-flow/tests/test_source_migration_acceptance.py

63 lines
2.1 KiB
Python

from __future__ import annotations
import hashlib
import json
from pathlib import Path, PurePosixPath
from gyxx_flow.catalog import WorkflowCatalog
from gyxx_flow.script_catalog import ScriptCatalog
PROJECT_ROOT = Path(__file__).resolve().parents[1]
MODULES = (
"content_marketing",
"product_commerce",
"shop_intelligence",
"supply_chain",
)
def _sha256(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as stream:
for chunk in iter(lambda: stream.read(1024 * 1024), b""):
digest.update(chunk)
return digest.hexdigest()
def test_all_source_manifests_use_one_schema_and_verify_target_hashes() -> None:
for module in MODULES:
path = PROJECT_ROOT / "config" / "source-manifests" / f"{module}.json"
manifest = json.loads(path.read_text(encoding="utf-8-sig"))
assert manifest["schema_version"] == 1
assert manifest["module"] == module
assert isinstance(manifest["snapshot"], str) and manifest["snapshot"]
assert manifest["files"]
assert isinstance(manifest["intentionally_excluded"], list)
for item in manifest["files"]:
target = PROJECT_ROOT.joinpath(
*PurePosixPath(item["target_relative_path"]).parts
)
assert target.is_file(), target
assert _sha256(target) == item["target_sha256"], target
def test_every_workflow_and_manual_script_resolves_inside_new_project() -> None:
catalog = WorkflowCatalog.load(PROJECT_ROOT / "config")
for workflow in catalog.workflows:
if workflow.trigger == "unavailable":
continue
target = (
PROJECT_ROOT
/ "src"
/ "gyxx_flow"
/ "modules"
/ workflow.module
/ workflow.entry
).resolve(strict=True)
assert target.is_relative_to(PROJECT_ROOT)
commands = ScriptCatalog.load_default()
assert len(commands.scripts) == 41
assert len(commands.command_ids) == len(set(commands.command_ids))
assert all(command.path.is_relative_to(PROJECT_ROOT) for command in commands.scripts)