diff --git a/src/service/logs/bulk.rs b/src/service/logs/bulk.rs index 25856266fe..b9d71121ed 100644 --- a/src/service/logs/bulk.rs +++ b/src/service/logs/bulk.rs @@ -104,6 +104,31 @@ pub async fn ingest( } (action, stream_name, doc_id) = ret.unwrap(); + if stream_name.is_empty() || stream_name == "_" || stream_name == "/" { + let err_msg = format!("Invalid stream name: {}", line); + log::warn!("{}", err_msg); + bulk_res.errors = true; + let err = BulkResponseError::new( + err_msg.to_string(), + stream_name.clone(), + err_msg, + "0".to_string(), + ); + let mut item = HashMap::new(); + item.insert( + action.clone(), + BulkResponseItem::new_failed( + stream_name.clone(), + doc_id.clone().unwrap_or_default(), + err, + Some(value), + stream_name.clone(), + ), + ); + bulk_res.items.push(item); + continue; // skip + } + if !cfg.common.skip_formatting_stream_name { stream_name = format_stream_name(&stream_name); } @@ -534,7 +559,7 @@ pub fn add_record_status( failure_type, stream_name.clone(), failure_reason.unwrap(), - "0".to_owned(), // TODO check + "0".to_owned(), ); item.insert( diff --git a/tests/api-testing/tests/test_bulk.py b/tests/api-testing/tests/test_bulk.py index cbdec1a6fe..87d2418300 100644 --- a/tests/api-testing/tests/test_bulk.py +++ b/tests/api-testing/tests/test_bulk.py @@ -30,33 +30,6 @@ def test_e2e_bulk_ingest(create_session, base_url): f"Expected no errors in the valid bulk request response, but got {resp_valid_bulk.json()}" ) - # **Negative Test Case 1: Invalid Bulk Payload (Missing Document Body)** - invalid_bulk_payload = ( - '{ "index" : { "_index" : "pytest10222", "_id" : "1" } }\n' - '{ "field1" : "value1" }' # Add a document body after the index action - ) - - resp_invalid_bulk_1 = session.post(url, data=invalid_bulk_payload, headers=headers) - - # Print the raw response and content for debugging - print("Negative Test Response (Raw Content):", resp_invalid_bulk_1.content) - - try: - response_json = resp_invalid_bulk_1.json() # Attempt to parse the JSON response - print("Negative Test Response (JSON):", response_json) - except Exception as e: - print("Error parsing JSON response:", str(e)) - - # Assert negative case for the first invalid payload - # Expecting a 200 response even for invalid data, but check that there's no error field - assert resp_invalid_bulk_1.status_code == 200, ( - f"Expected status code 200 for invalid bulk request, but got {resp_invalid_bulk_1.status_code}. " - f"Response: {resp_invalid_bulk_1.content}" - ) - assert not resp_invalid_bulk_1.json().get("errors"), ( - f"Expected no errors in the invalid bulk request response, but got {resp_invalid_bulk_1.json()}" - ) - # **Negative Test Case 2: Invalid Bulk Payload (Empty Index)** invalid_bulk_payload2 = [ { "index" : { "_index" : "" } }, # Empty index @@ -85,9 +58,9 @@ def test_e2e_bulk_ingest(create_session, base_url): f"Response: {resp_invalid_bulk_2.content}" ) - # Assert no errors in the response, as expected from Postman behavior - assert not resp_invalid_bulk_2.json().get("errors"), ( - f"Expected no errors in the invalid bulk request response, but got {resp_invalid_bulk_2.json()}" + # Assert errors is true in the response + assert resp_invalid_bulk_2.json().get("errors"), ( + f"Expected errors to be true in the invalid bulk request response, but got {resp_invalid_bulk_2.json()}" )