🧪 test(pypi): close the pypi and webhook mutation survivors

crates/peryx-ecosystem-pypi/src/cache/page_stream/mod.rs

@@ -123,6 +123,8 @@ pub async fn stream_detail(
 ) -> Result<PageOutcome, CacheError> {
     let index = state.index_at(position);
     index.policy.check_resource(PolicyAction::Serve, &project)?;
+    // A resource size limit makes a policy active, so this is also what keeps a size-limited index off
+    // the streaming path: it cannot total a project's bytes as it goes.
     if index.policy.active() || super::has_active_revocations(&state)? {
         return Ok(PageOutcome::Fallback);
     }
@@ -332,7 +334,6 @@ fn streaming_parts(
     project: &str,
 ) -> Result<Option<(String, UpstreamClient, bool, crate::stream::PageContext)>, CacheError> {
     match &index.kind {
-        _ if index.policy.has_resource_size_limit() => Ok(None),
         IndexKind::Cached { client, offline } => Ok(Some((
             index.name.clone(),
             client.clone(),

crates/peryx-ecosystem-pypi/src/search_pypi.rs

@@ -25,7 +25,6 @@ use peryx_search::{
 
 const IDENTITY_TEXT_BYTES: usize = INDEXED_TEXT_BYTES / 4;
 const CORE_METADATA_TEXT_BYTES: usize = INDEXED_TEXT_BYTES / 2;
-const CATALOG_TEXT_BYTES: usize = INDEXED_TEXT_BYTES - IDENTITY_TEXT_BYTES - CORE_METADATA_TEXT_BYTES;
 
 /// Produces `PyPI` search documents for the neutral search index.
 #[derive(Debug, Clone, Copy, Default)]
@@ -554,9 +553,11 @@ fn search_text(
         push_metadata(&mut core_metadata, metadata, CORE_METADATA_TEXT_BYTES);
     }
 
-    let catalog_limit = CATALOG_TEXT_BYTES
-        + IDENTITY_TEXT_BYTES.saturating_sub(identity.len())
-        + CORE_METADATA_TEXT_BYTES.saturating_sub(core_metadata.len());
+    // Catalog text is bounded by the whole allowance rather than by a share of it. Working out what
+    // identity and core metadata left unspent computed the same bound the assembly below already
+    // applies: the three sections are joined under `INDEXED_TEXT_BYTES`, and separators make that
+    // join bind a byte or two sooner than any share arithmetic could. One cap, applied once.
+    let catalog_limit = INDEXED_TEXT_BYTES;
     let mut catalog = String::with_capacity(512);
     push_unique_text(
         &mut catalog,
@@ -655,3 +656,7 @@ fn upload_key<'key>(key: &'key str, index: &str) -> Option<(&'key str, &'key str
     let (project, filename) = rest.split_once('/')?;
     (!project.is_empty() && !filename.is_empty()).then_some((project, filename))
 }
+
+#[cfg(test)]
+#[path = "../tests/unit/search_pypi/tests.rs"]
+mod tests;

crates/peryx-ecosystem-pypi/src/serving/mutate.rs

@@ -204,7 +204,7 @@ async fn yank_request(
     )
     .await;
     security_mutation_event(&audit, &result);
-    notify_mutation_webhook(state, &result);
+    peryx_events::webhook::notify_changed(state.as_ref(), &result);
     count_response(result)
 }
 
@@ -236,7 +236,7 @@ async fn restore_request(
     })
     .await;
     security_mutation_event(&audit, &result);
-    notify_mutation_webhook(state, &result);
+    peryx_events::webhook::notify_changed(state.as_ref(), &result);
     count_response(result)
 }
 
@@ -281,7 +281,7 @@ pub async fn pypi_dispatch_delete(
         )
         .await;
         security_mutation_event(&audit, &result);
-        notify_mutation_webhook(&state, &result);
+        peryx_events::webhook::notify_changed(state.as_ref(), &result);
         return count_response(result);
     }
     let (project, version) = match parse_project_version(spec) {
@@ -317,7 +317,7 @@ pub async fn pypi_dispatch_delete(
     )
     .await;
     security_mutation_event(&audit, &result);
-    notify_mutation_webhook(&state, &result);
+    peryx_events::webhook::notify_changed(state.as_ref(), &result);
     count_response(result)
 }
 
@@ -430,12 +430,6 @@ fn prepare_mutation_webhook(
     )
 }
 
-fn notify_mutation_webhook(state: &ServingState, result: &Result<usize, CacheError>) {
-    if result.as_ref().is_ok_and(|count| *count > 0) {
-        peryx_events::webhook::notify(state);
-    }
-}
-
 /// Peel a trailing `/{action}` off the spec, but only when a project segment precedes it. A project
 /// whose PEP 503 name is itself `yank`/`restore`/`promote` must stay addressable at the project
 /// level, so the action grammar never claims the whole spec.

crates/peryx-ecosystem-pypi/src/shadow/query.rs

@@ -90,3 +90,7 @@ fn paginate(mut candidates: Vec<ShadowCandidate>, query: &ShadowQuery) -> Shadow
         next_cursor,
     }
 }
+
+#[cfg(test)]
+#[path = "../../tests/unit/shadow/query_tests.rs"]
+mod tests;

crates/peryx-ecosystem-pypi/src/stream/transformer.rs

@@ -408,7 +408,12 @@ impl PageTransformer {
                         return;
                     }
                 };
-                let value = value << 4 | u16::from(digit);
+                // A shift-and-or over disjoint bit ranges is an addition wearing other clothes, and
+                // that is why swapping the operator changes nothing: with the low nibble always zero,
+                // `|`, `^` and `+` agree on every input. Spelled as the arithmetic it performs, a
+                // wrong operator shows up. Four hex digits is the most `\uXXXX` carries, so the
+                // accumulation tops out at `0xFFF * 16 + 15`, exactly `u16::MAX`.
+                let value = value * 16 + u16::from(digit);
                 if seen + 1 == 4 {
                     self.key_decode = KeyDecode::Literal;
                     self.push_key_codepoint(value);
@@ -561,7 +566,6 @@ impl PageTransformer {
                 }
             }
             b',' if self.depth == self.array_depth => {}
-            _ if self.capture.is_empty() && is_json_whitespace(byte) => {}
             _ => self.capture.push(byte),
         }
         Ok(())
@@ -843,3 +847,7 @@ fn supports_metadata_sibling(filename: &str) -> bool {
             .get(filename.len().saturating_sub(7)..)
             .is_some_and(|suffix| suffix.eq_ignore_ascii_case(".tar.gz"))
 }
+
+#[cfg(test)]
+#[path = "../../tests/unit/stream/transformer_bounds.rs"]
+mod bound_tests;

crates/peryx-ecosystem-pypi/src/stream/validator.rs

@@ -381,7 +381,12 @@ impl JsonValidator {
             self.failed = true;
             return;
         };
-        let value = (value << 4) | digit;
+        // A shift-and-or over disjoint bit ranges is an addition wearing other clothes: the low
+        // nibble is always zero, so `|`, `^` and `+` agree on every input and the operator's choice
+        // cannot be observed. Spelled as the arithmetic it performs, a wrong one shows. Four hex
+        // digits is all `\uXXXX` carries, so the accumulation tops out at `0xFFF * 16 + 15`, exactly
+        // `u16::MAX`.
+        let value = value * 16 + digit;
         if digits + 1 < 4 {
             self.state = State::Str {
                 escape: Escape::Unicode {

crates/peryx-ecosystem-pypi/tests/unit/cache/metadata/tests.rs

@@ -535,3 +535,128 @@ async fn test_virtual_index_surfaces_an_undecodable_hosted_record() {
 
     assert!(matches!(err, CacheError::Parse(_)), "{err:?}");
 }
+
+const RANGED_WHEEL: &str = "sample_pkg-1.0-py3-none-any.whl";
+const RANGED_METADATA: &str = "Metadata-Version: 2.1\nName: sample-pkg\nVersion: 1.0\n";
+
+fn ranged_wheel_bytes() -> Vec<u8> {
+    ranged_wheel_holding(RANGED_METADATA.as_bytes(), zip::CompressionMethod::Deflated)
+}
+
+fn ranged_wheel_holding(metadata: &[u8], method: zip::CompressionMethod) -> Vec<u8> {
+    use std::io::Write as _;
+    let mut bytes = Vec::new();
+    let mut archive = zip::ZipWriter::new(std::io::Cursor::new(&mut bytes));
+    let options = zip::write::SimpleFileOptions::default().compression_method(method);
+    archive.start_file("sample_pkg/__init__.py", options).unwrap();
+    archive.write_all(b"__version__ = \"1.0\"\n").unwrap();
+    archive
+        .start_file("sample_pkg-1.0.dist-info/METADATA", options)
+        .unwrap();
+    archive.write_all(metadata).unwrap();
+    archive.start_file("sample_pkg-1.0.dist-info/WHEEL", options).unwrap();
+    archive
+        .write_all(b"Wheel-Version: 1.0\nGenerator: peryx-test\nRoot-Is-Purelib: true\nTag: py3-none-any\n")
+        .unwrap();
+    archive.finish().unwrap();
+    bytes
+}
+
+async fn ranged_outcome(wheel: Vec<u8>) -> RemoteMetadata {
+    let server = wiremock::MockServer::start().await;
+    wiremock::Mock::given(wiremock::matchers::path(format!("/files/{RANGED_WHEEL}")))
+        .respond_with(RangedWheelServer(wheel))
+        .mount(&server)
+        .await;
+    let client = ArtifactClient::from(UpstreamClient::new(&format!("{}/", server.uri())).unwrap());
+    wheel_metadata_by_range(&client, &format!("{}/files/{RANGED_WHEEL}", server.uri()), RANGED_WHEEL)
+        .await
+        .unwrap()
+}
+
+/// The member budget admits a member of exactly its size and refuses only what passes it, and it asks
+/// that of both sizes a ZIP records: what the member occupies in the archive and what it becomes when
+/// decoded. Either one over the budget is enough to decline the ranged read, so a member that
+/// compresses small still cannot smuggle a large one past.
+///
+/// Only a member at the limit separates admitting from refusing it, which is why each case is exact.
+#[tokio::test]
+async fn test_a_member_of_exactly_the_budget_is_read() {
+    let metadata = vec![b'm'; usize::try_from(crate::archive::MAX_WHEEL_METADATA_BYTES).unwrap()];
+    let outcome = ranged_outcome(ranged_wheel_holding(&metadata, zip::CompressionMethod::Stored)).await;
+
+    assert!(
+        matches!(&outcome, RemoteMetadata::Found(read) if *read == metadata),
+        "a member at the budget is within it"
+    );
+}
+
+#[tokio::test]
+async fn test_a_member_one_byte_past_the_budget_is_declined() {
+    let metadata = vec![b'm'; usize::try_from(crate::archive::MAX_WHEEL_METADATA_BYTES).unwrap() + 1];
+    let outcome = ranged_outcome(ranged_wheel_holding(&metadata, zip::CompressionMethod::Stored)).await;
+
+    assert!(matches!(outcome, RemoteMetadata::Unsupported));
+}
+
+#[tokio::test]
+async fn test_a_member_that_compresses_small_is_still_judged_by_its_decoded_size() {
+    let metadata = vec![b'm'; usize::try_from(crate::archive::MAX_WHEEL_METADATA_BYTES).unwrap() + 1];
+    let wheel = ranged_wheel_holding(&metadata, zip::CompressionMethod::Deflated);
+    assert!(
+        wheel.len() < metadata.len(),
+        "the member has to compress well for this to mean anything"
+    );
+
+    assert!(matches!(ranged_outcome(wheel).await, RemoteMetadata::Unsupported));
+}
+
+/// A server that answers the ranged read the way a real one does: a `HEAD` carrying the length and a
+/// strong validator, then each `GET` returning exactly the bytes its `Range` asked for.
+struct RangedWheelServer(Vec<u8>);
+
+impl wiremock::Respond for RangedWheelServer {
+    fn respond(&self, request: &wiremock::Request) -> wiremock::ResponseTemplate {
+        let etag = "\"sample-pkg-1.0\"";
+        let Some(range) = request.headers.get("range") else {
+            return wiremock::ResponseTemplate::new(200)
+                .insert_header("etag", etag)
+                .insert_header("accept-ranges", "bytes")
+                .insert_header("content-length", self.0.len().to_string().as_str());
+        };
+        let spec = range.to_str().unwrap().trim_start_matches("bytes=").to_owned();
+        let (start, end) = spec.split_once('-').expect("a bounded range");
+        let (start, end): (usize, usize) = (start.parse().unwrap(), end.parse().unwrap());
+        wiremock::ResponseTemplate::new(206)
+            .insert_header("etag", etag)
+            .insert_header(
+                "content-range",
+                format!("bytes {start}-{end}/{}", self.0.len()).as_str(),
+            )
+            .set_body_bytes(self.0[start..=end].to_vec())
+    }
+}
+
+/// Reading a wheel's metadata over ranges is arithmetic on offsets and lengths: where the central
+/// directory starts and ends, where the member's data begins, and how far it runs. Every one of those
+/// ends is inclusive, so each is a length short of the next offset. Getting any of them wrong asks the
+/// server for the wrong bytes, and the answer is either a refused range or a member that will not
+/// decode, never the metadata.
+///
+/// Nothing drove that arithmetic before: the only ranged-read test rejected bad filenames without
+/// fetching anything.
+#[tokio::test]
+async fn test_a_wheel_member_is_read_back_whole_over_ranges() {
+    let server = wiremock::MockServer::start().await;
+    wiremock::Mock::given(wiremock::matchers::path(format!("/files/{RANGED_WHEEL}")))
+        .respond_with(RangedWheelServer(ranged_wheel_bytes()))
+        .mount(&server)
+        .await;
+    let client = ArtifactClient::from(UpstreamClient::new(&format!("{}/", server.uri())).unwrap());
+
+    let outcome = wheel_metadata_by_range(&client, &format!("{}/files/{RANGED_WHEEL}", server.uri()), RANGED_WHEEL)
+        .await
+        .unwrap();
+
+    assert!(matches!(&outcome, RemoteMetadata::Found(metadata) if metadata == RANGED_METADATA.as_bytes()));
+}

crates/peryx-ecosystem-pypi/tests/unit/cache/page_stream/tests.rs

@@ -141,3 +141,43 @@ async fn test_revalidation_keeps_the_stale_page_when_upstream_is_unparseable() {
     assert!(String::from_utf8(body).unwrap().contains("1.0"));
     drop(flight_gate(&state, "pypi/flask").try_lock_owned().unwrap());
 }
+
+/// The preflight reads 64 KiB before deciding how to serve a JSON page. Written as a product and read
+/// nowhere else, an operator changed inside it would resize that read without any test noticing.
+#[test]
+fn test_the_json_preflight_reads_sixty_four_kibibytes() {
+    assert_eq!(super::JSON_META_PREFLIGHT_BYTES, 65_536);
+}
+
+fn context_holding(local_versions: Vec<String>) -> crate::stream::PageContext {
+    crate::stream::page_context(
+        "root/pypi",
+        "demo",
+        peryx_policy::Policy::default(),
+        Vec::new(),
+        local_versions,
+        &std::collections::BTreeMap::new(),
+    )
+}
+
+/// An upstream that has nothing to say leaves peryx serving whatever it holds of its own. A project
+/// with no local files and no local versions has nothing, so the page is genuinely absent; holding
+/// either one is enough to answer from the local side instead. Requiring both to be missing is what
+/// separates those, and nothing exercised a project holding versions but no files.
+#[test]
+fn test_a_page_with_nothing_local_is_absent_rather_than_fallible() {
+    assert!(matches!(
+        missing_upstream_outcome(&context_holding(Vec::new())),
+        PageOutcome::NotFound
+    ));
+}
+
+#[test]
+fn test_a_page_holding_only_versions_still_falls_back_to_them() {
+    let outcome = missing_upstream_outcome(&context_holding(vec!["1.0".to_owned()]));
+
+    assert!(
+        matches!(outcome, PageOutcome::Fallback),
+        "a version peryx holds is something to serve"
+    );
+}

crates/peryx-ecosystem-pypi/tests/unit/cache/provenance/tests.rs

@@ -2,3 +2,65 @@
 fn test_artifact_project_falls_back_for_a_legacy_distribution() {
     assert_eq!(super::artifact_project("Legacy_Name-1.0.egg"), "legacy-name");
 }
+
+fn attestation_response(declared: Option<usize>, body: Vec<u8>) -> reqwest::Response {
+    let mut builder = axum::http::Response::builder()
+        .status(200)
+        .header(axum::http::header::CONTENT_TYPE, "application/json");
+    if let Some(declared) = declared {
+        builder = builder.header(axum::http::header::CONTENT_LENGTH, declared.to_string());
+    }
+    reqwest::Response::from(builder.body(body).unwrap())
+}
+
+fn refused_for_size(outcome: &Result<super::FetchOutcome, crate::cache::CacheError>) -> bool {
+    matches!(
+        outcome,
+        Err(crate::cache::CacheError::Upstream(
+            peryx_upstream::UpstreamError::ResponseTooLarge { .. }
+        ))
+    )
+}
+
+/// The limit admits a bundle of exactly its size and refuses only what passes it, and it says so twice:
+/// once from the length upstream declares, and again from the bytes that actually arrive, since a
+/// declared length is a claim rather than a measurement.
+#[tokio::test]
+async fn test_a_declared_length_of_exactly_the_limit_is_not_refused_for_size() {
+    let outcome = super::process_upstream_attestation_response(attestation_response(
+        Some(super::MAX_PROVENANCE_BYTES),
+        b"{}".to_vec(),
+    ))
+    .await;
+
+    assert!(!refused_for_size(&outcome), "a bundle at the limit is within it");
+}
+
+#[tokio::test]
+async fn test_a_declared_length_past_the_limit_is_refused_for_size() {
+    let outcome = super::process_upstream_attestation_response(attestation_response(
+        Some(super::MAX_PROVENANCE_BYTES + 1),
+        b"{}".to_vec(),
+    ))
+    .await;
+
+    assert!(refused_for_size(&outcome));
+}
+
+#[tokio::test]
+async fn test_arriving_bytes_of_exactly_the_limit_are_not_refused_for_size() {
+    let body = vec![b'x'; super::MAX_PROVENANCE_BYTES];
+
+    let outcome = super::process_upstream_attestation_response(attestation_response(None, body)).await;
+
+    assert!(!refused_for_size(&outcome), "bytes at the limit are within it");
+}
+
+#[tokio::test]
+async fn test_arriving_bytes_past_the_limit_are_refused_for_size() {
+    let body = vec![b'x'; super::MAX_PROVENANCE_BYTES + 1];
+
+    let outcome = super::process_upstream_attestation_response(attestation_response(None, body)).await;
+
+    assert!(refused_for_size(&outcome));
+}