9 import notmuch2._errors as errors
10 import notmuch2._database as dbmod
11 import notmuch2._message as message
16 with dbmod.Database.create(maildir.path) as db:
21 """Tests for reading the default database.
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.
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
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')
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()
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()
47 def test_default_path_parseerr(self, monkeypatch, tmppath):
48 cfg = tmppath / 'notmuch-config'
49 with cfg.open('w') as fp:
51 monkeypatch.setenv('NOTMUCH_CONFIG', str(cfg))
52 with pytest.raises(configparser.Error):
53 dbmod.Database.default_path()
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
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'
75 def test_create(self, tmppath, db):
76 assert tmppath.joinpath('.notmuch/xapian/').exists()
78 def test_create_already_open(self, tmppath, db):
79 with pytest.raises(errors.NotmuchError):
82 def test_create_existing(self, tmppath, db):
83 with pytest.raises(errors.DatabaseExistsError):
84 dbmod.Database.create(path=tmppath)
86 def test_close(self, db):
89 def test_del_noclose(self, db):
92 def test_close_del(self, db):
96 def test_closed_attr(self, db):
101 def test_ctx(self, db):
107 def test_path(self, db, tmppath):
108 assert db.path == tmppath
110 def test_version(self, db):
111 assert db.version > 0
113 def test_needs_upgrade(self, db):
114 assert db.needs_upgrade in (True, False)
119 def test_exit_early(self, db):
120 with pytest.raises(errors.UnbalancedAtomicError):
121 with db.atomic() as ctx:
124 def test_exit_late(self, db):
125 with db.atomic() as ctx:
127 with pytest.raises(errors.UnbalancedAtomicError):
130 def test_abort(self, db):
131 with db.atomic() as txn:
138 def test_single_rev(self, db):
140 assert isinstance(r, dbmod.DbRevision)
141 assert isinstance(r.rev, int)
142 assert isinstance(r.uuid, bytes)
150 def test_diff_db(self, tmppath):
151 dbpath0 = tmppath.joinpath('db0')
153 dbpath1 = tmppath.joinpath('db1')
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
162 def test_cmp(self, db, maildir):
164 _, pathname = maildir.deliver()
165 db.add(pathname, sync_flags=False)
169 assert not rev0 > rev1
170 assert not rev0 >= rev1
171 assert not rev0 == rev1
174 # XXX add tests for revisions comparisons
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
185 def test_add_message_str(self, db, maildir):
186 msgid, pathname = maildir.deliver()
187 msg, dup = db.add(str(pathname), sync_flags=False)
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)
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):
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):
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):
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
225 def test_find_message_notfound(self, db):
226 with pytest.raises(LookupError):
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
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
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
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.
254 def test_type(self, db):
255 assert isinstance(db.tags, collections.abc.Set)
257 def test_none(self, db):
258 itags = iter(db.tags)
259 with pytest.raises(StopIteration):
261 assert len(db.tags) == 0
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):
272 assert 'hello' in msg.tags
274 def test_cache(self, db):
275 assert db.tags is db.tags
277 def test_iters(self, db):
286 def db(self, maildir, notmuch):
287 """Return a read-only notmuch2.Database.
289 The database will have 3 messages, 2 threads.
291 msgid, _ = maildir.deliver(body='foo')
292 maildir.deliver(body='bar')
293 maildir.deliver(body='baz',
294 headers=[('In-Reply-To', '<{}>'.format(msgid))])
296 with dbmod.Database(maildir.path, 'rw') as db:
299 def test_count_messages(self, db):
300 assert db.count_messages('*') == 3
302 def test_messages_type(self, db):
303 msgs = db.messages('*')
304 assert isinstance(msgs, collections.abc.Iterator)
306 def test_message_no_results(self, db):
307 msgs = db.messages('not_a_matching_query')
308 with pytest.raises(StopIteration):
311 def test_message_match(self, db):
312 msgs = db.messages('*')
314 assert isinstance(msg, notmuch2.Message)
316 def test_count_threads(self, db):
317 assert db.count_threads('*') == 2
319 def test_threads_type(self, db):
320 threads = db.threads('*')
321 assert isinstance(threads, collections.abc.Iterator)
323 def test_threads_no_match(self, db):
324 threads = db.threads('not_a_matching_query')
325 with pytest.raises(StopIteration):
328 def test_threads_match(self, db):
329 threads = db.threads('*')
330 thread = next(threads)
331 assert isinstance(thread, notmuch2.Thread)
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)
340 assert isinstance(msg, notmuch2.Message)