Coverage for node / src / stigmem_node / cli / snapshot.py: 100%

1"""Snapshot CLI handlers.""" 

2 

3from __future__ import annotations 

4 

5import argparse 

6import sys 

7 

8 

9def _cmd_snapshot_create(args: argparse.Namespace) -> int: 

10 """Create a signed, content-addressed snapshot tarball.""" 

11 from pathlib import Path 

12 

13 from ..db import apply_migrations 

14 from ..settings import settings 

15 from ..snapshot import snapshot_create 

16 

17 db_path: str = args.db or settings.db_path 

18 apply_migrations(db_path=db_path) 

19 

20 out_path = Path(args.out) if args.out else None 

21 sign_with = Path(args.sign_with) if args.sign_with else None 

22 

23 result = snapshot_create(db_path=db_path, out_path=out_path, sign_with_key_path=sign_with) 

24 print(f"snapshot created: {result}", file=sys.stderr) 

25 return 0 

26 

27 

28def _cmd_snapshot_restore(args: argparse.Namespace) -> int: 

29 """Verify and restore a snapshot tarball.""" 

30 from pathlib import Path 

31 

32 from ..settings import settings 

33 from ..snapshot import SnapshotVerificationError, snapshot_restore 

34 

35 db_path: str = args.db or settings.db_path 

36 from_path = Path(args.from_path) 

37 trusted_keys = Path(args.trusted_keys) if args.trusted_keys else None 

38 

39 try: 

40 snapshot_restore( 

41 tarball_path=from_path, 

42 db_path=db_path, 

43 trusted_keys_path=trusted_keys, 

44 force_unverified=args.force_unverified, 

45 ) 

46 except SnapshotVerificationError as exc: 

47 print(f"error: {exc}", file=sys.stderr) 

48 return 1 

49 

50 print(f"snapshot restored to {db_path}", file=sys.stderr) 

51 return 0