Upload 9 files

beta_keys_schema.py

@@ -0,0 +1,119 @@
+"""
+VTX beta key schema, migration, and validation helpers.
+"""
+from __future__ import annotations
+
+import secrets
+import string
+from datetime import date
+from typing import Any
+
+VALID_STATUSES = frozenset({"ACTIVE", "REVOKED", "EXPIRED", "BANNED", "INVALID"})
+KEY_CHARS = string.ascii_uppercase + string.digits
+NEW_KEY_PREFIX = "VTX-"
+
+
+def today_iso() -> str:
+    return date.today().isoformat()
+
+
+def _random_key_segment(length: int = 6) -> str:
+    return "".join(secrets.choice(KEY_CHARS) for _ in range(length))
+
+
+def generate_unique_key(existing: dict[str, Any]) -> str:
+    for _ in range(5000):
+        candidate = f"{NEW_KEY_PREFIX}{_random_key_segment(6)}"
+        if candidate not in existing:
+            return candidate
+    raise RuntimeError("Unable to generate a unique beta key.")
+
+
+def normalize_record(key: str, record: dict[str, Any] | None) -> dict[str, Any]:
+    data = dict(record or {})
+    key_text = str(data.get("key") or key or "").strip()
+    active = bool(data.get("active"))
+    status = str(data.get("status") or "").strip().upper()
+    if not status:
+        status = "ACTIVE" if active else "REVOKED"
+    if status not in VALID_STATUSES:
+        status = "ACTIVE" if active else "REVOKED"
+    if status != "ACTIVE":
+        active = False
+    created = str(data.get("created") or data.get("created_date") or "").strip()
+    last_activation = data.get("last_activation")
+    if last_activation is not None and last_activation != "":
+        last_activation = str(last_activation).strip()
+    else:
+        last_activation = None
+    return {
+        "key": key_text,
+        "name": str(data.get("name") or data.get("owner") or "").strip(),
+        "notes": str(data.get("notes") or "").strip(),
+        "created": created,
+        "last_activation": last_activation,
+        "active": active,
+        "status": status,
+    }
+
+
+def migrate_keys_payload(payload: Any) -> dict[str, dict[str, Any]]:
+    if payload is None:
+        return {}
+    if isinstance(payload, list):
+        migrated: dict[str, dict[str, Any]] = {}
+        for item in payload:
+            if not isinstance(item, dict):
+                continue
+            key = str(item.get("key") or "").strip()
+            if not key:
+                continue
+            migrated[key] = normalize_record(key, item)
+        return migrated
+    if not isinstance(payload, dict):
+        return {}
+
+    if "keys" in payload and isinstance(payload.get("keys"), list):
+        return migrate_keys_payload(payload["keys"])
+
+    migrated: dict[str, dict[str, Any]] = {}
+    for raw_key, raw_record in payload.items():
+        key = str(raw_key or "").strip()
+        if not key:
+            continue
+        if isinstance(raw_record, str):
+            migrated[key] = normalize_record(
+                key,
+                {"key": key, "active": True, "status": "ACTIVE", "notes": raw_record},
+            )
+            continue
+        if not isinstance(raw_record, dict):
+            continue
+        migrated[key] = normalize_record(key, raw_record)
+    return migrated
+
+
+def validation_result(record: dict[str, Any] | None) -> dict[str, Any]:
+    if not record:
+        return {"valid": False, "active": False, "status": "INVALID"}
+    normalized = normalize_record(str(record.get("key") or ""), record)
+    active = bool(normalized.get("active"))
+    status = str(normalized.get("status") or "").strip().upper()
+    if status == "ACTIVE" and active:
+        return {"valid": True, "active": True, "status": "ACTIVE"}
+    if status == "REVOKED" or (not active and status == "ACTIVE"):
+        return {"valid": True, "active": False, "status": "REVOKED"}
+    if status in {"EXPIRED", "BANNED"}:
+        return {"valid": True, "active": False, "status": status}
+    return {"valid": True, "active": active, "status": status}
+
+
+def new_record(name: str, notes: str = "") -> dict[str, Any]:
+    return {
+        "name": str(name or "").strip(),
+        "notes": str(notes or "").strip(),
+        "created": today_iso(),
+        "last_activation": None,
+        "active": True,
+        "status": "ACTIVE",
+    }

beta_keys_store.py

@@ -0,0 +1,321 @@
+"""
+VTX beta keys — private Hugging Face dataset storage with local cache.
+
+Environment:
+  VTX_KEYS_SOURCE=local|dataset          (default: dataset)
+  VTX_KEYS_PATH=path/to/keys.json        (local mode / cache file override)
+  VTX_BETA_KEYS_DATASET=org/repo         (default: ManChildTechnologies/VTX-BetaKeys)
+  VTX_BETA_KEYS_FILENAME=keys.json
+  VTX_BETA_KEYS_REVISION=main
+  VTX_KEYS_REFRESH_SECONDS=30
+  HF_TOKEN / HUGGING_FACE_HUB_TOKEN
+"""
+from __future__ import annotations
+
+import io
+import json
+import os
+import threading
+import time
+from pathlib import Path
+from typing import Any
+
+from beta_keys_schema import migrate_keys_payload, normalize_record
+
+APP_DIR = Path(__file__).resolve().parent
+DEFAULT_KEYS_PATH = APP_DIR / "keys.json"
+DEFAULT_CACHE_PATH = APP_DIR / "keys_cache.json"
+DEFAULT_DATASET_REPO = "ManChildTechnologies/VTX-BetaKeys"
+DEFAULT_DATASET_FILENAME = "keys.json"
+
+_store_lock = threading.RLock()
+_keys_cache: dict[str, Any] = {
+    "source": None,
+    "signature": None,
+    "fetched_at": 0.0,
+    "data": {},
+    "error": "",
+    "using_stale_cache": False,
+}
+
+
+def keys_source() -> str:
+    raw = str(os.environ.get("VTX_KEYS_SOURCE", "dataset") or "dataset").strip().lower()
+    return "local" if raw == "local" else "dataset"
+
+
+def keys_path() -> Path:
+    custom = str(os.environ.get("VTX_KEYS_PATH", "") or "").strip()
+    if custom:
+        return Path(custom)
+    return DEFAULT_KEYS_PATH
+
+
+def cache_path() -> Path:
+    custom = str(os.environ.get("VTX_KEYS_CACHE_PATH", "") or "").strip()
+    if custom:
+        return Path(custom)
+    return DEFAULT_CACHE_PATH
+
+
+def dataset_repo() -> str:
+    return str(os.environ.get("VTX_BETA_KEYS_DATASET", DEFAULT_DATASET_REPO) or DEFAULT_DATASET_REPO).strip()
+
+
+def dataset_filename() -> str:
+    return str(os.environ.get("VTX_BETA_KEYS_FILENAME", DEFAULT_DATASET_FILENAME) or DEFAULT_DATASET_FILENAME).strip()
+
+
+def dataset_revision() -> str:
+    return str(os.environ.get("VTX_BETA_KEYS_REVISION", "main") or "main").strip() or "main"
+
+
+def refresh_seconds() -> int:
+    try:
+        return max(5, int(os.environ.get("VTX_KEYS_REFRESH_SECONDS", "30") or 30))
+    except Exception:
+        return 30
+
+
+def hf_token() -> str:
+    for key in ("HF_TOKEN", "HUGGING_FACE_HUB_TOKEN", "HUGGINGFACE_HUB_TOKEN"):
+        value = str(os.environ.get(key, "") or "").strip()
+        if value:
+            return value
+    return ""
+
+
+def _require_hf_token() -> str:
+    token = hf_token()
+    if not token:
+        raise RuntimeError(
+            "HF_TOKEN is required for VTX beta key dataset access. "
+            "Set HF_TOKEN as an environment variable or Hugging Face Space secret."
+        )
+    return token
+
+
+def _get_hf_api():
+    try:
+        from huggingface_hub import HfApi
+    except ImportError as exc:
+        raise RuntimeError("huggingface_hub is required for dataset key storage.") from exc
+    return HfApi(token=_require_hf_token())
+
+
+def _serialize_keys(keys: dict[str, dict[str, Any]]) -> str:
+    ordered = {
+        key: normalize_record(key, record)
+        for key, record in sorted(keys.items(), key=lambda item: item[0])
+    }
+    return json.dumps(ordered, indent=2) + "\n"
+
+
+def _read_json_file(path: Path) -> dict[str, dict[str, Any]]:
+    if not path.is_file():
+        return {}
+    try:
+        payload = json.loads(path.read_text(encoding="utf-8"))
+    except Exception:
+        return {}
+    return migrate_keys_payload(payload)
+
+
+def _write_json_file(path: Path, keys: dict[str, dict[str, Any]]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(_serialize_keys(keys), encoding="utf-8")
+
+
+def _local_signature(path: Path) -> str | None:
+    try:
+        if not path.is_file():
+            return None
+        stat = path.stat()
+        return f"local:{path}:{stat.st_mtime_ns}:{stat.st_size}"
+    except OSError:
+        return None
+
+
+def _fetch_dataset_keys() -> tuple[dict[str, dict[str, Any]], str]:
+    token = _require_hf_token()
+    try:
+        from huggingface_hub import hf_hub_download
+    except ImportError as exc:
+        raise RuntimeError("huggingface_hub is required for dataset key loading.") from exc
+
+    repo_id = dataset_repo()
+    filename = dataset_filename()
+    revision = dataset_revision()
+    cached_path = hf_hub_download(
+        repo_id=repo_id,
+        repo_type="dataset",
+        filename=filename,
+        revision=revision,
+        token=token,
+        force_download=True,
+    )
+    path = Path(cached_path)
+    data = _read_json_file(path)
+    signature = f"dataset:{repo_id}:{revision}:{filename}:{path.stat().st_mtime_ns}:{path.stat().st_size}"
+    _write_json_file(cache_path(), data)
+    return data, signature
+
+
+def push_keys_to_dataset(keys: dict[str, dict[str, Any]], commit_message: str) -> None:
+    token = _require_hf_token()
+    api = _get_hf_api()
+    payload = _serialize_keys(keys).encode("utf-8")
+    api.upload_file(
+        path_or_fileobj=io.BytesIO(payload),
+        path_in_repo=dataset_filename(),
+        repo_id=dataset_repo(),
+        repo_type="dataset",
+        revision=dataset_revision(),
+        commit_message=commit_message,
+        token=token,
+    )
+    _write_json_file(cache_path(), keys)
+    global _keys_cache
+    _keys_cache = {
+        "source": "dataset",
+        "signature": f"pushed:{time.time_ns()}",
+        "fetched_at": time.time(),
+        "data": dict(keys),
+        "error": "",
+        "using_stale_cache": False,
+    }
+
+
+def load_keys(force: bool = False) -> dict[str, dict[str, Any]]:
+    global _keys_cache
+
+    with _store_lock:
+        source = keys_source()
+        now = time.time()
+        ttl = refresh_seconds()
+
+        if source == "local":
+            path = keys_path()
+            signature = _local_signature(path)
+            if (
+                not force
+                and _keys_cache.get("source") == "local"
+                and signature == _keys_cache.get("signature")
+                and (now - float(_keys_cache.get("fetched_at") or 0.0)) < ttl
+            ):
+                return dict(_keys_cache.get("data") or {})
+
+            data = _read_json_file(path)
+            _keys_cache = {
+                "source": "local",
+                "signature": signature,
+                "fetched_at": now,
+                "data": data,
+                "error": "",
+                "using_stale_cache": False,
+            }
+            return dict(data)
+
+        if (
+            not force
+            and _keys_cache.get("source") == "dataset"
+            and (now - float(_keys_cache.get("fetched_at") or 0.0)) < ttl
+            and _keys_cache.get("data") is not None
+        ):
+            return dict(_keys_cache.get("data") or {})
+
+        try:
+            data, signature = _fetch_dataset_keys()
+            _keys_cache = {
+                "source": "dataset",
+                "signature": signature,
+                "fetched_at": now,
+                "data": data,
+                "error": "",
+                "using_stale_cache": False,
+            }
+            return dict(data)
+        except Exception as exc:
+            cached_file = _read_json_file(cache_path())
+            if cached_file:
+                _keys_cache = {
+                    "source": "dataset",
+                    "signature": _keys_cache.get("signature"),
+                    "fetched_at": now,
+                    "data": cached_file,
+                    "error": str(exc),
+                    "using_stale_cache": True,
+                }
+                return dict(cached_file)
+            _keys_cache = {
+                "source": "dataset",
+                "signature": None,
+                "fetched_at": now,
+                "data": {},
+                "error": str(exc),
+                "using_stale_cache": False,
+            }
+            raise
+
+
+def save_keys(keys: dict[str, dict[str, Any]], commit_message: str = "Update VTX beta keys") -> None:
+    normalized = {
+        key: normalize_record(key, record)
+        for key, record in (keys or {}).items()
+        if str(key or "").strip()
+    }
+    with _store_lock:
+        if keys_source() == "local":
+            _write_json_file(keys_path(), normalized)
+            global _keys_cache
+            _keys_cache = {
+                "source": "local",
+                "signature": _local_signature(keys_path()),
+                "fetched_at": time.time(),
+                "data": dict(normalized),
+                "error": "",
+                "using_stale_cache": False,
+            }
+            return
+        push_keys_to_dataset(normalized, commit_message)
+
+
+def update_key_record(
+    beta_key: str,
+    updater,
+    commit_message: str,
+) -> dict[str, dict[str, Any]]:
+    key = str(beta_key or "").strip()
+    if not key:
+        raise ValueError("Beta key is required.")
+
+    with _store_lock:
+        keys = load_keys(force=True)
+        record = dict(keys.get(key) or normalize_record(key, {"key": key}))
+        record = normalize_record(key, updater(record))
+        record["key"] = key
+        keys[key] = record
+        save_keys(keys, commit_message=commit_message)
+        return record
+
+
+def keys_status() -> dict[str, Any]:
+    source = keys_source()
+    status = {
+        "source": source,
+        "keys_loaded": len(_keys_cache.get("data") or {}),
+        "last_error": str(_keys_cache.get("error") or ""),
+        "cache_age_seconds": max(0, int(time.time() - float(_keys_cache.get("fetched_at") or 0.0))),
+        "using_stale_cache": bool(_keys_cache.get("using_stale_cache")),
+    }
+    if source == "local":
+        path = keys_path()
+        status["keys_path"] = str(path)
+        status["keys_exists"] = path.is_file()
+    else:
+        status["dataset_repo"] = dataset_repo()
+        status["dataset_filename"] = dataset_filename()
+        status["dataset_revision"] = dataset_revision()
+        status["hf_token_configured"] = bool(hf_token())
+        status["cache_path"] = str(cache_path())
+    return status

generate_beta_key.py

@@ -1,116 +1,113 @@
 """
 VTX Studio beta key administration utility.
 
-Manage keys.json locally, then redeploy the Hugging Face Space to publish changes.
-No accounts, billing, subscriptions, machine locking, or activation limits.
+Keys are stored in the private ManChildTechnologies/VTX-BetaKeys dataset.
 """
 from __future__ import annotations
 
 import argparse
 import json
-import secrets
-import string
 from pathlib import Path
 from typing import Any
 
+from beta_keys_schema import generate_unique_key, new_record, normalize_record
+from beta_keys_store import keys_path, load_keys, save_keys
+
 APP_DIR = Path(__file__).resolve().parent
 DEFAULT_KEYS_PATH = APP_DIR / "keys.json"
-KEY_PREFIX = "VTX-BETA"
-KEY_CHARS = string.ascii_uppercase + string.digits
-
-
-def keys_path(path: str | Path | None = None) -> Path:
-    if path:
-        return Path(path)
-    env = __import__("os").environ.get("VTX_KEYS_PATH", "").strip()
-    if env:
-        return Path(env)
-    return DEFAULT_KEYS_PATH
-
-
-def load_keys(path: str | Path | None = None) -> dict[str, dict[str, Any]]:
-    file_path = keys_path(path)
-    if not file_path.is_file():
-        return {}
-    data = json.loads(file_path.read_text(encoding="utf-8"))
-    if not isinstance(data, dict):
-        raise ValueError(f"keys file must be a JSON object: {file_path}")
-    return data
 
 
-def save_keys(keys: dict[str, dict[str, Any]], path: str | Path | None = None) -> Path:
-    file_path = keys_path(path)
-    file_path.parent.mkdir(parents=True, exist_ok=True)
-    file_path.write_text(json.dumps(keys, indent=2) + "\n", encoding="utf-8")
-    return file_path
-
-
-def _random_segment(length: int = 4) -> str:
-    return "".join(secrets.choice(KEY_CHARS) for _ in range(length))
+def generate_key(
+    *,
+    name: str,
+    notes: str = "",
+    path: str | Path | None = None,
+    push: bool = True,
+) -> str:
+    """Generate a new ACTIVE beta key and persist it to the dataset."""
+    if not str(name or "").strip():
+        raise ValueError("Tester name is required.")
+
+    keys = load_keys(force=True)
+    beta_key = generate_unique_key(keys)
+    record = new_record(name=name, notes=notes)
+    record["key"] = beta_key
+    keys[beta_key] = normalize_record(beta_key, record)
+    if push:
+        save_keys(keys, commit_message=f"Generate beta key for {name.strip()}")
+    elif path:
+        from beta_keys_store import _write_json_file
+
+        _write_json_file(Path(path), keys)
+    return beta_key
 
 
-def _new_key(existing: dict[str, Any]) -> str:
-    for _ in range(1000):
-        candidate = f"{KEY_PREFIX}-{_random_segment()}-{_random_segment()}"
-        if candidate not in existing:
-            return candidate
-    raise RuntimeError("Unable to generate a unique beta key.")
+def revoke_key(beta_key: str, path: str | Path | None = None, push: bool = True) -> dict[str, Any]:
+    key = str(beta_key or "").strip()
+    if not key:
+        raise ValueError("Beta key is required.")
 
+    keys = load_keys(force=True)
+    if key not in keys:
+        raise KeyError(f"Beta key not found: {key}")
 
-def generate_key(path: str | Path | None = None) -> str:
-    """Generate a new ACTIVE beta key and persist it to keys.json."""
-    keys = load_keys(path)
-    beta_key = _new_key(keys)
-    keys[beta_key] = {"active": True, "status": "ACTIVE"}
-    save_keys(keys, path)
-    return beta_key
+    record = normalize_record(key, keys[key])
+    record["active"] = False
+    record["status"] = "REVOKED"
+    keys[key] = record
+    if push:
+        save_keys(keys, commit_message=f"Revoke beta key {key}")
+    return record
 
 
-def revoke_key(beta_key: str, path: str | Path | None = None) -> dict[str, Any]:
-    """Revoke an existing beta key."""
+def reactivate_key(beta_key: str, path: str | Path | None = None, push: bool = True) -> dict[str, Any]:
     key = str(beta_key or "").strip()
     if not key:
         raise ValueError("Beta key is required.")
 
-    keys = load_keys(path)
+    keys = load_keys(force=True)
     if key not in keys:
         raise KeyError(f"Beta key not found: {key}")
 
-    keys[key] = {"active": False, "status": "REVOKED"}
-    save_keys(keys, path)
-    return keys[key]
+    record = normalize_record(key, keys[key])
+    record["active"] = True
+    record["status"] = "ACTIVE"
+    keys[key] = record
+    if push:
+        save_keys(keys, commit_message=f"Reactivate beta key {key}")
+    return record
 
 
-def reactivate_key(beta_key: str, path: str | Path | None = None) -> dict[str, Any]:
-    """Reactivate a previously revoked beta key."""
+def delete_key(beta_key: str, push: bool = True) -> None:
     key = str(beta_key or "").strip()
     if not key:
         raise ValueError("Beta key is required.")
 
-    keys = load_keys(path)
+    keys = load_keys(force=True)
     if key not in keys:
         raise KeyError(f"Beta key not found: {key}")
+    del keys[key]
+    if push:
+        save_keys(keys, commit_message=f"Delete beta key {key}")
 
-    keys[key] = {"active": True, "status": "ACTIVE"}
-    save_keys(keys, path)
-    return keys[key]
 
-
-def export_json(path: str | Path | None = None, output: str | Path | None = None) -> Path:
-    """Export keys.json to stdout path (defaults to keys.json itself)."""
-    keys = load_keys(path)
-    out = Path(output) if output else keys_path(path)
+def export_json(output: str | Path | None = None) -> Path:
+    keys = load_keys(force=True)
+    out = Path(output) if output else keys_path()
     out.parent.mkdir(parents=True, exist_ok=True)
-    out.write_text(json.dumps(keys, indent=2) + "\n", encoding="utf-8")
+    ordered = {key: normalize_record(key, record) for key, record in sorted(keys.items())}
+    out.write_text(json.dumps(ordered, indent=2) + "\n", encoding="utf-8")
     return out
 
 
 def main() -> int:
     parser = argparse.ArgumentParser(description="VTX Studio beta key manager")
-    parser.add_argument("--keys", default=str(DEFAULT_KEYS_PATH), help="Path to keys.json")
+    parser.add_argument("--keys", default="", help="Optional local export path")
     sub = parser.add_subparsers(dest="command", required=True)
 
-    sub.add_parser("generate", help="Generate a new ACTIVE beta key")
+    generate = sub.add_parser("generate", help="Generate a new ACTIVE beta key")
+    generate.add_argument("--name", required=True, help="Tester name")
+    generate.add_argument("--notes", default="", help="Optional notes")
 
     revoke = sub.add_parser("revoke", help="Revoke an existing beta key")
     revoke.add_argument("beta_key", help="Beta key to revoke")
@@ -118,34 +115,36 @@ def main() -> int:
     reactivate = sub.add_parser("reactivate", help="Reactivate a revoked beta key")
     reactivate.add_argument("beta_key", help="Beta key to reactivate")
 
-    export = sub.add_parser("export", help="Export keys.json")
-    export.add_argument(
-        "-o",
-        "--output",
-        default="",
-        help="Optional output path (defaults to keys file)",
-    )
+    delete = sub.add_parser("delete", help="Delete an existing beta key")
+    delete.add_argument("beta_key", help="Beta key to delete")
+
+    export = sub.add_parser("export", help="Export keys.json snapshot")
+    export.add_argument("-o", "--output", default="", help="Optional output path")
 
     args = parser.parse_args()
-    keys_file = args.keys
 
     if args.command == "generate":
-        beta_key = generate_key(keys_file)
+        beta_key = generate_key(name=args.name, notes=args.notes)
         print(beta_key)
         return 0
 
     if args.command == "revoke":
-        record = revoke_key(args.beta_key, keys_file)
-        print(json.dumps({args.beta_key: record}, indent=2))
+        record = revoke_key(args.beta_key)
+        print(json.dumps(record, indent=2))
         return 0
 
     if args.command == "reactivate":
-        record = reactivate_key(args.beta_key, keys_file)
-        print(json.dumps({args.beta_key: record}, indent=2))
+        record = reactivate_key(args.beta_key)
+        print(json.dumps(record, indent=2))
+        return 0
+
+    if args.command == "delete":
+        delete_key(args.beta_key)
+        print("deleted")
         return 0
 
     if args.command == "export":
-        out = export_json(keys_file, args.output or None)
+        out = export_json(args.output or None)
         print(str(out))
         return 0
 

requirements.txt

@@ -1,2 +1,3 @@
 fastapi>=0.110.0
-uvicorn>=0.27.0
+uvicorn>=0.27.0
+huggingface_hub>=0.23.0

tools/migrate_keys_to_dataset.py

@@ -0,0 +1,56 @@
+"""One-time migration: local keys.json -> private VTX-BetaKeys dataset."""
+from __future__ import annotations
+
+import argparse
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from beta_keys_schema import migrate_keys_payload
+from beta_keys_store import cache_path, dataset_repo, hf_token, push_keys_to_dataset
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="Migrate local keys.json to the private HF dataset")
+    parser.add_argument(
+        "--input",
+        default=str(Path(__file__).resolve().parents[1] / "keys.json"),
+        help="Path to legacy keys.json",
+    )
+    parser.add_argument(
+        "--dry-run",
+        action="store_true",
+        help="Print migrated key count without uploading",
+    )
+    args = parser.parse_args()
+
+    if not hf_token():
+        print("HF_TOKEN is required.", file=sys.stderr)
+        return 1
+
+    source = Path(args.input)
+    if not source.is_file():
+        print(f"Missing input file: {source}", file=sys.stderr)
+        return 1
+
+    import json
+
+    payload = json.loads(source.read_text(encoding="utf-8"))
+    migrated = migrate_keys_payload(payload)
+    print(f"Migrated {len(migrated)} keys from {source}")
+    print(f"Target dataset: {dataset_repo()}")
+
+    if args.dry_run:
+        for key, record in sorted(migrated.items()):
+            print(f"  {key}: {record.get('status')} name={record.get('name')!r}")
+        return 0
+
+    push_keys_to_dataset(migrated, commit_message="Migrate legacy keys.json to private dataset")
+    print(f"Uploaded keys.json to {dataset_repo()}")
+    print(f"Local cache written to {cache_path()}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tools/test_gui_import.py

@@ -0,0 +1,13 @@
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from beta_keys_schema import normalize_record
+import vtx_beta_key_manager as gui
+
+record = normalize_record("VTX-TEST01", {"active": True, "status": "ACTIVE"})
+assert record["status"] == "ACTIVE"
+assert gui.APP_NAME == "VTX Beta Key Manager"
+assert gui.APP_VERSION == "1.1.0"
+print("OK")

tools/test_openapi.py

@@ -0,0 +1,32 @@
+"""Verify OpenAPI documents ValidateRequest for Swagger."""
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from app import app
+
+schema = app.openapi()
+request_schema = (
+    schema["paths"]["/api/validate"]["post"]
+    .get("requestBody", {})
+    .get("content", {})
+    .get("application/json", {})
+    .get("schema", {})
+)
+
+ref = request_schema.get("$ref", "")
+assert ref.endswith("/ValidateRequest"), ref
+
+model = schema["components"]["schemas"]["ValidateRequest"]
+props = model.get("properties", {})
+assert "beta_key" in props, props
+
+examples = props["beta_key"].get("examples") or []
+if not examples:
+    example = props["beta_key"].get("example")
+    assert example == "VTX-BETA-XXXX-XXXX", props["beta_key"]
+else:
+    assert "VTX-BETA-XXXX-XXXX" in examples, examples
+
+print("OK")

tools/test_schema.py

@@ -0,0 +1,41 @@
+"""Schema migration and validation tests."""
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from beta_keys_schema import migrate_keys_payload, normalize_record, validation_result
+
+legacy = {
+    "VTX-BETA-7K4P-X92M": {"active": True, "status": "ACTIVE"},
+    "VTX-BETA-R8JL-Q4WT": {"active": False, "status": "REVOKED"},
+}
+migrated = migrate_keys_payload(legacy)
+assert "VTX-BETA-7K4P-X92M" in migrated
+record = migrated["VTX-BETA-7K4P-X92M"]
+assert record["key"] == "VTX-BETA-7K4P-X92M"
+assert record["name"] == ""
+assert record["notes"] == ""
+assert record["created"] == ""
+assert record["last_activation"] is None
+assert record["active"] is True
+assert validation_result(record)["status"] == "ACTIVE"
+assert validation_result(migrated["VTX-BETA-R8JL-Q4WT"])["status"] == "REVOKED"
+assert validation_result(None)["status"] == "INVALID"
+
+modern = normalize_record(
+    "VTX-4D8A9B",
+    {
+        "key": "VTX-4D8A9B",
+        "name": "Ramon",
+        "notes": "Beta Tester",
+        "created": "2026-06-12",
+        "last_activation": None,
+        "active": True,
+        "status": "ACTIVE",
+    },
+)
+assert modern["name"] == "Ramon"
+assert modern["notes"] == "Beta Tester"
+
+print("OK")

tools/test_validate.py

@@ -0,0 +1,23 @@
+"""Smoke test for validate_beta_key logic (local mode, no HF required)."""
+import os
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+os.environ["VTX_KEYS_SOURCE"] = "local"
+os.environ["VTX_KEYS_PATH"] = str(Path(__file__).resolve().parents[1] / "keys.json")
+
+from app import validate_beta_key
+
+active = validate_beta_key("VTX-BETA-7K4P-X92M", track_activation=False)
+revoked = validate_beta_key("VTX-BETA-R8JL-Q4WT", track_activation=False)
+unknown = validate_beta_key("VTX-BETA-NOT-REAL", track_activation=False)
+empty = validate_beta_key("", track_activation=False)
+
+assert active == {"valid": True, "active": True, "status": "ACTIVE"}, active
+assert revoked == {"valid": True, "active": False, "status": "REVOKED"}, revoked
+assert unknown == {"valid": False, "active": False, "status": "INVALID"}, unknown
+assert empty == {"valid": False, "active": False, "status": "INVALID"}, empty
+
+print("OK")