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
 131 class TestRevision:
 132
 133     def test_single_rev(self, db):
 134         r = db.revision()
 135         assert isinstance(r, dbmod.DbRevision)
 136         assert isinstance(r.rev, int)
 137         assert isinstance(r.uuid, bytes)
 138         assert r is r
 139         assert r == r
 140         assert r <= r
 141         assert r >= r
 142         assert not r < r
 143         assert not r > r
 144
 145     def test_diff_db(self, tmppath):
 146         dbpath0 = tmppath.joinpath('db0')
 147         dbpath0.mkdir()
 148         dbpath1 = tmppath.joinpath('db1')
 149         dbpath1.mkdir()
 150         db0 = dbmod.Database.create(path=dbpath0)
 151         db1 = dbmod.Database.create(path=dbpath1)
 152         r_db0 = db0.revision()
 153         r_db1 = db1.revision()
 154         assert r_db0 != r_db1
 155         assert r_db0.uuid != r_db1.uuid
 156
 157     def test_cmp(self, db, maildir):
 158         rev0 = db.revision()
 159         _, pathname = maildir.deliver()
 160         db.add(pathname, sync_flags=False)
 161         rev1 = db.revision()
 162         assert rev0 < rev1
 163         assert rev0 <= rev1
 164         assert not rev0 > rev1
 165         assert not rev0 >= rev1
 166         assert not rev0 == rev1
 167         assert rev0 != rev1
 168
 169     # XXX add tests for revisions comparisons
 170
 171 class TestMessages:
 172
 173     def test_add_message(self, db, maildir):
 174         msgid, pathname = maildir.deliver()
 175         msg, dup = db.add(pathname, sync_flags=False)
 176         assert isinstance(msg, message.Message)
 177         assert msg.path == pathname
 178         assert msg.messageid == msgid
 179
 180     def test_add_message_str(self, db, maildir):
 181         msgid, pathname = maildir.deliver()
 182         msg, dup = db.add(str(pathname), sync_flags=False)
 183
 184     def test_add_message_bytes(self, db, maildir):
 185         msgid, pathname = maildir.deliver()
 186         msg, dup = db.add(os.fsencode(bytes(pathname)), sync_flags=False)
 187
 188     def test_remove_message(self, db, maildir):
 189         msgid, pathname = maildir.deliver()
 190         msg, dup = db.add(pathname, sync_flags=False)
 191         assert db.find(msgid)
 192         dup = db.remove(pathname)
 193         with pytest.raises(LookupError):
 194             db.find(msgid)
 195
 196     def test_remove_message_str(self, db, maildir):
 197         msgid, pathname = maildir.deliver()
 198         msg, dup = db.add(pathname, sync_flags=False)
 199         assert db.find(msgid)
 200         dup = db.remove(str(pathname))
 201         with pytest.raises(LookupError):
 202             db.find(msgid)
 203
 204     def test_remove_message_bytes(self, db, maildir):
 205         msgid, pathname = maildir.deliver()
 206         msg, dup = db.add(pathname, sync_flags=False)
 207         assert db.find(msgid)
 208         dup = db.remove(os.fsencode(bytes(pathname)))
 209         with pytest.raises(LookupError):
 210             db.find(msgid)
 211
 212     def test_find_message(self, db, maildir):
 213         msgid, pathname = maildir.deliver()
 214         msg0, dup = db.add(pathname, sync_flags=False)
 215         msg1 = db.find(msgid)
 216         assert isinstance(msg1, message.Message)
 217         assert msg1.messageid == msgid == msg0.messageid
 218         assert msg1.path == pathname == msg0.path
 219
 220     def test_find_message_notfound(self, db):
 221         with pytest.raises(LookupError):
 222             db.find('foo')
 223
 224     def test_get_message(self, db, maildir):
 225         msgid, pathname = maildir.deliver()
 226         msg0, _ = db.add(pathname, sync_flags=False)
 227         msg1 = db.get(pathname)
 228         assert isinstance(msg1, message.Message)
 229         assert msg1.messageid == msgid == msg0.messageid
 230         assert msg1.path == pathname == msg0.path
 231
 232     def test_get_message_str(self, db, maildir):
 233         msgid, pathname = maildir.deliver()
 234         db.add(pathname, sync_flags=False)
 235         msg = db.get(str(pathname))
 236         assert msg.messageid == msgid
 237
 238     def test_get_message_bytes(self, db, maildir):
 239         msgid, pathname = maildir.deliver()
 240         db.add(pathname, sync_flags=False)
 241         msg = db.get(os.fsencode(bytes(pathname)))
 242         assert msg.messageid == msgid
 243
 244
 245 class TestTags:
 246     # We just want to test this behaves like a set at a hight level.
 247     # The set semantics are tested in detail in the test_tags module.
 248
 249     def test_type(self, db):
 250         assert isinstance(db.tags, collections.abc.Set)
 251
 252     def test_none(self, db):
 253         itags = iter(db.tags)
 254         with pytest.raises(StopIteration):
 255             next(itags)
 256         assert len(db.tags) == 0
 257         assert not db.tags
 258
 259     def test_some(self, db, maildir):
 260         _, pathname = maildir.deliver()
 261         msg, _ = db.add(pathname, sync_flags=False)
 262         msg.tags.add('hello')
 263         itags = iter(db.tags)
 264         assert next(itags) == 'hello'
 265         with pytest.raises(StopIteration):
 266             next(itags)
 267         assert 'hello' in msg.tags
 268
 269     def test_cache(self, db):
 270         assert db.tags is db.tags
 271
 272     def test_iters(self, db):
 273         i1 = iter(db.tags)
 274         i2 = iter(db.tags)
 275         assert i1 is not i2
 276
 277
 278 class TestQuery:
 279
 280     @pytest.fixture
 281     def db(self, maildir, notmuch):
 282         """Return a read-only notmuch2.Database.
 283
 284         The database will have 3 messages, 2 threads.
 285         """
 286         msgid, _ = maildir.deliver(body='foo')
 287         maildir.deliver(body='bar')
 288         maildir.deliver(body='baz',
 289                         headers=[('In-Reply-To', '<{}>'.format(msgid))])
 290         notmuch('new')
 291         with dbmod.Database(maildir.path, 'rw') as db:
 292             yield db
 293
 294     def test_count_messages(self, db):
 295         assert db.count_messages('*') == 3
 296
 297     def test_messages_type(self, db):
 298         msgs = db.messages('*')
 299         assert isinstance(msgs, collections.abc.Iterator)
 300
 301     def test_message_no_results(self, db):
 302         msgs = db.messages('not_a_matching_query')
 303         with pytest.raises(StopIteration):
 304             next(msgs)
 305
 306     def test_message_match(self, db):
 307         msgs = db.messages('*')
 308         msg = next(msgs)
 309         assert isinstance(msg, notmuch2.Message)
 310
 311     def test_count_threads(self, db):
 312         assert db.count_threads('*') == 2
 313
 314     def test_threads_type(self, db):
 315         threads = db.threads('*')
 316         assert isinstance(threads, collections.abc.Iterator)
 317
 318     def test_threads_no_match(self, db):
 319         threads = db.threads('not_a_matching_query')
 320         with pytest.raises(StopIteration):
 321             next(threads)
 322
 323     def test_threads_match(self, db):
 324         threads = db.threads('*')
 325         thread = next(threads)
 326         assert isinstance(thread, notmuch2.Thread)
 327
 328     def test_use_threaded_message_twice(self, db):
 329         thread = next(db.threads('*'))
 330         for msg in thread.toplevel():
 331             assert isinstance(msg, notmuch2.Message)
 332             assert msg.alive
 333             del msg
 334         for msg in thread:
 335             assert isinstance(msg, notmuch2.Message)
 336             assert msg.alive
 337             del msg