Coverage for integrations / social / sync_api.py: 28.1%

1""" 

2HevolveSocial - Sync & Backup API Blueprint 

3Endpoints for encrypted backup/restore and device management. 

4""" 

5import logging 

6from datetime import datetime 

7from flask import Blueprint, request, jsonify, g 

8 

9from .auth import require_auth 

10from .models import get_db, DeviceBinding 

11 

12logger = logging.getLogger('hevolve_social') 

13 

14sync_bp = Blueprint('sync', __name__, url_prefix='/api/social/sync') 

15 

16 

17def _ok(data=None, status=200): 

18 r = {'success': True} 

19 if data is not None: 

20 r['data'] = data 

21 return jsonify(r), status 

22 

23 

24def _err(msg, status=400): 

25 return jsonify({'success': False, 'error': msg}), status 

26 

27 

28# ─── Backup ─── 

29 

30@sync_bp.route('/backup', methods=['POST']) 

31@require_auth 

32def create_backup(): 

33 """Create an encrypted backup of user data.""" 

34 data = request.get_json(force=True, silent=True) or {} 

35 passphrase = data.get('passphrase', '').strip() 

36 if not passphrase or len(passphrase) < 8: 

37 return _err("Passphrase must be at least 8 characters") 

38 

39 db = get_db() 

40 try: 

41 from .backup_service import create_backup as _create 

42 result = _create(db, g.user.id, passphrase) 

43 return _ok(result, 201) 

44 except ValueError as e: 

45 return _err(str(e)) 

46 except Exception as e: 

47 logger.error(f"Backup creation failed: {e}") 

48 return _err("Backup creation failed") 

49 finally: 

50 db.close() 

51 

52 

53@sync_bp.route('/backup/metadata', methods=['GET']) 

54@require_auth 

55def get_backup_metadata(): 

56 """List all backup metadata for the current user.""" 

57 db = get_db() 

58 try: 

59 from .backup_service import list_backups 

60 backups = list_backups(db, g.user.id) 

61 return _ok(backups) 

62 finally: 

63 db.close() 

64 

65 

66@sync_bp.route('/restore', methods=['POST']) 

67@require_auth 

68def restore_backup(): 

69 """Restore user data from an encrypted backup.""" 

70 data = request.get_json(force=True, silent=True) or {} 

71 passphrase = data.get('passphrase', '').strip() 

72 backup_id = data.get('backup_id') # optional - defaults to latest 

73 if not passphrase: 

74 return _err("Passphrase required") 

75 

76 db = get_db() 

77 try: 

78 from .backup_service import restore_backup as _restore 

79 result = _restore(db, g.user.id, passphrase, backup_id) 

80 return _ok(result) 

81 except ValueError as e: 

82 return _err(str(e)) 

83 except Exception as e: 

84 logger.error(f"Backup restore failed: {e}") 

85 return _err("Restore failed") 

86 finally: 

87 db.close() 

88 

89 

90# ─── Device Management ─── 

91 

92@sync_bp.route('/link-device', methods=['POST']) 

93@require_auth 

94def link_device(): 

95 """Link a device to the current user for sync.""" 

96 data = request.get_json(force=True, silent=True) or {} 

97 device_id = data.get('device_id', '').strip() 

98 if not device_id: 

99 return _err("device_id required") 

100 

101 db = get_db() 

102 try: 

103 existing = db.query(DeviceBinding).filter_by( 

104 user_id=g.user.id, device_id=device_id).first() 

105 if existing: 

106 import json as _json 

107 existing.last_sync_at = datetime.utcnow() 

108 existing.is_active = True 

109 existing.device_name = data.get('device_name', existing.device_name) 

110 if 'form_factor' in data: 

111 existing.form_factor = data['form_factor'] 

112 caps = data.get('capabilities') 

113 if isinstance(caps, dict): 

114 existing.capabilities_json = _json.dumps(caps) 

115 db.commit() 

116 return _ok(existing.to_dict()) 

117 

118 import json as _json 

119 caps = data.get('capabilities') 

120 caps_json = _json.dumps(caps) if isinstance(caps, dict) else '{}' 

121 binding = DeviceBinding( 

122 user_id=g.user.id, 

123 device_id=device_id, 

124 device_name=data.get('device_name', ''), 

125 platform=data.get('platform', 'web'), 

126 form_factor=data.get('form_factor', 'phone'), 

127 capabilities_json=caps_json, 

128 ) 

129 db.add(binding) 

130 db.commit() 

131 return _ok(binding.to_dict(), 201) 

132 except Exception as e: 

133 db.rollback() 

134 logger.error(f"Device link failed: {e}") 

135 return _err("Device link failed") 

136 finally: 

137 db.close() 

138 

139 

140@sync_bp.route('/devices', methods=['GET']) 

141@require_auth 

142def list_devices(): 

143 """List all devices linked to the current user.""" 

144 db = get_db() 

145 try: 

146 devices = db.query(DeviceBinding).filter_by( 

147 user_id=g.user.id, is_active=True).all() 

148 return _ok([d.to_dict() for d in devices]) 

149 finally: 

150 db.close() 

151 

152 

153@sync_bp.route('/devices/<device_id>', methods=['DELETE']) 

154@require_auth 

155def unlink_device(device_id): 

156 """Unlink a device from the current user.""" 

157 db = get_db() 

158 try: 

159 binding = db.query(DeviceBinding).filter_by( 

160 id=device_id, user_id=g.user.id).first() 

161 if not binding: 

162 return _err("Device not found", 404) 

163 binding.is_active = False 

164 db.commit() 

165 return _ok({'message': 'Device unlinked'}) 

166 except Exception as e: 

167 db.rollback() 

168 logger.error(f"Device unlink failed: {e}") 

169 return _err("Device unlink failed") 

170 finally: 

171 db.close()