(environ, start_response)
| 88 | collection = client.test.get_collection(coll_name, codec_options=OPTS) |
| 89 | |
| 90 | def application(environ, start_response): |
| 91 | results = list(collection.find().batch_size(10)) |
| 92 | assert len(results) == ndocs, f"n_actual={len(results)} n_expected={ndocs}" |
| 93 | # Test encoding and decoding works (for sub interpreter support). |
| 94 | decoded = bson.decode(bson.encode(doc, codec_options=OPTS), codec_options=OPTS) |
| 95 | for key, value in doc.items(): |
| 96 | # Native regex objects are decoded as bson Regex. |
| 97 | if isinstance(value, re.Pattern): |
| 98 | value = Regex.from_native(value) |
| 99 | assert decoded[key] == value, f"failed on doc[{key!r}]: {decoded[key]!r} != {value!r}" |
| 100 | assert isinstance( |
| 101 | decoded[key], type(value) |
| 102 | ), f"failed on doc[{key}]: {decoded[key]!r} is not an instance of {type(value)}" |
| 103 | |
| 104 | output = ( |
| 105 | f" python {sys.version}, mod_wsgi {mod_wsgi_version}," |
| 106 | f" pymongo {pymongo.version}," |
| 107 | f' mod_wsgi.process_group = {environ["mod_wsgi.process_group"]!r}' |
| 108 | f' mod_wsgi.application_group = {environ["mod_wsgi.application_group"]!r}' |
| 109 | f' wsgi.multithread = {environ["wsgi.multithread"]!r}' |
| 110 | "\n" |
| 111 | ) |
| 112 | response_headers = [("Content-Length", str(len(output)))] |
| 113 | start_response("200 OK", response_headers) |
| 114 | return [output.encode("ascii")] |
nothing calls this directly
no test coverage detected