dictdb queries should [shallow] copy records

pop/edit actions on the results of queries should not affect the records themselves.

Hub.db_query pops buffers out of results in some situations, which would break future result_requests.  

Associated tests are included.

IPython/parallel/controller/dictdb.py

@@ -1,216 +1,216 b''
 """A Task logger that presents our DB interface,
 but exists entirely in memory and implemented with dicts.
 
 Authors:
 
 * Min RK
 
 
 TaskRecords are dicts of the form:
 {
     'msg_id' : str(uuid),
     'client_uuid' : str(uuid),
     'engine_uuid' : str(uuid) or None,
     'header' : dict(header),
     'content': dict(content),
     'buffers': list(buffers),
     'submitted': datetime,
     'started': datetime or None,
     'completed': datetime or None,
     'resubmitted': datetime or None,
     'result_header' : dict(header) or None,
     'result_content' : dict(content) or None,
     'result_buffers' : list(buffers) or None,
 }
 With this info, many of the special categories of tasks can be defined by query:
 
 pending:  completed is None
 client's outstanding: client_uuid = uuid && completed is None
 MIA: arrived is None (and completed is None)
 etc.
 
 EngineRecords are dicts of the form:
 {
     'eid' : int(id),
     'uuid': str(uuid)
 }
 This may be extended, but is currently.
 
 We support a subset of mongodb operators:
     $lt,$gt,$lte,$gte,$ne,$in,$nin,$all,$mod,$exists
 """
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2010-2011  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
-
+from copy import copy
 from datetime import datetime
 
 from IPython.config.configurable import LoggingConfigurable
 
 from IPython.utils.traitlets import Dict, Unicode, Instance
 
 filters = {
  '$lt' : lambda a,b: a < b,
  '$gt' : lambda a,b: b > a,
  '$eq' : lambda a,b: a == b,
  '$ne' : lambda a,b: a != b,
  '$lte': lambda a,b: a <= b,
  '$gte': lambda a,b: a >= b,
  '$in' : lambda a,b: a in b,
  '$nin': lambda a,b: a not in b,
  '$all': lambda a,b: all([ a in bb for bb in b ]),
  '$mod': lambda a,b: a%b[0] == b[1],
  '$exists' : lambda a,b: (b and a is not None) or (a is None and not b)
 }
 
 
 class CompositeFilter(object):
     """Composite filter for matching multiple properties."""
 
     def __init__(self, dikt):
         self.tests = []
         self.values = []
         for key, value in dikt.iteritems():
             self.tests.append(filters[key])
             self.values.append(value)
 
     def __call__(self, value):
         for test,check in zip(self.tests, self.values):
             if not test(value, check):
                 return False
         return True
 
 class BaseDB(LoggingConfigurable):
     """Empty Parent class so traitlets work on DB."""
     # base configurable traits:
     session = Unicode("")
 
 class DictDB(BaseDB):
     """Basic in-memory dict-based object for saving Task Records.
 
     This is the first object to present the DB interface
     for logging tasks out of memory.
 
     The interface is based on MongoDB, so adding a MongoDB
     backend should be straightforward.
     """
 
     _records = Dict()
 
     def _match_one(self, rec, tests):
         """Check if a specific record matches tests."""
         for key,test in tests.iteritems():
             if not test(rec.get(key, None)):
                 return False
         return True
 
     def _match(self, check):
         """Find all the matches for a check dict."""
         matches = []
         tests = {}
         for k,v in check.iteritems():
             if isinstance(v, dict):
                 tests[k] = CompositeFilter(v)
             else:
                 tests[k] = lambda o: o==v
 
         for rec in self._records.itervalues():
             if self._match_one(rec, tests):
-                matches.append(rec)
+                matches.append(copy(rec))
         return matches
 
     def _extract_subdict(self, rec, keys):
         """extract subdict of keys"""
         d = {}
         d['msg_id'] = rec['msg_id']
         for key in keys:
             d[key] = rec[key]
         return d
 
     def add_record(self, msg_id, rec):
         """Add a new Task Record, by msg_id."""
         if self._records.has_key(msg_id):
             raise KeyError("Already have msg_id %r"%(msg_id))
         self._records[msg_id] = rec
 
     def get_record(self, msg_id):
         """Get a specific Task Record, by msg_id."""
-        if not self._records.has_key(msg_id):
+        if not msg_id in self._records:
             raise KeyError("No such msg_id %r"%(msg_id))
-        return self._records[msg_id]
+        return copy(self._records[msg_id])
 
     def update_record(self, msg_id, rec):
         """Update the data in an existing record."""
         self._records[msg_id].update(rec)
 
     def drop_matching_records(self, check):
         """Remove a record from the DB."""
         matches = self._match(check)
         for m in matches:
             del self._records[m['msg_id']]
 
     def drop_record(self, msg_id):
         """Remove a record from the DB."""
         del self._records[msg_id]
 
 
     def find_records(self, check, keys=None):
         """Find records matching a query dict, optionally extracting subset of keys.
 
         Returns dict keyed by msg_id of matching records.
 
         Parameters
         ----------
 
         check: dict
             mongodb-style query argument
         keys: list of strs [optional]
             if specified, the subset of keys to extract.  msg_id will *always* be
             included.
         """
         matches = self._match(check)
         if keys:
             return [ self._extract_subdict(rec, keys) for rec in matches ]
         else:
             return matches
 
 
     def get_history(self):
         """get all msg_ids, ordered by time submitted."""
         msg_ids = self._records.keys()
         return sorted(msg_ids, key=lambda m: self._records[m]['submitted'])
 
 class NoDB(DictDB):
     """A blackhole db backend that actually stores no information.
     
     Provides the full DB interface, but raises KeyErrors on any
     method that tries to access the records.  This can be used to
     minimize the memory footprint of the Hub when its record-keeping
     functionality is not required.
     """
     
     def add_record(self, msg_id, record):
         pass
     
     def get_record(self, msg_id):
         raise KeyError("NoDB does not support record access")
     
     def update_record(self, msg_id, record):
         pass
     
     def drop_matching_records(self, check):
         pass
     
     def drop_record(self, msg_id):
         pass
     
     def find_records(self, check, keys=None):
         raise KeyError("NoDB does not store information")
     
     def get_history(self):
         raise KeyError("NoDB does not store information")
 

IPython/parallel/tests/test_client.py

@@ -1,319 +1,331 b''
 """Tests for parallel client.py
 
 Authors:
 
 * Min RK
 """
 
 #-------------------------------------------------------------------------------
 #  Copyright (C) 2011  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-------------------------------------------------------------------------------
 
 #-------------------------------------------------------------------------------
 # Imports
 #-------------------------------------------------------------------------------
 
 from __future__ import division
 
 import time
 from datetime import datetime
 from tempfile import mktemp
 
 import zmq
 
 from IPython.parallel.client import client as clientmod
 from IPython.parallel import error
 from IPython.parallel import AsyncResult, AsyncHubResult
 from IPython.parallel import LoadBalancedView, DirectView
 
 from clienttest import ClusterTestCase, segfault, wait, add_engines
 
 def setup():
     add_engines(4, total=True)
 
 class TestClient(ClusterTestCase):
     
     def test_ids(self):
         n = len(self.client.ids)
         self.add_engines(2)
         self.assertEquals(len(self.client.ids), n+2)
     
     def test_view_indexing(self):
         """test index access for views"""
         self.minimum_engines(4)
         targets = self.client._build_targets('all')[-1]
         v = self.client[:]
         self.assertEquals(v.targets, targets)
         t = self.client.ids[2]
         v = self.client[t]
         self.assert_(isinstance(v, DirectView))
         self.assertEquals(v.targets, t)
         t = self.client.ids[2:4]
         v = self.client[t]
         self.assert_(isinstance(v, DirectView))
         self.assertEquals(v.targets, t)
         v = self.client[::2]
         self.assert_(isinstance(v, DirectView))
         self.assertEquals(v.targets, targets[::2])
         v = self.client[1::3]
         self.assert_(isinstance(v, DirectView))
         self.assertEquals(v.targets, targets[1::3])
         v = self.client[:-3]
         self.assert_(isinstance(v, DirectView))
         self.assertEquals(v.targets, targets[:-3])
         v = self.client[-1]
         self.assert_(isinstance(v, DirectView))
         self.assertEquals(v.targets, targets[-1])
         self.assertRaises(TypeError, lambda : self.client[None])
     
     def test_lbview_targets(self):
         """test load_balanced_view targets"""
         v = self.client.load_balanced_view()
         self.assertEquals(v.targets, None)
         v = self.client.load_balanced_view(-1)
         self.assertEquals(v.targets, [self.client.ids[-1]])
         v = self.client.load_balanced_view('all')
         self.assertEquals(v.targets, None)
     
     def test_dview_targets(self):
         """test direct_view targets"""
         v = self.client.direct_view()
         self.assertEquals(v.targets, 'all')
         v = self.client.direct_view('all')
         self.assertEquals(v.targets, 'all')
         v = self.client.direct_view(-1)
         self.assertEquals(v.targets, self.client.ids[-1])
     
     def test_lazy_all_targets(self):
         """test lazy evaluation of rc.direct_view('all')"""
         v = self.client.direct_view()
         self.assertEquals(v.targets, 'all')
         
         def double(x):
             return x*2
         seq = range(100)
         ref = [ double(x) for x in seq ]
         
         # add some engines, which should be used
         self.add_engines(1)
         n1 = len(self.client.ids)
         
         # simple apply
         r = v.apply_sync(lambda : 1)
         self.assertEquals(r, [1] * n1)
         
         # map goes through remotefunction
         r = v.map_sync(double, seq)
         self.assertEquals(r, ref)
 
         # add a couple more engines, and try again
         self.add_engines(2)
         n2 = len(self.client.ids)
         self.assertNotEquals(n2, n1)
         
         # apply
         r = v.apply_sync(lambda : 1)
         self.assertEquals(r, [1] * n2)
         
         # map
         r = v.map_sync(double, seq)
         self.assertEquals(r, ref)
     
     def test_targets(self):
         """test various valid targets arguments"""
         build = self.client._build_targets
         ids = self.client.ids
         idents,targets = build(None)
         self.assertEquals(ids, targets)
     
     def test_clear(self):
         """test clear behavior"""
         self.minimum_engines(2)
         v = self.client[:]
         v.block=True
         v.push(dict(a=5))
         v.pull('a')
         id0 = self.client.ids[-1]
         self.client.clear(targets=id0, block=True)
         a = self.client[:-1].get('a')
         self.assertRaisesRemote(NameError, self.client[id0].get, 'a')
         self.client.clear(block=True)
         for i in self.client.ids:
             self.assertRaisesRemote(NameError, self.client[i].get, 'a')
     
     def test_get_result(self):
         """test getting results from the Hub."""
         c = clientmod.Client(profile='iptest')
         t = c.ids[-1]
         ar = c[t].apply_async(wait, 1)
         # give the monitor time to notice the message
         time.sleep(.25)
         ahr = self.client.get_result(ar.msg_ids)
         self.assertTrue(isinstance(ahr, AsyncHubResult))
         self.assertEquals(ahr.get(), ar.get())
         ar2 = self.client.get_result(ar.msg_ids)
         self.assertFalse(isinstance(ar2, AsyncHubResult))
         c.close()
     
     def test_ids_list(self):
         """test client.ids"""
         ids = self.client.ids
         self.assertEquals(ids, self.client._ids)
         self.assertFalse(ids is self.client._ids)
         ids.remove(ids[-1])
         self.assertNotEquals(ids, self.client._ids)
     
     def test_queue_status(self):
         ids = self.client.ids
         id0 = ids[0]
         qs = self.client.queue_status(targets=id0)
         self.assertTrue(isinstance(qs, dict))
         self.assertEquals(sorted(qs.keys()), ['completed', 'queue', 'tasks'])
         allqs = self.client.queue_status()
         self.assertTrue(isinstance(allqs, dict))
         intkeys = list(allqs.keys())
         intkeys.remove('unassigned')
         self.assertEquals(sorted(intkeys), sorted(self.client.ids))
         unassigned = allqs.pop('unassigned')
         for eid,qs in allqs.items():
             self.assertTrue(isinstance(qs, dict))
             self.assertEquals(sorted(qs.keys()), ['completed', 'queue', 'tasks'])
 
     def test_shutdown(self):
         ids = self.client.ids
         id0 = ids[0]
         self.client.shutdown(id0, block=True)
         while id0 in self.client.ids:
             time.sleep(0.1)
             self.client.spin()
         
         self.assertRaises(IndexError, lambda : self.client[id0])
         
     def test_result_status(self):
         pass
         # to be written
     
     def test_db_query_dt(self):
         """test db query by date"""
         hist = self.client.hub_history()
         middle = self.client.db_query({'msg_id' : hist[len(hist)//2]})[0]
         tic = middle['submitted']
         before = self.client.db_query({'submitted' : {'$lt' : tic}})
         after = self.client.db_query({'submitted' : {'$gte' : tic}})
         self.assertEquals(len(before)+len(after),len(hist))
         for b in before:
             self.assertTrue(b['submitted'] < tic)
         for a in after:
             self.assertTrue(a['submitted'] >= tic)
         same = self.client.db_query({'submitted' : tic})
         for s in same:
             self.assertTrue(s['submitted'] == tic)
     
     def test_db_query_keys(self):
         """test extracting subset of record keys"""
         found = self.client.db_query({'msg_id': {'$ne' : ''}},keys=['submitted', 'completed'])
         for rec in found:
             self.assertEquals(set(rec.keys()), set(['msg_id', 'submitted', 'completed']))
     
     def test_db_query_default_keys(self):
         """default db_query excludes buffers"""
         found = self.client.db_query({'msg_id': {'$ne' : ''}})
         for rec in found:
             keys = set(rec.keys())
             self.assertFalse('buffers' in keys, "'buffers' should not be in: %s" % keys)
             self.assertFalse('result_buffers' in keys, "'result_buffers' should not be in: %s" % keys)
     
     def test_db_query_msg_id(self):
         """ensure msg_id is always in db queries"""
         found = self.client.db_query({'msg_id': {'$ne' : ''}},keys=['submitted', 'completed'])
         for rec in found:
             self.assertTrue('msg_id' in rec.keys())
         found = self.client.db_query({'msg_id': {'$ne' : ''}},keys=['submitted'])
         for rec in found:
             self.assertTrue('msg_id' in rec.keys())
         found = self.client.db_query({'msg_id': {'$ne' : ''}},keys=['msg_id'])
         for rec in found:
             self.assertTrue('msg_id' in rec.keys())
     
+    def test_db_query_get_result(self):
+        """pop in db_query shouldn't pop from result itself"""
+        self.client[:].apply_sync(lambda : 1)
+        found = self.client.db_query({'msg_id': {'$ne' : ''}})
+        rc2 = clientmod.Client(profile='iptest')
+        # If this bug is not fixed, this call will hang:
+        ar = rc2.get_result(self.client.history[-1])
+        ar.wait(2)
+        self.assertTrue(ar.ready())
+        ar.get()
+        rc2.close()
+    
     def test_db_query_in(self):
         """test db query with '$in','$nin' operators"""
         hist = self.client.hub_history()
         even = hist[::2]
         odd = hist[1::2]
         recs = self.client.db_query({ 'msg_id' : {'$in' : even}})
         found = [ r['msg_id'] for r in recs ]
         self.assertEquals(set(even), set(found))
         recs = self.client.db_query({ 'msg_id' : {'$nin' : even}})
         found = [ r['msg_id'] for r in recs ]
         self.assertEquals(set(odd), set(found))
     
     def test_hub_history(self):
         hist = self.client.hub_history()
         recs = self.client.db_query({ 'msg_id' : {"$ne":''}})
         recdict = {}
         for rec in recs:
             recdict[rec['msg_id']] = rec
         
         latest = datetime(1984,1,1)
         for msg_id in hist:
             rec = recdict[msg_id]
             newt = rec['submitted']
             self.assertTrue(newt >= latest)
             latest = newt
         ar = self.client[-1].apply_async(lambda : 1)
         ar.get()
         time.sleep(0.25)
         self.assertEquals(self.client.hub_history()[-1:],ar.msg_ids)
     
     def test_resubmit(self):
         def f():
             import random
             return random.random()
         v = self.client.load_balanced_view()
         ar = v.apply_async(f)
         r1 = ar.get(1)
         # give the Hub a chance to notice:
         time.sleep(0.5)
         ahr = self.client.resubmit(ar.msg_ids)
         r2 = ahr.get(1)
         self.assertFalse(r1 == r2)
 
     def test_resubmit_inflight(self):
         """ensure ValueError on resubmit of inflight task"""
         v = self.client.load_balanced_view()
         ar = v.apply_async(time.sleep,1)
         # give the message a chance to arrive
         time.sleep(0.2)
         self.assertRaisesRemote(ValueError, self.client.resubmit, ar.msg_ids)
         ar.get(2)
 
     def test_resubmit_badkey(self):
         """ensure KeyError on resubmit of nonexistant task"""
         self.assertRaisesRemote(KeyError, self.client.resubmit, ['invalid'])
 
     def test_purge_results(self):
         # ensure there are some tasks
         for i in range(5):
             self.client[:].apply_sync(lambda : 1)
         # Wait for the Hub to realise the result is done:
         # This prevents a race condition, where we
         # might purge a result the Hub still thinks is pending.
         time.sleep(0.1)
         rc2 = clientmod.Client(profile='iptest')
         hist = self.client.hub_history()
         ahr = rc2.get_result([hist[-1]])
         ahr.wait(10)
         self.client.purge_results(hist[-1])
         newhist = self.client.hub_history()
         self.assertEquals(len(newhist)+1,len(hist))
         rc2.spin()
         rc2.close()
         
     def test_purge_all_results(self):
         self.client.purge_results('all')
         hist = self.client.hub_history()
         self.assertEquals(len(hist), 0)
 

IPython/parallel/tests/test_db.py

@@ -1,194 +1,224 b''
 """Tests for db backends
 
 Authors:
 
 * Min RK
 """
 
 #-------------------------------------------------------------------------------
 #  Copyright (C) 2011  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-------------------------------------------------------------------------------
 
 #-------------------------------------------------------------------------------
 # Imports
 #-------------------------------------------------------------------------------
 
 from __future__ import division
 
 import tempfile
 import time
 
 from datetime import datetime, timedelta
 from unittest import TestCase
 
 from IPython.parallel import error
 from IPython.parallel.controller.dictdb import DictDB
 from IPython.parallel.controller.sqlitedb import SQLiteDB
 from IPython.parallel.controller.hub import init_record, empty_record
 
 from IPython.testing import decorators as dec
 from IPython.zmq.session import Session
 
 
 #-------------------------------------------------------------------------------
 # TestCases
 #-------------------------------------------------------------------------------
 
 class TestDictBackend(TestCase):
     def setUp(self):
         self.session = Session()
         self.db = self.create_db()
         self.load_records(16)
     
     def create_db(self):
         return DictDB()
     
     def load_records(self, n=1):
         """load n records for testing"""
         #sleep 1/10 s, to ensure timestamp is different to previous calls
         time.sleep(0.1)
         msg_ids = []
         for i in range(n):
             msg = self.session.msg('apply_request', content=dict(a=5))
             msg['buffers'] = []
             rec = init_record(msg)
             msg_id = msg['header']['msg_id']
             msg_ids.append(msg_id)
             self.db.add_record(msg_id, rec)
         return msg_ids
     
     def test_add_record(self):
         before = self.db.get_history()
         self.load_records(5)
         after = self.db.get_history()
         self.assertEquals(len(after), len(before)+5)
         self.assertEquals(after[:-5],before)
         
     def test_drop_record(self):
         msg_id = self.load_records()[-1]
         rec = self.db.get_record(msg_id)
         self.db.drop_record(msg_id)
         self.assertRaises(KeyError,self.db.get_record, msg_id)
     
     def _round_to_millisecond(self, dt):
         """necessary because mongodb rounds microseconds"""
         micro = dt.microsecond
         extra = int(str(micro)[-3:])
         return dt - timedelta(microseconds=extra)
     
     def test_update_record(self):
         now = self._round_to_millisecond(datetime.now())
         # 
         msg_id = self.db.get_history()[-1]
         rec1 = self.db.get_record(msg_id)
         data = {'stdout': 'hello there', 'completed' : now}
         self.db.update_record(msg_id, data)
         rec2 = self.db.get_record(msg_id)
         self.assertEquals(rec2['stdout'], 'hello there')
         self.assertEquals(rec2['completed'], now)
         rec1.update(data)
         self.assertEquals(rec1, rec2)
     
     # def test_update_record_bad(self):
     #     """test updating nonexistant records"""
     #     msg_id = str(uuid.uuid4())
     #     data = {'stdout': 'hello there'}
     #     self.assertRaises(KeyError, self.db.update_record, msg_id, data)
 
     def test_find_records_dt(self):
         """test finding records by date"""
         hist = self.db.get_history()
         middle = self.db.get_record(hist[len(hist)//2])
         tic = middle['submitted']
         before = self.db.find_records({'submitted' : {'$lt' : tic}})
         after = self.db.find_records({'submitted' : {'$gte' : tic}})
         self.assertEquals(len(before)+len(after),len(hist))
         for b in before:
             self.assertTrue(b['submitted'] < tic)
         for a in after:
             self.assertTrue(a['submitted'] >= tic)
         same = self.db.find_records({'submitted' : tic})
         for s in same:
             self.assertTrue(s['submitted'] == tic)
     
     def test_find_records_keys(self):
         """test extracting subset of record keys"""
         found = self.db.find_records({'msg_id': {'$ne' : ''}},keys=['submitted', 'completed'])
         for rec in found:
             self.assertEquals(set(rec.keys()), set(['msg_id', 'submitted', 'completed']))
     
     def test_find_records_msg_id(self):
         """ensure msg_id is always in found records"""
         found = self.db.find_records({'msg_id': {'$ne' : ''}},keys=['submitted', 'completed'])
         for rec in found:
             self.assertTrue('msg_id' in rec.keys())
         found = self.db.find_records({'msg_id': {'$ne' : ''}},keys=['submitted'])
         for rec in found:
             self.assertTrue('msg_id' in rec.keys())
         found = self.db.find_records({'msg_id': {'$ne' : ''}},keys=['msg_id'])
         for rec in found:
             self.assertTrue('msg_id' in rec.keys())
     
     def test_find_records_in(self):
         """test finding records with '$in','$nin' operators"""
         hist = self.db.get_history()
         even = hist[::2]
         odd = hist[1::2]
         recs = self.db.find_records({ 'msg_id' : {'$in' : even}})
         found = [ r['msg_id'] for r in recs ]
         self.assertEquals(set(even), set(found))
         recs = self.db.find_records({ 'msg_id' : {'$nin' : even}})
         found = [ r['msg_id'] for r in recs ]
         self.assertEquals(set(odd), set(found))
     
     def test_get_history(self):
         msg_ids = self.db.get_history()
         latest = datetime(1984,1,1)
         for msg_id in msg_ids:
             rec = self.db.get_record(msg_id)
             newt = rec['submitted']
             self.assertTrue(newt >= latest)
             latest = newt
         msg_id = self.load_records(1)[-1]
         self.assertEquals(self.db.get_history()[-1],msg_id)
     
     def test_datetime(self):
         """get/set timestamps with datetime objects"""
         msg_id = self.db.get_history()[-1]
         rec = self.db.get_record(msg_id)
         self.assertTrue(isinstance(rec['submitted'], datetime))
         self.db.update_record(msg_id, dict(completed=datetime.now()))
         rec = self.db.get_record(msg_id)
         self.assertTrue(isinstance(rec['completed'], datetime))
 
     def test_drop_matching(self):
         msg_ids = self.load_records(10)
         query = {'msg_id' : {'$in':msg_ids}}
         self.db.drop_matching_records(query)
         recs = self.db.find_records(query)
         self.assertEquals(len(recs), 0)
     
     def test_null(self):
         """test None comparison queries"""
         msg_ids = self.load_records(10)
 
         query = {'msg_id' : None}
         recs = self.db.find_records(query)
         self.assertEquals(len(recs), 0)
 
         query = {'msg_id' : {'$ne' : None}}
         recs = self.db.find_records(query)
         self.assertTrue(len(recs) >= 10)
+    
+    def test_pop_safe_get(self):
+        """editing query results shouldn't affect record [get]"""
+        msg_id = self.db.get_history()[-1]
+        rec = self.db.get_record(msg_id)
+        rec.pop('buffers')
+        rec['garbage'] = 'hello'
+        rec2 = self.db.get_record(msg_id)
+        self.assertTrue('buffers' in rec2)
+        self.assertFalse('garbage' in rec2)
+    
+    def test_pop_safe_find(self):
+        """editing query results shouldn't affect record [find]"""
+        msg_id = self.db.get_history()[-1]
+        rec = self.db.find_records({'msg_id' : msg_id})[0]
+        rec.pop('buffers')
+        rec['garbage'] = 'hello'
+        rec2 = self.db.find_records({'msg_id' : msg_id})[0]
+        self.assertTrue('buffers' in rec2)
+        self.assertFalse('garbage' in rec2)
+
+    def test_pop_safe_find_keys(self):
+        """editing query results shouldn't affect record [find+keys]"""
+        msg_id = self.db.get_history()[-1]
+        rec = self.db.find_records({'msg_id' : msg_id}, keys=['buffers'])[0]
+        rec.pop('buffers')
+        rec['garbage'] = 'hello'
+        rec2 = self.db.find_records({'msg_id' : msg_id})[0]
+        self.assertTrue('buffers' in rec2)
+        self.assertFalse('garbage' in rec2)
 
 
 class TestSQLiteBackend(TestDictBackend):
 
     @dec.skip_without('sqlite3')
     def create_db(self):
         return SQLiteDB(location=tempfile.gettempdir())
     
     def tearDown(self):
         self.db._db.close()