PYTHON-952 - Add spec test runner and fix bugs

Author: behackett

pymongo/bulk.py

@@ -337,14 +337,16 @@ def execute_no_results(self, sock_info, generator):
  check_keys,
  operation['multi'],
  write_concern=write_concern,
- op_id=op_id)
+ op_id=op_id,
+ ordered=self.ordered)
  else:
  for operation in run.ops:
  coll._delete(sock_info,
  operation['q'],
  not operation['limit'],
  write_concern,
- op_id)
+ op_id,
+ self.ordered)
  except OperationFailure:
  if self.ordered:
  break
@@ -373,6 +375,7 @@ def execute_legacy(self, sock_info, generator, write_concern):
  if run.op_type == _INSERT:
  coll._insert(sock_info,
  operation,
+ self.ordered,
  write_concern=write_concern,
  op_id=op_id)
  result = {}
@@ -388,13 +391,15 @@ def execute_legacy(self, sock_info, generator, write_concern):
  check_keys,
  operation['multi'],
  write_concern=write_concern,
- op_id=op_id)
+ op_id=op_id,
+ ordered=self.ordered)
  else:
  result = coll._delete(sock_info,
  operation['q'],
  not operation['limit'],
  write_concern,
- op_id)
+ op_id,
+ self.ordered)
  _merge_legacy(run, full_result, result, idx)
  except DocumentTooLarge as exc:
  # MongoDB 2.6 uses error code 2 for "too large".

pymongo/collection.py

@@ -411,17 +411,19 @@ def _legacy_write(
  dur, details, name, rqst_id, sock_info.address, op_id)
  raise
  if publish:
- # Comply with APM spec.
- reply = {'ok': 1}
- if result:
+ if result is not None:
  reply = message._convert_write_result(name, cmd, result)
+ else:
+ # Comply with APM spec.
+ reply = {'ok': 1}
  duration = (datetime.datetime.now() - start) + duration
  monitoring.publish_command_success(
  duration, reply, name, rqst_id, sock_info.address, op_id)
  return result
 
  def _insert_one(
- self, sock_info, doc, check_keys, manipulate, write_concern, op_id):
+ self, sock_info, doc, ordered,
+ check_keys, manipulate, write_concern, op_id):
  """Internal helper for inserting a single document."""
  if manipulate:
  doc = self.__database._apply_incoming_manipulators(doc, self)
@@ -432,7 +434,7 @@ def _insert_one(
  concern = (write_concern or self.write_concern).document
  acknowledged = concern.get("w") != 0
  command = SON([('insert', self.name),
- ('ordered', True),
+ ('ordered', ordered),
  ('documents', [doc])])
  if concern:
  command['writeConcern'] = concern
@@ -457,7 +459,8 @@ def _insert(self, sock_info, docs, ordered=True, check_keys=True,
  """Internal insert helper."""
  if isinstance(docs, collections.MutableMapping):
  return self._insert_one(
- sock_info, docs, check_keys, manipulate, write_concern, op_id)
+ sock_info, docs, ordered,
+ check_keys, manipulate, write_concern, op_id)
 
  ids = []
 
@@ -580,15 +583,15 @@ def gen():
 
  def _update(self, sock_info, criteria, document, upsert=False,
  check_keys=True, multi=False, manipulate=False,
- write_concern=None, op_id=None):
+ write_concern=None, op_id=None, ordered=True):
  """Internal update / replace helper."""
  common.validate_boolean("upsert", upsert)
  if manipulate:
  document = self.__database._fix_incoming(document, self)
  concern = (write_concern or self.write_concern).document
  acknowledged = concern.get("w") != 0
  command = SON([('update', self.name),
- ('ordered', True),
+ ('ordered', ordered),
  ('updates', [SON([('q', criteria),
  ('u', document),
  ('multi', multi),
@@ -758,13 +761,14 @@ def drop(self):
  self.__database.drop_collection(self.__name)
 
  def _delete(
- self, sock_info, criteria, multi, write_concern=None, op_id=None):
+ self, sock_info, criteria, multi,
+ write_concern=None, op_id=None, ordered=True):
  """Internal delete helper."""
  common.validate_is_mapping("filter", criteria)
  concern = (write_concern or self.write_concern).document
  acknowledged = concern.get("w") != 0
  command = SON([('delete', self.name),
- ('ordered', True),
+ ('ordered', ordered),
  ('deletes', [SON([('q', criteria),
  ('limit', int(not multi))])])])
  if concern:

pymongo/message.py

@@ -115,7 +115,8 @@ def _convert_write_result(operation, command, result):
  elif result.get("updatedExisting") is False and affected == 1:
  # If _id is in both the update document *and* the query spec
  # the update document _id takes precedence.
- _id = command["u"].get("_id", command["q"].get("_id"))
+ update = command['updates'][0]
+ _id = update["u"].get("_id", update["q"].get("_id"))
  res["upserted"] = [{"index": 0, "_id": _id}]
  return res
 
@@ -479,14 +480,16 @@ def legacy_write(self, request_id, msg, max_doc_size, acknowledged, docs):
  cmd = self._start(request_id, docs)
  start = datetime.datetime.now()
  try:
- reply = self.sock_info.legacy_write(
+ result = self.sock_info.legacy_write(
  request_id, msg, max_doc_size, acknowledged)
  if self.publish:
  duration = (datetime.datetime.now() - start) + duration
- self._succeed(
- request_id,
- _convert_write_result(self.name, cmd, reply),
- duration)
+ if result is not None:
+ reply = _convert_write_result(self.name, cmd, result)
+ else:
+ # Comply with APM spec.
+ reply = {'ok': 1}
+ self._succeed(request_id, reply, duration)
  except OperationFailure as exc:
  if self.publish:
  duration = (datetime.datetime.now() - start) + duration
@@ -498,7 +501,7 @@ def legacy_write(self, request_id, msg, max_doc_size, acknowledged, docs):
  raise
  finally:
  self.start_time = datetime.datetime.now()
- return reply
+ return result
 
  def write_command(self, request_id, msg, docs):
  """A proxy for SocketInfo.write_command that handles event publishing.

test/command_monitoring/bulkWrite.json

@@ -0,0 +1,161 @@
+{
+ "data": [
+ {
+ "_id": 1,
+ "x": 11
+ },
+ {
+ "_id": 2,
+ "x": 22
+ },
+ {
+ "_id": 3,
+ "x": 33
+ }
+ ],
+ "collection_name": "test",
+ "database_name": "command-monitoring-tests",
+ "tests": [
+ {
+ "description": "A successful mixed bulk write",
+ "operation": {
+ "name": "bulkWrite",
+ "arguments": {
+ "requests": [
+ {
+ "insertOne": {
+ "document": {
+ "_id": 4,
+ "x": 44
+ }
+ }
+ },
+ {
+ "updateOne": {
+ "filter": {
+ "_id": 3
+ },
+ "update": {
+ "$set": {
+ "x": 333
+ }
+ }
+ }
+ }
+ ]
+ }
+ },
+ "expectations": [
+ {
+ "command_started_event": {
+ "command": {
+ "insert": "test",
+ "documents": [
+ {
+ "_id": 4,
+ "x": 44
+ }
+ ],
+ "ordered": true
+ },
+ "command_name": "insert",
+ "database_name": "command-monitoring-tests"
+ }
+ },
+ {
+ "command_succeeded_event": {
+ "reply": {
+ "ok": 1.0,
+ "n": 1
+ },
+ "command_name": "insert"
+ }
+ },
+ {
+ "command_started_event": {
+ "command": {
+ "update": "test",
+ "updates": [
+ {
+ "q": {
+ "_id": 3
+ },
+ "u": {
+ "$set": {
+ "x": 333
+ }
+ },
+ "upsert": false,
+ "multi": false
+ }
+ ],
+ "ordered": true
+ },
+ "command_name": "update",
+ "database_name": "command-monitoring-tests"
+ }
+ },
+ {
+ "command_succeeded_event": {
+ "reply": {
+ "ok": 1.0,
+ "n": 1
+ },
+ "command_name": "update"
+ }
+ }
+ ]
+ },
+ {
+ "description": "A successful unordered bulk write with an unacknowledged write concern",
+ "operation": {
+ "name": "bulkWrite",
+ "arguments": {
+ "requests": [
+ {
+ "insertOne": {
+ "document": {
+ "_id": 4,
+ "x": 44
+ }
+ }
+ }
+ ],
+ "ordered": false,
+ "writeConcern": {
+ "w": 0
+ }
+ }
+ },
+ "expectations": [
+ {
+ "command_started_event": {
+ "command": {
+ "insert": "test",
+ "documents": [
+ {
+ "_id": 4,
+ "x": 44
+ }
+ ],
+ "ordered": false,
+ "writeConcern": {
+ "w": 0
+ }
+ },
+ "command_name": "insert",
+ "database_name": "command-monitoring-tests"
+ }
+ },
+ {
+ "command_succeeded_event": {
+ "reply": {
+ "ok": 1.0
+ },
+ "command_name": "insert"
+ }
+ }
+ ]
+ }
+ ]
+}