git.notmuchmail.org Git - notmuch/blob - bindings/python-cffi/tests/test_database.py

   1 import collections
   2 import configparser
   3 import os
   4 import pathlib
   5
   6 import pytest
   7
   8 import notmuch2
   9 import notmuch2._errors as errors
  10 import notmuch2._database as dbmod
  11 import notmuch2._message as message
  12
  13
  14 @pytest.fixture
  15 def db(maildir):
  16     with dbmod.Database.create(maildir.path) as db:
  17         yield db
  18
  19
  20 class TestDefaultDb:
  21     """Tests for reading the default database.
  22
  23     The error cases are fairly undefined, some relevant Python error
  24     will come out if you give it a bad filename or if the file does
  25     not parse correctly.  So we're not testing this too deeply.
  26     """
  27
  28     def test_config_pathname_default(self, monkeypatch):
  29         monkeypatch.delenv('NOTMUCH_CONFIG', raising=False)
  30         user = pathlib.Path('~/.notmuch-config').expanduser()
  31         assert dbmod._config_pathname() == user
  32
  33     def test_config_pathname_env(self, monkeypatch):
  34         monkeypatch.setenv('NOTMUCH_CONFIG', '/some/random/path')
  35         assert dbmod._config_pathname() == pathlib.Path('/some/random/path')
  36
  37     def test_default_path_nocfg(self, monkeypatch, tmppath):
  38         monkeypatch.setenv('NOTMUCH_CONFIG', str(tmppath/'foo'))
  39         with pytest.raises(FileNotFoundError):
  40             dbmod.Database.default_path()
  41
  42     def test_default_path_cfg_is_dir(self, monkeypatch, tmppath):
  43         monkeypatch.setenv('NOTMUCH_CONFIG', str(tmppath))
  44         with pytest.raises(IsADirectoryError):
  45             dbmod.Database.default_path()
  46
  47     def test_default_path_parseerr(self, monkeypatch, tmppath):
  48         cfg = tmppath / 'notmuch-config'
  49         with cfg.open('w') as fp:
  50             fp.write('invalid')
  51         monkeypatch.setenv('NOTMUCH_CONFIG', str(cfg))
  52         with pytest.raises(configparser.Error):
  53             dbmod.Database.default_path()
  54
  55     def test_default_path_parse(self, monkeypatch, tmppath):
  56         cfg = tmppath / 'notmuch-config'
  57         with cfg.open('w') as fp:
  58             fp.write('[database]\n')
  59             fp.write('path={!s}'.format(tmppath))
  60         monkeypatch.setenv('NOTMUCH_CONFIG', str(cfg))
  61         assert dbmod.Database.default_path() == tmppath
  62
  63     def test_default_path_param(self, monkeypatch, tmppath):
  64         cfg_dummy = tmppath / 'dummy'
  65         monkeypatch.setenv('NOTMUCH_CONFIG', str(cfg_dummy))
  66         cfg_real = tmppath / 'notmuch_config'
  67         with cfg_real.open('w') as fp:
  68             fp.write('[database]\n')
  69             fp.write('path={!s}'.format(cfg_real/'mail'))
  70         assert dbmod.Database.default_path(cfg_real) == cfg_real/'mail'
  71
  72
  73 class TestCreate:
  74
  75     def test_create(self, tmppath, db):
  76         assert tmppath.joinpath('.notmuch/xapian/').exists()
  77
  78     def test_create_already_open(self, tmppath, db):
  79         with pytest.raises(errors.NotmuchError):
  80             db.create(tmppath)
  81
  82     def test_create_existing(self, tmppath, db):
  83         with pytest.raises(errors.FileError):
  84             dbmod.Database.create(path=tmppath)
  85
  86     def test_close(self, db):
  87         db.close()
  88
  89     def test_del_noclose(self, db):
  90         del db
  91
  92     def test_close_del(self, db):
  93         db.close()
  94         del db
  95
  96     def test_closed_attr(self, db):
  97         assert not db.closed
  98         db.close()
  99         assert db.closed
 100
 101     def test_ctx(self, db):
 102         with db as ctx:
 103             assert ctx is db
 104             assert not db.closed
 105         assert db.closed
 106
 107     def test_path(self, db, tmppath):
 108         assert db.path == tmppath
 109
 110     def test_version(self, db):
 111         assert db.version > 0
 112
 113     def test_needs_upgrade(self, db):
 114         assert db.needs_upgrade in (True, False)
 115
 116
 117 class TestAtomic:
 118
 119     def test_exit_early(self, db):
 120         with pytest.raises(errors.UnbalancedAtomicError):
 121             with db.atomic() as ctx:
 122                 ctx.force_end()
 123
 124     def test_exit_late(self, db):
 125         with db.atomic() as ctx:
 126             pass
 127         with pytest.raises(errors.UnbalancedAtomicError):
 128             ctx.force_end()
 129
 130     def test_abort(self, db):
 131         with db.atomic() as txn:
 132             txn.abort()
 133         assert db.closed
 134
 135
 136 class TestRevision:
 137
 138     def test_single_rev(self, db):
 139         r = db.revision()
 140         assert isinstance(r, dbmod.DbRevision)
 141         assert isinstance(r.rev, int)
 142         assert isinstance(r.uuid, bytes)
 143         assert r is r
 144         assert r == r
 145         assert r <= r
 146         assert r >= r
 147         assert not r < r
 148         assert not r > r
 149
 150     def test_diff_db(self, tmppath):
 151         dbpath0 = tmppath.joinpath('db0')
 152         dbpath0.mkdir()
 153         dbpath1 = tmppath.joinpath('db1')
 154         dbpath1.mkdir()
 155         db0 = dbmod.Database.create(path=dbpath0)
 156         db1 = dbmod.Database.create(path=dbpath1)
 157         r_db0 = db0.revision()
 158         r_db1 = db1.revision()
 159         assert r_db0 != r_db1
 160         assert r_db0.uuid != r_db1.uuid
 161
 162     def test_cmp(self, db, maildir):
 163         rev0 = db.revision()
 164         _, pathname = maildir.deliver()
 165         db.add(pathname, sync_flags=False)
 166         rev1 = db.revision()
 167         assert rev0 < rev1
 168         assert rev0 <= rev1
 169         assert not rev0 > rev1
 170         assert not rev0 >= rev1
 171         assert not rev0 == rev1
 172         assert rev0 != rev1
 173
 174     # XXX add tests for revisions comparisons
 175
 176 class TestMessages:
 177
 178     def test_add_message(self, db, maildir):
 179         msgid, pathname = maildir.deliver()
 180         msg, dup = db.add(pathname, sync_flags=False)
 181         assert isinstance(msg, message.Message)
 182         assert msg.path == pathname
 183         assert msg.messageid == msgid
 184
 185     def test_add_message_str(self, db, maildir):
 186         msgid, pathname = maildir.deliver()
 187         msg, dup = db.add(str(pathname), sync_flags=False)
 188
 189     def test_add_message_bytes(self, db, maildir):
 190         msgid, pathname = maildir.deliver()
 191         msg, dup = db.add(os.fsencode(bytes(pathname)), sync_flags=False)
 192
 193     def test_remove_message(self, db, maildir):
 194         msgid, pathname = maildir.deliver()
 195         msg, dup = db.add(pathname, sync_flags=False)
 196         assert db.find(msgid)
 197         dup = db.remove(pathname)
 198         with pytest.raises(LookupError):
 199             db.find(msgid)
 200
 201     def test_remove_message_str(self, db, maildir):
 202         msgid, pathname = maildir.deliver()
 203         msg, dup = db.add(pathname, sync_flags=False)
 204         assert db.find(msgid)
 205         dup = db.remove(str(pathname))
 206         with pytest.raises(LookupError):
 207             db.find(msgid)
 208
 209     def test_remove_message_bytes(self, db, maildir):
 210         msgid, pathname = maildir.deliver()
 211         msg, dup = db.add(pathname, sync_flags=False)
 212         assert db.find(msgid)
 213         dup = db.remove(os.fsencode(bytes(pathname)))
 214         with pytest.raises(LookupError):
 215             db.find(msgid)
 216
 217     def test_find_message(self, db, maildir):
 218         msgid, pathname = maildir.deliver()
 219         msg0, dup = db.add(pathname, sync_flags=False)
 220         msg1 = db.find(msgid)
 221         assert isinstance(msg1, message.Message)
 222         assert msg1.messageid == msgid == msg0.messageid
 223         assert msg1.path == pathname == msg0.path
 224
 225     def test_find_message_notfound(self, db):
 226         with pytest.raises(LookupError):
 227             db.find('foo')
 228
 229     def test_get_message(self, db, maildir):
 230         msgid, pathname = maildir.deliver()
 231         msg0, _ = db.add(pathname, sync_flags=False)
 232         msg1 = db.get(pathname)
 233         assert isinstance(msg1, message.Message)
 234         assert msg1.messageid == msgid == msg0.messageid
 235         assert msg1.path == pathname == msg0.path
 236
 237     def test_get_message_str(self, db, maildir):
 238         msgid, pathname = maildir.deliver()
 239         db.add(pathname, sync_flags=False)
 240         msg = db.get(str(pathname))
 241         assert msg.messageid == msgid
 242
 243     def test_get_message_bytes(self, db, maildir):
 244         msgid, pathname = maildir.deliver()
 245         db.add(pathname, sync_flags=False)
 246         msg = db.get(os.fsencode(bytes(pathname)))
 247         assert msg.messageid == msgid
 248
 249
 250 class TestTags:
 251     # We just want to test this behaves like a set at a hight level.
 252     # The set semantics are tested in detail in the test_tags module.
 253
 254     def test_type(self, db):
 255         assert isinstance(db.tags, collections.abc.Set)
 256
 257     def test_none(self, db):
 258         itags = iter(db.tags)
 259         with pytest.raises(StopIteration):
 260             next(itags)
 261         assert len(db.tags) == 0
 262         assert not db.tags
 263
 264     def test_some(self, db, maildir):
 265         _, pathname = maildir.deliver()
 266         msg, _ = db.add(pathname, sync_flags=False)
 267         msg.tags.add('hello')
 268         itags = iter(db.tags)
 269         assert next(itags) == 'hello'
 270         with pytest.raises(StopIteration):
 271             next(itags)
 272         assert 'hello' in msg.tags
 273
 274     def test_cache(self, db):
 275         assert db.tags is db.tags
 276
 277     def test_iters(self, db):
 278         i1 = iter(db.tags)
 279         i2 = iter(db.tags)
 280         assert i1 is not i2
 281
 282
 283 class TestQuery:
 284
 285     @pytest.fixture
 286     def db(self, maildir, notmuch):
 287         """Return a read-only notmuch2.Database.
 288
 289         The database will have 3 messages, 2 threads.
 290         """
 291         msgid, _ = maildir.deliver(body='foo')
 292         maildir.deliver(body='bar')
 293         maildir.deliver(body='baz',
 294                         headers=[('In-Reply-To', '<{}>'.format(msgid))])
 295         notmuch('new')
 296         with dbmod.Database(maildir.path, 'rw') as db:
 297             yield db
 298
 299     def test_count_messages(self, db):
 300         assert db.count_messages('*') == 3
 301
 302     def test_messages_type(self, db):
 303         msgs = db.messages('*')
 304         assert isinstance(msgs, collections.abc.Iterator)
 305
 306     def test_message_no_results(self, db):
 307         msgs = db.messages('not_a_matching_query')
 308         with pytest.raises(StopIteration):
 309             next(msgs)
 310
 311     def test_message_match(self, db):
 312         msgs = db.messages('*')
 313         msg = next(msgs)
 314         assert isinstance(msg, notmuch2.Message)
 315
 316     def test_count_threads(self, db):
 317         assert db.count_threads('*') == 2
 318
 319     def test_threads_type(self, db):
 320         threads = db.threads('*')
 321         assert isinstance(threads, collections.abc.Iterator)
 322
 323     def test_threads_no_match(self, db):
 324         threads = db.threads('not_a_matching_query')
 325         with pytest.raises(StopIteration):
 326             next(threads)
 327
 328     def test_threads_match(self, db):
 329         threads = db.threads('*')
 330         thread = next(threads)
 331         assert isinstance(thread, notmuch2.Thread)
 332
 333     def test_use_threaded_message_twice(self, db):
 334         thread = next(db.threads('*'))
 335         for msg in thread.toplevel():
 336             assert isinstance(msg, notmuch2.Message)
 337             assert msg.alive
 338             del msg
 339         for msg in thread:
 340             assert isinstance(msg, notmuch2.Message)
 341             assert msg.alive
 342             del msg