(params, rq, endpoint, record_rq_by_uid, silent=True)
| 1038 | |
| 1039 | |
| 1040 | def get_record_v3_response(params, rq, endpoint, record_rq_by_uid, silent=True): |
| 1041 | # type: (KeeperParams, Any, str, dict[str, dict], bool) -> Optional[bool] |
| 1042 | rs = communicate_rest(params, rq, endpoint) |
| 1043 | records_modify_rs = record_pb2.RecordsModifyResponse() |
| 1044 | records_modify_rs.ParseFromString(rs) |
| 1045 | |
| 1046 | for r in records_modify_rs.records: |
| 1047 | ruid = loginv3.CommonHelperMethods.bytes_to_url_safe_str(r.record_uid) |
| 1048 | record_rq = record_rq_by_uid[ruid] |
| 1049 | success = (r.status == record_pb2.RecordModifyResult.DESCRIPTOR.values_by_name['RS_SUCCESS'].number) |
| 1050 | status = record_pb2.RecordModifyResult.DESCRIPTOR.values_by_number[r.status].name |
| 1051 | |
| 1052 | if not success: |
| 1053 | logging.error(bcolors.FAIL + 'Error: Record update failed with status - %s' + bcolors.ENDC, status) |
| 1054 | return False |
| 1055 | |
| 1056 | new_revision = 0 |
| 1057 | if success: |
| 1058 | new_revision = records_modify_rs.revision |
| 1059 | |
| 1060 | if new_revision == 0: |
| 1061 | logging.error('Error: Revision not updated') |
| 1062 | return False |
| 1063 | |
| 1064 | if new_revision == record_rq['revision']: |
| 1065 | logging.error('Error: Revision did not change') |
| 1066 | return False |
| 1067 | |
| 1068 | if not silent: |
| 1069 | logging.info( |
| 1070 | 'Update record successful for record_uid=%s, revision=%d, new_revision=%s', |
| 1071 | ruid, record_rq['revision'], new_revision |
| 1072 | ) |
| 1073 | |
| 1074 | record_rq['revision'] = new_revision |
| 1075 | |
| 1076 | params.sync_data = True |
| 1077 | return True |
| 1078 | |
| 1079 | |
| 1080 | def update_record_v3(params, rec, **kwargs): # type: (KeeperParams, Record, ...) -> Optional[bool] |
no test coverage detected