Buckets:
| """Tests for app.assets.services.cursor. | |
| Cursors are opaque tokens internal to this server — these tests cover | |
| round-tripping, validation, and length caps, not any particular wire | |
| byte layout. | |
| """ | |
| from __future__ import annotations | |
| import base64 | |
| from datetime import datetime, timedelta, timezone | |
| import pytest | |
| from app.assets.services.cursor import ( | |
| MAX_CURSOR_ID_LENGTH, | |
| MAX_CURSOR_VALUE_LENGTH, | |
| MAX_ENCODED_CURSOR_LENGTH, | |
| CursorPayload, | |
| InvalidCursorError, | |
| decode_cursor, | |
| decode_cursor_int, | |
| decode_cursor_time, | |
| encode_cursor, | |
| encode_cursor_from_time, | |
| ) | |
| ALLOWED = ("created_at", "updated_at", "name", "size") | |
| class TestRoundTrip: | |
| def test_encode_decode(self, sort_field, value, id): | |
| encoded = encode_cursor(sort_field, value, id) | |
| assert encoded != "" | |
| payload = decode_cursor(encoded, ALLOWED) | |
| assert payload.sort_field == sort_field | |
| assert payload.value == value | |
| assert payload.id == id | |
| class TestTimeCursor: | |
| def test_microsecond_precision_preserved(self): | |
| # Pick a time with non-zero microseconds — encoding at ms would lose the µs. | |
| ts = datetime(2024, 5, 20, 12, 53, 20, 123456, tzinfo=timezone.utc) | |
| encoded = encode_cursor_from_time("created_at", ts, "id-1") | |
| payload = decode_cursor(encoded, ALLOWED) | |
| # Value must be a microsecond integer string, not a millisecond one. | |
| assert payload.value == "1716209600123456" | |
| decoded = decode_cursor_time(payload) | |
| assert decoded == ts | |
| def test_decode_returns_utc(self): | |
| payload = CursorPayload(sort_field="created_at", value="1716200000123456", id="id-1", order="desc") | |
| decoded = decode_cursor_time(payload) | |
| assert decoded.tzinfo == timezone.utc | |
| def test_naive_datetime_rejected_on_encode(self): | |
| naive = datetime(2024, 5, 20, 12, 0, 0) | |
| with pytest.raises(ValueError): | |
| encode_cursor_from_time("created_at", naive, "id-1") | |
| def test_non_integer_value_rejected_on_decode(self): | |
| with pytest.raises(InvalidCursorError): | |
| decode_cursor_time(CursorPayload("created_at", "not-a-number", "id-1", "desc")) | |
| def test_none_payload_rejected(self): | |
| with pytest.raises(InvalidCursorError): | |
| decode_cursor_time(None) | |
| def test_non_utc_aware_normalized(self): | |
| # Same instant, different timezone — must encode to the same micros. | |
| utc_ts = datetime(2024, 5, 20, 12, 0, 0, tzinfo=timezone.utc) | |
| offset_ts = utc_ts.astimezone(timezone(timedelta(hours=-5))) | |
| assert encode_cursor_from_time("created_at", utc_ts, "x") == encode_cursor_from_time( | |
| "created_at", offset_ts, "x" | |
| ) | |
| class TestIntCursor: | |
| def test_decode_int(self): | |
| assert decode_cursor_int(CursorPayload("size", "1024", "id-1", "desc")) == 1024 | |
| def test_decode_int_rejects_non_int(self): | |
| with pytest.raises(InvalidCursorError): | |
| decode_cursor_int(CursorPayload("size", "abc", "id-1", "desc")) | |
| def test_decode_int_rejects_none(self): | |
| with pytest.raises(InvalidCursorError): | |
| decode_cursor_int(None) | |
| class TestInvalidInputs: | |
| def test_oversized_cursor(self): | |
| oversized = "a" * (MAX_ENCODED_CURSOR_LENGTH + 1) | |
| with pytest.raises(InvalidCursorError, match="maximum length"): | |
| decode_cursor(oversized, ALLOWED) | |
| def test_not_base64(self): | |
| with pytest.raises(InvalidCursorError): | |
| decode_cursor("not base64!!!", ALLOWED) | |
| def test_not_json(self): | |
| encoded = base64.urlsafe_b64encode(b"definitely not json").rstrip(b"=").decode("ascii") | |
| with pytest.raises(InvalidCursorError): | |
| decode_cursor(encoded, ALLOWED) | |
| def test_empty_id(self): | |
| # Encoder rejects empty id symmetrically with the decoder, so build the | |
| # payload manually to exercise the decoder's missing-id branch. | |
| raw = b'{"s":"created_at","v":"1","id":"","o":"desc"}' | |
| encoded = base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii") | |
| with pytest.raises(InvalidCursorError, match="missing id"): | |
| decode_cursor(encoded, ALLOWED) | |
| def test_oversized_id(self): | |
| # Encoder enforces the cap symmetrically; hand-build to exercise decode. | |
| big_id = "a" * (MAX_CURSOR_ID_LENGTH + 1) | |
| raw = ('{"s":"created_at","v":"1","id":"' + big_id + '","o":"desc"}').encode("ascii") | |
| encoded = base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii") | |
| with pytest.raises(InvalidCursorError, match="id exceeds maximum length"): | |
| decode_cursor(encoded, ALLOWED) | |
| def test_oversized_value(self): | |
| # Encoder enforces the cap symmetrically; hand-build to exercise decode. | |
| big_v = "v" * (MAX_CURSOR_VALUE_LENGTH + 1) | |
| raw = ('{"s":"created_at","v":"' + big_v + '","id":"id-1","o":"desc"}').encode("ascii") | |
| encoded = base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii") | |
| with pytest.raises(InvalidCursorError, match="value exceeds maximum length"): | |
| decode_cursor(encoded, ALLOWED) | |
| def test_unsupported_sort_field(self): | |
| encoded = encode_cursor("execution_time", "1", "id-1") | |
| with pytest.raises(InvalidCursorError, match="unsupported sort field"): | |
| decode_cursor(encoded, ALLOWED) | |
| def test_no_allowed_fields_rejects_everything(self): | |
| encoded = encode_cursor("created_at", "1", "id-1") | |
| with pytest.raises(InvalidCursorError): | |
| decode_cursor(encoded, ()) | |
| def test_non_dict_payload_rejected(self): | |
| encoded = base64.urlsafe_b64encode(b'["array","not","dict"]').rstrip(b"=").decode("ascii") | |
| with pytest.raises(InvalidCursorError, match="expected object"): | |
| decode_cursor(encoded, ALLOWED) | |
| class TestEncodeAtCapsFits: | |
| def test_max_field_lengths_fit_wire_cap(self): | |
| # Worst-case payload: value and id at their per-field caps, with a long | |
| # sort field name. The encoded cursor must fit within MAX_ENCODED_CURSOR_LENGTH | |
| # so the wire cap cannot reject a cursor the encoder mints at the per-field caps. | |
| value = "v" * MAX_CURSOR_VALUE_LENGTH | |
| id = "i" * MAX_CURSOR_ID_LENGTH | |
| sort_field = "very_long_sort_field_name" | |
| encoded = encode_cursor(sort_field, value, id) | |
| assert len(encoded) <= MAX_ENCODED_CURSOR_LENGTH | |
| payload = decode_cursor(encoded, (sort_field,)) | |
| assert payload.value == value | |
| assert payload.id == id | |
| class TestDatetimeOverflow: | |
| """Crafted cursors with extreme micros must map to InvalidCursorError, | |
| not OverflowError/OSError leaking as 500. | |
| """ | |
| def test_out_of_range_micros_rejected(self, micros_str): | |
| encoded = encode_cursor("created_at", micros_str, "asset-x") | |
| payload = decode_cursor(encoded, ALLOWED) | |
| with pytest.raises(InvalidCursorError): | |
| decode_cursor_time(payload) | |
| class TestEncoderDecoderSymmetry: | |
| """The encoder must never mint a cursor the decoder would reject, or the | |
| same server would 400 on a cursor it just handed out. Per-field caps keep | |
| the encoded length below the wire cap, so a freshly minted cursor always | |
| round-trips. | |
| """ | |
| def test_long_name_within_cap_round_trips(self): | |
| """Assets allow names up to 512 chars (`String(512)`); the cursor | |
| encoder must round-trip a value at that cap so a freshly minted | |
| cursor never fails decode on the next request.""" | |
| long_name = "n" * MAX_CURSOR_VALUE_LENGTH | |
| encoded = encode_cursor("name", long_name, "asset-x") | |
| payload = decode_cursor(encoded, ALLOWED) | |
| assert payload.value == long_name | |
| def test_encoder_rejects_empty_id(self): | |
| with pytest.raises(InvalidCursorError, match="id must be non-empty"): | |
| encode_cursor("created_at", "1", "") | |
| def test_encoder_rejects_oversized_id(self): | |
| with pytest.raises(InvalidCursorError, match="id exceeds maximum length"): | |
| encode_cursor("created_at", "1", "a" * (MAX_CURSOR_ID_LENGTH + 1)) | |
| def test_encoder_rejects_oversized_value(self): | |
| with pytest.raises(InvalidCursorError, match="value exceeds maximum length"): | |
| encode_cursor("name", "v" * (MAX_CURSOR_VALUE_LENGTH + 1), "id-1") | |
| def test_multibyte_value_at_cap_round_trips(self): | |
| """A value at the char-count cap made of multibyte characters | |
| (e.g. 'é' = 2 UTF-8 bytes) stays under the wire cap, so it mints and | |
| round-trips — the per-field caps, not a mint-time length check, are | |
| what bound cursor size.""" | |
| value = "é" * MAX_CURSOR_VALUE_LENGTH | |
| encoded = encode_cursor("name", value, "asset-multibyte") | |
| assert len(encoded) <= MAX_ENCODED_CURSOR_LENGTH | |
| payload = decode_cursor(encoded, ALLOWED) | |
| assert payload.value == value | |
| def test_escape_heavy_value_at_cap_round_trips(self): | |
| """JSON escape expansion is the worst case: each control character | |
| serializes to the six-byte `\\uXXXX` form. A value of 512 of them is | |
| the largest a cursor can get, and it still fits the wire cap, mints, | |
| and round-trips.""" | |
| value = "\x01" * MAX_CURSOR_VALUE_LENGTH | |
| encoded = encode_cursor("name", value, "asset-escape") | |
| assert len(encoded) <= MAX_ENCODED_CURSOR_LENGTH | |
| payload = decode_cursor(encoded, ALLOWED) | |
| assert payload.value == value | |
| class TestOrderBinding: | |
| def test_order_baked_into_payload(self): | |
| encoded = encode_cursor("created_at", "1", "id-1", order="asc") | |
| payload = decode_cursor(encoded, ALLOWED) | |
| assert payload.order == "asc" | |
| def test_mismatched_order_rejected(self): | |
| encoded = encode_cursor("created_at", "1", "id-1", order="desc") | |
| with pytest.raises(InvalidCursorError, match="does not match request order"): | |
| decode_cursor(encoded, ALLOWED, expected_order="asc") | |
| def test_matching_order_accepted(self): | |
| encoded = encode_cursor("created_at", "1", "id-1", order="desc") | |
| payload = decode_cursor(encoded, ALLOWED, expected_order="desc") | |
| assert payload.order == "desc" | |
| def test_invalid_order_token_rejected_on_encode(self): | |
| with pytest.raises(ValueError): | |
| encode_cursor("created_at", "1", "id-1", order="sideways") | |
| def test_invalid_order_token_rejected_on_decode(self): | |
| # Hand-craft a payload with an illegal `o` value. | |
| raw = b'{"s":"name","v":"x","id":"id-1","o":"sideways"}' | |
| encoded = base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii") | |
| with pytest.raises(InvalidCursorError, match="unsupported order"): | |
| decode_cursor(encoded, ALLOWED) | |
| def test_cursor_without_order_rejected(self): | |
| """`o` is mandatory. A cursor minted without it is rejected as | |
| malformed rather than silently walking the keyset in whatever | |
| direction the request happens to ask for.""" | |
| raw = b'{"s":"name","v":"x","id":"id-1"}' | |
| encoded = base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii") | |
| with pytest.raises(InvalidCursorError, match="missing or non-string o"): | |
| decode_cursor(encoded, ALLOWED, expected_order="desc") | |
Xet Storage Details
- Size:
- 12 kB
- Xet hash:
- 888c8db3d2024f442712365434cb8632e7ad11277b69d0f90c9881c41721cee1
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.