(v, pos)
| 219 | |
| 220 | |
| 221 | def _decode(v, pos): |
| 222 | code = six.indexbytes(v, pos) |
| 223 | if code == NULL_CODE: |
| 224 | return None, pos + 1 |
| 225 | elif code == BYTES_CODE: |
| 226 | end = _find_terminator(v, pos + 1) |
| 227 | return v[pos + 1:end].replace(b"\x00\xFF", b"\x00"), end + 1 |
| 228 | elif code == STRING_CODE: |
| 229 | end = _find_terminator(v, pos + 1) |
| 230 | return v[pos + 1:end].replace(b"\x00\xFF", b"\x00").decode("utf-8"), end + 1 |
| 231 | elif code >= INT_ZERO_CODE and code < POS_INT_END: |
| 232 | n = code - 20 |
| 233 | end = pos + 1 + n |
| 234 | return struct.unpack(">Q", b'\x00' * (8 - n) + v[pos + 1:end])[0], end |
| 235 | elif code > NEG_INT_START and code < INT_ZERO_CODE: |
| 236 | n = 20 - code |
| 237 | end = pos + 1 + n |
| 238 | return struct.unpack(">Q", b'\x00' * (8 - n) + v[pos + 1:end])[0] - _size_limits[n], end |
| 239 | elif code == POS_INT_END: # 0x1d; Positive 9-255 byte integer |
| 240 | length = six.indexbytes(v, pos + 1) |
| 241 | val = 0 |
| 242 | for i in _range(length): |
| 243 | val = val << 8 |
| 244 | val += six.indexbytes(v, pos + 2 + i) |
| 245 | return val, pos + 2 + length |
| 246 | elif code == NEG_INT_START: # 0x0b; Negative 9-255 byte integer |
| 247 | length = six.indexbytes(v, pos + 1) ^ 0xff |
| 248 | val = 0 |
| 249 | for i in _range(length): |
| 250 | val = val << 8 |
| 251 | val += six.indexbytes(v, pos + 2 + i) |
| 252 | return val - (1 << (length * 8)) + 1, pos + 2 + length |
| 253 | elif code == FLOAT_CODE: |
| 254 | return SingleFloat(struct.unpack(">f", _float_adjust(v[pos + 1:pos + 5], False))[0]), pos + 5 |
| 255 | elif code == DOUBLE_CODE: |
| 256 | return struct.unpack(">d", _float_adjust(v[pos + 1:pos + 9], False))[0], pos + 9 |
| 257 | elif code == UUID_CODE: |
| 258 | return uuid.UUID(bytes=v[pos + 1:pos + 17]), pos + 17 |
| 259 | elif code == FALSE_CODE: |
| 260 | if fdb.is_api_version_selected() and fdb.get_api_version() < 500: |
| 261 | raise ValueError("Invalid API version " + str(fdb._version) + " for boolean types") |
| 262 | return False, pos + 1 |
| 263 | elif code == TRUE_CODE: |
| 264 | if fdb.is_api_version_selected() and fdb.get_api_version() < 500: |
| 265 | raise ValueError("Invalid API version " + str(fdb._version) + " for boolean types") |
| 266 | return True, pos + 1 |
| 267 | elif code == VERSIONSTAMP_CODE: |
| 268 | return Versionstamp.from_bytes(v, pos + 1), pos + 1 + Versionstamp.LENGTH |
| 269 | elif code == NESTED_CODE: |
| 270 | ret = [] |
| 271 | end_pos = pos + 1 |
| 272 | while end_pos < len(v): |
| 273 | if six.indexbytes(v, end_pos) == 0x00: |
| 274 | if end_pos + 1 < len(v) and six.indexbytes(v, end_pos + 1) == 0xff: |
| 275 | ret.append(None) |
| 276 | end_pos += 2 |
| 277 | else: |
| 278 | break |
no test coverage detected