Falcon - Difference in stream type between unittests and actual API on post

Question

I'm trying to write unittests for my falcon api, and I encountered a really weird issue when I tried reading the body I added to the unittests.

This is my unittest:

class TestDetectionApi(DetectionApiSetUp):
    def test_valid_detection(self):
        headers = {"Content-Type": "application/x-www-form-urlencoded"}
        body = {'test': 'test'}
        detection_result = self.simulate_post('/environments/e6ce2a50-f68f-4a7a-8562-ca50822b805d/detectionEvaluations',
                                              body=urlencode(body), headers=headers)
        self.assertEqual(detection_result.json, None)

and this is the part in my API that reads the body:

    def _get_request_body(request: falcon.Request) -> dict:
        request_stream = request.stream.read()
        request_body = json.loads(request_stream)
        validate(request_body, REQUEST_VALIDATION_SCHEMA)
        return request_body

Now for the weird part, my function for reading the body is working without any issue when I run the API, but when I run the unittests the stream type seems to be different which affect the reading of it. The stream type when running the API is gunicorn.http.body.Body and using unittests: wsgiref.validate.InputWrapper.

So when reading the body from the api all I need to do it request.stream.read() but when using the unittests I need to do request.stream.input.read() which is pretty annoying since I need to change my original code to work with both cases and I don't want to do it.

Is there a way to fix this issue? Thanks!!

Answer 1

It seems like issue was with how I read it. instead of using stream I used bounded_stream which seemed to work, also I removed the headers and just decoded my body.

my unittest:

class TestDetectionApi(DetectionApiSetUp):
    def test_valid_detection(self):
        body = '''{'test': 'test'}'''
        detection_result = self.simulate_post('/environments/e6ce2a50-f68f-4a7a-8562-ca50822b805d/detectionEvaluations',
                                              body=body.encode(), headers=headers)
        self.assertEqual(detection_result.json, None)

how I read it:

    def _get_request_body(request: falcon.Request) -> dict:
        request_stream = request.bounded_stream.read()
        request_body = json.loads(request_stream)
        validate(request_body, REQUEST_VALIDATION_SCHEMA)
        return request_body